Learning Hub

How to Verify Webhook Signatures in Python and Stop Forged, Tampered, and Replayed Events

Build a FastAPI webhook receiver that rejects forged, edited, replayed and duplicate events, using HMAC-SHA256, a signed timestamp, atomic duplicate detection and secret rotation, with every defence...

October 7, 2026 35 Min Read
21

By the end of this tutorial you will have a small webhook receiver, written in Python with FastAPI, that accepts genuine events from a provider and turns away everything else: forged events, events edited in transit, old events replayed by someone who captured them, duplicate deliveries, and requests that only work because a secret was never configured. You will build it one defence at a time and attack it after every step, so you can see each rule earn its place.

Table Of Content

A webhook is an HTTP request that another service sends to your server when something happens on its side, for example “payment succeeded” or “code pushed”. Your endpoint usually has to be reachable from the internet, which means anyone can send it a request that claims to come from the provider. The only thing that separates a real event from a forged one is a signature: a short value the provider computes from the message and a secret that only the provider and you know. If your code skips the check, or performs it slightly wrong, an attacker can make your system believe a payment arrived.

Four separate questions hide inside “is this event real?”. Did the sender know the secret (authenticity)? Are these the bytes that were signed (integrity)? Is the message recent (freshness)? Have you already handled it (uniqueness)? A signature alone answers only the first two. Steps 5 and 6 add the other two, and Steps 3, 4 and 7 cover the places where correct-looking code quietly fails.

Prerequisites

You need Python 3.10 or newer (everything below ran on Python 3.13.14 on Windows 11), a terminal, and basic comfort reading Python and HTTP requests. No cloud account, provider account or public URL is required: a few lines of Python play the part of the provider, and a server on 127.0.0.1 plays your endpoint. Make a folder called webhook-lab, open a terminal in it, and create a virtual environment so nothing touches your system Python.

# Windows (PowerShell)
python -m venv .venv
.venv\Scripts\Activate.ps1

# macOS or Linux
python3 -m venv .venv
source .venv/bin/activate

python -m pip install fastapi uvicorn requests pytest standardwebhooks

The versions used here were FastAPI 0.142.2, Starlette 1.7.0, Pydantic 2.13.5, uvicorn 0.54.0, requests 2.34.2, pytest 9.1.1 and standardwebhooks 1.1.0 (only Step 10 needs the last one). Every file in this tutorial goes into the same folder, and every script is run with python name.py from that folder.

One helper for the demos

Each step attacks a real HTTP server, so we need a way to start one from inside a script. The helper below runs uvicorn in a background thread of the same process, waits until it is listening, and shuts it down when the with block ends. Because the server shares memory with the demo script, the script can also look inside the application afterwards. Save it as labserver.py.

# labserver.py
import threading
import time

import uvicorn


class LiveServer:
    """Run a FastAPI app on 127.0.0.1 in a background thread for the demos."""

    def __init__(self, app, port):
        config = uvicorn.Config(app, host="127.0.0.1", port=port, log_level="critical")
        self.server = uvicorn.Server(config)
        self.thread = threading.Thread(target=self.server.run, daemon=True)
        self.url = f"http://127.0.0.1:{port}"

    def __enter__(self):
        self.thread.start()
        while not self.server.started:
            if not self.thread.is_alive():
                raise RuntimeError("the server did not start (is the port already in use?)")
            time.sleep(0.02)
        return self

    def __exit__(self, *exc):
        self.server.should_exit = True
        self.thread.join(timeout=5)

Step 1: See the attack before you defend

The fastest way to understand why signatures matter is to watch an unsigned endpoint lose. The app below pretends to be a shop. Orders start as awaiting_payment, and a webhook from the payment provider flips them to paid. The handler believes whatever JSON it receives. Save it as step01_naive.py.

# step01_naive.py
import requests
from fastapi import FastAPI, Request

from labserver import LiveServer

ORDERS = {"A-1001": "awaiting_payment", "A-1002": "awaiting_payment"}

app = FastAPI()


@app.post("/webhooks/payments")
async def payments(request: Request):
    event = await request.json()
    if event["type"] == "payment.succeeded":
        ORDERS[event["order_id"]] = "paid"
    return {"ok": True}


if __name__ == "__main__":
    with LiveServer(app, 8101) as srv:
        print("before:", ORDERS)
        forged = {"id": "evt_forged_1", "type": "payment.succeeded", "order_id": "A-1001", "amount_cents": 49900}
        reply = requests.post(srv.url + "/webhooks/payments", json=forged, timeout=5)
        print("attacker got:", reply.status_code, reply.json())
        print("after: ", ORDERS)

Run python step01_naive.py. The “attacker” is the requests.post call at the bottom: it knows the URL and the field names, and nothing else.

before: {'A-1001': 'awaiting_payment', 'A-1002': 'awaiting_payment'}
attacker got: 200 {'ok': True}
after:  {'A-1001': 'paid', 'A-1002': 'awaiting_payment'}

Order A-1001 is now marked paid, and nobody paid anything. The attacker needed no secret and no stolen credentials, only an address and a guess at the field names (providers publish their event formats). HTTPS does not change this: it encrypts the request on its way to you, but it says nothing about who wrote the request.

Check that it worked

Your output should show A-1001 changing from awaiting_payment to paid while A-1002 stays untouched. If the script stops with “the server did not start (is the port already in use?)”, another program is using port 8101; change the number in the script.

Step 2: Sign and verify the body with HMAC-SHA256

What an HMAC is

An HMAC (hash-based message authentication code) mixes a secret key into a hash function. Anyone can compute the SHA-256 hash of a message, but only someone who knows the key can compute its HMAC-SHA256. So the provider computes the HMAC of the request body with a secret you both hold and sends the result in a header. You recompute it over the bytes you received. If the two values match, the sender knew the secret and the bytes are the ones that were signed. If you have read our TOTP tutorial, this is the same primitive that powers one-time codes, used here to vouch for a whole message.

GitHub’s webhook design is the simplest widely used example. GitHub sends the hex digest in an X-Hub-Signature-256 header with a sha256= prefix, and its documentation says that “you should validate the webhook signature before processing the delivery further” (GitHub Docs, Validating webhook deliveries). It also says to choose “a random string of text with high entropy” as the secret, which matters more than it sounds, as Step 7 shows. Save the functions below as step02_hmac.py.

# step02_hmac.py
import hashlib
import hmac


def sign_body(secret: bytes, body: bytes) -> str:
    digest = hmac.new(secret, body, hashlib.sha256).hexdigest()
    return "sha256=" + digest


def verify_body(secret: bytes, body: bytes, header: str) -> bool:
    return sign_body(secret, body) == header  # plain == for now; Step 4 replaces it


if __name__ == "__main__":
    secret = b"It's a Secret to Everybody"
    body = b"Hello, World!"
    published = "sha256=757107ea0eb2509fc211221cce984b8a37570b6d7586c22c46f4379c8b043e17"

    header = sign_body(secret, body)
    print("computed:  ", header)
    print("GitHub doc:", published)
    print("match:     ", header == published)
    print()
    print("untouched body verifies:  ", verify_body(secret, body, header))
    print("one byte changed verifies:", verify_body(secret, b"Hello, World?", header))
    print("wrong secret verifies:    ", verify_body(b"a guess", body, header))

Notice the plain == in verify_body. It is deliberately wrong and Step 4 replaces it, but it lets us concentrate on the HMAC first. The __main__ block uses a known-answer test: GitHub’s documentation publishes a test secret, a test payload, and the signature a correct implementation must produce. Run python step02_hmac.py.

computed:   sha256=757107ea0eb2509fc211221cce984b8a37570b6d7586c22c46f4379c8b043e17
GitHub doc: sha256=757107ea0eb2509fc211221cce984b8a37570b6d7586c22c46f4379c8b043e17
match:      True

