Merge lp:~wgrant/launchpad/webhook-delivery-tweaks into lp:launchpad

[William Grant]

Webhook delivery tweaks: 30 second timeout, User-Agent, payload signatures, and faster retries.

Webhook delivery tweaks: 30 second timeout, User-Agent, payload signatures, and faster retries.

[Colin Watson (cjwatson)]

review: Approve

=== modified file 'lib/lp/services/job/tests/test_celery_configuration.py'
--- lib/lp/services/job/tests/test_celery_configuration.py	2015-08-03 10:16:49 +0000
+++ lib/lp/services/job/tests/test_celery_configuration.py	2015-08-04 00:12:30 +0000
@@ -4,6 +4,8 @@
 from contextlib import contextmanager
 from testtools.matchers import MatchesRegex
 
+from testtools.matchers import MatchesRegex
+
 from lp.services.config import config
 from lp.testing import TestCase
 from lp.testing.layers import RabbitMQLayer

=== modified file 'lib/lp/services/webhooks/client.py'
--- lib/lp/services/webhooks/client.py	2015-07-12 23:36:55 +0000
+++ lib/lp/services/webhooks/client.py	2015-08-04 00:12:30 +0000
@@ -8,16 +8,32 @@
     'WebhookClient',
     ]
 
+import hashlib
+import hmac
+import json
+
 import requests
 from zope.interface import implementer
 
 from lp.services.webhooks.interfaces import IWebhookClient
 
 
+def create_request(user_agent, secret, payload):
+    body = json.dumps(payload)
+    headers = {
+        'User-Agent': user_agent,
+        'Content-Type': 'application/json',
+        }
+    if secret is not None:
+        hexdigest = hmac.new(secret, body, digestmod=hashlib.sha1).hexdigest()
+        headers['X-Hub-Signature'] = 'sha1=%s' % hexdigest
+    return (body, headers)
+
+
 @implementer(IWebhookClient)
 class WebhookClient:
 
-    def deliver(self, url, proxy, payload):
+    def deliver(self, url, proxy, user_agent, timeout, secret, payload):
         """See `IWebhookClient`."""
         # We never want to execute a job if there's no proxy configured, as
         # we'd then be sending near-arbitrary requests from a trusted
@@ -32,8 +48,11 @@
         session = requests.Session()
         session.trust_env = False
         session.headers = {}
-        preq = session.prepare_request(
-            requests.Request('POST', url, json=payload))
+
+        body, headers = create_request(user_agent, secret, payload)
+        preq = session.prepare_request(requests.Request(
+            'POST', url, data=body, headers=headers))
+
         result = {
             'request': {
                 'url': url,
@@ -43,7 +62,7 @@
                 },
             }
         try:
-            resp = session.send(preq, proxies=proxies)
+            resp = session.send(preq, proxies=proxies, timeout=timeout)
             result['response'] = {
                 'status_code': resp.status_code,
                 'headers': dict(resp.headers),

=== modified file 'lib/lp/services/webhooks/interfaces.py'
--- lib/lp/services/webhooks/interfaces.py	2015-07-29 08:37:51 +0000
+++ lib/lp/services/webhooks/interfaces.py	2015-08-04 00:12:30 +0000
@@ -249,7 +249,7 @@
 
 class IWebhookClient(Interface):
 
-    def deliver(self, url, proxy, payload):
+    def deliver(self, url, proxy, user_agent, timeout, secret, payload):
         """Deliver a payload to a webhook endpoint.
 
         Returns a dict of request and response details. The 'request' key
@@ -260,6 +260,13 @@
         cannot be the fault of the remote endpoint. For example, a 404 will
         return a response, and a DNS error returns a connection_error, but
         the proxy being offline will raise an exception.
+
+        The timeout is just given to the underlying requests library, so
+        it only provides connect and inter-read timeouts. A reliable
+        overall request timeout will require another mechanism.
+
+        If secret is not None, a PubSubHubbub-compatible X-Hub-Signature
+        header will be sent using HMAC-SHA1.
         """
 
 patch_collection_property(IWebhook, 'deliveries', IWebhookDeliveryJob)

=== modified file 'lib/lp/services/webhooks/model.py'
--- lib/lp/services/webhooks/model.py	2015-07-29 08:31:06 +0000
+++ lib/lp/services/webhooks/model.py	2015-08-04 00:12:30 +0000
@@ -38,6 +38,7 @@
     )
 from zope.security.proxy import removeSecurityProxy
 
