Add a verify_sendgrid_webhook_signature decorator to make Sendgrid Events webhook development easier (#138)

blag · web-flow · commit 76cc12da94e2 · 2025-05-14T09:42:16.000-04:00
* Add a verify_sendgrid_webhook_signature decorator to make Sendgrid Events webhook development easier

* Add tests for webhook signature verification decorator

* Document new webhook signature verification decorator
diff --git a/README.md b/README.md
@@ -109,6 +109,59 @@ msg.ip_pool_name = 'my-ip-pool'
 msg.send(fail_silently=False)
 ```
 
+### Webhook Helpers
+
+Version 6 of the `sendgrid` package or later includes some helper functions to
+cryptographically verify the signature and contents of events from the Sendgrid
+Events webhook.
+
+This project includes some additional helpers for Sendgrid's webhook signature
+verification.
+
+1. Enable signature verification for Sendgrid webhooks (see [Sendgrid docs](https://www.twilio.com/docs/sendgrid/for-developers/tracking-events/getting-started-event-webhook-security-features#enable-signature-verification)). Once you have saved the webhook and edited it again, copy the verification key.
+2. Modify your project's `settings.py` and set `SENDGRID_WEBHOOK_VERIFICATION_KEY` to your verification key value.
+3. Setup a project URLConf and view. Below is an example view you can adapt to your needs.
+
+```python
+import json
+from datetime import datetime
+
+from django.db import transaction
+from django.http import HttpRequest, HttpResponse
+from django.views.decorators.csrf import csrf_exempt
+from django.views.decorators.http import require_POST
+from post_office.models import Email, Log as EmailLog, STATUS
+from pytz import utc
+from sendgrid_backend.decorators import verify_sendgrid_webhook_signature
+
+EVENTS = {'delivered': STATUS.sent, 'bounce': STATUS.failed, 'blocked': STATUS.failed}
+
+@csrf_exempt
+@require_POST
+@verify_sendgrid_webhook_signature
+def sendgrid_deliverability_webhook_handler(request: HttpRequest) -> HttpResponse:
+    """
+    Example webhook handler to save delivered, bounce, and blocked events to
+    the email log.
+    """
+    for msg_dict in reversed(json.loads(request.body)):
+        if event := EVENTS.get(msg_dict.get('event', None), None):
+            event_timestamp = datetime.fromtimestamp(msg_dict.get('timestamp'), tz=utc)
+            with transaction.atomic():
+                Email.objects.filter(message_id=msg_dict.get('smtp-id', None)).update(
+                    last_updated=event_timestamp,
+                    status=event,
+                )
+
+                EmailLog.objects.create(
+                    email__message_id=msg_dict.get('smtp-id', None),
+                    date=event_timestamp,
+                    status=event,
+                    message=json.dumps(msg_dict),
+                )
+    return HttpResponse("ok")
+```
+
 
 ### FAQ
 **How to change a Sender's Name ?**
diff --git a/pyproject.toml b/pyproject.toml
@@ -77,6 +77,7 @@ legacy_tox_ini = """
         django51: Django>=5.1,<5.2
         sendgrid5: sendgrid>=5,<6
         sendgrid6: sendgrid>=6,<7
+        starkbank-ecdsa
         pytest-cov
 
     commands =
diff --git a/sendgrid_backend/decorators.py b/sendgrid_backend/decorators.py
@@ -0,0 +1,46 @@
+from functools import wraps
+from inspect import iscoroutinefunction
+from typing import Callable
+
+from django.conf import settings
+from django.http import HttpResponseNotFound
+
+from sendgrid_backend.util import SENDGRID_6
+
+if SENDGRID_6:
+    from sendgrid.helpers.eventwebhook import EventWebhook
+    from sendgrid.helpers.eventwebhook.eventwebhook_header import EventWebhookHeader
+
+    # Adapted from:
+    # https://stackoverflow.com/a/71672552
+    def check_sendgrid_signature(request):
+        event_webhook = EventWebhook()
+        key = settings.SENDGRID_WEBHOOK_VERIFICATION_KEY
+        ec_public_key = event_webhook.convert_public_key_to_ecdsa(key)
+
+        return event_webhook.verify_signature(
+            request.body.decode("utf-8"),
+            request.headers[EventWebhookHeader.SIGNATURE],
+            request.headers[EventWebhookHeader.TIMESTAMP],
+            ec_public_key,
+        )
+
+    def verify_sendgrid_webhook_signature(func: Callable) -> Callable:
+        """Check a view for a valid sendgrid webhook"""
+        if iscoroutinefunction(func):
+
+            @wraps(func)
+            async def inner(request, *args, **kwargs):
+                if not check_sendgrid_signature(request):
+                    return HttpResponseNotFound()
+                return await func(request, *args, **kwargs)
+
+        else:
+
+            @wraps(func)
+            def inner(request, *args, **kwargs):
+                if not check_sendgrid_signature(request):
+                    return HttpResponseNotFound()
+                return func(request, *args, **kwargs)
+
+        return inner
diff --git a/test/test_decorators.py b/test/test_decorators.py
@@ -0,0 +1,104 @@
+from unittest.mock import MagicMock
+
+from django.http import HttpResponseNotFound
+from django.test import SimpleTestCase
+
+from sendgrid_backend.util import SENDGRID_6
+
+if SENDGRID_6:
+    from ellipticcurve.ecdsa import Ecdsa
+    from ellipticcurve.privateKey import PrivateKey
+    from sendgrid.helpers.eventwebhook.eventwebhook_header import EventWebhookHeader
+
+    from sendgrid_backend.decorators import (
+        check_sendgrid_signature,
+        verify_sendgrid_webhook_signature,
+    )
+
+    # Generate privateKey from PEM string
+    PRIVATE_KEY = PrivateKey.fromPem(
+        """
+        -----BEGIN EC PARAMETERS-----
+        BgUrgQQACg==
+        -----END EC PARAMETERS-----
+        -----BEGIN EC PRIVATE KEY-----
+        MHQCAQEEIODvZuS34wFbt0X53+P5EnSj6tMjfVK01dD1dgDH02RzoAcGBSuBBAAK
+        oUQDQgAE/nvHu/SQQaos9TUljQsUuKI15Zr5SabPrbwtbfT/408rkVVzq8vAisbB
+        RmpeRREXj5aog/Mq8RrdYy75W9q/Ig==
+        -----END EC PRIVATE KEY-----
+    """
+    )
+    PUBLIC_KEY_STRING = "".join(PRIVATE_KEY.publicKey().toPem().splitlines()[2:-1])
+    TIMESTAMP = "2025-05-13 07:42:18.792332+00:00"
+
+    class TestDecoratorTestCase(SimpleTestCase):
+        @classmethod
+        def setUpClass(cls):
+            cls.message = "this data should be signed"
+
+            cls.good_request = MagicMock()
+            cls.good_request.body.decode.return_value = "this data should be signed"
+            cls.good_request.headers = {
+                EventWebhookHeader.SIGNATURE: Ecdsa.sign(
+                    TIMESTAMP + cls.message, PRIVATE_KEY
+                ).toBase64(),
+                EventWebhookHeader.TIMESTAMP: TIMESTAMP,
+            }
+
+            cls.bad_request = MagicMock()
+            cls.bad_request.body.decode.return_value = "this data should be signed"
+            cls.bad_request.headers = {
+                EventWebhookHeader.SIGNATURE: Ecdsa.sign(
+                    TIMESTAMP + cls.message, PRIVATE_KEY
+                ).toBase64(),
+                EventWebhookHeader.TIMESTAMP: TIMESTAMP + "A",  # One character off
+            }
+
+        def test_check_sendgrid_signature(self):
+            with self.settings(SENDGRID_WEBHOOK_VERIFICATION_KEY=PUBLIC_KEY_STRING):
+                self.assertTrue(check_sendgrid_signature(self.good_request))
+
+        def test_check_sendgrid_signature_bad_signature(self):
+            with self.settings(SENDGRID_WEBHOOK_VERIFICATION_KEY=PUBLIC_KEY_STRING):
+                self.assertFalse(check_sendgrid_signature(self.bad_request))
+
+        def test_verify_sendgrid_webhook_signature_decorator(self):
+            @verify_sendgrid_webhook_signature
+            def test_func(request):
+                return "The function was successfully run"
+
+            with self.settings(SENDGRID_WEBHOOK_VERIFICATION_KEY=PUBLIC_KEY_STRING):
+                self.assertEqual(
+                    test_func(self.good_request), "The function was successfully run"
+                )
+
+        def test_verify_sendgrid_webhook_signature_decorator_bad_signature(self):
+            @verify_sendgrid_webhook_signature
+            def test_func(request):
+                return "The function was successfully run"
+
+            with self.settings(SENDGRID_WEBHOOK_VERIFICATION_KEY=PUBLIC_KEY_STRING):
+                self.assertIsInstance(test_func(self.bad_request), HttpResponseNotFound)
+
+        async def test_async_verify_sendgrid_webhook_signature_decorator(self):
+            @verify_sendgrid_webhook_signature
+            async def test_func(request):
+                return "The function was successfully run"
+
+            with self.settings(SENDGRID_WEBHOOK_VERIFICATION_KEY=PUBLIC_KEY_STRING):
+                self.assertEqual(
+                    await test_func(self.good_request),
+                    "The function was successfully run",
+                )
+
+        async def test_async_verify_sendgrid_webhook_signature_decorator_bad_signature(
+            self,
+        ):
+            @verify_sendgrid_webhook_signature
+            async def test_func(request):
+                return "The function was successfully run"
+
+            with self.settings(SENDGRID_WEBHOOK_VERIFICATION_KEY=PUBLIC_KEY_STRING):
+                self.assertIsInstance(
+                    await test_func(self.bad_request), HttpResponseNotFound
+                )
