fix: Separate an unreadable webhook payload from a forged one

README.rst

@@ -597,12 +597,20 @@ 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 Svix's ``WebhookVerificationError``, re-exported as
+``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 +620,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

@@ -9,6 +9,7 @@
     SeamError,
     SeamHttpApiError,
     SeamHttpInvalidResponseError,
+    SeamInvalidWebhookPayloadError,
     SeamHttpUnauthorizedError,
     SeamHttpInvalidInputError,
     SeamValidationError,

seam/exceptions.py

@@ -1,5 +1,6 @@
 from dataclasses import dataclass
 from typing import Any, Dict, List, Optional
+
 from .resources import ActionAttempt, ErrorActionAttempt, PendingActionAttempt
 
 
@@ -15,6 +16,18 @@ class SeamError(Exception):
     """Base exception for all errors raised by the Seam SDK."""
 
 
+# Webhook
+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 SeamHttpInvalidResponseError(SeamError):
     """

seam/seam_webhook.py

@@ -1,10 +1,19 @@
+from json import JSONDecodeError
 from typing import Dict
+
 from svix.webhooks import Webhook
+
+from .exceptions import SeamInvalidWebhookPayloadError
 from .resources import SeamEvent, seam_event_from_dict
 
 
 class SeamWebhook:
-    """Verifies and parses incoming Seam webhook events using the Svix library."""
+    """Verifies and parses incoming Seam webhook events using the Svix library.
+
+    Verification failures raise svix's ``WebhookVerificationError``, which is
+    re-exported as ``SeamWebhookVerificationError``. A verified payload that
+    is not a readable event raises ``SeamInvalidWebhookPayloadError``.
+    """
 
     def __init__(self, secret: str):
         """
@@ -25,9 +34,32 @@ 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 = {str(key).lower(): value for key, value in headers.items()}
+
+        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
+
+        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,150 @@
+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_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_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_verification_failures_raise_the_svix_error():
+    # The webhook handler is svix, so a failed signature raises svix's own
+    # error rather than an SDK-specific wrapper.
+    assert SeamWebhookVerificationError is WebhookVerificationError
+
+
+def test_an_invalid_payload_is_a_seam_error():
+    assert issubclass(SeamInvalidWebhookPayloadError, SeamError)