From 09bdb8aaadd8bf46e3a98fd37e08216e38a20f97 Mon Sep 17 00:00:00 2001 From: KM Koushik Date: Sun, 8 Feb 2026 08:18:14 +1100 Subject: [PATCH] feat(python-sdk): add webhook verification and event handling (#344) * feat(python-sdk): add webhook verification and event handling Add webhook support to the Python SDK matching the JS SDK implementation: - Add Webhooks class with verify() and construct_event() methods - Implement HMAC-SHA256 signature verification with timing-safe comparison - Add timestamp validation with configurable tolerance (default 5 minutes) - Add comprehensive webhook event types (18 events: email, contact, domain, test) - Add WebhookVerificationError with typed error codes - Export webhook constants (headers) and types * fix(python-sdk): harden webhook parsing and typing Normalize invalid UTF-8 webhook payloads to INVALID_BODY errors so verify() safely returns false, and narrow base email webhook event types to avoid discriminated-union overlap. Add regression tests for both paths. * chore(python-sdk): bump package version to 0.2.9 * feat(python-sdk): add local webhook test example project Add a runnable Flask receiver and signed webhook sender under packages/python-sdk/example, and link it from the Python SDK README for local verification. --------- Co-authored-by: Claude --- packages/python-sdk/README.md | 6 + .../example/webhook-test-project/README.md | 43 ++ .../example/webhook-test-project/receiver.py | 37 ++ .../webhook-test-project/requirements.txt | 2 + .../webhook-test-project/send_test_webhook.py | 63 +++ packages/python-sdk/pyproject.toml | 2 +- packages/python-sdk/tests/test_webhooks.py | 103 +++++ packages/python-sdk/usesend/__init__.py | 22 +- packages/python-sdk/usesend/types.py | 415 ++++++++++++++++++ packages/python-sdk/usesend/usesend.py | 29 ++ packages/python-sdk/usesend/webhooks.py | 326 ++++++++++++++ 11 files changed, 1046 insertions(+), 2 deletions(-) create mode 100644 packages/python-sdk/example/webhook-test-project/README.md create mode 100644 packages/python-sdk/example/webhook-test-project/receiver.py create mode 100644 packages/python-sdk/example/webhook-test-project/requirements.txt create mode 100644 packages/python-sdk/example/webhook-test-project/send_test_webhook.py create mode 100644 packages/python-sdk/tests/test_webhooks.py create mode 100644 packages/python-sdk/usesend/webhooks.py diff --git a/packages/python-sdk/README.md b/packages/python-sdk/README.md index 56a3917..5605cc8 100644 --- a/packages/python-sdk/README.md +++ b/packages/python-sdk/README.md @@ -84,6 +84,12 @@ else: print("ok:", raw) ``` +## Webhook Local Example + +For a runnable webhook verification demo project, see: + +- `example/webhook-test-project/README.md` + ## Development This package is managed with Poetry. Models are maintained in-repo under diff --git a/packages/python-sdk/example/webhook-test-project/README.md b/packages/python-sdk/example/webhook-test-project/README.md new file mode 100644 index 0000000..5dbdd30 --- /dev/null +++ b/packages/python-sdk/example/webhook-test-project/README.md @@ -0,0 +1,43 @@ +# Webhook Test Project (Flask) + +This example project helps you validate Python SDK webhook signature verification locally. + +## What it includes + +- `receiver.py`: local webhook endpoint that verifies and parses events +- `send_test_webhook.py`: sends a signed test webhook request to your local endpoint + +## Setup + +```bash +cd packages/python-sdk/example/webhook-test-project +python -m venv .venv +source .venv/bin/activate +pip install -r requirements.txt +``` + +## Run + +Terminal 1: + +```bash +python receiver.py +``` + +Terminal 2: + +```bash +python send_test_webhook.py +``` + +You should see: + +- `200` response from the receiver +- parsed webhook event output in the receiver terminal + +## Environment variables + +- `USESEND_WEBHOOK_SECRET` (default: `whsec_test`) +- `WEBHOOK_URL` (default: `http://127.0.0.1:8000/webhook`) + +Use the same `USESEND_WEBHOOK_SECRET` for both scripts. diff --git a/packages/python-sdk/example/webhook-test-project/receiver.py b/packages/python-sdk/example/webhook-test-project/receiver.py new file mode 100644 index 0000000..8823293 --- /dev/null +++ b/packages/python-sdk/example/webhook-test-project/receiver.py @@ -0,0 +1,37 @@ +from __future__ import annotations + +import os + +from flask import Flask, jsonify, request +from flask.typing import ResponseReturnValue + +from usesend import UseSend, WebhookVerificationError # type: ignore[import-not-found] + + +WEBHOOK_SECRET = os.getenv("USESEND_WEBHOOK_SECRET", "whsec_test") + +app = Flask(__name__) +usesend = UseSend("us_test") +webhooks = usesend.webhooks(WEBHOOK_SECRET) + + +@app.post("/webhook") +def webhook() -> ResponseReturnValue: + raw_body = request.get_data() + + try: + event = webhooks.construct_event(raw_body, headers=request.headers) + except WebhookVerificationError as exc: + return jsonify({"ok": False, "code": exc.code, "message": str(exc)}), 400 + + print(f"Received event: {event['type']}") + + if event["type"] == "email.bounced": + bounce = event["data"].get("bounce", {}) + print("Bounce details:", bounce) + + return jsonify({"ok": True, "type": event["type"]}), 200 + + +if __name__ == "__main__": + app.run(host="127.0.0.1", port=8000) diff --git a/packages/python-sdk/example/webhook-test-project/requirements.txt b/packages/python-sdk/example/webhook-test-project/requirements.txt new file mode 100644 index 0000000..2d45018 --- /dev/null +++ b/packages/python-sdk/example/webhook-test-project/requirements.txt @@ -0,0 +1,2 @@ +-e ../.. +flask>=3.0,<4.0 diff --git a/packages/python-sdk/example/webhook-test-project/send_test_webhook.py b/packages/python-sdk/example/webhook-test-project/send_test_webhook.py new file mode 100644 index 0000000..e4d8c4b --- /dev/null +++ b/packages/python-sdk/example/webhook-test-project/send_test_webhook.py @@ -0,0 +1,63 @@ +from __future__ import annotations + +import hashlib +import hmac +import json +import os +import time +import uuid + +import requests + + +WEBHOOK_SECRET = os.getenv("USESEND_WEBHOOK_SECRET", "whsec_test") +WEBHOOK_URL = os.getenv("WEBHOOK_URL", "http://127.0.0.1:8000/webhook") + + +def _signature(secret: str, timestamp_ms: str, body: str) -> str: + digest = hmac.new( + secret.encode("utf-8"), + f"{timestamp_ms}.{body}".encode("utf-8"), + hashlib.sha256, + ).hexdigest() + return f"v1={digest}" + + +def main() -> None: + payload = { + "id": f"evt_{uuid.uuid4().hex[:8]}", + "type": "email.bounced", + "createdAt": "2026-02-08T10:00:00.000Z", + "data": { + "id": "email_123", + "status": "BOUNCED", + "from": "sender@example.com", + "to": ["recipient@example.com"], + "occurredAt": "2026-02-08T10:00:00.000Z", + "bounce": { + "type": "Permanent", + "subType": "General", + "message": "Mailbox unavailable", + }, + }, + } + + body = json.dumps(payload) + timestamp = str(int(time.time() * 1000)) + signature = _signature(WEBHOOK_SECRET, timestamp, body) + + headers = { + "Content-Type": "application/json", + "X-UseSend-Signature": signature, + "X-UseSend-Timestamp": timestamp, + "X-UseSend-Event": payload["type"], + "X-UseSend-Call": f"call_{uuid.uuid4().hex[:10]}", + } + + response = requests.post(WEBHOOK_URL, data=body, headers=headers, timeout=10) + print("Status:", response.status_code) + print("Body:", response.text) + + +if __name__ == "__main__": + main() diff --git a/packages/python-sdk/pyproject.toml b/packages/python-sdk/pyproject.toml index f598be3..faee6a5 100644 --- a/packages/python-sdk/pyproject.toml +++ b/packages/python-sdk/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "usesend" -version = "0.2.8" +version = "0.2.9" description = "Python SDK for the UseSend API" authors = ["UseSend"] license = "MIT" diff --git a/packages/python-sdk/tests/test_webhooks.py b/packages/python-sdk/tests/test_webhooks.py new file mode 100644 index 0000000..04563d9 --- /dev/null +++ b/packages/python-sdk/tests/test_webhooks.py @@ -0,0 +1,103 @@ +import hashlib +import hmac +import json +import time +from typing import get_args, get_type_hints + +import pytest + +from usesend import types +from usesend.webhooks import ( + WEBHOOK_SIGNATURE_HEADER, + WEBHOOK_TIMESTAMP_HEADER, + WebhookVerificationError, + Webhooks, +) + + +def _sign(secret: str, timestamp: str, body: str) -> str: + digest = hmac.new( + secret.encode("utf-8"), + f"{timestamp}.{body}".encode("utf-8"), + hashlib.sha256, + ).hexdigest() + return f"v1={digest}" + + +def test_verify_returns_false_for_non_utf8_bytes_body() -> None: + webhooks = Webhooks("whsec_test") + timestamp = str(int(time.time() * 1000)) + + is_valid = webhooks.verify( + b"\xff", + headers={ + WEBHOOK_SIGNATURE_HEADER: "v1=deadbeef", + WEBHOOK_TIMESTAMP_HEADER: timestamp, + }, + ) + + assert is_valid is False + + +def test_construct_event_raises_invalid_body_for_non_utf8_bytes() -> None: + webhooks = Webhooks("whsec_test") + timestamp = str(int(time.time() * 1000)) + + with pytest.raises(WebhookVerificationError) as exc: + webhooks.construct_event( + b"\xff", + headers={ + WEBHOOK_SIGNATURE_HEADER: "v1=deadbeef", + WEBHOOK_TIMESTAMP_HEADER: timestamp, + }, + ) + + assert exc.value.code == "INVALID_BODY" + + +def test_email_webhook_event_type_excludes_specialized_events() -> None: + email_event_type = get_type_hints(types.EmailWebhookEvent)["type"] + supported = set(get_args(email_event_type)) + + assert "email.delivered" in supported + assert "email.bounced" not in supported + assert "email.failed" not in supported + assert "email.suppressed" not in supported + assert "email.opened" not in supported + assert "email.clicked" not in supported + + +def test_construct_event_parses_bounced_event_with_valid_signature() -> None: + secret = "whsec_test" + webhooks = Webhooks(secret) + timestamp = str(int(time.time() * 1000)) + + payload = { + "id": "evt_123", + "type": "email.bounced", + "createdAt": "2026-02-08T10:00:00.000Z", + "data": { + "id": "email_123", + "status": "BOUNCED", + "from": "from@example.com", + "to": ["to@example.com"], + "occurredAt": "2026-02-08T10:00:00.000Z", + "bounce": { + "type": "Permanent", + "subType": "General", + }, + }, + } + body = json.dumps(payload) + signature = _sign(secret, timestamp, body) + + event = webhooks.construct_event( + body, + headers={ + WEBHOOK_SIGNATURE_HEADER: signature, + WEBHOOK_TIMESTAMP_HEADER: timestamp, + }, + ) + + assert event["type"] == "email.bounced" + assert event["data"]["bounce"]["type"] == "Permanent" diff --git a/packages/python-sdk/usesend/__init__.py b/packages/python-sdk/usesend/__init__.py index 9194975..f3bb914 100644 --- a/packages/python-sdk/usesend/__init__.py +++ b/packages/python-sdk/usesend/__init__.py @@ -3,6 +3,26 @@ from .usesend import UseSend, UseSendHTTPError from .domains import Domains # type: ignore from .campaigns import Campaigns # type: ignore +from .webhooks import ( + Webhooks, + WebhookVerificationError, + WEBHOOK_SIGNATURE_HEADER, + WEBHOOK_TIMESTAMP_HEADER, + WEBHOOK_EVENT_HEADER, + WEBHOOK_CALL_HEADER, +) from . import types -__all__ = ["UseSend", "UseSendHTTPError", "types", "Domains", "Campaigns"] +__all__ = [ + "UseSend", + "UseSendHTTPError", + "types", + "Domains", + "Campaigns", + "Webhooks", + "WebhookVerificationError", + "WEBHOOK_SIGNATURE_HEADER", + "WEBHOOK_TIMESTAMP_HEADER", + "WEBHOOK_EVENT_HEADER", + "WEBHOOK_CALL_HEADER", +] diff --git a/packages/python-sdk/usesend/types.py b/packages/python-sdk/usesend/types.py index 6ec188d..f5c9553 100644 --- a/packages/python-sdk/usesend/types.py +++ b/packages/python-sdk/usesend/types.py @@ -449,3 +449,418 @@ class CampaignActionResponse(TypedDict, total=False): class APIError(TypedDict): code: str message: str + + +# --------------------------------------------------------------------------- +# Webhook Events +# --------------------------------------------------------------------------- + +# Event type literals +ContactWebhookEventType = Literal[ + "contact.created", + "contact.updated", + "contact.deleted", +] + +DomainWebhookEventType = Literal[ + "domain.created", + "domain.verified", + "domain.updated", + "domain.deleted", +] + +EmailWebhookEventType = Literal[ + "email.queued", + "email.sent", + "email.delivery_delayed", + "email.delivered", + "email.bounced", + "email.rejected", + "email.rendering_failure", + "email.complained", + "email.failed", + "email.cancelled", + "email.suppressed", + "email.opened", + "email.clicked", +] + +EmailBaseWebhookEventType = Literal[ + "email.queued", + "email.sent", + "email.delivery_delayed", + "email.delivered", + "email.rejected", + "email.rendering_failure", + "email.complained", + "email.cancelled", +] + +WebhookTestEventType = Literal["webhook.test"] + +WebhookEventType = Literal[ + # Contact events + "contact.created", + "contact.updated", + "contact.deleted", + # Domain events + "domain.created", + "domain.verified", + "domain.updated", + "domain.deleted", + # Email events + "email.queued", + "email.sent", + "email.delivery_delayed", + "email.delivered", + "email.bounced", + "email.rejected", + "email.rendering_failure", + "email.complained", + "email.failed", + "email.cancelled", + "email.suppressed", + "email.opened", + "email.clicked", + # Test event + "webhook.test", +] + +# Email status for webhook payloads +WebhookEmailStatus = Literal[ + "QUEUED", + "SENT", + "DELIVERY_DELAYED", + "DELIVERED", + "BOUNCED", + "REJECTED", + "RENDERING_FAILURE", + "COMPLAINED", + "FAILED", + "CANCELLED", + "SUPPRESSED", + "OPENED", + "CLICKED", + "SCHEDULED", +] + + +# Webhook payload types +class EmailBasePayload(TypedDict, total=False): + """Base payload for email webhook events.""" + + id: str + status: WebhookEmailStatus + # Note: 'from' is a reserved keyword, using alternative access + to: List[str] + occurredAt: str + campaignId: Optional[str] + contactId: Optional[str] + domainId: Optional[int] + subject: str + templateId: str + metadata: Dict[str, Any] + + +# Using functional syntax for 'from' field +EmailBasePayloadFull = TypedDict( + "EmailBasePayloadFull", + { + "id": str, + "status": WebhookEmailStatus, + "from": str, + "to": List[str], + "occurredAt": str, + "campaignId": NotRequired[Optional[str]], + "contactId": NotRequired[Optional[str]], + "domainId": NotRequired[Optional[int]], + "subject": NotRequired[str], + "templateId": NotRequired[str], + "metadata": NotRequired[Dict[str, Any]], + }, +) + + +class ContactWebhookPayload(TypedDict, total=False): + """Payload for contact webhook events.""" + + id: str + email: str + contactBookId: str + subscribed: bool + properties: Dict[str, Any] + firstName: Optional[str] + lastName: Optional[str] + createdAt: str + updatedAt: str + + +class DomainWebhookPayload(TypedDict, total=False): + """Payload for domain webhook events.""" + + id: int + name: str + status: str + region: str + createdAt: str + updatedAt: str + clickTracking: bool + openTracking: bool + subdomain: Optional[str] + sesTenantId: Optional[str] + dkimStatus: Optional[str] + spfDetails: Optional[str] + dmarcAdded: Optional[bool] + + +BounceType = Literal["Transient", "Permanent", "Undetermined"] +BounceSubType = Literal[ + "General", + "NoEmail", + "Suppressed", + "OnAccountSuppressionList", + "MailboxFull", + "MessageTooLarge", + "ContentRejected", + "AttachmentRejected", +] + + +class BounceDetails(TypedDict, total=False): + """Bounce details for email.bounced events.""" + + type: BounceType + subType: BounceSubType + message: str + + +class FailureDetails(TypedDict): + """Failure details for email.failed events.""" + + reason: str + + +SuppressionType = Literal["Bounce", "Complaint", "Manual"] + + +class SuppressionDetails(TypedDict, total=False): + """Suppression details for email.suppressed events.""" + + type: SuppressionType + reason: str + source: str + + +class OpenDetails(TypedDict, total=False): + """Open tracking details for email.opened events.""" + + timestamp: str + userAgent: str + ip: str + platform: str + + +class ClickDetails(TypedDict, total=False): + """Click tracking details for email.clicked events.""" + + timestamp: str + url: str + userAgent: str + ip: str + platform: str + + +# Extended email payloads with additional details +EmailBouncedPayload = TypedDict( + "EmailBouncedPayload", + { + "id": str, + "status": WebhookEmailStatus, + "from": str, + "to": List[str], + "occurredAt": str, + "campaignId": NotRequired[Optional[str]], + "contactId": NotRequired[Optional[str]], + "domainId": NotRequired[Optional[int]], + "subject": NotRequired[str], + "templateId": NotRequired[str], + "metadata": NotRequired[Dict[str, Any]], + "bounce": BounceDetails, + }, +) + +EmailFailedPayload = TypedDict( + "EmailFailedPayload", + { + "id": str, + "status": WebhookEmailStatus, + "from": str, + "to": List[str], + "occurredAt": str, + "campaignId": NotRequired[Optional[str]], + "contactId": NotRequired[Optional[str]], + "domainId": NotRequired[Optional[int]], + "subject": NotRequired[str], + "templateId": NotRequired[str], + "metadata": NotRequired[Dict[str, Any]], + "failed": FailureDetails, + }, +) + +EmailSuppressedPayload = TypedDict( + "EmailSuppressedPayload", + { + "id": str, + "status": WebhookEmailStatus, + "from": str, + "to": List[str], + "occurredAt": str, + "campaignId": NotRequired[Optional[str]], + "contactId": NotRequired[Optional[str]], + "domainId": NotRequired[Optional[int]], + "subject": NotRequired[str], + "templateId": NotRequired[str], + "metadata": NotRequired[Dict[str, Any]], + "suppression": SuppressionDetails, + }, +) + +EmailOpenedPayload = TypedDict( + "EmailOpenedPayload", + { + "id": str, + "status": WebhookEmailStatus, + "from": str, + "to": List[str], + "occurredAt": str, + "campaignId": NotRequired[Optional[str]], + "contactId": NotRequired[Optional[str]], + "domainId": NotRequired[Optional[int]], + "subject": NotRequired[str], + "templateId": NotRequired[str], + "metadata": NotRequired[Dict[str, Any]], + "open": OpenDetails, + }, +) + +EmailClickedPayload = TypedDict( + "EmailClickedPayload", + { + "id": str, + "status": WebhookEmailStatus, + "from": str, + "to": List[str], + "occurredAt": str, + "campaignId": NotRequired[Optional[str]], + "contactId": NotRequired[Optional[str]], + "domainId": NotRequired[Optional[int]], + "subject": NotRequired[str], + "templateId": NotRequired[str], + "metadata": NotRequired[Dict[str, Any]], + "click": ClickDetails, + }, +) + + +class WebhookTestPayload(TypedDict): + """Payload for webhook.test events.""" + + test: bool + webhookId: str + sentAt: str + + +# Webhook event structures +class EmailWebhookEvent(TypedDict): + """Structure for email webhook events.""" + + id: str + type: EmailBaseWebhookEventType + createdAt: str + data: EmailBasePayloadFull + + +class EmailBouncedEvent(TypedDict): + """Structure for email.bounced webhook events.""" + + id: str + type: Literal["email.bounced"] + createdAt: str + data: EmailBouncedPayload + + +class EmailFailedEvent(TypedDict): + """Structure for email.failed webhook events.""" + + id: str + type: Literal["email.failed"] + createdAt: str + data: EmailFailedPayload + + +class EmailSuppressedEvent(TypedDict): + """Structure for email.suppressed webhook events.""" + + id: str + type: Literal["email.suppressed"] + createdAt: str + data: EmailSuppressedPayload + + +class EmailOpenedEvent(TypedDict): + """Structure for email.opened webhook events.""" + + id: str + type: Literal["email.opened"] + createdAt: str + data: EmailOpenedPayload + + +class EmailClickedEvent(TypedDict): + """Structure for email.clicked webhook events.""" + + id: str + type: Literal["email.clicked"] + createdAt: str + data: EmailClickedPayload + + +class ContactWebhookEvent(TypedDict): + """Structure for contact webhook events.""" + + id: str + type: ContactWebhookEventType + createdAt: str + data: ContactWebhookPayload + + +class DomainWebhookEvent(TypedDict): + """Structure for domain webhook events.""" + + id: str + type: DomainWebhookEventType + createdAt: str + data: DomainWebhookPayload + + +class WebhookTestEvent(TypedDict): + """Structure for webhook.test events.""" + + id: str + type: Literal["webhook.test"] + createdAt: str + data: WebhookTestPayload + + +# Union type for all webhook events +WebhookEventData = Union[ + EmailWebhookEvent, + EmailBouncedEvent, + EmailFailedEvent, + EmailSuppressedEvent, + EmailOpenedEvent, + EmailClickedEvent, + ContactWebhookEvent, + DomainWebhookEvent, + WebhookTestEvent, +] diff --git a/packages/python-sdk/usesend/usesend.py b/packages/python-sdk/usesend/usesend.py index f65c1f4..471b40c 100644 --- a/packages/python-sdk/usesend/usesend.py +++ b/packages/python-sdk/usesend/usesend.py @@ -74,6 +74,34 @@ class UseSend: self.domains = Domains(self) self.campaigns = Campaigns(self) + # ------------------------------------------------------------------ + # Webhooks + # ------------------------------------------------------------------ + def webhooks(self, secret: str) -> "Webhooks": + """Create a Webhooks instance for verifying webhook signatures. + + Parameters + ---------- + secret: + The webhook signing secret (starts with 'whsec_'). + + Returns + ------- + Webhooks + A Webhooks instance for verifying signatures and constructing events. + + Example + ------- + ```python + usesend = UseSend("us_12345") + webhooks = usesend.webhooks("whsec_xxx") + + # In your webhook handler + event = webhooks.construct_event(body, headers=request.headers) + ``` + """ + return Webhooks(secret) + # ------------------------------------------------------------------ # Internal request helper # ------------------------------------------------------------------ @@ -160,3 +188,4 @@ from .emails import Emails # noqa: E402 pylint: disable=wrong-import-position from .contacts import Contacts # noqa: E402 pylint: disable=wrong-import-position from .domains import Domains # type: ignore # noqa: E402 from .campaigns import Campaigns # type: ignore # noqa: E402 +from .webhooks import Webhooks # noqa: E402 pylint: disable=wrong-import-position diff --git a/packages/python-sdk/usesend/webhooks.py b/packages/python-sdk/usesend/webhooks.py new file mode 100644 index 0000000..12beaa0 --- /dev/null +++ b/packages/python-sdk/usesend/webhooks.py @@ -0,0 +1,326 @@ +"""Webhook verification and event construction for UseSend webhooks. + +This module provides secure webhook signature verification using HMAC-SHA256, +timestamp validation, and type-safe event parsing. +""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import time +from typing import Any, Dict, Literal, Mapping, Optional, Union + +from .types import WebhookEventData + + +# Webhook header names +WEBHOOK_SIGNATURE_HEADER = "X-UseSend-Signature" +WEBHOOK_TIMESTAMP_HEADER = "X-UseSend-Timestamp" +WEBHOOK_EVENT_HEADER = "X-UseSend-Event" +WEBHOOK_CALL_HEADER = "X-UseSend-Call" + +# Signature format +SIGNATURE_PREFIX = "v1=" + +# Default tolerance: 5 minutes in milliseconds +DEFAULT_TOLERANCE_MS = 5 * 60 * 1000 + + +WebhookVerificationErrorCode = Literal[ + "MISSING_SIGNATURE", + "MISSING_TIMESTAMP", + "INVALID_SIGNATURE_FORMAT", + "INVALID_TIMESTAMP", + "TIMESTAMP_OUT_OF_RANGE", + "SIGNATURE_MISMATCH", + "INVALID_BODY", + "INVALID_JSON", +] + + +class WebhookVerificationError(Exception): + """Error raised when webhook verification fails. + + Attributes: + code: The error code indicating the type of verification failure. + message: A human-readable description of the error. + """ + + def __init__(self, code: WebhookVerificationErrorCode, message: str) -> None: + self.code = code + self.message = message + super().__init__(message) + + def __str__(self) -> str: + return f"[{self.code}] {self.message}" + + +class Webhooks: + """Webhook verification and event construction. + + This class provides methods to verify webhook signatures and parse + webhook events with type safety. + + Parameters + ---------- + secret: + The webhook signing secret (starts with 'whsec_'). + + Example + ------- + ```python + from usesend import UseSend + + usesend = UseSend("us_12345") + webhooks = usesend.webhooks("whsec_xxx") + + # Flask example + @app.route("/webhook", methods=["POST"]) + def handle_webhook(): + try: + event = webhooks.construct_event( + request.data, + headers=request.headers + ) + + if event["type"] == "email.delivered": + print(f"Email delivered to {event['data']['to']}") + + return "OK", 200 + except WebhookVerificationError as e: + return str(e), 400 + ``` + """ + + def __init__(self, secret: str) -> None: + self._secret = secret + + def verify( + self, + body: Union[str, bytes], + *, + headers: Mapping[str, Any], + secret: Optional[str] = None, + tolerance: Optional[int] = None, + ) -> bool: + """Verify webhook signature without parsing the event. + + Parameters + ---------- + body: + Raw webhook body (string or bytes). + headers: + Request headers (dict-like object). + secret: + Optional override for the webhook secret. + tolerance: + Optional tolerance in milliseconds for timestamp validation. + Defaults to 5 minutes. Set to -1 to disable timestamp validation. + + Returns + ------- + bool + True if signature is valid, False otherwise. + + Example + ------- + ```python + is_valid = webhooks.verify(body, headers=request.headers) + + if not is_valid: + return "Invalid signature", 401 + ``` + """ + try: + self._verify_internal(body, headers=headers, secret=secret, tolerance=tolerance) + return True + except WebhookVerificationError: + return False + + def construct_event( + self, + body: Union[str, bytes], + *, + headers: Mapping[str, Any], + secret: Optional[str] = None, + tolerance: Optional[int] = None, + ) -> WebhookEventData: + """Verify and parse a webhook event. + + Parameters + ---------- + body: + Raw webhook body (string or bytes). + headers: + Request headers (dict-like object). + secret: + Optional override for the webhook secret. + tolerance: + Optional tolerance in milliseconds for timestamp validation. + Defaults to 5 minutes. Set to -1 to disable timestamp validation. + + Returns + ------- + WebhookEventData + Verified and typed webhook event. + + Raises + ------ + WebhookVerificationError + If the webhook signature is invalid or the payload cannot be parsed. + + Example + ------- + ```python + # Flask + event = webhooks.construct_event( + request.data, + headers=request.headers + ) + + # Django + event = webhooks.construct_event( + request.body, + headers=request.headers + ) + + # FastAPI + event = webhooks.construct_event( + await request.body(), + headers=dict(request.headers) + ) + + # Type-safe event handling + if event["type"] == "email.delivered": + print(event["data"]["to"]) + elif event["type"] == "email.bounced": + print(event["data"]["bounce"]["type"]) + ``` + """ + self._verify_internal(body, headers=headers, secret=secret, tolerance=tolerance) + + body_string = _to_string(body) + try: + return json.loads(body_string) + except (json.JSONDecodeError, ValueError) as e: + raise WebhookVerificationError( + "INVALID_JSON", + f"Webhook payload is not valid JSON: {e}", + ) from e + + def _verify_internal( + self, + body: Union[str, bytes], + *, + headers: Mapping[str, Any], + secret: Optional[str] = None, + tolerance: Optional[int] = None, + ) -> None: + """Internal verification logic.""" + webhook_secret = secret if secret is not None else self._secret + signature = _get_header(headers, WEBHOOK_SIGNATURE_HEADER) + timestamp = _get_header(headers, WEBHOOK_TIMESTAMP_HEADER) + + if not signature: + raise WebhookVerificationError( + "MISSING_SIGNATURE", + f"Missing {WEBHOOK_SIGNATURE_HEADER} header", + ) + + if not timestamp: + raise WebhookVerificationError( + "MISSING_TIMESTAMP", + f"Missing {WEBHOOK_TIMESTAMP_HEADER} header", + ) + + if not signature.startswith(SIGNATURE_PREFIX): + raise WebhookVerificationError( + "INVALID_SIGNATURE_FORMAT", + "Signature header must start with v1=", + ) + + try: + timestamp_num = int(timestamp) + except ValueError as e: + raise WebhookVerificationError( + "INVALID_TIMESTAMP", + "Timestamp header must be a number (milliseconds since epoch)", + ) from e + + tolerance_ms = tolerance if tolerance is not None else DEFAULT_TOLERANCE_MS + now = int(time.time() * 1000) + + if tolerance_ms >= 0 and abs(now - timestamp_num) > tolerance_ms: + raise WebhookVerificationError( + "TIMESTAMP_OUT_OF_RANGE", + "Webhook timestamp is outside the allowed tolerance", + ) + + body_string = _to_string(body) + expected = _compute_signature(webhook_secret, timestamp, body_string) + + if not _safe_compare(expected, signature): + raise WebhookVerificationError( + "SIGNATURE_MISMATCH", + "Webhook signature does not match", + ) + + +def _compute_signature(secret: str, timestamp: str, body: str) -> str: + """Compute the HMAC-SHA256 signature for webhook verification.""" + message = f"{timestamp}.{body}" + signature = hmac.new( + secret.encode("utf-8"), + message.encode("utf-8"), + hashlib.sha256, + ).hexdigest() + return f"{SIGNATURE_PREFIX}{signature}" + + +def _to_string(body: Union[str, bytes]) -> str: + """Convert body to UTF-8 string.""" + if isinstance(body, str): + return body + if isinstance(body, bytes): + try: + return body.decode("utf-8") + except UnicodeDecodeError as e: + raise WebhookVerificationError( + "INVALID_BODY", + "Webhook body must be valid UTF-8.", + ) from e + raise WebhookVerificationError( + "INVALID_BODY", + f"Unsupported body type: {type(body).__name__}. Expected str or bytes.", + ) + + +def _get_header(headers: Mapping[str, Any], name: str) -> Optional[str]: + """Get header value in a case-insensitive manner.""" + if headers is None: + return None + + # Try direct access first + if name in headers: + value = headers[name] + if isinstance(value, list): + return value[0] if value else None + return str(value) if value is not None else None + + # Case-insensitive lookup + lower_name = name.lower() + for key in headers: + if key.lower() == lower_name: + value = headers[key] + if isinstance(value, list): + return value[0] if value else None + return str(value) if value is not None else None + + return None + + +def _safe_compare(a: str, b: str) -> bool: + """Timing-safe string comparison to prevent timing attacks.""" + return hmac.compare_digest(a.encode("utf-8"), b.encode("utf-8"))