gib/GibSend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(python-sdk): add webhook verification and event handling (#344)

Browse Source 
* 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 <noreply@anthropic.com>
This commit is contained in:
KM Koushik
2026-02-08 08:18:14 +11:00
committed by GitHub
parent e246d32ef9
commit 09bdb8aaad
11 changed files with 1046 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/python-sdk/README.md
+6
View File
@@ -84,6 +84,12 @@ else:
print("ok:", raw) print("ok:", raw)
``` ```
## Webhook Local Example
For a runnable webhook verification demo project, see:
- `example/webhook-test-project/README.md`
## Development ## Development
This package is managed with Poetry. Models are maintained in-repo under This package is managed with Poetry. Models are maintained in-repo under
packages/python-sdk/example/webhook-test-project/README.md
+43
View File
@@ -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.
packages/python-sdk/example/webhook-test-project/receiver.py
+37
View File
@@ -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)
packages/python-sdk/example/webhook-test-project/requirements.txt
+2
View File
@@ -0,0 +1,2 @@
-e ../..
flask>=3.0,<4.0
packages/python-sdk/example/webhook-test-project/send_test_webhook.py
+63
View File
@@ -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()
packages/python-sdk/pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[tool.poetry] [tool.poetry]
name = "usesend" name = "usesend"
version = "0.2.8" version = "0.2.9"
description = "Python SDK for the UseSend API" description = "Python SDK for the UseSend API"
authors = ["UseSend"] authors = ["UseSend"]
license = "MIT" license = "MIT"
packages/python-sdk/tests/test_webhooks.py
+103
View File
@@ -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"
packages/python-sdk/usesend/__init__.py
+21 -1
View File
@@ -3,6 +3,26 @@
from .usesend import UseSend, UseSendHTTPError from .usesend import UseSend, UseSendHTTPError
from .domains import Domains # type: ignore from .domains import Domains # type: ignore
from .campaigns import Campaigns # 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 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",
]
packages/python-sdk/usesend/types.py
+415
View File
@@ -449,3 +449,418 @@ class CampaignActionResponse(TypedDict, total=False):
class APIError(TypedDict): class APIError(TypedDict):
code: str code: str
message: 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,
]
packages/python-sdk/usesend/usesend.py
+29
View File
@@ -74,6 +74,34 @@ class UseSend:
self.domains = Domains(self) self.domains = Domains(self)
self.campaigns = Campaigns(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 # 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 .contacts import Contacts # noqa: E402 pylint: disable=wrong-import-position
from .domains import Domains # type: ignore # noqa: E402 from .domains import Domains # type: ignore # noqa: E402
from .campaigns import Campaigns # type: ignore # noqa: E402 from .campaigns import Campaigns # type: ignore # noqa: E402
from .webhooks import Webhooks # noqa: E402 pylint: disable=wrong-import-position
packages/python-sdk/usesend/webhooks.py
+326
View File
@@ -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"))