untouched body verifies:   True
one byte changed verifies: False
wrong secret verifies:     False

Our computed header is identical to the value in GitHub’s documentation, so the primitive is wired correctly. Never trust a verifier you have not run against a published answer. The last three lines show the two properties we wanted: the untouched body verifies, and changing one byte or using the wrong secret does not.

Common mistake

Digest encodings differ between providers. GitHub uses lowercase hex with a sha256= prefix; the scheme we build in Step 5 uses base64 with a v1, prefix. A verifier that computes the right HMAC and compares it in the wrong encoding fails on every genuine delivery, which is the usual moment people switch verification off “temporarily”. Always start from the provider’s test vector.

Step 3: Verify the exact bytes, not the parsed JSON

A signature is computed over bytes. Web frameworks make it tempting to work with parsed JSON instead, and that is where genuine deliveries start failing. Stripe’s documentation states the rule bluntly: “Stripe requires the raw body of the request to perform signature verification. If you’re using a framework, make sure it doesn’t manipulate the raw body” (Stripe Docs, Webhooks). The Standard Webhooks specification explains why this is such a common failure: “many webhook consumers often accidentally parse the body as json, and then serialize it again”, and “even a stray space can cause the signature to be invalid” (Standard Webhooks specification). GitHub adds that you should handle the payload as UTF-8, because “webhook payloads can contain unicode characters”.

The script below hosts two endpoints. /broken parses the JSON and serializes it again before checking the signature. /raw checks the bytes exactly as they arrived. Two different senders post the same event, which includes the name “Zoë”, to both. Save it as step03_rawbody.py.

# step03_rawbody.py
import json

import requests
from fastapi import FastAPI, Request

from labserver import LiveServer
from step02_hmac import sign_body, verify_body

SECRET = b"lab-secret-for-step-three-only"
app = FastAPI()


@app.post("/broken")
async def broken(request: Request):
    payload = await request.json()  # parse the JSON...
    rebuilt = json.dumps(payload).encode()  # ...then serialize it again
    ok = verify_body(SECRET, rebuilt, request.headers.get("x-signature", ""))
    return {"verified": ok}


@app.post("/raw")
async def raw(request: Request):
    body = await request.body()  # the exact bytes the sender signed
    ok = verify_body(SECRET, body, request.headers.get("x-signature", ""))
    return {"verified": ok}


def deliver(url: str, body: bytes, signed_body: bytes | None = None) -> bool:
    signature = sign_body(SECRET, body if signed_body is None else signed_body)
    headers = {"Content-Type": "application/json", "X-Signature": signature}
    return requests.post(url, data=body, headers=headers, timeout=5).json()["verified"]


if __name__ == "__main__":
    event = {"type": "payment.succeeded", "customer": "Zoë", "amount_cents": 4999}
    senders = {
        "sender A: json.dumps defaults": json.dumps(event).encode(),
        "sender B: compact JSON, UTF-8 text": json.dumps(event, separators=(",", ":"), ensure_ascii=False).encode("utf-8"),
    }
    with LiveServer(app, 8103) as srv:
        for label, body in senders.items():
            print(label)
            print("  bytes on the wire:", body)
            print("  /broken verified: ", deliver(srv.url + "/broken", body))
            print("  /raw verified:    ", deliver(srv.url + "/raw", body))
        original = senders["sender B: compact JSON, UTF-8 text"]
        tampered = original.replace(b"4999", b"9999")
        print("tampered after signing, /raw verified:", deliver(srv.url + "/raw", tampered, signed_body=original))

Run python step03_rawbody.py.

sender A: json.dumps defaults
  bytes on the wire: b'{"type": "payment.succeeded", "customer": "Zo\\u00eb", "amount_cents": 4999}'
  /broken verified:  True
  /raw verified:     True
sender B: compact JSON, UTF-8 text
  bytes on the wire: b'{"type":"payment.succeeded","customer":"Zo\xc3\xab","amount_cents":4999}'
  /broken verified:  False
  /raw verified:     True
tampered after signing, /raw verified: False

Sender A calls json.dumps with its defaults, which puts a space after each comma and colon and writes “ë” as a six-character escape sequence. Re-serializing the parsed event with the same defaults happens to reproduce the same bytes, so /broken passes. Sender B writes compact JSON with the character as real UTF-8. When /broken re-serializes that event it inserts spaces and swaps the character for an escape sequence, the bytes no longer match what was signed, and a perfectly genuine delivery is rejected. /raw accepts both, and still rejects the tampered copy in the last line.

This is the classic works-in-testing, fails-in-production bug: your own test sender is probably written in Python and uses the same defaults as your re-serializer, so your tests can pass until a real provider with different formatting arrives.

Common mistake

Do not “fix” a signature mismatch by turning verification off, or by comparing only some fields. Read the raw bytes first (await request.body() in FastAPI), verify them, and only then parse those same bytes. If your route needs a Pydantic model, accept a Request and call model_validate_json(body) yourself after the check, which is what the final receiver does.

Check that it worked

The /raw lines must say True for both senders and False for the tampered copy. If /raw fails for sender B, something between the socket and your handler is changing the body, for example middleware that decodes and re-encodes it.

Step 4: Compare in constant time and treat headers as hostile

Why == is the wrong operator

Python’s == on strings stops at the first character that differs. Comparing a secret-derived value this way makes the running time depend on how many leading characters an attacker guessed correctly, which in principle lets them recover a valid signature one character at a time. The standard library’s documentation warns about exactly this: it is “recommended to use the compare_digest() function instead of the == operator to reduce the vulnerability to timing attacks” (Python docs, hmac). GitHub is blunter (“Never use a plain == operator”), and the Standard Webhooks specification says that skipping constant-time comparison can “turn them into signing oracles”.

Let us measure instead of assuming. The script below times == and hmac.compare_digest when the two strings differ at the first character and at the last. Save it as step04b_timing.py and run it; the numbers will differ on your machine, but the pattern should not.

# step04b_timing.py
import hmac
import timeit


def measure(statements: dict, scope: dict, number: int, rounds: int = 9) -> dict:
    """Best-of-rounds nanoseconds per call; the statements alternate so drift hits all of them equally."""
    best = {label: float("inf") for label in statements}
    for _ in range(rounds):
        for label, stmt in statements.items():
            elapsed = timeit.timeit(stmt, globals=scope, number=number)
            best[label] = min(best[label], elapsed / number * 1e9)
    return best


if __name__ == "__main__":
    for length, number in ((64, 1_000_000), (1_000_000, 500)):
        a = "a" * length
        first, last = "b" + a[1:], a[:-1] + "b"
        scope = {"hmac": hmac, "a": a, "first": first, "last": last,
                 "ab": a.encode(), "bfirst": first.encode(), "blast": last.encode()}
        results = measure({
            "==             mismatch at the first character": "a == first",
            "==             mismatch at the last character": "a == last",
            "compare_digest mismatch at the first character": "hmac.compare_digest(ab, bfirst)",
            "compare_digest mismatch at the last character": "hmac.compare_digest(ab, blast)",
        }, scope, number)
        print(f"{length:,} characters, nanoseconds per comparison")
        for label, ns in results.items():
            print(f"  {label}: {ns:>12,.1f}")
64 characters, nanoseconds per comparison
  ==             mismatch at the first character:          9.1
  ==             mismatch at the last character:         10.1
  compare_digest mismatch at the first character:         44.8
  compare_digest mismatch at the last character:         44.8
1,000,000 characters, nanoseconds per comparison
  ==             mismatch at the first character:         12.2
  ==             mismatch at the last character:     30,442.0
  compare_digest mismatch at the first character:    409,776.6
  compare_digest mismatch at the last character:    409,640.0

