fix: Separate an unreadable webhook payload from a forged one

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y1RzepycXEYA3LStfjt8cY

Author: claude

README.rst

@@ -597,12 +597,19 @@ Refer to the `Svix docs on Consuming Webhooks <https://docs.svix.com/receiving/i
 This example is for `Flask <https://flask.palletsprojects.com/>`_,
 see the `Svix docs for more examples in specific frameworks <https://docs.svix.com/receiving/verifying-payloads/how>`_.
 
+Verification failures raise a ``SeamWebhookVerificationError``: treat the
+payload as forged and respond with an error status so Svix retries. A payload
+that is correctly signed but unreadable raises a
+``SeamInvalidWebhookPayloadError`` instead: it is genuinely from Seam and will
+never become readable, so log it as a bug rather than reporting a verification
+failure and letting Svix retry it through its full backoff schedule.
+
 .. code-block:: python
 
   import os
 
   from flask import Flask, request
-  from seam import SeamWebhook
+  from seam import SeamInvalidWebhookPayloadError, SeamWebhook, SeamWebhookVerificationError
 
   app = Flask(__name__)
 
@@ -612,8 +619,11 @@ see the `Svix docs for more examples in specific frameworks <https://docs.svix.c
   def handle_webhook():
       try:
           data = webhook.verify(request.get_data(), request.headers)
-      except Exception:
+      except SeamWebhookVerificationError:
           return 'Bad Request', 400
+      except SeamInvalidWebhookPayloadError:
+          app.logger.exception('Unreadable Seam webhook payload')
+          return '', 204
 
       try:
           store_event(data)

seam/__init__.py

@@ -8,6 +8,8 @@
 from .exceptions import (
     SeamError,
     SeamHttpApiError,
+    SeamInvalidWebhookPayloadError,
+    SeamWebhookVerificationError,
     SeamHttpUnauthorizedError,
     SeamHttpInvalidInputError,
     SeamValidationError,
@@ -16,7 +18,6 @@
     SeamActionAttemptTimeoutError,
 )
 from .seam_webhook import SeamWebhook
-from svix.webhooks import WebhookVerificationError as SeamWebhookVerificationError
 from .null import NULL, Null
 from .url_search_params_serializer import UnserializableParamError, UrlSearchParams
 from .strict_url_search_params_serializer import (

seam/exceptions.py

@@ -1,5 +1,8 @@
 from dataclasses import dataclass
 from typing import Any, Dict, List, Optional
+
+from svix.webhooks import WebhookVerificationError
+
 from .resources import ActionAttempt, ErrorActionAttempt, PendingActionAttempt
 
 
@@ -15,6 +18,27 @@ class SeamError(Exception):
     """Base exception for all errors raised by the Seam SDK."""
 
 
+# Webhook
+class SeamWebhookVerificationError(WebhookVerificationError, SeamError):
+    """
+    Exception raised when a webhook payload fails signature verification.
+
+    Treat the payload as forged: respond with an error status so the sender
+    retries, and investigate if the failures persist.
+    """
+
+
+class SeamInvalidWebhookPayloadError(SeamError):
+    """
+    Exception raised when a webhook payload passes signature verification
+    but cannot be read as a Seam event.
+
+    The payload is genuinely from Seam and will never become readable, so
+    report it as a bug instead of letting the sender retry it, and do not
+    treat it as forgery.
+    """
+
+
 # HTTP
 class SeamHttpApiError(SeamError):
     """

seam/seam_webhook.py

@@ -1,5 +1,9 @@
+from json import JSONDecodeError
 from typing import Dict
-from svix.webhooks import Webhook
+
+from svix.webhooks import Webhook, WebhookVerificationError
+
+from .exceptions import SeamInvalidWebhookPayloadError, SeamWebhookVerificationError
 from .resources import SeamEvent, seam_event_from_dict
 
 
@@ -25,9 +29,45 @@ def verify(self, payload: str, headers: Dict[str, str]) -> SeamEvent:
         :type headers: Dict[str, str]
         :return: The SeamEvent object created from the verified payload.
         :rtype: SeamEvent
-        :raises WebhookVerificationError: If the webhook signature verification fails.
+        :raises SeamWebhookVerificationError: If the webhook signature
+            verification fails. Respond with an error status so the sender
+            retries.
+        :raises SeamInvalidWebhookPayloadError: If the payload is correctly
+            signed but cannot be read as a Seam event. The payload will never
+            become readable, so report it as a bug instead of letting the
+            sender retry it.
         """
-        normalized_headers = {k.lower(): v for k, v in headers.items()}
-        res = self._webhook.verify(payload, normalized_headers)
+        normalized_headers: Dict[str, str] = {}
+        for key, value in headers.items():
+            name = str(key).lower()
+            if name in normalized_headers and normalized_headers[name] != value:
+                raise SeamWebhookVerificationError(
+                    f"Conflicting values for webhook header {name}"
+                )
+            normalized_headers[name] = value
+
+        try:
+            res = self._webhook.verify(payload, normalized_headers)
+        except JSONDecodeError as error:
+            # The signature already checked out, so the payload is genuinely
+            # from Seam but permanently unreadable.
+            raise SeamInvalidWebhookPayloadError(
+                f"The verified webhook payload is not valid JSON: {error}"
+            ) from error
+        except WebhookVerificationError as error:
+            raise SeamWebhookVerificationError(str(error)) from error
+        except ValueError as error:
+            # A malformed signature or timestamp header can leak a bare
+            # ValueError out of the verifier; it is a verification failure.
+            raise SeamWebhookVerificationError(str(error)) from error
+
+        if (
+            not isinstance(res, dict)
+            or not isinstance(res.get("event_id"), str)
+            or not isinstance(res.get("event_type"), str)
+        ):
+            raise SeamInvalidWebhookPayloadError(
+                "The verified webhook payload did not contain a Seam event"
+            )
 
         return seam_event_from_dict(res)

test/seam_webhook_test.py

@@ -0,0 +1,168 @@
+import json
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from svix.webhooks import Webhook, WebhookVerificationError
+
+from seam import (
+    SeamError,
+    SeamInvalidWebhookPayloadError,
+    SeamWebhook,
+    SeamWebhookVerificationError,
+)
+
+SECRET = "MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw"
+
+EVENT_PAYLOAD = json.dumps(
+    {
+        "event_id": "11111111-1111-1111-1111-111111111111",
+        "event_type": "device.connected",
+        "workspace_id": "22222222-2222-2222-2222-222222222222",
+        "device_id": "33333333-3333-3333-3333-333333333333",
+        "created_at": "2026-08-27T00:00:00.000Z",
+        "occurred_at": "2026-08-27T00:00:00.000Z",
+    }
+)
+
+
+def sign_headers(payload, msg_id="msg_1", timestamp=None, secret=SECRET):
+    timestamp = timestamp or datetime.now(timezone.utc)
+    signature = Webhook(secret).sign(msg_id, timestamp, payload)
+
+    return {
+        "svix-id": msg_id,
+        "svix-timestamp": str(int(timestamp.timestamp())),
+        "svix-signature": signature,
+    }
+
+
+def test_verifies_and_parses_a_signed_event():
+    webhook = SeamWebhook(SECRET)
+
+    event = webhook.verify(EVENT_PAYLOAD, sign_headers(EVENT_PAYLOAD))
+
+    assert event.event_type == "device.connected"
+    assert event.event_id == "11111111-1111-1111-1111-111111111111"
+
+
+def test_accepts_mixed_case_headers():
+    webhook = SeamWebhook(SECRET)
+    headers = {k.upper(): v for k, v in sign_headers(EVENT_PAYLOAD).items()}
+
+    event = webhook.verify(EVENT_PAYLOAD, headers)
+
+    assert event.event_type == "device.connected"
+
+
+def test_a_tampered_payload_fails_verification():
+    webhook = SeamWebhook(SECRET)
+    headers = sign_headers(EVENT_PAYLOAD)
+    tampered = EVENT_PAYLOAD.replace("device.connected", "device.disconnected")
+
+    with pytest.raises(SeamWebhookVerificationError, match="No matching signature"):
+        webhook.verify(tampered, headers)
+
+
+def test_a_wrong_secret_fails_verification():
+    webhook = SeamWebhook(SECRET)
+    headers = sign_headers(EVENT_PAYLOAD, secret="WrongQ9r8GKYqrTwjUPD8ILPZIo2LaLa")
+
+    with pytest.raises(SeamWebhookVerificationError, match="No matching signature"):
+        webhook.verify(EVENT_PAYLOAD, headers)
+
+
+def test_an_expired_timestamp_fails_verification():
+    webhook = SeamWebhook(SECRET)
+    headers = sign_headers(
+        EVENT_PAYLOAD,
+        timestamp=datetime.now(timezone.utc) - timedelta(hours=1),
+    )
+
+    with pytest.raises(SeamWebhookVerificationError, match="too old"):
+        webhook.verify(EVENT_PAYLOAD, headers)
+
+
+@pytest.mark.parametrize("missing", ["svix-id", "svix-timestamp", "svix-signature"])
+def test_a_missing_header_fails_verification(missing):
+    webhook = SeamWebhook(SECRET)
+    headers = sign_headers(EVENT_PAYLOAD)
+    del headers[missing]
+
+    with pytest.raises(SeamWebhookVerificationError, match="Missing required headers"):
+        webhook.verify(EVENT_PAYLOAD, headers)
+
+
+def test_conflicting_duplicate_headers_fail_verification():
+    webhook = SeamWebhook(SECRET)
+    headers = sign_headers(EVENT_PAYLOAD)
+    headers["SVIX-ID"] = "msg_other"
+
+    with pytest.raises(
+        SeamWebhookVerificationError,
+        match="Conflicting values for webhook header svix-id",
+    ):
+        webhook.verify(EVENT_PAYLOAD, headers)
+
+
+def test_identical_duplicate_headers_are_accepted():
+    webhook = SeamWebhook(SECRET)
+    headers = sign_headers(EVENT_PAYLOAD)
+    headers["SVIX-ID"] = headers["svix-id"]
+
+    event = webhook.verify(EVENT_PAYLOAD, headers)
+
+    assert event.event_type == "device.connected"
+
+
+def test_a_malformed_signature_header_fails_verification():
+    webhook = SeamWebhook(SECRET)
+    headers = sign_headers(EVENT_PAYLOAD)
+    headers["svix-signature"] = "garbage-without-a-version"
+
+    with pytest.raises(SeamWebhookVerificationError):
+        webhook.verify(EVENT_PAYLOAD, headers)
+
+
+def test_a_signed_but_unparseable_payload_is_not_forgery():
+    webhook = SeamWebhook(SECRET)
+    payload = '{"event_id": "trailing-comma",}'
+
+    with pytest.raises(
+        SeamInvalidWebhookPayloadError,
+        match="The verified webhook payload is not valid JSON",
+    ):
+        webhook.verify(payload, sign_headers(payload))
+
+
+@pytest.mark.parametrize("payload", ["null", "[1]", "42", '"event"', "{}"])
+def test_a_signed_non_event_payload_is_not_forgery(payload):
+    webhook = SeamWebhook(SECRET)
+
+    with pytest.raises(
+        SeamInvalidWebhookPayloadError,
+        match="The verified webhook payload did not contain a Seam event",
+    ):
+        webhook.verify(payload, sign_headers(payload))
+
+
+def test_an_unknown_event_type_still_parses():
+    webhook = SeamWebhook(SECRET)
+    payload = json.dumps(
+        {
+            "event_id": "11111111-1111-1111-1111-111111111111",
+            "event_type": "future.event_type",
+            "future_field": {"nested": True},
+        }
+    )
+
+    event = webhook.verify(payload, sign_headers(payload))
+
+    assert event.event_type == "future.event_type"
+    assert event.future_field.nested is True
+
+
+def test_webhook_errors_are_seam_errors():
+    assert issubclass(SeamWebhookVerificationError, SeamError)
+    assert issubclass(SeamInvalidWebhookPayloadError, SeamError)
+    # Existing code catching the svix error keeps working.
+    assert issubclass(SeamWebhookVerificationError, WebhookVerificationError)