+import lp.app.versioninfo
 from lp.registry.model.person import Person
 from lp.services.config import config
 from lp.services.database.bulk import load_related
@@ -296,6 +297,13 @@
     retry_error_types = (WebhookDeliveryRetry,)
     user_error_types = (WebhookDeliveryFailure,)
 
+    # The request timeout is 30 seconds, requests timeouts aren't
+    # totally reliable so we also have a relatively low celery timeout
+    # as a backup. The celery timeout and lease expiry have a bit of
+    # slack to cope with slow job start/finish without conflicts.
+    soft_time_limit = timedelta(seconds=45)
+    lease_duration = timedelta(seconds=60)
+
     # Effectively infinite, as we give up by checking
     # retry_automatically and raising a fatal exception instead.
     max_retries = 1000
@@ -366,14 +374,20 @@
 
     @property
     def retry_delay(self):
-        if self._time_since_first_attempt < timedelta(hours=1):
+        if self._time_since_first_attempt < timedelta(minutes=10):
+            return timedelta(minutes=1)
+        elif self._time_since_first_attempt < timedelta(hours=1):
             return timedelta(minutes=5)
         else:
             return timedelta(hours=1)
 
     def run(self):
+        user_agent = '%s-Webhooks/r%s' % (
+            config.vhost.mainsite.hostname, lp.app.versioninfo.revno)
+        secret = self.webhook.secret
         result = getUtility(IWebhookClient).deliver(
             self.webhook.delivery_url, config.webhooks.http_proxy,
+            user_agent, 30, secret.encode('utf-8') if secret else None,
             self.payload)
         # Request and response headers and body may be large, so don't
         # store them in the frequently-used JSON. We could store them in

=== modified file 'lib/lp/services/webhooks/tests/test_webhookjob.py'
--- lib/lp/services/webhooks/tests/test_webhookjob.py	2015-07-29 08:31:06 +0000
+++ lib/lp/services/webhooks/tests/test_webhookjob.py	2015-08-04 00:12:30 +0000
@@ -5,12 +5,16 @@
 
 __metaclass__ = type
 
-from datetime import timedelta
+from datetime import (
+    datetime,
+    timedelta,
+    )
 
 from httmock import (
     HTTMock,
     urlmatch,
     )
+from pytz import utc
 import requests
 from storm.store import Store
 from testtools import TestCase
@@ -23,18 +27,24 @@
     KeysEqual,
     LessThan,
     MatchesAll,
+    MatchesDict,
     MatchesStructure,
     Not,
     )
 import transaction
 from zope.component import getUtility
+from zope.security.proxy import removeSecurityProxy
 
+from lp.app.versioninfo import revno
 from lp.services.features.testing import FeatureFixture
 from lp.services.job.interfaces.job import JobStatus
 from lp.services.job.runner import JobRunner
 from lp.services.job.tests import block_on_job
 from lp.services.scripts.tests import run_script
-from lp.services.webhooks.client import WebhookClient
+from lp.services.webhooks.client import (
+    create_request,
+    WebhookClient,
+    )
 from lp.services.webhooks.interfaces import (
     IWebhookClient,
     IWebhookDeliveryJob,
@@ -129,7 +139,7 @@
     def sendToWebhook(self, response_status=200, raises=None):
         reqs = []
 
-        @urlmatch(netloc='hookep.com')
+        @urlmatch(netloc='example.com')
         def endpoint_mock(url, request):
             if raises:
                 raise raises
@@ -138,58 +148,89 @@
 
         with HTTMock(endpoint_mock):
             result = WebhookClient().deliver(
-                'http://hookep.com/foo',
-                {'http': 'http://squid.example.com:3128'},
-                {'foo': 'bar'})
+                'http://example.com/ep', 'http://squid.example.com:3128',
+                'TestWebhookClient', 30, 'sekrit', {'foo': 'bar'})
 
         return reqs, result
 
+    @property
+    def request_matcher(self):
+        return MatchesDict({
+            'url': Equals('http://example.com/ep'),
+            'method': Equals('POST'),
+            'headers': Equals(
+                {'Content-Type': 'application/json',
+                 'Content-Length': '14',
+                 'User-Agent': 'TestWebhookClient',
+                 'X-Hub-Signature':
+                    'sha1=de75f136c37d89f5eb24834468c1ecd602fa95dd',
+                 }),
+            'body': Equals('{"foo": "bar"}'),
+            })
+
     def test_sends_request(self):
         [request], result = self.sendToWebhook()
