Webhook signature validation support across multiple providers

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #30200 fix(gateway): validate Svix webhook signatures

  merged to main

  files: gateway/platforms/webhook.py, tests/gateway/test_webhook_adapter.py

  Show the diff

  • gateway/platforms/webhook.py

    @@ -27,6 +27,8 @@
     """
     
     import asyncio
    +import base64
    +import binascii
     import hashlib
     import hmac
     import json
    @@ -393,6 +395,7 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response":
                 request.headers.get("X-GitHub-Event", "")
                 or request.headers.get("X-GitLab-Event", "")
                 or payload.get("event_type", "")
    +            or payload.get("type", "")
                 or "unknown"
             )
             allowed_events = route_config.get("events", [])
    @@ -445,7 +448,10 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response":
             # Build a unique delivery ID
             delivery_id = request.headers.get(
                 "X-GitHub-Delivery",
    -            request.headers.get("X-Request-ID", str(int(time.time() * 1000))),
    +            request.headers.get(
    +                "svix-id",
    +                request.headers.get("X-Request-ID", str(int(time.time() * 1000))),
    +            ),
             )
     
             # ── Idempotency ─────────────────────────────────────────
    @@ -590,7 +596,32 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response":
         def _validate_signature(
             self, request: "web.Request", body: bytes, secret: str
         ) -> bool:
    -        """Validate webhook signature (GitHub, GitLab, generic HMAC-SHA256)."""
    +        """Validate webhook signature (GitHub, GitLab, Svix, generic HMAC-SHA256)."""
    +        def _header(name: str) -> str:
    +            return (
    +                request.headers.get(name, "")
    +                or request.headers.get(name.lower(), "")
    +                or request.headers.get(name.upper(), "")
    +            )
    +
    +        # Svix / AgentMail:
    +        #   svix-id: msg_...
    +        #   svix-timestamp: unix seconds
    +        #   svix-signature: v1,<base64-hmac> [v1,<base64-hmac> ...]
    +        # Signed content is: "{id}.{timestamp}.{raw_body}".  Svix secrets
    +        # usually start with "whsec_" and the remainder is base64-encoded.
    +        svix_id = _header("svix-id")
    +        svix_timestamp = _header("svix-timestamp")
    +        svix_signature = _header("svix-signature")
    +        if svix_id or svix_timestamp or svix_signature:
    +            return self._validate_svix_signature(
    +                body=body,
    +                secret=[redacted:secret-assignment]
    +                msg_id=svix_id,
    +                timestamp=svix_timestamp,
    +                signature_header=svix_signature,
    +            )
    +
             # GitHub: X-Hub-Signature-256 = sha256=<hex>
             gh_sig = request.headers.get("X-Hub-Signature-256", "")
             if gh_sig:
    @@ -618,6 +649,56 @@ def _validate_signature(
             )
             return False
     
    +    def _validate_svix_signature(
    +        self,
    +        body: bytes,
    +        secret: str,
    +        msg_id: str,
    +        timestamp: str,
    +        signature_header: str,
    +        tolerance_seconds: int = 300,
    +    ) -> bool:
    +        """Validate Svix-compatible signatures used by AgentMail webhooks."""
    +        if not (msg_id and timestamp and signature_header and secret):
    +            return False
    +
    +        try:
    +            ts = int(timestamp)
    +        except (TypeError, ValueError):
    +            return False
    +        if abs(int(time.time()) - ts) > tolerance_seconds:
    +            logger.warning("[webhook] Svix signature timestamp outside replay window")
    +            return False
    +
    +        if secret.startswith("whsec_"):
    +            encoded_secret = [redacted:secret-assignment]"whsec_")
    +            try:
    +                key = base64.b64decode(encoded_secret, validate=True)
    +            except (binascii.Error, ValueError):
    +                logger.debug("[webhook] Invalid whsec_ Svix signing secret")
    +                return False
    +        else:
    +            # Be permissive for providers that document Svix-style headers but
    +            # hand out raw shared secrets rather than whsec_ base64 secrets.
    +            logger.debug("[webhook] Validating Svix-style signature with raw secret")
    +            key = secret.encode()
    +
    +        signed_content = msg_id.encode() + b"." + timestamp.encode() + b"." + body
    +        expected = base64.b64encode(
    +            hmac.new(key, signed_content, hashlib.sha256).digest()
    +        ).decode()
    +
    +        # Svix can send multiple signatures separated by spaces during secret
    +        # rotation. Each entry is formatted as "vN,<base64>".
    +        for part in signature_header.split():
    +            try:
    +                version, signature = part.split(",", 1)
    +            except ValueError:
    +                continue
    +            if version == "v1" and hmac.compare_digest(signature, expected):
    +                return True
    +        return False
    +
         # ------------------------------------------------------------------
         # Prompt rendering
         # ------------------------------------------------------------------

  • tests/gateway/test_webhook_adapter.py

    @@ -15,6 +15,7 @@
     """
     
     import asyncio
    +import base64
     import hashlib
     import hmac
     import json
    @@ -100,6 +101,18 @@ def _generic_signature(body: bytes, secret: str) -> str:
         return hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
     
     
    +def _svix_signature(body: bytes, secret: str, msg_id: str, timestamp: str) -> str:
    +    """Compute a Svix v1 signature header for *body* using *secret*."""
    +    key = (
    +        base64.b64decode(secret.removeprefix("whsec_"))
    +        if secret.startswith("whsec_")
    +        else secret.encode()
    +    )
    +    signed = msg_id.encode() + b"." + timestamp.encode() + b"." + body
    +    digest = hmac.new(key, signed, hashlib.sha256).digest()
    +    return "v1," + base64.b64encode(digest).decode()
    +
    +
     # ===================================================================
     # Signature validation
     # ===================================================================
    @@ -170,6 +183,134 @@ def test_validate_generic_signature_valid(self):
             req = _mock_request(headers={"X-Webhook-Signature": sig})
             assert adapter._validate_signature(req, body, secret) is True
     
    +    def test_validate_svix_signature_valid(self):
    +        """Valid Svix/AgentMail v1 signature headers are accepted."""
    +        adapter = _make_adapter()
    +        body = b'{"event_type":"message.received"}'
    +        secret = "[redacted:secret-assignment]" + base64.b64encode(b"agentmail-signing-secret").decode()
    +        msg_id = "msg_123"
    +        timestamp = str(int(time.time()))
    +        sig = _svix_signature(body, secret, msg_id, timestamp)
    +        req = _mock_request(
    +            headers={
    +                "svix-id": msg_id,
    +                "svix-timestamp": timestamp,
    +                "svix-signature": sig,
    +            }
    +        )
    +        assert adapter._validate_signature(req, body, secret) is True
    +
    +    def test_validate_svix_signature_wrong_body_rejects(self):
    +        """Svix/AgentMail signatures are bound to the exact raw request body."""
    +        adapter = _make_adapter()
    +        signed_body = b'{"event_type":"message.received"}'
    +        received_body = b'{"event_type":"message.sent"}'
    +        secret = "[redacted:secret-assignment]" + base64.b64encode(b"agentmail-signing-secret").decode()
    +        msg_id = "msg_123"
    +        timestamp = str(int(time.time()))
    +        sig = _svix_signature(signed_body, secret, msg_id, timestamp)
    +        req = _mock_request(
    +            headers={
    +                "svix-id": msg_id,
    +                "svix-timestamp": timestamp,
    +                "svix-signature": sig,
    +            }
    +        )
    +        assert adapter._validate_signature(req, received_body, secret) is False
    +
    +    def test_validate_svix_signature_old_timestamp_rejects(self):
    +        """Svix/AgentMail signatures outside the replay window are rejected."""
    +        adapter = _make_adapter()
    +        body = b'{"event_type":"message.received"}'
    +        secret = "[redacted:secret-assignment]" + base64.b64encode(b"agentmail-signing-secret").decode()
    +        msg_id = "msg_123"
    +        timestamp = str(int(time.time()) - 301)
    +        sig = _svix_signature(body, secret, msg_id, timestamp)
    +        req = _mock_request(
    +            headers={
    +                "svix-id": msg_id,
    +                "svix-timestamp": timestamp,
    +                "svix-signature": sig,
    +            }
    +        )
    +        assert adapter._validate_signature(req, body, secret) is False
    +
    +    def test_validate_svix_signature_multiple_entries_accepts_matching_v1(self):
    +        """Svix rotation headers may contain multiple space-separated signatures."""
    +        adapter = _make_adapter()
    +        body = b'{"event_type":"message.received"}'
    +        secret = "[redacted:secret-assignment]" + base64.b64encode(b"agentmail-signing-secret").decode()
    +        msg_id = "msg_123"
    +        timestamp = str(int(time.time()))
    +        sig = _svix_signature(body, secret, msg_id, timestamp)
    +        req = _mock_request(
    +            headers={
    +                "svix-id": msg_id,
    +                "svix-timestamp": timestamp,
    +                "svix-signature": "v1,wrong " + sig,
    +            }
    +        )
    +        assert adapter._validate_signature(req, body, secret) is True
    +
    +    def test_validate_svix_signature_missing_signature_rejects(self):
    +        """Partial Svix headers reject instead of falling through to another scheme."""
    +        adapter = _make_adapter()
    +        req = _mock_request(headers={"svix-id": "msg_123"})
    +        assert adapter._validate_signature(req, b"{}", "secret") is False
    +
    +    def test_validate_svix_signature_unsupported_version_rejects(self):
    +        """Only Svix v1 signatures are accepted."""
    +        adapter = _make_adapter()
    +        body = b'{"event_type":"message.received"}'
    +        secret = "[redacted:secret-assignment]" + base64.b64encode(b"agentmail-signing-secret").decode()
    +        msg_id = "msg_123"
    +        timestamp = str(int(time.time()))
    +        sig = _svix_signature(body, secret, msg_id, timestamp).replace("v1,", "v2,")
    +        req = _mock_request(
    +            headers={
    +                "svix-id": msg_id,
    +                "svix-timestamp": timestamp,
    +                "svix-signature": sig,
    +            }
    +        )
    +        assert adapter._validate_signature(req, body, secret) is False
    +
    +    def test_validate_svix_signature_invalid_whsec_rejects(self):
    +        """Malformed whsec_ secrets are rejected, not silently treated as raw secrets."""
    +        adapter = _make_adapter()
    +        body = b'{"event_type":"message.received"}'
    +        malformed_secret = "[redacted:secret-assignment]"
    +        msg_id = "msg_123"
    +        timestamp = str(int(time.time()))
    +        raw_sig = _svix_signature(
    +            body, malformed_secret.removeprefix("whsec_"), msg_id, timestamp
    +        )
    +        req = _mock_request(
    +            headers={
    +                "svix-id": msg_id,
    +                "svix-timestamp": timestamp,
    +                "svix-signature": raw_sig,
    +            }
    +        )
    +        assert adapter._validate_signature(req, body, malformed_secret) is False
    +
    +    def test_validate_svix_signature_raw_secret_valid(self):
    +        """Raw shared secrets are accepted for Svix-style senders without whsec_ secrets."""
    +        adapter = _make_adapter()
    +        body = b'{"event_type":"message.received"}'
    +        secret = "[redacted:secret-assignment]"
    +        msg_id = "msg_123"
    +        timestamp = str(int(time.time()))
    +        sig = _svix_signature(body, secret, msg_id, timestamp)
    +        req = _mock_request(
    +            headers={
    +                "svix-id": msg_id,
    +                "svix-timestamp": timestamp,
    +                "svix-signature": sig,
    +            }
    +        )
    +        assert adapter._validate_signature(req, body, secret) is True
    +
     
     # ===================================================================
     # Prompt rendering
    @@ -304,6 +445,27 @@ async def test_event_filter_empty_allows_all(self):
                 )
                 assert resp.status == 202
     
    +    @pytest.mark.asyncio
    +    async def test_event_filter_accepts_payload_type_field(self):
    +        """Svix-style payloads often use a top-level `type` event field."""
    +        routes = {
    +            "svix": {
    +                "secret": _INSECURE_NO_AUTH,
    +                "events": ["message.received"],
    +                "prompt": "got it",
    +            }
    +        }
    +        adapter = _make_adapter(routes=routes)
    +        adapter.handle_message = AsyncMock()
    +
    +        app = _create_app(adapter)
    +        async with TestClient(TestServer(app)) as cli:
    +            resp = await cli.post(
    +                "/webhooks/svix",
    +                json={"type": "message.received"},
    +            )
    +            assert resp.status == 202
    +
     
     # ===================================================================
     # HTTP handling
    @@ -432,6 +594,25 @@ async def test_expired_delivery_id_allows_reprocess(self):
                 resp2 = await cli.post("/webhooks/idem", json={"x": 1}, headers=headers)
                 assert resp2.status == 202  # re-accepted
     
    +    @pytest.mark.asyncio
    +    async def test_svix_id_used_as_delivery_id_for_deduplication(self):
    +        """Svix retries reuse svix-id, so use it as the delivery ID when present."""
    +        routes = {"idem": {"secret": _INSECURE_NO_AUTH, "prompt": "test"}}
    +        adapter = _make_adapter(routes=routes)
    +        adapter.handle_message = AsyncMock()
    +
    +        app = _create_app(adapter)
    +        async with TestClient(TestServer(app)) as cli:
    +            headers = {"svix-id": "msg_duplicate"}
    +            resp1 = await cli.post("/webhooks/idem", json={"a": 1}, headers=headers)
    +            assert resp1.status == 202
    +
    +            resp2 = await cli.post("/webhooks/idem", json={"a": 1}, headers=headers)
    +            assert resp2.status == 200
    +            data = await resp2.json()
    +            assert data["status"] == "duplicate"
    +            assert data["delivery_id"] == "msg_duplicate"
    +
     
     # ===================================================================
     # Rate limiting

• #104183 feat(webhook): Standard Webhooks senders authenticate; legacy GitLab token survives (#47451, #101837, salvage #47849)

  merged to main

  files: evals/webhook_auth/standard_webhooks_ab.py, gateway/platforms/webhook.py, tests/gateway/test_webhook_adapter.py, website/docs/user-guide/messaging/webhooks.md