diff --git a/README.rst b/README.rst index 3bbbb486..37ebce40 100644 --- a/README.rst +++ b/README.rst @@ -597,12 +597,20 @@ Refer to the `Svix docs on Consuming Webhooks `_, see the `Svix docs for more examples in specific frameworks `_. +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 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) diff --git a/test/seam_webhook_test.py b/test/seam_webhook_test.py new file mode 100644 index 00000000..0819c36f --- /dev/null +++ b/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)