Webhook Security: Verify Signatures, Stop Replays, Handle Retries

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:

1. Is this really from the sender? That's forgery, and signatures answer it.

2. Is someone replaying a real request they captured? That's replay, and timestamps plus event IDs answer it.

3. 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

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:

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 seconds

All 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:

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 success

The 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 duplicate

What to return in each case:

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 True

A 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 moment

Run them as two separate processes, with the secret in an environment variable:

WEBHOOK_SECRET=whsec_... uvicorn main:app --port 8000
python run_worker.py

Also 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