Read this honestly. For a 64-character digest, the kind a hex HMAC produces, == takes about 9 nanoseconds when the first character differs and about 10 when only the last does. That gap is far below what network jitter would let an attacker measure, so I am not claiming that == is trivially exploitable here. The 1,000,000-character rows show the mechanism is real: when the strings are long enough, == takes about 12 nanoseconds if the first character differs and about 30 microseconds if only the last does, while compare_digest takes about 410 microseconds either way. The point is not that you can exploit the short case today. The point is that compare_digest costs nothing you will notice, is designed to take the same time wherever the strings differ (its documentation describes “avoiding content-based short circuiting behaviour”), and does not rely on how a particular interpreter happens to compare strings.

The non-ASCII trap

Switching to compare_digest introduces a different problem. Its documentation says that “a and b must both be of the same type: either str (ASCII only, as e.g. returned by HMAC.hexdigest()), or a bytes-like object.” The signature header comes from the caller, and callers choose what they send. Save the script below as step04_compare.py. It first shows the exception, then sends the same hostile header, a signature made of the letter “é”, to two endpoints: one compares str values and one compares bytes.

# step04_compare.py
import hashlib
import hmac

import requests
from fastapi import FastAPI, Request, Response

from labserver import LiveServer

SECRET = b"lab-secret-for-step-four-only"
app = FastAPI()


def expected_header(body: bytes) -> str:
    return "sha256=" + hmac.new(SECRET, body, hashlib.sha256).hexdigest()


@app.post("/str-compare")
async def str_compare(request: Request):
    body = await request.body()
    given = request.headers.get("x-signature", "")
    if not hmac.compare_digest(expected_header(body), given):  # str against str
        return Response(status_code=401)
    return {"verified": True}


@app.post("/bytes-compare")
async def bytes_compare(request: Request):
    body = await request.body()
    given = request.headers.get("x-signature", "")
    if not hmac.compare_digest(expected_header(body).encode(), given.encode()):  # bytes against bytes
        return Response(status_code=401)
    return {"verified": True}


if __name__ == "__main__":
    print("== a non-ASCII str in compare_digest")
    try:
        hmac.compare_digest("sha256=abc", "sha256=abé")
    except TypeError as exc:
        print("TypeError:", exc)
    print("same text encoded to bytes first:", hmac.compare_digest("sha256=abc".encode(), "sha256=abé".encode()))

    print("== the same hostile header against two endpoints")
    body = b'{"type":"payment.succeeded"}'
    hostile = "sha256=" + "é" * 64
    with LiveServer(app, 8104) as srv:
        for path in ("/str-compare", "/bytes-compare"):
            good = requests.post(srv.url + path, data=body, headers={"X-Signature": expected_header(body)}, timeout=5)
            bad = requests.post(srv.url + path, data=body, headers={"X-Signature": hostile}, timeout=5)
            print(f"{path:<15} valid signature -> {good.status_code}   hostile header -> {bad.status_code}")

Run python step04_compare.py.

== a non-ASCII str in compare_digest
TypeError: comparing strings with non-ASCII characters is not supported
same text encoded to bytes first: False
== the same hostile header against two endpoints
/str-compare    valid signature -> 200   hostile header -> 500
/bytes-compare  valid signature -> 200   hostile header -> 401

The string version crashes with an unhandled TypeError, so the caller gets a 500 and your logs get a traceback, on demand, from anyone. That is not an authentication bypass, but it is a crash path that an attacker can trigger at will, and any error handler that swallows exceptions and carries on would turn it into one. The bytes version treats the hostile header as what it is, a wrong signature, and answers 401. The final verifier in Step 5 goes one step further and decodes the header’s base64 into bytes before comparing, so it never compares text at all.

Check that it worked

The hostile header must produce 401 from /bytes-compare, and compare_digest must show the same figure in both of its rows. Your absolute numbers will differ, and for 64 characters the two == rows can even swap order from run to run; the 1,000,000-character rows are the ones that show the mechanism.

Step 5: Stop replays by signing an id and a timestamp

The replay problem

A valid signature proves who wrote a message, not when. A replay attack is resending a genuine, correctly signed request later. HTTPS protects deliveries on the wire, but a valid request can still end up in a proxy log, a debugging tool or a compromised hop, and anyone who ever sees one can send it again. The Step 2 scheme signs only the body, so every copy verifies. The first block of output below, produced by the script we are about to save, shows a wallet credit being applied three times from one captured request.

The scheme: sign the id, the timestamp and the body together

The fix is to put freshness inside the signed data. We will follow the format described by the Standard Webhooks specification, which three headers carry: webhook-id (a unique id for the message), webhook-timestamp (seconds since the epoch) and webhook-signature (one or more entries shaped like v1,<base64 HMAC>). The signed string is msg_id.timestamp.body, with full stops between the parts. Because the timestamp is inside the signed bytes, an attacker cannot refresh it. As Stripe’s documentation puts it, “an attacker can’t change the timestamp without invalidating the signature”. The receiver then rejects anything outside a tolerance window; Stripe’s libraries default to 5 minutes, and so does our code.

Create webhooks.py with the first part of the library below. It will grow in Steps 6 and 7.

# webhooks.py
import base64
import hashlib
import hmac
import re
import sqlite3  # used from Step 6 on
import time

ID_PATTERN = re.compile(r"[A-Za-z0-9_-]{1,128}")


class WebhookError(Exception):
    """Verification failed. The message is for our logs, never for the caller."""


def sign(secret: bytes, msg_id: str, timestamp: int, body: bytes) -> str:
    """What the sender does: sign msg_id.timestamp.body and return a v1 header value."""
    content = f"{msg_id}.{timestamp}.".encode() + body
    return "v1," + base64.b64encode(hmac.new(secret, content, hashlib.sha256).digest()).decode()


def verify(secret, msg_id, timestamp_header, signature_header, body, now=None, tolerance=300):
    """What the receiver does. Returns None for an authentic delivery, raises WebhookError otherwise."""
    if tolerance <= 0:
        raise ValueError("tolerance must be a positive number of seconds")
    if not ID_PATTERN.fullmatch(msg_id):
        raise WebhookError("message id is missing or has unexpected characters")
    if not (timestamp_header.isascii() and timestamp_header.isdigit() and len(timestamp_header) <= 12):
        raise WebhookError("timestamp is not a plain integer")
    now = time.time() if now is None else now
    if abs(now - int(timestamp_header)) > tolerance:
        raise WebhookError("timestamp is outside the tolerance window")
    content = f"{msg_id}.{timestamp_header}.".encode() + body
    expected = hmac.new(secret, content, hashlib.sha256).digest()
    for entry in signature_header.split():
        version, _, encoded = entry.partition(",")
        if version != "v1":
            continue
        try:
            candidate = base64.b64decode(encoded, validate=True)
        except ValueError:
            continue
        if hmac.compare_digest(expected, candidate):
            return
    raise WebhookError("no v1 signature matched")

sign is what the provider does; you would normally only write verify, but having both lets us test the pair. verify runs its cheap checks first and the HMAC last:

First, a tolerance of zero or less raises ValueError, because Stripe warns that “using a tolerance value of 0 disables the recency check entirely”, and we would rather crash at configuration time than silently accept stale messages. Second, the message id must match a strict pattern. Third, the timestamp must be a plain run of ASCII digits of sane length, and its distance from the current time, in either direction, must be within the tolerance. Only then does it compute the HMAC over the header text exactly as received and walk the space-separated entries in webhook-signature. Entries with an unknown version are skipped, entries whose base64 is invalid are skipped, and each remaining digest is compared with hmac.compare_digest on bytes, which also avoids the Step 4 trap. The Standard Webhooks specification describes the header as a “space delimited list of signatures” so that a sender can sign with an old and a new key during rotation, which Step 7 uses.

Now save step05_replay.py. It replays the Step 2 scheme, then runs the new scheme through six situations with an injected clock so the results are repeatable, then demonstrates two traps.

