Webhooks done right: verify HMAC signatures and block replays

A webhook is a request that another service sends to you: "an order was paid", "a build finished". Your endpoint must be public so they can reach it, which means anyone on the internet can POST to it too. If you trust the body blindly, an attacker can send a fake "order.paid" and get a free product. The defence is to verify a signature. The code and outputs below were run with Python 3 before publishing.

How signing works

1. You and the sender share a secret (given when you register the webhook).
2. The sender computes an HMAC-SHA256 of the request, using the secret, and sends it in a header.
3. You compute the same HMAC over what you received. If the two match, the request came from someone who knows the secret and was not changed on the way.

HMAC is a keyed hash. Without the secret, an attacker cannot produce a valid signature for a modified body.

A working example

Save as webhook.py (the sender's sign and the receiver's check use the same function):

import hashlib
import hmac
import json
import time

TOLERANCE = 300  # seconds


def sign(secret: bytes, timestamp: int, body: bytes) -> str:
    """Signature over 'timestamp.body' so the time cannot be swapped."""
    message = str(timestamp).encode() + b"." + body
    return hmac.new(secret, message, hashlib.sha256).hexdigest()


def build_headers(secret: bytes, body: bytes, now=None) -> dict:
    ts = int(now if now is not None else time.time())
    return {"X-Timestamp": str(ts), "X-Signature": "sha256=" + sign(secret, ts, body)}


class Receiver:
    def __init__(self, secret: bytes, now=time.time):
        self.secret = secret
        self.now = now
        self.seen = set()  # processed event ids (use a database in real life)
        self.processed = []

    def handle(self, headers: dict, raw_body: bytes):
        try:
            ts = int(headers["X-Timestamp"])
            received = headers["X-Signature"].removeprefix("sha256=")
        except (KeyError, ValueError):
            return 400, "missing or bad headers"

        if abs(self.now() - ts) > TOLERANCE:
            return 400, "timestamp outside tolerance"

        expected = sign(self.secret, ts, raw_body)
        if not hmac.compare_digest(expected, received):
            return 401, "bad signature"

        event = json.loads(raw_body)  # parse only AFTER verification
        if event["id"] in self.seen:
            return 200, "duplicate ignored"
        self.seen.add(event["id"])
        self.processed.append(event["type"])
        return 200, "ok"


if __name__ == "__main__":
    secret = b"whsec_demo_only_not_real"
    clock = [1_800_000_000]
    rx = Receiver(secret, now=lambda: clock[0])
    body = json.dumps({"id": "evt_1", "type": "order.paid"}).encode()
    good = build_headers(secret, body, now=clock[0])

    print("valid            :", rx.handle(good, body))
    print("same event again :", rx.handle(good, body))
    tampered = body.replace(b"order.paid", b"order.free")
    print("tampered body    :", rx.handle(good, tampered))
    print("wrong secret     :", rx.handle(build_headers(b"other", body, now=clock[0]), body))
    clock[0] += 3600
    print("replayed 1h later:", rx.handle(good, body))
    print("no headers       :", rx.handle({}, body))
    print("processed events :", rx.processed)

Output of python3 webhook.py:

valid            : (200, 'ok')
same event again : (200, 'duplicate ignored')
tampered body    : (401, 'bad signature')
wrong secret     : (401, 'bad signature')
replayed 1h later: (400, 'timestamp outside tolerance')
no headers       : (400, 'missing or bad headers')
processed events : ['order.paid']

What each case shows

• Valid: accepted and processed once.
• Same event again: providers retry, so duplicates are normal. We remember event ids and answer 200 without processing twice (the same idea as idempotency keys).
• Tampered body: one changed word, signature no longer matches, 401.
• Wrong secret: an attacker without the secret cannot sign, 401.
• Replayed an hour later: a captured, genuinely signed request is useless after the time window, 400.
• No headers: rejected before any work.

The details that matter

