An accepted send request and a delivery event answer different questions. The send endpoint returns HTTP 202 when the API has accepted a request; that does not establish delivery or inbox placement. A signed status event tells you what happened afterwards. This post walks through a small Python receiver that verifies the signature, stores a receipt once, and refuses to acknowledge anything it did not durably commit.
Keep two kinds of idempotency separate. Use a stable Idempotency-Key for one business-event send request, with the same request body on retries, so a retry cannot create another intended send (see designing idempotent sends). A webhook consumer has a second job: duplicate deliveries of one event_id must not repeat a downstream action, while a new event for the same message_id still needs processing.
It helps to treat four facts as separate. The API accepted a request. A verified event arrived from your configured subscription. Your consumer durably stored a receipt, which lets it deduplicate that event after a restart. A trusted worker applied the intended business effect. The receiver below establishes the second and third; the fourth is your application's job.
Every status webhook carries a PulsePigeon-Signature: t=<unix seconds>,v1=<hex> header. t is when the delivery was enqueued, and v1 is the HMAC-SHA256 of the timestamp, a dot, and the JSON body serialized with sorted keys and no whitespace, keyed with the subscription secret. Verify against that canonical form rather than the raw bytes you received, which may be formatted differently. The webhook documentation has the exact definition and golden test vectors.
The timestamp is set once, when the delivery is enqueued, and retries reuse the original signature. The receiver tolerance therefore has to cover queue delay, retry backoff, and clock skew. The example defaults to 3600 seconds, the same default the official SDK helpers use, and takes a tolerance_seconds argument. A retry that arrives later than your tolerance fails even though its signature is valid; see retries and dead-letter. Secret rotation has the same overlap: deliveries enqueued after a rotation are signed with the new secret, while queued retries keep the signature made with the old one, so the function takes the current secret plus the previous one (one or two secrets) and accepts a signature that matches either. See secret rotation.
The receiver checks the body size and the signature format, enforces the timestamp window, and parses the JSON while rejecting duplicate keys and non-finite numbers. It recomputes the canonical form and compares digests in constant time before it trusts any field. It then requires the expected schema version and a delivery event type, and inserts the event ID, message ID, event type, and canonical payload into a SQLite table keyed by event ID. A repeat of the same event returns duplicate. A reused event ID with different content raises ReceiptRejected, because that is a conflict, not a retry.
Return a success response only after accept_receipt returns stored or duplicate. A non-2xx response makes PulsePigeon retry, and after the configured number of consecutive failures the delivery is marked dead-lettered. Never return success for a rejected request or for a database failure.
Understand the example's limits. Use a separate private receipt database per trusted tenant or subscription, and never pick that scope from the event body. The example persists each verified event's canonical payload so a worker has the data it needs after a crash. That payload includes the recipient address, so restrict access to the database directory; the example never expires receipts, so apply your own retention. It handles the six delivery event types only. A subscription can also receive inbound.received, subscription.changed, and smtp.submission_error, which carry different payloads and need their own handling. The example refuses them, so choose the event types on your subscription to match what you handle, or extend the receiver.
A worker should read the stored payloads and project them using each event's occurred_at and your ordering policy, because events can arrive out of order. A late queued event must not erase an outcome you have already recorded as final. If you need to notify another system, write a durable internal job in the same transaction as the trusted business change and deliver it idempotently.
The example comes with tests that run against what the platform actually signs. They sign events with the platform's own signer and verify the published golden vectors. They also cover tampering, the wrong secret, rotation with a previous secret, the platform's own test delivery, timestamps just inside and just outside the tolerance, malformed headers, a reused event ID with different content, concurrent duplicate deliveries that store once, and hostile bodies. These are local tests, not evidence of deployed webhook delivery. To exercise your real receiver, use the subscription test endpoint, POST /v1/webhooks/status/{subscription_id}/test, which sends a signed synthetic pulsepigeon.test delivery; the receiver verifies and acknowledges it without storing it.
webhook_receiver.py (Python 3.11+, standard library only)
"""Offline example: verify a PulsePigeon status webhook and store its receipt once.
No listener, network client or sender. The signature contract matches the platform:
``PulsePigeon-Signature: t=<unix seconds>,v1=<hex>`` where ``v1`` is HMAC-SHA256 over
``<t>.<canonical JSON>`` and canonical JSON is
``json.dumps(payload, sort_keys=True, separators=(",", ":"))`` (ASCII-escaped).
"""
from __future__ import annotations
import hashlib
import hmac
import json
import math
import re
import sqlite3
from contextlib import closing
from pathlib import Path
from typing import Any
MAX_BODY_BYTES = 1_048_576
# The platform signs when a delivery is queued and reuses that signature on retries, so a
# receiver tolerance must cover queue delay, retry backoff and clock skew. 3600 seconds is
# the default the official SDK helpers use; tune it to your own subscription settings.
DEFAULT_TOLERANCE_SECONDS = 3600
SCHEMA_VERSION = "2026-08-14"
TEST_EVENT_TYPE = "pulsepigeon.test" # sent by POST /v1/webhooks/status/{id}/test
SIGNATURE_RE = re.compile(r"t=([0-9]{1,12}),v1=([0-9a-f]{64})")
# The delivery events this example handles. Other event types a subscription can receive
# (inbound.received, subscription.changed, smtp.submission_error) need their own handling.
EVENT_TYPES = frozenset({"accepted", "queued", "delivered", "deferred", "bounced", "complained"})
class ReceiptRejected(ValueError):
"""The caller must not acknowledge rejected or uncommitted input."""
def _unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
result: dict[str, Any] = {}
for key, value in pairs:
if key in result:
raise ReceiptRejected("duplicate_json_key")
result[key] = value
return result
def _finite_float(value: str) -> float:
result = float(value)
if not math.isfinite(result):
raise ReceiptRejected("nonfinite_json")
return result
def _reject_constant(value: str) -> Any:
raise ReceiptRejected("nonfinite_json")
def accept_receipt(
body: bytes,
signature: str,
*,
secrets: tuple[bytes, ...],
now: int,
database: Path,
tolerance_seconds: int = DEFAULT_TOLERANCE_SECONDS,
) -> str:
"""Return "stored" or "duplicate" only after verification and a durable commit.
``secrets`` is the current subscription secret, plus the previous one while a rotation
is in flight (queued retries keep the signature made with the old secret). Pass one or
two secrets. Use a separate database per trusted tenant or subscription, and never
derive that scope from the event JSON. The platform's test delivery
(``pulsepigeon.test``) is verified and acknowledged as "test" without being stored.
A worker should project stored events into business state using each event's own
times and an ordering policy; this example persists the verified payload for it.
"""
if not isinstance(secrets, tuple) or not 1 <= len(secrets) <= 2:
raise ReceiptRejected("secret_count")
if any(not isinstance(secret, bytes) or len(secret) < 16 for secret in secrets):
raise ReceiptRejected("secret_too_short")
if not isinstance(body, bytes) or len(body) > MAX_BODY_BYTES:
raise ReceiptRejected("body_limit")
match = SIGNATURE_RE.fullmatch(signature) if isinstance(signature, str) else None
if not match or abs(now - int(match[1])) > tolerance_seconds:
raise ReceiptRejected("signature_timestamp")
try:
payload = json.loads(
body,
object_pairs_hook=_unique_object,
parse_constant=_reject_constant,
parse_float=_finite_float,
)
except (ValueError, UnicodeDecodeError, RecursionError) as exc:
raise ReceiptRejected("invalid_json") from exc
if not isinstance(payload, dict):
raise ReceiptRejected("object_required")
try:
canonical = json.dumps(
payload, sort_keys=True, separators=(",", ":"), allow_nan=False
).encode("utf-8")
except (ValueError, RecursionError) as exc:
raise ReceiptRejected("invalid_json") from exc
signed = match[1].encode() + b"." + canonical
digests = [hmac.new(secret, signed, hashlib.sha256).hexdigest() for secret in secrets]
# Evaluate every candidate so timing does not reveal which secret matched.
matched = [hmac.compare_digest(digest, match[2]) for digest in digests]
if not any(matched):
raise ReceiptRejected("invalid_signature")
for key in ("event_id", "schema_version"):
if not isinstance(payload.get(key), str) or not 1 <= len(payload[key]) <= 128:
raise ReceiptRejected("invalid_" + key)
if payload["schema_version"] != SCHEMA_VERSION:
raise ReceiptRejected("unsupported_event")
if payload.get("type") == TEST_EVENT_TYPE:
return "test"
if not isinstance(payload.get("message_id"), str) or not 1 <= len(payload["message_id"]) <= 128:
raise ReceiptRejected("invalid_message_id")
if not isinstance(payload.get("type"), str) or payload["type"] not in EVENT_TYPES:
raise ReceiptRejected("unsupported_event")
database.parent.mkdir(parents=True, exist_ok=True)
# The stored payload includes the recipient address and event metadata. Restrict
# access to the database directory and apply your own retention to it.
with closing(sqlite3.connect(database, timeout=5)) as connection, connection:
connection.execute(
"CREATE TABLE IF NOT EXISTS receipts (event_id TEXT PRIMARY KEY, "
"message_id TEXT NOT NULL, event_type TEXT NOT NULL, "
"payload TEXT NOT NULL)"
)
try:
connection.execute(
"INSERT INTO receipts VALUES (?, ?, ?, ?)",
(
payload["event_id"],
payload["message_id"],
payload["type"],
canonical.decode("utf-8"),
),
)
except sqlite3.IntegrityError:
saved = connection.execute(
"SELECT payload FROM receipts WHERE event_id = ?",
(payload["event_id"],),
).fetchone()
if saved is None or not hmac.compare_digest(saved[0].encode("utf-8"), canonical):
raise ReceiptRejected("event_id_conflict") from None
return "duplicate"
return "stored"
Calling the receiver from your HTTP handler
from pathlib import Path
from webhook_receiver import ReceiptRejected, accept_receipt
# Load secrets from your secret store, never from the incoming payload. During a rotation,
# pass the current secret first and the previous one second; drop the second when it ends.
try:
status = accept_receipt(
raw_body, # the exact bytes of the request body
request.headers["PulsePigeon-Signature"],
secrets=(current_secret_bytes, previous_secret_bytes),
now=current_unix_seconds,
database=Path("private-receipts/subscription-42.sqlite3"),
)
except ReceiptRejected:
... # respond with a non-2xx status; never acknowledge
else:
... # status is "stored", "duplicate" or "test": respond 2xx