# step05_replay.py
import json

from step02_hmac import sign_body, verify_body
from webhooks import WebhookError, sign, verify

SECRET = bytes(range(32))
NOW = 1_700_000_000  # a fixed "current time" keeps the demo repeatable


def replay_with_the_step_two_scheme():
    print("== Step 2 scheme: the same request sent three times")
    wallet = 0
    body = json.dumps({"type": "wallet.credit", "amount_cents": 5000}).encode()
    header = sign_body(SECRET, body)  # one genuine delivery, seen by someone it was not meant for
    for attempt in range(1, 4):
        if verify_body(SECRET, body, header):
            wallet += 5000
        print(f"  resend {attempt}: signature valid, wallet is {wallet}")


def replay_with_a_signed_timestamp():
    print("== signed id and timestamp: the same attacks")
    msg_id, sent_at = "msg_77", NOW
    body = json.dumps({"type": "wallet.credit", "amount_cents": 5000}).encode()
    signature = sign(SECRET, msg_id, sent_at, body)

    def attempt(label, now, timestamp=sent_at):
        try:
            verify(SECRET, msg_id, str(timestamp), signature, body, now=now)
            print(f"  {label:<46} accepted")
        except WebhookError as exc:
            print(f"  {label:<46} rejected: {exc}")

    attempt("delivered 2 seconds after signing", NOW + 2)
    attempt("replayed 10 minutes later", NOW + 600)
    attempt("replayed, timestamp header rewritten", NOW + 600, timestamp=NOW + 600)
    attempt("receiver clock 4 minutes behind the sender", NOW - 240)
    attempt("receiver clock 6 minutes behind the sender", NOW - 360)
    attempt("exact copy replayed 30 seconds later", NOW + 30)


def traps():
    print("== traps")

    def content(msg_id, timestamp, body):
        return f"{msg_id}.{timestamp}.".encode() + body

    one = content("evt", "1.2", b"x")
    two = content("evt.1", "2", b"x")
    print("  two different field splits, one signed string:", one, two, one == two)
    for label, args in (("id with a dot", ("evt.1", "2")), ("timestamp with a dot", ("evt", "1.2"))):
        try:
            verify(SECRET, args[0], args[1], "v1,AAAA", b"x", now=2)
        except WebhookError as exc:
            print(f"  {label}: {exc}")
    try:
        verify(SECRET, "msg_77", str(NOW), "v1,AAAA", b"x", now=NOW, tolerance=0)
    except ValueError as exc:
        print("  tolerance=0:", exc)


if __name__ == "__main__":
    replay_with_the_step_two_scheme()
    print()
    replay_with_a_signed_timestamp()
    print()
    traps()

Run python step05_replay.py. The first block is the replay against the Step 2 scheme.

== Step 2 scheme: the same request sent three times
  resend 1: signature valid, wallet is 5000
  resend 2: signature valid, wallet is 10000
  resend 3: signature valid, wallet is 15000

Three identical requests credited the wallet three times. Now the same attacks against the signed timestamp:

== signed id and timestamp: the same attacks
  delivered 2 seconds after signing              accepted
  replayed 10 minutes later                      rejected: timestamp is outside the tolerance window
  replayed, timestamp header rewritten           rejected: no v1 signature matched
  receiver clock 4 minutes behind the sender     accepted
  receiver clock 6 minutes behind the sender     rejected: timestamp is outside the tolerance window
  exact copy replayed 30 seconds later           accepted

Read the rows from the top. A delivery 2 seconds after signing is accepted. The same bytes ten minutes later are rejected by the window. An attacker who rewrites the timestamp header to look fresh is rejected for a different reason: the signature no longer matches, which proves the timestamp is really covered. The two clock-skew rows show the window works in both directions, so a receiver whose clock is slow is still tolerated for 5 minutes but not 6, which is why you should keep servers on NTP. The last row is the important one: an exact copy replayed 30 seconds later is accepted, because it is signed, recent and unmodified. A timestamp shrinks the replay window but cannot close it. Step 6 closes it.

Two traps in the signed string

== traps
  two different field splits, one signed string: b'evt.1.2.x' b'evt.1.2.x' True
  id with a dot: message id is missing or has unexpected characters
  timestamp with a dot: timestamp is not a plain integer
  tolerance=0: tolerance must be a positive number of seconds

The first line is a delimiter ambiguity. If the message id could contain a full stop, the id evt with timestamp 1.2 and the id evt.1 with timestamp 2 would produce the same signed string, so a signature for one would verify the other. The Standard Webhooks specification asks that the message id and timestamp not be user controlled, or at least not contain a full stop. Our verifier simply refuses ids outside [A-Za-z0-9_-] and timestamps that are not plain digits, as the next two lines show. The last line shows the zero-tolerance guard from earlier.

What retries do to the timestamp

Retries produce new attempts. Stripe notes that when it retries an event it generates a new signature and timestamp for the new delivery attempt, so the tolerance check judges each attempt on its own. That is also why an id-based check, as in the next step, is needed to recognise the same event arriving twice.

Check that it worked

Three rows must be accepted (the fresh delivery, 4 minutes of clock skew and the 30-second copy) and three rejected (the ten-minute replay, the rewritten timestamp and 6 minutes of skew). If the rewritten timestamp is accepted, your signed string does not include the timestamp.

Step 6: Process each event once, atomically

The Standard Webhooks specification recommends using the message id “as an idempotency key to prevent accidentally processing the same webhook more than once”. An operation is idempotent when doing it twice has the same effect as doing it once. Duplicates are not only a malicious problem. The same specification says the id lets a consumer process an event once “even if sent multiple times maliciously, in error, or due to networking issues”, so a delivery whose acknowledgement never reached the provider will simply be sent again. The task is to remember which ids you have handled, and the subtle part is when to remember them.

Add the second part of the library to webhooks.py:

# webhooks.py (continued)
class EventStore:
    """Remembers which webhook ids were handled and applies each event in the same transaction."""

    def __init__(self, path=":memory:", keep_seconds=7 * 24 * 3600):
        self.db = sqlite3.connect(path, check_same_thread=False, isolation_level=None)
        self.db.execute("CREATE TABLE IF NOT EXISTS seen (msg_id TEXT PRIMARY KEY, seen_at INTEGER NOT NULL)")
        self.keep_seconds = keep_seconds

    def process_once(self, msg_id, apply, now=None):
        """Run apply(db) once per msg_id. True means applied, False means a duplicate."""
        now = int(time.time() if now is None else now)
        self.db.execute("BEGIN IMMEDIATE")
        try:
            self.db.execute("DELETE FROM seen WHERE seen_at < ?", (now - self.keep_seconds,))
            claimed = self.db.execute(
                "INSERT OR IGNORE INTO seen (msg_id, seen_at) VALUES (?, ?)", (msg_id, now)
            ).rowcount
            if claimed:
                apply(self.db)
            self.db.execute("COMMIT")
            return bool(claimed)
        except BaseException:
            if self.db.in_transaction:
                self.db.execute("ROLLBACK")
            raise

process_once does everything inside one SQLite transaction. BEGIN IMMEDIATE starts a write transaction straight away (the SQLite documentation says it “causes the database connection to start a new write immediately”), so the claim and the work happen under the database’s write lock (SQLite, Transactions). This tutorial uses one connection, called from FastAPI’s event loop, which is enough here. With several worker processes you would give each its own connection to a shared database file, and a delivery that finds the lock taken has to wait or retry, because the documentation notes that BEGIN IMMEDIATE “might fail with SQLITE_BUSY if another write transaction is already active”. INSERT OR IGNORE claims the id, and its rowcount is 0 when the id was already there. Only when the claim succeeds do we call apply(db), the function that changes your data. If anything raises, the rollback removes the claim and the effect together. Old ids are deleted as we go, after keep_seconds.

