{"schemaVersion":"1.0","type":"TechArticle","types":["Article","TechArticle"],"slug":"webhook-security-verify-signatures-stop-replays-handle-retries-x05wj","url":"https://zyvop.com/webhook-security-verify-signatures-stop-replays-handle-retries-x05wj","title":"Webhook Security: Verify Signatures, Stop Replays, Handle Retries","subtitle":"How to accept webhooks safely: verify the signature on the raw body, reject replays with timestamps and event IDs, and survive retries, with tested Python code.","tldr":"Anyone who finds your webhook URL can post to it. Here's how to verify HMAC signatures, stop replayed requests, and handle duplicate and retried deliveries, with code checked against GitHub's and Standard Webhooks' published test vectors.","keywords":["webhooks","Python","security","API Design","FastAPI","DevOps"],"entities":["Ankit Singh","Software Engineer","webhooks","Python","security","API Design","FastAPI","DevOps","ZyVOP"],"keyTakeaways":["A webhook endpoint is a public URL that your server trusts to make things happen.","A payment provider posts to it, and you credit an account.","A Git host posts to it, and you deploy."],"headings":["What you're defending against","How signing works","Step 1: Verify the signature","1. Verify the raw bytes","2. Compare in constant time","3. Check the timestamp, in both directions","4. Accept more than one secret","5. Trust only the version you know","6. Say nothing useful to attackers","Checked against the real thing","Step 2: Stop replays","Step 3: Handle retries","Acknowledge fast, process later","The inbox","The worker","Make the handler idempotent","Don't rely on order","Wire it up","See it work","Defense in depth","If you're the one sending webhooks","Checklist","The tests"],"outboundLinks":["https://docs.github.com/en/webhooks/using-webhooks/validating-webhook-deliveries","https://docs.github.com/en/webhooks/using-webhooks/best-practices-for-using-webhooks","https://docs.stripe.com/webhooks","https://docs.stripe.com/webhooks/signatures","https://www.standardwebhooks.com/","https://docs.python.org/3/library/hmac.html"],"contentText":"A webhook endpoint is a public URL that your server trusts to make things happen. A payment provider posts to it, and you credit an account. A Git host posts to it, and you deploy. The URL isn't secret for long. It shows up in logs, dashboards and screenshots. And anyone who knows it can send it a request. That leaves three questions you have to answer on every delivery: Is this really from the sender? That's forgery, and signatures answer it. Is someone replaying a real request they captured? That's replay, and timestamps plus event IDs answer it. Is the sender delivering the same event again, late, or out of order? That's retries, and a fast acknowledgement plus an idempotent handler answer it. This post builds the receiving side in Python. The signature code is checked against test vectors that GitHub and the Standard Webhooks spec publish, so you can see that it computes what the real senders compute. What you're defending against Attack or problem What happens The defense Forgery Someone posts a made-up event Verify an HMAC signature made with a shared secret Tampering Someone changes a real request on the way The signature covers the exact body Replay Someone re-sends a captured, validly signed request A timestamp window, plus event ID deduplication Timing attack Guessing a signature one byte at a time from response times Constant-time comparison Resource exhaustion Huge bodies, or floods of junk A size limit before any crypto, plus rate limits at the edge A leaked secret The shared secret escapes Rotation, so you accept the old and new secret for a while And here's the pipeline every request will pass through: flowchart TD A[\"POST /webhooks/payments\"] --> B{\"Body too large?\"} B -->|\"yes\"| X1[\"413\"] B -->|\"no\"| C{\"Headers present and<br/>timestamp inside the window?\"} C -->|\"no\"| X2[\"400\"] C -->|\"yes\"| D{\"Signature valid for<br/>one of our secrets?\"} D -->|\"no\"| X2 D -->|\"yes\"| E{\"Event ID already stored?\"} E -->|\"yes\"| F[\"200 duplicate\"] E -->|\"no\"| G[\"Store it, then 200 accepted\"]How signing works The sender and you share a secret. For each delivery, the sender computes an HMAC of the request using that secret, and sends the result in a header. HMAC-SHA256 is a keyed hash. Only someone holding the secret can produce the right value for a given body. You recompute it from what you received and compare. If the two match, the sender knew the secret and the body wasn't changed. The idea is the same everywhere, but the details differ: GitHub Stripe Standard Webhooks Header X-Hub-Signature-256 Stripe-Signature webhook-signature What's signed The body timestamp.body id.timestamp.body Format sha256= and hex t=...,v1=... with hex v1, and base64 Key The secret as text The whsec_... secret as text The base64 part of whsec_..., decoded Timestamp None In the header and signed Its own header, and signed Where's the event ID? X-GitHub-Delivery header id in the JSON body webhook-id header, and signed Standard Webhooks is an open spec that several services follow. We'll build on it, because it signs the ID and the timestamp as well as the body. That gives you replay protection almost for free. GitHub signs only the body, so with GitHub the delivery ID does all the replay work. Step 1: Verify the signature Create signing.py: import base64 import hashlib import hmac from dataclasses import dataclass class InvalidWebhook(Exception): \"\"\"The request can't be trusted. `reason` is for your logs, not for the sender.\"\"\" def __init__(self, reason): super().__init__(reason) self.reason = reason @dataclass(frozen=True) class Verified: message_id: str timestamp: int def _lower(headers): return {str(name).lower(): value for name, value in headers.items()} def _equal(a: str, b: str) -> bool: \"\"\"Constant-time comparison. Bytes, so a non-ASCII header can't raise a TypeError.\"\"\" return hmac.compare_digest(a.encode(\"utf-8\", \"replace\"), b.encode(\"utf-8\", \"replace\")) def _fresh(timestamp: str, now: float, tolerance: int) -> int: try: seconds = int(timestamp) except ValueError: raise InvalidWebhook(\"bad-timestamp\") from None if abs(now - seconds) > tolerance: # too old (a replay) or too far in the future raise InvalidWebhook(\"stale-timestamp\") return seconds # ---- Standard Webhooks: webhook-id, webhook-timestamp, webhook-signature -------------------- def standard_signature(secret: str, message_id: str, timestamp: str, body: bytes) -> str: key = base64.b64decode(secret.removeprefix(\"whsec_\")) signed = f\"{message_id}.{timestamp}.\".encode() + body # id.timestamp.raw-body return \"v1,\" + base64.b64encode(hmac.new(key, signed, hashlib.sha256).digest()).decode() def verify_standard(secrets, headers, body: bytes, now: float, tolerance: int = 300) -> Verified: \"\"\"`secrets` is a list, so you can accept the old and new secret while rotating.\"\"\" h = _lower(headers) try: message_id, timestamp, signatures = h[\"webhook-id\"], h[\"webhook-timestamp\"], h[\"webhook-signature\"] except KeyError: raise InvalidWebhook(\"missing-headers\") from None seconds = _fresh(timestamp, now, tolerance) # the cheap check goes first for secret in secrets: expected = standard_signature(secret, message_id, timestamp, body) for candidate in signatures.split(): # the header can hold several: \"v1,aaa v1,bbb\" if candidate.startswith(\"v1,\") and _equal(candidate, expected): return Verified(message_id, seconds) raise InvalidWebhook(\"bad-signature\") # ---- GitHub: X-Hub-Signature-256 (no timestamp, so dedupe on X-GitHub-Delivery) ------------- def verify_github(secret: str, headers, body: bytes) -> None: given = _lower(headers).get(\"x-hub-signature-256\") if given is None: raise InvalidWebhook(\"missing-headers\") expected = \"sha256=\" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest() if not _equal(given, expected): raise InvalidWebhook(\"bad-signature\") # ---- Stripe: Stripe-Signature: t=...,v1=...,v0=... ------------------------------------------ def verify_stripe(secret: str, header: str, body: bytes, now: float, tolerance: int = 300) -> int: parts = [item.split(\"=\", 1) for item in header.split(\",\") if \"=\" in item] timestamps = [value for key, value in parts if key.strip() == \"t\"] signatures = [value for key, value in parts if key.strip() == \"v1\"] # ignore v0 and anything else if not timestamps or not signatures: raise InvalidWebhook(\"missing-headers\") seconds = _fresh(timestamps[0], now, tolerance) expected = hmac.new(secret.encode(), f\"{timestamps[0]}.\".encode() + body, hashlib.sha256).hexdigest() if not any(_equal(signature, expected) for signature in signatures): raise InvalidWebhook(\"bad-signature\") return secondsAll three schemes are in there. Each one raises InvalidWebhook with a reason, and the reason is for your logs. The code follows six rules. Each one stops a real mistake. 1. Verify the raw bytes The signature covers the exact bytes the sender transmitted. If your framework parses the JSON and you serialize it again, the bytes can change. Whitespace, key order and number formatting are all fair game. test_re_serialised_json_breaks_the_signature shows it. A body with two spaces in it verifies fine. Parse it and dump it again, and the signature no longer matches. In FastAPI, take the Request and call await request.body(). Don't declare a Pydantic body parameter, because that parses the body before you ever see the raw bytes. 2. Compare in constant time == stops at the first difference. Given enough tries, an attacker can use response time to find a valid signature byte by byte. hmac.compare_digest is designed to prevent that. The Python docs recommend it for exactly this. One detail: both arguments must be the same type, and a str must be ASCII-only. A non-ASCII character in a header would raise a TypeError, and your endpoint would return a 500 instead of a rejection. So the code compares bytes. test_a_non_ascii_signature_is_rejected_not_a_crash covers it. 3. Check the timestamp, in both directions A request that's minutes old is suspicious, and so is one dated in the future. The window here is five minutes either way, which is Stripe's default and what Standard Webhooks recommends. Stripe warns against setting the tolerance to zero, and says to keep your server's clock synced with NTP. The cheap timestamp check runs before the HMAC. 4. Accept more than one secret When you rotate a secret, there's a gap where the sender uses the old one and you've deployed the new one, or the other way round. Stripe handles this by sending a signature for each active secret. A header can hold several signatures, and one valid match is enough. So verify_standard takes a list of secrets, and it tries every signature in the header. Once the old secret is retired, take it off the list. test_secret_rotation_accepts_old_and_new_and_ignores_unknown_versions checks both sides. 5. Trust only the version you know Signature headers carry a version label, like v1. Ignore any scheme you don't recognize. Stripe also sends a v0 signature for test events, and says never to trust it for live traffic. 6. Say nothing useful to attackers Every failure gets the same 400 and the same body. The reason (bad signature, stale timestamp, missing headers) goes to your logs. Otherwise your error messages become a guide for working out which check someone is failing. Checked against the real thing A verifier that agrees with itself proves little. So the tests check two published vectors. GitHub's docs publish a secret (It's a Secret to Everybody), a payload (Hello, World!) and the signature they should produce. The Standard Webhooks spec publishes a secret, message ID, timestamp, payload and signature. The code in this post reproduces both signatures exactly. You can see it in test_github_matches_its_published_vector and test_standard_webhooks_matches_its_published_vector. Step 2: Stop replays Say an attacker captures a real, correctly signed request. They can't forge a new one, but they can send the captured one again. If it credited an account once, will it credit it twice? Two layers answer that, and they cover different things. The timestamp window limits how long a captured request stays useful. After five minutes it's rejected. This only works if the timestamp is inside the signed content, because otherwise the attacker just edits it. Stripe and Standard Webhooks sign it. GitHub's signature has no timestamp at all. Deduplicating event IDs makes a replay inside the window harmless. If the ID has been seen, the request is acknowledged and nothing else happens. test_a_replay_is_harmless_inside_the_window_and_rejected_after_it shows both. The same bytes sent 100 seconds later are acknowledged as a duplicate. Sent 400 seconds after that, they're rejected as stale. The two windows are not the same length. The timestamp window defends against attackers, so it's short. The ID window defends against the sender's own retries, so it has to last as long as the sender can redeliver. Stripe retries for up to three days in live mode. Keep IDs at least that long. Step 3: Handle retries Senders retry, and not always the way you'd expect: Sender When it retries Notes Stripe After any response that isn't a 2xx, including redirects, and after timeouts. In live mode, for up to three days with exponential backoff Each attempt gets a new timestamp and signature GitHub It doesn't redeliver failed deliveries on its own. You redeliver them from the UI or the API You must answer with a 2xx within 10 seconds. A redelivery keeps the same X-GitHub-Delivery ID Standard Webhooks senders Typically exponential backoff with jitter The webhook-id stays the same across attempts, while the timestamp and signature are fresh Two things follow. First, a retry is a new request with a valid signature, a fresh timestamp and an ID you've already seen. The replay defenses above will accept it. That's right, because the dedupe layer turns it into a harmless duplicate. Second, the sender is probably waiting on you. A slow handler leads to a timeout, which leads to a retry, which leads to more load on the slow handler. Acknowledge fast, process later So do as little as possible in the request. Verify, store, answer: import logging import time from signing import InvalidWebhook, verify_standard log = logging.getLogger(\"webhooks\") class WebhookReceiver: \"\"\"Verify, store, answer. Nothing slow happens here.\"\"\" def __init__(self, inbox, secrets, clock=time.time, tolerance=300, max_body=1_000_000, on_reject=None): self.inbox, self.secrets, self.clock = inbox, secrets, clock self.tolerance, self.max_body = tolerance, max_body self.on_reject = on_reject or (lambda reason: log.warning(\"rejected webhook: %s\", reason)) def handle(self, headers, body: bytes): \"\"\"Returns (status code, JSON body).\"\"\" if len(body) > self.max_body: return 413, {\"error\": \"payload too large\"} # refuse before spending CPU on an HMAC try: verified = verify_standard(self.secrets, headers, body, self.clock(), self.tolerance) except InvalidWebhook as error: self.on_reject(error.reason) # the reason goes to your logs... return 400, {\"error\": \"invalid webhook\"} # ...and the sender learns nothing is_new = self.inbox.store(\"standard\", verified.message_id, body) return 200, {\"status\": \"accepted\" if is_new else \"duplicate\"} # a duplicate is still a successThe handler never runs here. The event goes into an inbox table, and a worker picks it up afterwards: sequenceDiagram participant S as Sender participant R as Your receiver participant I as Inbox (database) participant W as Worker S->>R: POST event (id evt_1) R->>R: verify signature, check timestamp R->>I: store evt_1 R-->>S: 200 accepted, in milliseconds W->>I: lease the next due event W->>W: run your handler Note over W: if it fails, retry later with a growing delay W->>I: mark it done S->>R: retry of evt_1 (the sender missed the 200) R->>I: store evt_1 (already there) R-->>S: 200 duplicateWhat to return in each case: Situation Status Why Valid, new 200 You've stored it, so you've got it Valid, already seen 200 A duplicate isn't an error. A non-2xx here would just make the sender retry again Bad signature, stale timestamp, missing headers 400 The same answer for all of them Body too large 413 Refused before any crypto You couldn't store it 500 The one time you want a retry An event type you don't handle 200 Acknowledge and ignore. Don't make the sender retry something you'll never process The inbox Here's the table behind it: import random import sqlite3 import time from contextlib import closing from dataclasses import dataclass @dataclass class Job: source: str event_id: str body: bytes attempts: int # which attempt this is, starting at 1 class Inbox: \"\"\"Every verified delivery lands here first. The primary key makes duplicates impossible.\"\"\" def __init__(self, path, lease=60, max_attempts=6, clock=time.time, rng=random.random): self.path, self.lease, self.max_attempts, self.clock, self.rng = path, lease, max_attempts, clock, rng with closing(self._connect()) as db: db.execute( \"\"\"CREATE TABLE IF NOT EXISTS inbox ( source TEXT NOT NULL, event_id TEXT NOT NULL, body BLOB NOT NULL, status TEXT NOT NULL, -- pending, done or dead attempts INTEGER NOT NULL, next_attempt_at REAL NOT NULL, last_error TEXT, received_at REAL NOT NULL, PRIMARY KEY (source, event_id) )\"\"\" ) def _connect(self): return sqlite3.connect(self.path, timeout=10, isolation_level=None) def store(self, source: str, event_id: str, body: bytes) -> bool: \"\"\"True for a new event, False for a duplicate.\"\"\" now = self.clock() with closing(self._connect()) as db: return db.execute( \"INSERT OR IGNORE INTO inbox VALUES (?, ?, ?, 'pending', 0, ?, NULL, ?)\", (source, event_id, body, now, now), ).rowcount == 1 def next_job(self): \"\"\"Lease the next due job. If the worker dies, the lease runs out and it comes back.\"\"\" now = self.clock() with closing(self._connect()) as db: db.execute(\"BEGIN IMMEDIATE\") # one worker at a time can take a job row = db.execute( \"SELECT source, event_id, body, attempts FROM inbox\" \" WHERE status = 'pending' AND next_attempt_at <= ? ORDER BY next_attempt_at LIMIT 1\", (now,), ).fetchone() if row: db.execute( \"UPDATE inbox SET attempts = attempts + 1, next_attempt_at = ? WHERE source = ? AND event_id = ?\", (now + self.lease, row[0], row[1]), ) db.execute(\"COMMIT\") return Job(row[0], row[1], row[2], row[3] + 1) if row else None def _update(self, job, status, next_attempt_at, error=None): with closing(self._connect()) as db: db.execute( \"UPDATE inbox SET status = ?, next_attempt_at = ?, last_error = ? WHERE source = ? AND event_id = ?\", (status, next_attempt_at, error, job.source, job.event_id), ) def done(self, job): self._update(job, \"done\", 0) def dead(self, job, error): self._update(job, \"dead\", 0, error) def fail(self, job, error): if job.attempts >= self.max_attempts: return self.dead(job, error) # give up, and keep it for a human to look at self._update(job, \"pending\", self.clock() + self.backoff(job.attempts), error) def backoff(self, attempt: int) -> float: delay = min(3600, 30 * 2 ** (attempt - 1)) # 30s, 1m, 2m, 4m ... capped at an hour return delay / 2 + delay / 2 * self.rng() # jitter, so retries don't all line up def counts(self) -> dict: with closing(self._connect()) as db: return dict(db.execute(\"SELECT status, COUNT(*) FROM inbox GROUP BY status\"))It does two jobs at once. Deduplication. (source, event_id) is the primary key, and INSERT OR IGNORE reports whether a row was actually inserted. The database decides who was first, so eight simultaneous copies of one delivery produce one stored event. test_eight_simultaneous_copies_are_stored_once fires them from eight threads at once. A work queue. next_job leases the next due event inside a transaction, so two workers never take the same one. If a worker dies mid-job, the lease runs out and the event comes back. The worker import json class Worker: \"\"\"Takes events out of the inbox and runs your handlers, retrying failures.\"\"\" def __init__(self, inbox, handlers): self.inbox, self.handlers = inbox, handlers def run_once(self) -> bool: job = self.inbox.next_job() if job is None: return False try: event = json.loads(job.body) kind = event[\"type\"] except (ValueError, KeyError, TypeError) as error: self.inbox.dead(job, f\"unreadable event: {error!r}\") # retrying can't fix a bad payload return True try: handler = self.handlers.get(kind) if handler is not None: # events you don't handle are acknowledged and ignored handler(event) except Exception as error: self.inbox.fail(job, repr(error)) # retried later, with a growing delay else: self.inbox.done(job) return TrueA failing handler isn't a reason to lose the event. It's retried with a growing delay, 30 seconds, then a minute, then two, capped at an hour, with some jitter so retries don't line up. After max_attempts it's set aside as a dead letter, and it stays in the table for a human to look at. An unreadable payload goes straight to the dead letter pile, because retrying can't make bad JSON valid. One setting to get right is the lease, which defaults to 60 seconds. Make it longer than your slowest handler. If it's shorter, a second worker can pick up the event while the first is still busy, and your handler runs twice at the same time. Make the handler idempotent Everything so far delivers each event at least once, never exactly once. The receiver dedupes, but a worker can still crash after your handler has done its work and before it marks the event done. When the lease expires, the handler runs again. So the handler has to be safe to run twice. The simplest way is to record the event ID with the effect, and let the database refuse the second one: import sqlite3 from contextlib import closing class Ledger: \"\"\"An idempotent handler: an account is credited once per event, however often it's delivered.\"\"\" def __init__(self, path): self.path = path with closing(self._connect()) as db: db.execute(\"CREATE TABLE IF NOT EXISTS credits (event_id TEXT PRIMARY KEY, account TEXT NOT NULL, cents INTEGER NOT NULL)\") def _connect(self): return sqlite3.connect(self.path, timeout=10, isolation_level=None) def credit(self, event_id: str, account: str, cents: int) -> bool: with closing(self._connect()) as db: return db.execute(\"INSERT OR IGNORE INTO credits VALUES (?, ?, ?)\", (event_id, account, cents)).rowcount == 1 def balance(self, account: str) -> int: with closing(self._connect()) as db: return db.execute(\"SELECT COALESCE(SUM(cents), 0) FROM credits WHERE account = ?\", (account,)).fetchone()[0]test_a_crash_after_the_side_effect_is_recovered_and_credits_once plays out the awkward case. A worker leases an event and credits the account, then dies. The lease expires, the event runs again, and the balance is still 700 cents, not 1,400. It's the same idea as idempotency keys on an API: let a unique constraint decide whether the work has already happened. Don't rely on order Stripe's docs say plainly that events can arrive out of order, and not to use the created timestamp to order them. Write handlers that don't care. For anything important, treat the event as a signal to go and look. Fetch the object's current state from the provider's API, and act on that. Wire it up The FastAPI route is thin: import os from fastapi import FastAPI, Request from fastapi.concurrency import run_in_threadpool from fastapi.responses import JSONResponse from inbox import Inbox from receiver import WebhookReceiver app = FastAPI() receiver = WebhookReceiver(Inbox(\"webhooks.db\"), secrets=[os.environ[\"WEBHOOK_SECRET\"]]) @app.post(\"/webhooks/payments\") async def payments_webhook(request: Request): body = await request.body() # the RAW bytes: the signature covers exactly these status, payload = await run_in_threadpool(receiver.handle, request.headers, body) return JSONResponse(payload, status_code=status)And the worker is a loop: import time from inbox import Inbox from ledger import Ledger from worker import Worker ledger = Ledger(\"webhooks.db\") def payment_succeeded(event): ledger.credit(event[\"id\"], event[\"account\"], event[\"cents\"]) worker = Worker(Inbox(\"webhooks.db\"), {\"payment.succeeded\": payment_succeeded}) while True: if not worker.run_once(): time.sleep(1) # nothing due: wait a momentRun them as two separate processes, with the secret in an environment variable: WEBHOOK_SECRET=whsec_... uvicorn main:app --port 8000 python run_worker.pyAlso put a body size limit on your reverse proxy, such as client_max_body_size in nginx. The receiver's max_body check runs after the framework has read the whole body. See it work This script sends the receiver seven different requests, with a clock it can move forward: import json import os import tempfile from inbox import Inbox from receiver import WebhookReceiver from signing import standard_signature SECRET = \"whsec_\" + \"dGVzdC1zZWNyZXQtMDEyMzQ1Njc4OWFiY2RlZg==\" now = [1_800_000_000] # a clock we can move reasons = [] inbox = Inbox(os.path.join(tempfile.mkdtemp(), \"demo.db\"), clock=lambda: now[0]) receiver = WebhookReceiver(inbox, [SECRET], clock=lambda: now[0], on_reject=reasons.append) body = json.dumps({\"id\": \"evt_1\", \"type\": \"payment.succeeded\", \"account\": \"acct_1\", \"cents\": 500}).encode() def signed(message_id=\"msg_1\", payload=body, secret=SECRET): timestamp = str(now[0]) return {\"webhook-id\": message_id, \"webhook-timestamp\": timestamp, \"webhook-signature\": standard_signature(secret, message_id, timestamp, payload)} def show(label, headers, payload=body): status, answer = receiver.handle(headers, payload) detail = answer.get(\"status\") or f\"{answer['error']} ({reasons[-1]})\" print(f\"{label:<34} -> {status} {detail}\") captured = signed() show(\"a valid delivery\", captured) show(\"the same request again\", captured) now[0] += 30 show(\"the sender's retry (new timestamp)\", signed()) show(\"a tampered body\", signed(), body.replace(b\"500\", b\"900\")) show(\"signed with the wrong secret\", signed(secret=\"whsec_\" + \"d3Jvbmctc2VjcmV0LXdyb25nLXNlY3JldC0wMDAwMA==\")) show(\"no signature headers at all\", {}) now[0] += 600 show(\"the captured request, 10 minutes on\", captured) print(\"stored events:\", inbox.counts())a valid delivery -> 200 accepted the same request again -> 200 duplicate the sender's retry (new timestamp) -> 200 duplicate a tampered body -> 400 invalid webhook (bad-signature) signed with the wrong secret -> 400 invalid webhook (bad-signature) no signature headers at all -> 400 invalid webhook (missing-headers) the captured request, 10 minutes on -> 400 invalid webhook (stale-timestamp) stored events: {'pending': 1}The first delivery is accepted. The same request again, and the sender's retry with a new timestamp, are acknowledged as duplicates. A tampered body, a wrong secret and missing headers are all rejected, each for its own reason. And the captured request replayed ten minutes later is stale. At the end, exactly one event is stored. Defense in depth Signatures are the real control. These add layers: HTTPS only. Both GitHub and Stripe expect it, and you should verify certificates when you're the sender. IP allowlists. GitHub and Stripe publish the addresses their webhooks come from. An allowlist narrows who can reach you, but the addresses change, so it never replaces the signature. A separate endpoint and secret per integration. A leak then stays contained. No secrets in logs. Not the secret, and not full signature headers. Skip CSRF protection on the route. Webhooks aren't browser requests, and the signature takes CSRF's place. If you're the one sending webhooks The same lessons apply from the other side: Sign the ID and the timestamp, not just the body. Use one secret per endpoint, and support rotation by signing with both during the changeover. Retry with exponential backoff and jitter. Keep the ID the same and give each attempt a fresh timestamp and signature. Use short timeouts. Treat anything that isn't a 2xx, redirects included, as a failure. Validate customer-supplied URLs, so your system can't be tricked into calling internal addresses. Checklist Verify the signature on the raw body. Compare with hmac.compare_digest, on bytes. Reject old and future timestamps, and accept a list of secrets. Deduplicate on the event ID, and keep IDs as long as the sender can retry. Answer fast: verify, store, return 2xx. Process from a queue, with backoff and a dead letter pile. Make handlers idempotent, and don't depend on event order. Log the reason for a rejection, and tell the sender nothing. The tests Here are all 23. The vector tests are the first ones to read: import base64 import hashlib import hmac import json import os import sys import tempfile import threading import time from unittest import mock from inbox import Inbox from ledger import Ledger from receiver import WebhookReceiver from signing import InvalidWebhook, standard_signature, verify_github, verify_standard, verify_stripe from worker import Worker NOW = 1_800_000_000 SECRET = \"whsec_\" + base64.b64encode(b\"a-secret-for-tests-0123456789abc\").decode() OLD_SECRET = \"whsec_\" + base64.b64encode(b\"the-previous-secret-0123456789ab\").decode() # The published Standard Webhooks test vector, and GitHub's from its docs. SPEC = dict(secret=\"whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw\", id=\"msg_p5jXN8AQM9LWM0D4loKWxJek\", timestamp=\"1614265330\", body=b'{\"test\": 2432232314}', signature=\"v1,g0hM9SsE+OTPJTGt/tmIKtSyZlE3uFJELVlNIOLJ1OE=\") GITHUB = dict(secret=\"It's a Secret to Everybody\", body=b\"Hello, World!\", header=\"sha256=757107ea0eb2509fc211221cce984b8a37570b6d7586c22c46f4379c8b043e17\") class FakeClock: def __init__(self, now=NOW): self.now = now def __call__(self): return self.now def advance(self, seconds): self.now += seconds def reason(call, *args): \"\"\"The reason a verification was refused, or None if it passed.\"\"\" try: call(*args) except InvalidWebhook as error: return error.reason return None def standard_headers(message_id=\"msg_1\", body=b\"{}\", timestamp=NOW, secret=SECRET): return {\"webhook-id\": message_id, \"webhook-timestamp\": str(timestamp), \"webhook-signature\": standard_signature(secret, message_id, str(timestamp), body)} # ---- signatures, checked against the published vectors ------------------------------------------- def test_github_matches_its_published_vector(): headers = {\"X-Hub-Signature-256\": GITHUB[\"header\"]} assert reason(verify_github, GITHUB[\"secret\"], headers, GITHUB[\"body\"]) is None assert reason(verify_github, GITHUB[\"secret\"], headers, b\"Hello, World?\") == \"bad-signature\" assert reason(verify_github, \"wrong secret\", headers, GITHUB[\"body\"]) == \"bad-signature\" assert reason(verify_github, GITHUB[\"secret\"], {}, GITHUB[\"body\"]) == \"missing-headers\" sha1 = {\"X-Hub-Signature-256\": \"sha1=\" + GITHUB[\"header\"][7:]} assert reason(verify_github, GITHUB[\"secret\"], sha1, GITHUB[\"body\"]) == \"bad-signature\" def test_standard_webhooks_matches_its_published_vector(): assert standard_signature(SPEC[\"secret\"], SPEC[\"id\"], SPEC[\"timestamp\"], SPEC[\"body\"]) == SPEC[\"signature\"] headers = {\"webhook-id\": SPEC[\"id\"], \"webhook-timestamp\": SPEC[\"timestamp\"], \"webhook-signature\": SPEC[\"signature\"]} verified = verify_standard([SPEC[\"secret\"]], headers, SPEC[\"body\"], now=1_614_265_330) assert (verified.message_id, verified.timestamp) == (SPEC[\"id\"], 1_614_265_330) def test_the_timestamp_window_is_five_minutes_either_way(): h = {\"webhook-id\": SPEC[\"id\"], \"webhook-timestamp\": SPEC[\"timestamp\"], \"webhook-signature\": SPEC[\"signature\"]} ts = int(SPEC[\"timestamp\"]) check = lambda now: reason(verify_standard, [SPEC[\"secret\"]], h, SPEC[\"body\"], now) assert check(ts + 300) is None and check(ts - 300) is None assert check(ts + 301) == \"stale-timestamp\" and check(ts - 301) == \"stale-timestamp\" assert reason(verify_standard, [SPEC[\"secret\"]], {**h, \"webhook-timestamp\": \"yesterday\"}, SPEC[\"body\"], ts) == \"bad-timestamp\" def test_everything_that_was_signed_must_match(): body = b'{\"amount\": 500}' good = standard_headers(\"msg_1\", body) verify = lambda headers, payload=body, secrets=(SECRET,): reason(verify_standard, list(secrets), headers, payload, NOW) assert verify(good) is None assert verify(good, b'{\"amount\": 900}') == \"bad-signature\" # a changed body assert verify({**good, \"webhook-id\": \"msg_2\"}) == \"bad-signature\" # a changed id assert verify({**good, \"webhook-timestamp\": str(NOW + 1)}) == \"bad-signature\" # a changed timestamp assert verify(good, secrets=(OLD_SECRET,)) == \"bad-signature\" # the wrong secret def test_secret_rotation_accepts_old_and_new_and_ignores_unknown_versions(): body = b\"{}\" signed_with_old = standard_headers(\"msg_1\", body, secret=OLD_SECRET) assert reason(verify_standard, [SECRET, OLD_SECRET], signed_with_old, body, NOW) is None assert reason(verify_standard, [SECRET], signed_with_old, body, NOW) == \"bad-signature\" # once the old one is retired good = standard_headers(\"msg_1\", body)[\"webhook-signature\"] two = {**standard_headers(\"msg_1\", body), \"webhook-signature\": \"v1,AAAA \" + good} assert reason(verify_standard, [SECRET], two, body, NOW) is None # one valid signature is enough other_version = {**standard_headers(\"msg_1\", body), \"webhook-signature\": good.replace(\"v1,\", \"v2,\")} assert reason(verify_standard, [SECRET], other_version, body, NOW) == \"bad-signature\" def test_a_non_ascii_signature_is_rejected_not_a_crash(): headers = {**standard_headers(\"msg_1\"), \"webhook-signature\": \"v1,caf\\u00e9\\u2713\"} assert reason(verify_standard, [SECRET], headers, b\"{}\", NOW) == \"bad-signature\" assert reason(verify_github, \"s\", {\"x-hub-signature-256\": \"sha256=\\u2713\"}, b\"\") == \"bad-signature\" def test_header_names_are_case_insensitive(): headers = {name.title(): value for name, value in standard_headers().items()} assert reason(verify_standard, [SECRET], headers, b\"{}\", NOW) is None def test_signatures_are_compared_in_constant_time(): with mock.patch.object(hmac, \"compare_digest\", wraps=hmac.compare_digest) as compare: verify_standard([SECRET], standard_headers(), b\"{}\", NOW) verify_github(GITHUB[\"secret\"], {\"x-hub-signature-256\": GITHUB[\"header\"]}, GITHUB[\"body\"]) assert compare.call_count == 2 def test_re_serialised_json_breaks_the_signature(): raw = b'{\"id\": \"evt_1\", \"type\": \"payment.succeeded\"}' # two spaces: perfectly valid JSON headers = standard_headers(\"msg_1\", raw) assert reason(verify_standard, [SECRET], headers, raw, NOW) is None parsed_and_dumped = json.dumps(json.loads(raw)).encode() assert parsed_and_dumped != raw assert reason(verify_standard, [SECRET], headers, parsed_and_dumped, NOW) == \"bad-signature\" def stripe_header(secret, body, timestamp=NOW, extra=\"\"): digest = hmac.new(secret.encode(), f\"{timestamp}.\".encode() + body, hashlib.sha256).hexdigest() return f\"t={timestamp},v1={digest}{extra}\" def test_stripe_style_signatures(): secret, body = \"whsec_stripe_test\", b'{\"id\": \"evt_1\"}' header = stripe_header(secret, body) assert reason(verify_stripe, secret, header, body, NOW) is None assert reason(verify_stripe, secret, header, b'{\"id\": \"evt_2\"}', NOW) == \"bad-signature\" assert reason(verify_stripe, secret, header, body, NOW + 301) == \"stale-timestamp\" assert reason(verify_stripe, secret, f\"t={NOW},v0=abc\", body, NOW) == \"missing-headers\" # v0 is never trusted assert reason(verify_stripe, secret, \"v1=abc\", body, NOW) == \"missing-headers\" assert reason(verify_stripe, secret, stripe_header(secret, body, extra=\",v1=\" + \"0\" * 64), body, NOW) is None # two v1s # ---- the receiver: verify, store, answer ------------------------------------------------------------------ def make(**kwargs): clock = FakeClock() rejected = [] inbox = Inbox(os.path.join(tempfile.mkdtemp(), \"w.db\"), clock=clock, rng=lambda: 1.0, **kwargs) receiver = WebhookReceiver(inbox, [SECRET], clock=clock, on_reject=rejected.append) return receiver, inbox, clock, rejected def event(event_id=\"evt_1\", cents=500): return json.dumps({\"id\": event_id, \"type\": \"payment.succeeded\", \"account\": \"acct_1\", \"cents\": cents}).encode() def test_a_valid_delivery_is_stored_and_acknowledged(): receiver, inbox, _, _ = make() body = event() assert receiver.handle(standard_headers(\"msg_1\", body), body) == (200, {\"status\": \"accepted\"}) assert inbox.counts() == {\"pending\": 1} # stored for later: nothing was processed yet def test_a_retry_is_acknowledged_but_stored_once(): receiver, inbox, clock, _ = make() body = event() receiver.handle(standard_headers(\"msg_1\", body, clock.now), body) clock.advance(30) # the sender retries: same id, fresh timestamp, fresh signature status, payload = receiver.handle(standard_headers(\"msg_1\", body, clock.now), body) assert (status, payload) == (200, {\"status\": \"duplicate\"}) assert inbox.counts() == {\"pending\": 1} def test_eight_simultaneous_copies_are_stored_once(): receiver, inbox, _, _ = make() body, results, barrier = event(), [], threading.Barrier(8) def deliver(): barrier.wait() results.append(receiver.handle(standard_headers(\"msg_1\", body), body)[1][\"status\"]) threads = [threading.Thread(target=deliver) for _ in range(8)] for t in threads: t.start() for t in threads: t.join() assert sorted(results) == [\"accepted\"] + [\"duplicate\"] * 7 and inbox.counts() == {\"pending\": 1} def test_rejections_say_nothing_to_the_sender_and_store_nothing(): receiver, inbox, clock, rejected = make() body = event() forged = standard_headers(\"msg_1\", body, secret=OLD_SECRET) stale = standard_headers(\"msg_2\", body, clock.now - 1000) responses = [receiver.handle(forged, body), receiver.handle(stale, body), receiver.handle({}, body)] assert responses == [(400, {\"error\": \"invalid webhook\"})] * 3 # the same answer every time assert rejected == [\"bad-signature\", \"stale-timestamp\", \"missing-headers\"] # the reasons are for you assert inbox.counts() == {} def test_oversized_bodies_are_refused_before_any_hmac_is_computed(): receiver, inbox, _, _ = make() huge = b\"x\" * 2_000_000 headers = standard_headers(\"msg_1\", huge) # signed correctly, so only the size can be the problem with mock.patch.object(hmac, \"new\", side_effect=AssertionError(\"computed an HMAC\")): assert receiver.handle(headers, huge)[0] == 413 assert inbox.counts() == {} def test_a_replay_is_harmless_inside_the_window_and_rejected_after_it(): receiver, inbox, clock, rejected = make() body = event() captured = standard_headers(\"msg_1\", body, clock.now) assert receiver.handle(captured, body)[1] == {\"status\": \"accepted\"} clock.advance(100) assert receiver.handle(captured, body) == (200, {\"status\": \"duplicate\"}) # the same bytes again: nothing happens clock.advance(300) assert receiver.handle(captured, body)[0] == 400 and rejected == [\"stale-timestamp\"] assert inbox.counts() == {\"pending\": 1} # ---- the worker: retries, dead letters, crashes --------------------------------------------------------------------- def deliver(receiver, clock, event_id, cents=500, message_id=None): body = event(event_id, cents) message_id = message_id or \"msg_\" + event_id return receiver.handle(standard_headers(message_id, body, clock.now), body) def test_the_worker_runs_the_handler_and_marks_the_event_done(): receiver, inbox, clock, _ = make() deliver(receiver, clock, \"evt_1\") seen = [] worker = Worker(inbox, {\"payment.succeeded\": lambda e: seen.append(e[\"id\"])}) assert worker.run_once() is True and worker.run_once() is False assert seen == [\"evt_1\"] and inbox.counts() == {\"done\": 1} def test_unknown_events_are_ignored_and_unreadable_ones_are_set_aside(): receiver, inbox, clock, _ = make() unknown = json.dumps({\"id\": \"evt_1\", \"type\": \"something.else\"}).encode() receiver.handle(standard_headers(\"msg_1\", unknown, clock.now), unknown) garbage = b\"not json at all\" receiver.handle(standard_headers(\"msg_2\", garbage, clock.now), garbage) called = [] worker = Worker(inbox, {\"payment.succeeded\": called.append}) while worker.run_once(): pass assert called == [] and inbox.counts() == {\"done\": 1, \"dead\": 1} # a bad payload won't improve with retries def test_failures_are_retried_with_growing_delays(): receiver, inbox, clock, _ = make() deliver(receiver, clock, \"evt_1\") attempts = [] def flaky(e): attempts.append(clock.now) if len(attempts) < 3: raise RuntimeError(\"database is down\") worker = Worker(inbox, {\"payment.succeeded\": flaky}) assert worker.run_once() is True # attempt 1 fails: wait 30 seconds assert worker.run_once() is False clock.advance(30) assert worker.run_once() is True # attempt 2 fails: wait 60 seconds clock.advance(59) assert worker.run_once() is False clock.advance(1) assert worker.run_once() is True # attempt 3 works assert len(attempts) == 3 and inbox.counts() == {\"done\": 1} def test_an_event_that_keeps_failing_is_set_aside_after_the_last_attempt(): receiver, inbox, clock, _ = make(max_attempts=3) deliver(receiver, clock, \"evt_1\") def broken(e): raise RuntimeError(\"always\") worker = Worker(inbox, {\"payment.succeeded\": broken}) for _ in range(3): assert worker.run_once() is True clock.advance(3600) assert worker.run_once() is False and inbox.counts() == {\"dead\": 1} def test_a_crash_after_the_side_effect_is_recovered_and_credits_once(): receiver, inbox, clock, _ = make() ledger = Ledger(os.path.join(tempfile.mkdtemp(), \"ledger.db\")) deliver(receiver, clock, \"evt_1\", cents=700) job = inbox.next_job() # a worker leases the event... ledger.credit(\"evt_1\", \"acct_1\", 700) # ...runs the handler... del job # ...and dies before it can mark the event done worker = Worker(inbox, {\"payment.succeeded\": lambda e: ledger.credit(e[\"id\"], e[\"account\"], e[\"cents\"])}) assert worker.run_once() is False # still leased: nobody else touches it yet clock.advance(61) # the lease runs out assert worker.run_once() is True # the event runs again assert ledger.balance(\"acct_1\") == 700 # but the account was credited once assert inbox.counts() == {\"done\": 1} def test_two_workers_never_run_the_same_event(): receiver, inbox, clock, _ = make() for i in range(60): deliver(receiver, clock, f\"evt_{i}\") handled, lock, barrier = [], threading.Lock(), threading.Barrier(8) def handler(e): with lock: handled.append(e[\"id\"]) def work(): worker = Worker(inbox, {\"payment.succeeded\": handler}) barrier.wait() # all eight start together while worker.run_once(): pass switch_interval = sys.getswitchinterval() sys.setswitchinterval(1e-6) # make threads trade places as often as possible try: threads = [threading.Thread(target=work) for _ in range(8)] for t in threads: t.start() for t in threads: t.join() finally: sys.setswitchinterval(switch_interval) assert sorted(handled) == sorted(f\"evt_{i}\" for i in range(60)) and inbox.counts() == {\"done\": 60} def test_end_to_end_retries_and_duplicates_credit_each_payment_once(): receiver, inbox, clock, _ = make() ledger = Ledger(os.path.join(tempfile.mkdtemp(), \"ledger.db\")) deliver(receiver, clock, \"evt_1\", cents=500) deliver(receiver, clock, \"evt_2\", cents=250) clock.advance(30) deliver(receiver, clock, \"evt_1\", cents=500) # the sender retried evt_1 worker = Worker(inbox, {\"payment.succeeded\": lambda e: ledger.credit(e[\"id\"], e[\"account\"], e[\"cents\"])}) while worker.run_once(): pass assert ledger.balance(\"acct_1\") == 750 and inbox.counts() == {\"done\": 2}Run them with pytest -q. They use SQLite and a fake clock, so they need no network and no web framework. Checked against the GitHub and Stripe documentation, the Standard Webhooks spec and the Python docs on November 7, 2026. Providers change their signing details occasionally, so check the current docs for the sender you integrate with. Sources GitHub: Validating webhook deliveries GitHub: Best practices for using webhooks Stripe: Receive Stripe events in your webhook endpoint Stripe: Check the webhook signatures Standard Webhooks Python docs: hmac","contentHash":"sha256:5372bc0e979a46e288adb6f618b920228f4ecd0b917e5c20872e4c512f03c11b","authorName":"Ankit Singh","authorUrl":"https://zyvop.com/author/ankit","authorSameAs":[],"category":"DevOps","tags":["webhooks","Python","security","API Design","FastAPI"],"audience":"Software engineers and developers building applications with DevOps","tone":"Instructional, practical, code-first","readingTimeMinutes":21,"wordCount":5535,"faqs":null,"primaryTopic":"DevOps","publishedAt":"2026-11-11T05:43:31.753Z","updatedAt":"2026-11-11T10:46:36.189Z","canonicalUrl":"https://zyvop.com/webhook-security-verify-signatures-stop-replays-handle-retries-x05wj"}