Verify the raw bytes. Compute the HMAC over the exact body bytes you received. If your framework parses the JSON first and you re-serialise it, spacing or key order can change and the signature will not match. Read the raw body, verify, and only then parse (see the comment in the code).

Constant-time comparison. Use hmac.compare_digest, not ==. A normal comparison can stop at the first different character, and timing differences can leak information about the correct signature.

Sign the timestamp too. We sign timestamp.body. If the timestamp were outside the signature, an attacker could replay an old request with a fresh timestamp. The test test_swapped_timestamp proves that changing it breaks the signature.

Timestamp tolerance. Reject requests older (or newer) than a few minutes. This limits how long a captured request can be replayed. Your server clock must be reasonably accurate for this.

Deduplicate by event id. A valid signature proves who sent it, not that it is the first time. Store processed ids (in a database, with a unique constraint) and ignore repeats. Our in-memory set is only for the demo.

Keep the secret secret. Store it in an environment variable or secrets manager, never in the repository. Support rotation: during a change, accept signatures from the old and the new secret for a short period.

Tests

Save as test_webhook.py:

import json
import unittest

from webhook import Receiver, build_headers

SECRET = b"s3cret"
NOW = 1_800_000_000


def make(event_id="e1"):
    body = json.dumps({"id": event_id, "type": "t"}).encode()
    return body, build_headers(SECRET, body, now=NOW)


class WebhookTests(unittest.TestCase):
    def setUp(self):
        self.rx = Receiver(SECRET, now=lambda: NOW)

    def test_valid(self):
        body, h = make()
        self.assertEqual(self.rx.handle(h, body)[0], 200)
        self.assertEqual(self.rx.processed, ["t"])

    def test_tampered_body(self):
        body, h = make()
        self.assertEqual(self.rx.handle(h, body + b" ")[0], 401)
        self.assertEqual(self.rx.processed, [])

    def test_swapped_timestamp(self):
        body, h = make()
        h["X-Timestamp"] = str(NOW + 10)
        self.assertEqual(self.rx.handle(h, body)[0], 401)

    def test_old_timestamp(self):
        body = b'{"id":"e","type":"t"}'
        h = build_headers(SECRET, body, now=NOW - 1000)
        self.assertEqual(self.rx.handle(h, body)[0], 400)

    def test_duplicate_event(self):
        body, h = make()
        self.rx.handle(h, body)
        self.assertEqual(self.rx.handle(h, body), (200, "duplicate ignored"))
        self.assertEqual(self.rx.processed, ["t"])


if __name__ == "__main__":
    unittest.main()

$ python3 -m unittest -v
Ran 5 tests in 0.002s

OK

Receiving webhooks in practice

• Respond fast. Verify, save the event, return 2xx, and do the slow work in a background job. Senders often time out after a few seconds and retry.
• Use the right status codes. 401 or 400 for bad requests. A 5xx tells the sender to retry later, which is what you want for your own temporary failures.
• Handle out-of-order events. Delivery order is usually not guaranteed. Check the object's current state instead of assuming "created" arrives before "updated".
• Log failures (without logging the secret or full signature) so you can debug.
• Use HTTPS so the body is not readable in transit. The signature proves integrity and origin, not secrecy.

Common mistakes

• Skipping verification "because the URL is secret". URLs leak in logs and tools.
• Comparing signatures with ==.
• Verifying a re-serialised body instead of the raw one.
• Not checking a timestamp, which allows replay.
• Processing before verifying, or parsing untrusted JSON before the check.
• Doing slow work before replying, which causes timeouts and duplicate deliveries.

Limits of this example

Each provider has its own header names and signing format (some sign only the body, some include the URL, some use different encodings). Always follow the provider's documentation and, where available, use their official verification library. This example shows the general pattern, not any specific provider's exact scheme.

Main idea: treat every webhook as untrusted until its HMAC signature, computed over the raw body and timestamp, matches in constant time; then check the timestamp window and deduplicate by event id.

---

Backend series, Part 3 (reliable APIs): 2 of 5. Previous: Idempotency keys: make API retries safe (so nobody is charged twice).