To see why the transaction matters, save step06_dedupe.py. It runs the store, then two plausible-looking alternatives that each fail in a different way.

# step06_dedupe.py
import sqlite3

from webhooks import EventStore


def make_wallet(db):
    db.execute("CREATE TABLE IF NOT EXISTS wallet (id INTEGER PRIMARY KEY, balance_cents INTEGER NOT NULL)")
    db.execute("INSERT OR IGNORE INTO wallet VALUES (1, 0)")


def balance(db):
    return db.execute("SELECT balance_cents FROM wallet WHERE id = 1").fetchone()[0]


def credit(cents, crash=None):
    def apply(db):
        if crash == "before":
            raise RuntimeError("the ledger connection dropped")
        db.execute("UPDATE wallet SET balance_cents = balance_cents + ? WHERE id = 1", (cents,))
        if crash == "after":
            raise RuntimeError("the process died right after the update")

    return apply


def duplicates():
    print("== an exact copy inside the tolerance window")
    store = EventStore()
    make_wallet(store.db)
    print("  first delivery applied:", store.process_once("msg_77", credit(5000)))
    print("  exact copy applied:    ", store.process_once("msg_77", credit(5000)))
    print("  wallet balance:        ", balance(store.db))


def autocommit_db():
    db = sqlite3.connect(":memory:", isolation_level=None)  # autocommit: every statement commits at once
    db.execute("CREATE TABLE seen (msg_id TEXT PRIMARY KEY)")
    make_wallet(db)
    return db


def claim_first():
    print("== claim the id first, do the work second")
    db = autocommit_db()

    def handle(msg_id):
        if db.execute("SELECT 1 FROM seen WHERE msg_id = ?", (msg_id,)).fetchone():
            return "duplicate, skipped"
        db.execute("INSERT INTO seen VALUES (?)", (msg_id,))
        credit(2500, crash="before")(db)
        return "credited"

    try:
        handle("msg_78")
    except RuntimeError as exc:
        print("  attempt 1 crashed:", exc)
    print("  the provider retries:", handle("msg_78"))
    print("  wallet balance:", balance(db), "for a 2500 event, and the event is gone for good")


def work_first():
    print("== do the work first, record the id second")
    db = autocommit_db()
    attempts = []

    def handle(msg_id):
        if db.execute("SELECT 1 FROM seen WHERE msg_id = ?", (msg_id,)).fetchone():
            return "duplicate, skipped"
        credit(2500)(db)
        attempts.append(msg_id)
        if len(attempts) == 1:
            raise RuntimeError("the process died before it recorded the id")
        db.execute("INSERT INTO seen VALUES (?)", (msg_id,))
        return "credited"

    try:
        handle("msg_79")
    except RuntimeError as exc:
        print("  attempt 1 crashed:", exc)
    print("  the provider retries:", handle("msg_79"))
    print("  wallet balance:", balance(db), "for a 2500 event")


def one_transaction():
    print("== claim and work in one transaction")
    store = EventStore()
    make_wallet(store.db)
    try:
        store.process_once("msg_80", credit(2500, crash="after"))
    except RuntimeError as exc:
        print("  attempt 1 crashed:", exc)
    print("  wallet after the crash:", balance(store.db))
    print("  the provider retries, applied:", store.process_once("msg_80", credit(2500)))
    print("  a second retry, applied:      ", store.process_once("msg_80", credit(2500)))
    print("  wallet balance:", balance(store.db))


if __name__ == "__main__":
    duplicates()
    print()
    claim_first()
    print()
    work_first()
    print()
    one_transaction()

Run python step06_dedupe.py. The first block shows the happy path.

== an exact copy inside the tolerance window
  first delivery applied: True
  exact copy applied:     False
  wallet balance:         5000

The exact copy is recognised and the wallet credits once. Now the failure modes of doing it by hand:

== claim the id first, do the work second
  attempt 1 crashed: the ledger connection dropped
  the provider retries: duplicate, skipped
  wallet balance: 0 for a 2500 event, and the event is gone for good

== do the work first, record the id second
  attempt 1 crashed: the process died before it recorded the id
  the provider retries: credited
  wallet balance: 5000 for a 2500 event

If you record the id first and crash before doing the work, the provider’s retry is treated as a duplicate and the event is gone for good. If you do the work first and crash before recording the id, the retry is processed again and the customer is credited twice. Each ordering fails on one side of the crash. The single transaction removes the choice:

== claim and work in one transaction
  attempt 1 crashed: the process died right after the update
  wallet after the crash: 0
  the provider retries, applied: True
  a second retry, applied:       False
  wallet balance: 2500

The crash happened after the UPDATE, yet the wallet shows 0 afterwards because the rollback undid the update and the claim together. The provider’s retry then applies the event once, and a second retry is recognised as a duplicate. The same idea appears in our transactional outbox tutorial, which solves the related problem of writing to a database and sending a message without losing either, and our guide to retrying failed API calls shows the sending side of the retries you are protecting yourself against.

How long to remember ids

Keep them at least as long as your tolerance window, since an older replay already fails the timestamp check. Ideally keep them as long as your provider can keep retrying a failed delivery, which you should read in its documentation. Standard Webhooks gives a short example (“save the IDs in redis for 5 minutes”); the code above defaults to seven days because a table of ids is cheap. With a different database, use the same pattern: a unique constraint on the event id inside the transaction that applies the effect.

Check that it worked

Look at the four balances: 5,000 after the duplicate (credited once, which is correct), 0 for the claim-first run (the event is lost), 5,000 for a 2,500 event in the work-first run (credited twice), and 2,500 in the transaction run (exactly once). Only the transaction run is what a production handler should do.

Step 7: Load secrets safely and rotate them without downtime

The empty secret trap

HMAC accepts an empty key without complaint, and an attacker can compute an HMAC with an empty key as easily as you can. So a receiver whose secret comes from an unset environment variable, os.environ.get("WEBHOOK_SECRET", ""), does not merely lose protection: it makes the empty string a key that every attacker already knows. The specification’s table says a signing secret should be random and “between 24 bytes (192 bits) and 64 bytes (512 bits)”, and serialized as base64 with a whsec_ prefix. We will enforce the minimum and refuse to start without a valid secret. Add the third part of the library to webhooks.py:

# webhooks.py (continued)
def load_secrets(value: str) -> list[bytes]:
    """Parse comma-separated whsec_ secrets, newest first. Refuse anything weak or missing."""
    keys = []
    for item in value.split(","):
        item = item.strip()
        if not item:
            continue
        try:
            key = base64.b64decode(item.removeprefix("whsec_"), validate=True)
        except ValueError:
            raise ValueError("a webhook secret is not valid base64") from None
        if len(key) < 24:
            raise ValueError("a webhook secret must decode to at least 24 bytes")
        keys.append(key)
    if not keys:
        raise ValueError("no webhook secret configured")
    return keys


def verify_with_any(secrets, msg_id, timestamp_header, signature_header, body, now=None, tolerance=300):
    """Accept a delivery signed with any configured secret. An empty list fails closed."""
    error = WebhookError("no secrets configured")
    for secret in secrets:
        try:
            return verify(secret, msg_id, timestamp_header, signature_header, body, now, tolerance)
        except WebhookError as exc:
            error = exc
    raise error

load_secrets accepts a comma-separated list, newest first, strips the prefix, decodes the base64 and refuses anything shorter than 24 bytes. verify_with_any tries each secret in turn and fails closed when the list is empty. Save the demo as step07_secrets.py.

# step07_secrets.py
import base64
import secrets

from webhooks import WebhookError, load_secrets, sign, verify, verify_with_any  # noqa: F401

NOW = 1_700_000_000
BODY = b'{"type":"payment.succeeded"}'