-        self.assertEqual(
-            {'Content-Type': 'application/json', 'Content-Length': '14'},
-            result['request']['headers'])
-        self.assertEqual('{"foo": "bar"}', result['request']['body'])
-        self.assertEqual(200, result['response']['status_code'])
-        self.assertEqual({}, result['response']['headers'])
-        self.assertEqual('Content', result['response']['body'])
+        self.assertThat(
+            result,
+            MatchesDict({
+                'request': self.request_matcher,
+                'response': MatchesDict({
+                    'status_code': Equals(200),
+                    'headers': Equals({}),
+                    'body': Equals('Content'),
+                    }),
+                }))
 
     def test_accepts_404(self):
         [request], result = self.sendToWebhook(response_status=404)
-        self.assertEqual(
-            {'Content-Type': 'application/json', 'Content-Length': '14'},
-            result['request']['headers'])
-        self.assertEqual('{"foo": "bar"}', result['request']['body'])
-        self.assertEqual(404, result['response']['status_code'])
-        self.assertEqual({}, result['response']['headers'])
-        self.assertEqual('Content', result['response']['body'])
+        self.assertThat(
+            result,
+            MatchesDict({
+                'request': self.request_matcher,
+                'response': MatchesDict({
+                    'status_code': Equals(404),
+                    'headers': Equals({}),
+                    'body': Equals('Content'),
+                    }),
+                }))
 
     def test_connection_error(self):
         # Attempts that fail to connect have a connection_error rather
         # than a response.
         reqs, result = self.sendToWebhook(
             raises=requests.ConnectionError('Connection refused'))
-        self.assertNotIn('response', result)
-        self.assertEqual(
-            'Connection refused', result['connection_error'])
+        self.assertThat(
+            result,
+            MatchesDict({
+                'request': self.request_matcher,
+                'connection_error': Equals('Connection refused'),
+                }))
         self.assertEqual([], reqs)
 
 
-class MockWebhookClient:
+class MockWebhookClient(WebhookClient):
 
     def __init__(self, response_status=200, raises=None):
         self.response_status = response_status
         self.raises = raises
         self.requests = []
 
-    def deliver(self, url, proxy, payload):
-        result = {'request': {}}
+    def deliver(self, url, proxy, user_agent, timeout, secret, payload):
+        body, headers = create_request(user_agent, secret, payload)
+        result = {
+            'request': {
+                'url': url,
+                'method': 'POST',
+                'headers': headers,
+                'body': body,
+                },
+            }
         if isinstance(self.raises, requests.ConnectionError):
             result['connection_error'] = str(self.raises)
         elif self.raises is not None:
             raise self.raises
         else:
-            self.requests.append(('POST', url))
+            self.requests.append(('POST', url, result['request']['headers']))
             result['response'] = {'status_code': self.response_status}
         return result
 
@@ -199,8 +240,10 @@
 
     layer = ZopelessDatabaseLayer
 
-    def makeAndRunJob(self, response_status=200, raises=None, mock=True):
-        hook = self.factory.makeWebhook(delivery_url=u'http://hookep.com/foo')
+    def makeAndRunJob(self, response_status=200, raises=None, mock=True,
+                      secret=None):
+        hook = self.factory.makeWebhook(
+            delivery_url=u'http://example.com/ep', secret=secret)
         job = WebhookDeliveryJob.create(hook, payload={'foo': 'bar'})
 
         client = MockWebhookClient(
@@ -217,6 +260,21 @@
         self.assertProvides(
             WebhookDeliveryJob.create(hook, payload={}), IWebhookDeliveryJob)
 
+    def test_short_lease_and_timeout(self):
+        # Webhook jobs have a request timeout of 30 seconds, a celery
+        # timeout of 45 seconds, and a lease of 60 seconds, to give
+        # reasonable time for sluggish things to catch up.
+        hook = self.factory.makeWebhook()
+        job = hook.ping()
+        job.acquireLease()
+        self.assertThat(
+            job.lease_expires - datetime.now(utc),
+            MatchesAll(
+                GreaterThan(timedelta(seconds=50)),
+                LessThan(timedelta(seconds=60))))
+        self.assertEqual(
+            timedelta(seconds=45), removeSecurityProxy(job).soft_time_limit)
+
     def test_run_200(self):
         # A request that returns 200 is a success.
         with CaptureOops() as oopses:
@@ -235,7 +293,26 @@
                             {'response': ContainsDict(
                                 {'status_code': Equals(200)})}))})))
         self.assertEqual(1, len(reqs))