def whsec(key: bytes) -> str:
    return "whsec_" + base64.b64encode(key).decode()


def empty_secret_trap():
    print("== an unset environment variable becomes the secret")
    forged = sign(b"", "msg_1", NOW, BODY)  # the attacker signs with an empty key
    verify(b"", "msg_1", str(NOW), forged, BODY, now=NOW)
    print("  verify() with an empty secret accepted the attacker's signature")
    for label, value in (
        ("empty", ""),
        ("5 bytes", whsec(b"short")),
        ("not base64", "whsec_!!!"),
        ("32 random bytes", whsec(secrets.token_bytes(32))),
    ):
        try:
            print(f"  load_secrets({label}) ->", len(load_secrets(value)), "secret loaded")
        except ValueError as exc:
            print(f"  load_secrets({label}) -> refused: {exc}")


def rotation():
    print("== rotation: the receiver holds the new secret first and the old one second")
    old, new, stranger = (secrets.token_bytes(32) for _ in range(3))
    configured = load_secrets(f"{whsec(new)},{whsec(old)}")

    def attempt(label, signature, held=configured):
        try:
            verify_with_any(held, "msg_9", str(NOW), signature, BODY, now=NOW)
            print(f"  {label:<48} accepted")
        except WebhookError as exc:
            print(f"  {label:<48} rejected: {exc}")

    attempt("signed with the old secret", sign(old, "msg_9", NOW, BODY))
    attempt("signed with the new secret", sign(new, "msg_9", NOW, BODY))
    attempt("signed with a secret nobody configured", sign(stranger, "msg_9", NOW, BODY))
    attempt("sender signs with both, receiver knows only old", sign(new, "msg_9", NOW, BODY) + " " + sign(old, "msg_9", NOW, BODY), held=[old])
    attempt("junk entries before a valid one", "v2,abc v1,!!!! v1a,xyz " + sign(new, "msg_9", NOW, BODY))
    attempt("only junk", "v1,!!!! v2,abc nonsense")
    attempt("no secrets configured at all", sign(new, "msg_9", NOW, BODY), held=[])


if __name__ == "__main__":
    empty_secret_trap()
    print()
    rotation()

Run python step07_secrets.py.

== an unset environment variable becomes the secret
  verify() with an empty secret accepted the attacker's signature
  load_secrets(empty) -> refused: no webhook secret configured
  load_secrets(5 bytes) -> refused: a webhook secret must decode to at least 24 bytes
  load_secrets(not base64) -> refused: a webhook secret is not valid base64
  load_secrets(32 random bytes) -> 1 secret loaded

The first line is the problem in one sentence: with an empty secret, the attacker’s signature is accepted. The four load_secrets lines show the cure, since an empty value, a 5-byte secret and invalid base64 are all refused at startup, and only a 32-byte random secret loads.

Rotating a secret without dropping deliveries

Secrets should change from time to time, and immediately if one leaks. A hard cutover would reject every in-flight delivery signed with the old secret. The specification’s answer is that the signature header is a list: “webhook consumers can try to verify each signature until one matches”, and a sender can sign with the current and the previous key for a while. Our verifier supports both halves of that design: it walks every entry in the header, and it can hold more than one secret.

== rotation: the receiver holds the new secret first and the old one second
  signed with the old secret                       accepted
  signed with the new secret                       accepted
  signed with a secret nobody configured           rejected: no v1 signature matched
  sender signs with both, receiver knows only old  accepted
  junk entries before a valid one                  accepted
  only junk                                        rejected: no v1 signature matched
  no secrets configured at all                     rejected: no secrets configured

With the new secret first and the old one second, deliveries signed with either are accepted, and a stranger’s secret is not. A sender that signs with both keys is accepted by a receiver that only knows the old one. Junk entries and unknown versions in front of a valid signature are skipped rather than crashing the check, and a header made only of junk is rejected. With no secrets configured at all, verification fails closed.

The usual procedure is: add the new secret in front of the old one and deploy the receiver; make the provider switch to the new secret; wait until the old one has stopped appearing (your tolerance window plus the provider’s retry period); then delete the old secret and deploy again. If you are rotating because a secret leaked, skip the overlap and remove the old one at once, accepting that some deliveries will fail.

Check that it worked

Empty, short and non-base64 secrets must all be refused, and every rejected row in the rotation block must be one you expect: the stranger’s secret, the junk-only header and the empty secret list.

Step 8: Put it together in a receiver and attack it

Now wire the library into a real endpoint. Save this as receiver.py. It reads WEBHOOK_SECRETS from the environment, which makes the design decisions below visible.

# receiver.py
import logging
import os

from fastapi import FastAPI, Request, Response
from pydantic import BaseModel, ValidationError

from webhooks import EventStore, WebhookError, load_secrets, verify_with_any

log = logging.getLogger("receiver")
MAX_BODY_BYTES = 64 * 1024

SECRETS = load_secrets(os.environ.get("WEBHOOK_SECRETS", ""))  # the app refuses to start without a good secret
store = EventStore()
store.db.execute("CREATE TABLE wallet (id INTEGER PRIMARY KEY, balance_cents INTEGER NOT NULL)")
store.db.execute("INSERT INTO wallet VALUES (1, 0)")
app = FastAPI()


class PaymentEvent(BaseModel):
    type: str
    order_id: str
    amount_cents: int


class BodyTooLarge(Exception):
    pass


async def read_limited(request: Request, limit: int) -> bytes:
    declared = request.headers.get("content-length", "")
    if declared.isascii() and declared.isdigit() and int(declared) > limit:
        raise BodyTooLarge
    chunks, total = [], 0
    async for chunk in request.stream():
        total += len(chunk)
        if total > limit:
            raise BodyTooLarge
        chunks.append(chunk)
    return b"".join(chunks)


def apply_payment(event: PaymentEvent):
    def apply(db):
        if event.type == "payment.succeeded":
            db.execute("UPDATE wallet SET balance_cents = balance_cents + ? WHERE id = 1", (event.amount_cents,))

    return apply


@app.post("/webhooks/payments")
async def payments(request: Request):
    try:
        body = await read_limited(request, MAX_BODY_BYTES)
    except BodyTooLarge:
        return Response(status_code=413)
    headers = request.headers
    msg_id = headers.get("webhook-id", "")
    try:
        verify_with_any(
            SECRETS, msg_id, headers.get("webhook-timestamp", ""), headers.get("webhook-signature", ""), body
        )
    except WebhookError as exc:
        log.warning("rejected delivery %r: %s", msg_id[:40], exc)
        return Response(status_code=401)  # the same empty answer for every reason
    try:
        event = PaymentEvent.model_validate_json(body)  # parse the bytes we just verified
    except ValidationError:
        return Response(status_code=422)
    applied = store.process_once(msg_id, apply_payment(event))
    return {"ok": True, "duplicate": not applied}

The decisions in this file

Cap the body before you hash it. An unauthenticated caller should not be able to make you read and hash an unbounded request. read_limited rejects a declared Content-Length above 64 KB straight away and also counts the bytes of chunked uploads, which carry no length header.

Verify first, parse second. The route reads the raw bytes, verifies them with verify_with_any, and only afterwards parses those same bytes with PaymentEvent.model_validate_json. Nothing in the payload influences the verification.

Say as little as possible when you refuse. Every verification failure produces the same empty 401, so the response does not tell an attacker whether the id, the timestamp or the signature was wrong. The reason goes to your log instead, with the id printed through %r and truncated so a hostile id cannot forge log lines. Never log the secret or the signature header value.

Acknowledge duplicates. A duplicate id returns 200 with "duplicate": true, which tells the provider to stop retrying without applying the event again. Authentic events we cannot use return 422. Stripe’s documentation says it retries when an endpoint replies with a non-2xx status, so decide per provider whether a permanent 422 is better than acknowledging and discarding.

Refuse to start without a good secret. Importing the module without WEBHOOK_SECRETS set shows the Step 7 guard in action:

ValueError: no webhook secret configured

Run the attack matrix

Save the script below as step08_attack_matrix.py. It generates three random secrets (the receiver’s new and old secrets, and one that belongs to the attacker), starts the receiver on port 8108, and sends fourteen requests, printing the HTTP status, whether the receiver called the event a duplicate, and the wallet balance after each.

# step08_attack_matrix.py
import base64
import os
import secrets
import time

import requests

NEW_KEY, OLD_KEY, ATTACKER_KEY = (secrets.token_bytes(32) for _ in range(3))
os.environ["WEBHOOK_SECRETS"] = ",".join("whsec_" + base64.b64encode(k).decode() for k in (NEW_KEY, OLD_KEY))

from labserver import LiveServer  # noqa: E402  (after the environment variable is set)
from receiver import app, store  # noqa: E402
from webhooks import sign  # noqa: E402

NOW = int(time.time())
EVENT = b'{"type":"payment.succeeded","order_id":"A-1001","amount_cents":5000}'


def signed_headers(msg_id, key=NEW_KEY, timestamp=NOW, body=EVENT):
    return {
        "webhook-id": msg_id,
        "webhook-timestamp": str(timestamp),
        "webhook-signature": sign(key, msg_id, timestamp, body),
    }


def balance():
    return store.db.execute("SELECT balance_cents FROM wallet WHERE id = 1").fetchone()[0]


SCENARIOS = [
    ("genuine delivery", EVENT, signed_headers("msg_001")),
    ("same request sent again", EVENT, signed_headers("msg_001")),
    ("no signature headers", EVENT, {}),
    ("forged with the attacker's own key", EVENT, signed_headers("msg_002", key=ATTACKER_KEY)),
    ("amount edited after signing", EVENT.replace(b"5000", b"500000"), signed_headers("msg_003")),
    ("captured request replayed 10 minutes later", EVENT, signed_headers("msg_004", timestamp=NOW - 600)),
    ("replay with the timestamp header refreshed", EVENT, {**signed_headers("msg_005", timestamp=NOW - 600), "webhook-timestamp": str(NOW)}),
    ("hostile non-ASCII signature header", EVENT, {**signed_headers("msg_006"), "webhook-signature": "v1," + "é" * 44}),
    ("signed with the retired secret", EVENT, signed_headers("msg_007", key=OLD_KEY)),
    ("message id containing a dot", EVENT, signed_headers("msg.008")),
    ("200 KB body", b"x" * 200_000, signed_headers("msg_009", body=b"x" * 200_000)),
    ("chunked 200 KB upload, no length header", (b"x" * 10_000 for _ in range(20)), signed_headers("msg_012", body=b"x" * 200_000)),
    ("authentic bytes that are not JSON", b"not json", signed_headers("msg_010", body=b"not json")),
    ("authentic JSON with a field missing", b'{"type":"payment.succeeded"}', signed_headers("msg_011", body=b'{"type":"payment.succeeded"}')),
]

if __name__ == "__main__":
    with LiveServer(app, 8108) as srv:
        print(f"{'scenario':<45} {'status':>6}  {'duplicate':<9} wallet")
        for label, body, headers in SCENARIOS:
            try:
                reply = requests.post(srv.url + "/webhooks/payments", data=body, headers=headers, timeout=5)
                status = str(reply.status_code)
                duplicate = str(reply.json().get("duplicate", "")) if reply.headers.get("content-type", "").startswith("application/json") else ""
            except requests.exceptions.ConnectionError:
                status, duplicate = "reset", ""
            print(f"{label:<45} {status:>6}  {duplicate:<9} {balance()}")

Run python step08_attack_matrix.py.

scenario                                      status  duplicate wallet
genuine delivery                                 200  False     5000
same request sent again                          200  True      5000
no signature headers                             401            5000
forged with the attacker's own key               401            5000
amount edited after signing                      401            5000
captured request replayed 10 minutes later       401            5000
replay with the timestamp header refreshed       401            5000
hostile non-ASCII signature header               401            5000
signed with the retired secret                   200  False     10000
message id containing a dot                      401            10000
200 KB body                                      413            10000
chunked 200 KB upload, no length header          413            10000
authentic bytes that are not JSON                422            10000
authentic JSON with a field missing              422            10000

The two genuine deliveries (the first row, and the one signed with the retired secret that is still on the receiver’s list) each credit 5,000 and nothing else changes the wallet, so it ends at 10,000. The row that sends the first request again gets 200 with duplicate set to True and no change. Every attack in the 401 rows gets the same empty answer: no headers, a forged key, an edited amount, a ten-minute-old replay, a replay with a refreshed timestamp, a hostile header (401 instead of the 500 from Step 4), and an id with a full stop. The two oversized bodies get 413 before anything is hashed, the first on its declared length alone and the chunked upload, which declares none, once the running count passes 64 KB. Two authentic requests that the application cannot use, bytes that are not JSON and JSON with a missing field, get 422 because they passed verification but failed validation.

Check that it worked

The statuses in the second column and the wallet in the last column are what matter; the random secrets change on every run, but the verdicts must not. If the hostile-header row shows 500, you are comparing text instead of bytes somewhere. If the replay rows show 200, your timestamp check or tolerance is not being applied.

Step 9: Lock it in with tests

Every lesson above is easy to undo by accident, such as someone “simplifying” verify six months from now. A test suite pins each one down. The tests below use an injected clock instead of the real time, so they are fast and repeatable. They cover GitHub’s published vector, the edges of the tolerance window (exactly 300 seconds passes, 301 fails, in both directions), malformed ids and timestamps, hostile headers, rotation, weak secrets, duplicate handling, rollback when the work fails, and pruning of old ids. Save them as test_webhooks.py.

# test_webhooks.py
import base64

import pytest

from step02_hmac import sign_body
from webhooks import EventStore, WebhookError, load_secrets, sign, verify, verify_with_any

KEY = bytes(range(32))
OTHER = bytes(range(32, 64))
BODY = b'{"type":"payment.succeeded","order_id":"A-1","amount_cents":5000}'
NOW = 1_700_000_000


def delivery(key=KEY, msg_id="msg_1", timestamp=NOW, body=BODY):
    return msg_id, str(timestamp), sign(key, msg_id, timestamp, body)


def whsec(key):
    return "whsec_" + base64.b64encode(key).decode()


def test_github_published_vector():
    secret, payload = b"It's a Secret to Everybody", b"Hello, World!"
    assert sign_body(secret, payload) == "sha256=757107ea0eb2509fc211221cce984b8a37570b6d7586c22c46f4379c8b043e17"


def test_authentic_delivery_is_accepted():
    msg_id, ts, signature = delivery()
    assert verify(KEY, msg_id, ts, signature, BODY, now=NOW + 1) is None


def test_changed_body_is_rejected():
    msg_id, ts, signature = delivery()
    with pytest.raises(WebhookError, match="no v1 signature"):
        verify(KEY, msg_id, ts, signature, BODY.replace(b"5000", b"9999"), now=NOW)


def test_wrong_secret_is_rejected():
    msg_id, ts, signature = delivery(key=OTHER)
    with pytest.raises(WebhookError):
        verify(KEY, msg_id, ts, signature, BODY, now=NOW)


@pytest.mark.parametrize("offset, accepted", [(0, True), (300, True), (301, False), (-300, True), (-301, False)])
def test_tolerance_window_edges(offset, accepted):
    msg_id, ts, signature = delivery()
    if accepted:
        verify(KEY, msg_id, ts, signature, BODY, now=NOW + offset)
    else:
        with pytest.raises(WebhookError, match="tolerance"):
            verify(KEY, msg_id, ts, signature, BODY, now=NOW + offset)