-        self.assertEqual([('POST', 'http://hookep.com/foo')], reqs)
+        self.assertEqual([
+            ('POST', 'http://example.com/ep',
+             {'Content-Type': 'application/json',
+              'User-Agent': 'launchpad.dev-Webhooks/r%s' % revno}),
+            ], reqs)
+        self.assertEqual([], oopses.oopses)
+
+    def test_run_signature(self):
+        # If the webhook has a secret, the request is signed in a
+        # PubSubHubbub-compatible way.
+        with CaptureOops() as oopses:
+            job, reqs = self.makeAndRunJob(
+                response_status=200, secret=u'sekrit')
+        self.assertEqual([
+            ('POST', 'http://example.com/ep',
+             {'Content-Type': 'application/json',
+              'User-Agent': 'launchpad.dev-Webhooks/r%s' % revno,
+              'X-Hub-Signature':
+                'sha1=de75f136c37d89f5eb24834468c1ecd602fa95dd'}),
+            ], reqs)
         self.assertEqual([], oopses.oopses)
 
     def test_run_404(self):
@@ -318,9 +395,15 @@
         self.assertEqual(orig_first_sent, job.date_first_sent)
 
     def test_retry_delay(self):
-        # Deliveries are retried every 5 minutes for the first hour, and
-        # every hour thereafter.
+        # Deliveries are retried every minute for the first 10 minutes,
+        # every 5 minutes up to an hour, and every hour thereafter.
         job, reqs = self.makeAndRunJob(response_status=404)
+        self.assertEqual(timedelta(minutes=1), job.retry_delay)
+        job.json_data['date_first_sent'] = (
+            job.date_first_sent - timedelta(minutes=5)).isoformat()
+        self.assertEqual(timedelta(minutes=1), job.retry_delay)
+        job.json_data['date_first_sent'] = (
+            job.date_first_sent - timedelta(minutes=5)).isoformat()
         self.assertEqual(timedelta(minutes=5), job.retry_delay)
         job.json_data['date_first_sent'] = (
             job.date_first_sent - timedelta(minutes=30)).isoformat()
@@ -436,7 +519,7 @@
     layer = ZopelessDatabaseLayer
 
     def test_run_from_cronscript(self):
-        hook = self.factory.makeWebhook(delivery_url=u'http://hookep.com/foo')
+        hook = self.factory.makeWebhook(delivery_url=u'http://example.com/ep')
         job = WebhookDeliveryJob.create(hook, payload={'foo': 'bar'})
         self.assertEqual(JobStatus.WAITING, job.status)
         transaction.commit()
@@ -462,7 +545,7 @@
 
     def test_WebhookDeliveryJob(self):
         """WebhookDeliveryJob runs under Celery."""
-        hook = self.factory.makeWebhook(delivery_url=u'http://hookep.com/foo')
+        hook = self.factory.makeWebhook(delivery_url=u'http://example.com/ep')
 
         self.useFixture(FeatureFixture(
             {'jobs.celery.enabled_classes': 'WebhookDeliveryJob'}))

=== modified file 'lib/lp/testing/factory.py'
--- lib/lp/testing/factory.py	2015-07-30 14:57:06 +0000
+++ lib/lp/testing/factory.py	2015-08-04 00:12:30 +0000
@@ -4525,14 +4525,13 @@
         return ProxyFactory(
             LiveFSFile(livefsbuild=livefsbuild, libraryfile=libraryfile))
 
-    def makeWebhook(self, target=None, delivery_url=None):
+    def makeWebhook(self, target=None, delivery_url=None, secret=None):
         if target is None:
             target = self.makeGitRepository()
         if delivery_url is None:
             delivery_url = self.getUniqueURL().decode('utf-8')
         return getUtility(IWebhookSource).new(
-            target, self.makePerson(), delivery_url, [], True,
-            self.getUniqueUnicode())
+            target, self.makePerson(), delivery_url, [], True, secret)
 
     def makeSnap(self, registrant=None, owner=None, distroseries=None,
                  name=None, branch=None, git_ref=None,