def test_refreshing_the_timestamp_breaks_the_signature():
    msg_id, ts, signature = delivery(timestamp=NOW - 600)
    with pytest.raises(WebhookError, match="no v1 signature"):
        verify(KEY, msg_id, str(NOW), signature, BODY, now=NOW)


def test_zero_tolerance_is_a_configuration_error():
    msg_id, ts, signature = delivery()
    with pytest.raises(ValueError):
        verify(KEY, msg_id, ts, signature, BODY, now=NOW, tolerance=0)


@pytest.mark.parametrize("msg_id, ts", [("evt.1", "2"), ("evt", "1.2"), ("", "2"), ("evt", "-5"), ("evt", "٣٤٥"), ("evt", "9" * 40)])
def test_malformed_id_or_timestamp(msg_id, ts):
    with pytest.raises(WebhookError):
        verify(KEY, msg_id, ts, "v1,AAAA", BODY, now=2)


def test_non_ascii_signature_header_fails_cleanly():
    msg_id, ts, _ = delivery()
    with pytest.raises(WebhookError):
        verify(KEY, msg_id, ts, "v1," + "é" * 44, BODY, now=NOW)


def test_junk_and_unknown_versions_are_skipped():
    msg_id, ts, signature = delivery()
    verify(KEY, msg_id, ts, "v2,abc v1,!!!! nonsense " + signature, BODY, now=NOW)


def test_any_signature_in_the_list_may_match():
    msg_id, ts, good = delivery()
    _, _, other = delivery(key=OTHER)
    verify(KEY, msg_id, ts, other + " " + good, BODY, now=NOW)
    verify(OTHER, msg_id, ts, other + " " + good, BODY, now=NOW)


def test_verify_with_any_old_new_neither_and_empty():
    msg_id, ts, signed_by_old = delivery(key=KEY)
    verify_with_any([OTHER, KEY], msg_id, ts, signed_by_old, BODY, now=NOW)
    with pytest.raises(WebhookError):
        verify_with_any([OTHER], msg_id, ts, signed_by_old, BODY, now=NOW)
    with pytest.raises(WebhookError, match="no secrets"):
        verify_with_any([], msg_id, ts, signed_by_old, BODY, now=NOW)


def test_load_secrets_refuses_weak_input():
    for bad in ("", " , ", whsec(b"short"), "whsec_!!!"):
        with pytest.raises(ValueError):
            load_secrets(bad)
    assert load_secrets(f" {whsec(KEY)} , ,{whsec(OTHER)}") == [KEY, OTHER]


def test_store_applies_once():
    store, calls = EventStore(), []
    assert store.process_once("a", lambda db: calls.append(1), now=NOW) is True
    assert store.process_once("a", lambda db: calls.append(1), now=NOW + 5) is False
    assert calls == [1]


def test_store_rolls_back_the_claim_when_the_work_fails():
    store = EventStore()

    def boom(db):
        raise RuntimeError("ledger down")

    with pytest.raises(RuntimeError):
        store.process_once("a", boom, now=NOW)
    assert store.db.execute("SELECT COUNT(*) FROM seen").fetchone()[0] == 0
    assert store.process_once("a", lambda db: None, now=NOW + 60) is True


def test_store_forgets_ids_older_than_keep_seconds():
    store = EventStore(keep_seconds=100)
    store.process_once("old", lambda db: None, now=NOW)
    store.process_once("new", lambda db: None, now=NOW + 101)
    assert [r[0] for r in store.db.execute("SELECT msg_id FROM seen")] == ["new"]

Run python -m pytest -q.

.........................                                                                                                                      [100%]
25 passed in 0.05s

Step 10: Cross-check against a reference implementation

Passing your own tests proves you are consistent with yourself. A stronger check is interoperating with someone else’s code. The Standard Webhooks project publishes an official Python library, standardwebhooks, and the script below tests both directions: our sign against its verify, and its sign against our verify. Save it as step10_interop.py.

# step10_interop.py
import base64
import datetime
import importlib.metadata
import secrets
import time

from standardwebhooks import Webhook

from webhooks import WebhookError, sign, verify

key = secrets.token_bytes(32)
official = Webhook("whsec_" + base64.b64encode(key).decode())
msg_id = "msg_2KWPBgLlAfxdpx2AI54pPJ85f4W"
body = '{"type":"payment.succeeded","order_id":"A-1001","amount_cents":5000}'
now = int(time.time())
print("standardwebhooks", importlib.metadata.version("standardwebhooks"))

ours = sign(key, msg_id, now, body.encode())
headers = {"webhook-id": msg_id, "webhook-timestamp": str(now), "webhook-signature": ours}
print("their verify() accepts our signature:", official.verify(body, headers)["order_id"] == "A-1001")

theirs = official.sign(msg_id, datetime.datetime.fromtimestamp(now, tz=datetime.timezone.utc), body)
verify(key, msg_id, str(now), theirs, body.encode())
print("our verify() accepts their signature: True")
print("the two signature strings are identical:", ours == theirs)

print("== a header with a junk entry in front of a valid signature")
junk = "nonsense " + ours
try:
    verify(key, msg_id, str(now), junk, body.encode())
    print("ours:   accepted")
except WebhookError as exc:
    print("ours:   rejected,", exc)
try:
    official.verify(body, {**headers, "webhook-signature": junk})
    print("theirs: accepted")
except Exception as exc:
    print("theirs:", type(exc).__name__, "-", exc)

Run python step10_interop.py.

standardwebhooks 1.1.0
their verify() accepts our signature: True
our verify() accepts their signature: True
the two signature strings are identical: True
== a header with a junk entry in front of a valid signature
ours:   accepted
theirs: ValueError - not enough values to unpack (expected 2, got 1)

Both libraries accept each other’s signatures, and the two signature strings are byte-for-byte identical, so our implementation is compatible with the specification’s reference code. The last block shows a difference in robustness. Given a header with a junk entry in front of a valid signature, our verifier skips the junk and accepts the genuine signature, while version 1.1.0 of the reference library raised a ValueError instead of its own verification error. If you use that library, catch Exception around verify (or check the header’s shape first), so that a malformed header becomes a 401 rather than a 500.

Confirm the whole thing end to end

Run the scripts in order, step01_naive.py through step10_interop.py, and compare each with the output printed above. Timings and the random secrets differ from run to run; the verdicts must not. Then run python -m pytest -q and expect all tests to pass. If all of that holds, you have a receiver that rejects forged, edited, stale and duplicate events, handles hostile headers without crashing, refuses to run with a weak secret, and can rotate secrets without dropping deliveries.

Where to go next

Use your provider’s own library or test tooling when it has one, and keep this tutorial’s checklist for auditing it: raw bytes, constant-time comparison, a signed timestamp with a tolerance that is not zero, an idempotency key applied in the same transaction as the effect, and a secret that cannot be empty. GitHub’s documentation has pages on redelivering webhooks and testing with its CLI, which are the quickest way to try a real delivery against your endpoint.

Signatures are one layer. Stripe’s documentation recommends that, in addition to signature verification, you “configure your server to only accept webhook requests from Stripe’s IP addresses” (Stripe Docs, Manage webhook endpoints), and it advises returning a 2xx response quickly before slow logic runs. For slow work, store the event and a job in one transaction and process the job later, as in our outbox tutorial. If you do not control both ends of the connection, the specification recommends asymmetric signatures, so that a leak on your side cannot be used to sign events.

The same habit of treating every input as hostile runs through our other security tutorials, including preventing server-side request forgery, where your server makes the untrusted request, and verifying Cloudflare Turnstile tokens, another case where the check only counts if it happens on the server.

Sources

Tags:

API SecurityApplication SecurityFastAPIHMACPythonWebhooks

Share

No Comment! Be the first one.

Leave a Reply

Your email address will not be published. Required fields are marked *

Related Posts