Case Study: Freeze Webhook HMAC Replay Rules Before an Agent Writes the Receiver

You should freeze webhook HMAC replay rules in tests before any coding agent writes the HTTP receiver. Agents invent header names, skip timestamp windows, and accept unsigned bodies when no contract is already failing. This case study walks one small webhook project from a frozen test file through a generated handler. You can keep the same tests on your laptop even if you never use a hosted coding assistant.

Background

Webhook providers retry deliveries, replay delayed events, and sometimes send the same POST long after success. A receiver that only checks a shared secret still accepts captured requests outside the allowed time window. Coding agents trained on mixed tutorials often hash re-serialized JSON instead of the raw request body bytes. They also rename headers to Signature or X-Hub-Signature-256 because those names dominate public code samples.

You need a contract that names the headers, the canonical bytes, the hash algorithm, and allowed clock skew. Without that file, an agent writes something that looks secure in review and still fails under replay. This project stays small on purpose so you can read every failing assertion before a handler exists. The work is a teaching slice, not a billing platform and not a vendor SDK rewrite.

Goal

The goal is a tiny FastAPI receiver that verifies HMAC-SHA256 signatures and rejects stale or future timestamps. You will not ask an agent for handler code until pytest already encodes header names, canonical bytes, and skew. A public callback only helps provider sandboxes; it is not required for the contract tests themselves. Local pytest remains the source of truth even when a remote process later serves the same application module.

Success for this case study means four checks, and none of them is a generated README:

1. Missing or malformed signature headers fail closed with HTTP 401.
2. Timestamps older or newer than 300 seconds fail closed with HTTP 401.
3. Canonical bytes are the ascii timestamp, one dot, and the raw body, never re-serialized JSON.
4. A valid signature with a fresh timestamp returns HTTP 204 and does not parse the body twice.

The frozen contract

Treat the decision table as the only protocol the agent is allowed to implement in this exercise. If a generated handler invents extra headers or a different payload encoding, you discard that draft immediately. Use a fictional vendor prefix so you do not copy a live provider document into the repository. Pin max_skew_seconds to 300 inside the test module rather than burying the number inside a chat prompt.

Case	X-Acme-Timestamp	X-Acme-Signature	Body	Clock vs timestamp	Status
valid	1700000000	sha256= plus matching hex	{"ok":true}	0s	204
missing signature	1700000000	absent	{"ok":true}	0s	401
wrong prefix	1700000000	hex without sha256=	{"ok":true}	0s	401
replay	1700000000	valid for that timestamp	{"ok":true}	+301s	401
future	1700000301	valid for that timestamp	{"ok":true}	-301s	401
mutated body	1700000000	signature for the original body	{"ok":false}	0s	401

Canonical bytes

You must authenticate exactly one byte string: ascii timestamp, one ASCII dot, then the untouched raw body. Prefer hashing timestamp.encode("ascii") + b"." + body, and never call json.dumps before computing the digest. Agents pretty-print JSON by default, and pretty-printed JSON will never match what the provider actually signed. Compare digests with hmac.compare_digest so invalid signatures do not become cheap timing oracles.

Implementation

Write the tests before the route

The pytest module below is the original artifact for this case study, not a sketch you should paraphrase later. Save it as tests/test_webhook_replay.py and run it before app.py exists so collection failures stay honest. Add a tiny stub that raises NotImplementedError if you want collection to succeed while every assertion still fails. That stub is cheaper than asking an agent to invent both the tests and the implementation in one pass.

# tests/test_webhook_replay.py
# Proposed example: pin replay rules before any agent writes the handler.
from __future__ import annotations

import hashlib
import hmac

import pytest
from fastapi.testclient import TestClient

from webhook_app import MAX_SKEW_SECONDS, app

SECRET = b"test-secret-not-for-production"
BODY = b'{"ok":true}'
TS = "1700000000"


def sign(timestamp: str, body: bytes) -> str:
    mac = hmac.new(
        SECRET,
        timestamp.encode("ascii") + b"." + body,
        hashlib.sha256,
    ).hexdigest()
    return f"sha256={mac}"


@pytest.fixture
def client(monkeypatch: pytest.MonkeyPatch) -> TestClient:
    monkeypatch.setenv("WEBHOOK_SECRET", SECRET.decode("ascii"))
    return TestClient(app)


def test_valid_signature_returns_204(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr("webhook_app.now_seconds", lambda: 1_700_000_000)
    response = client.post(
        "/webhooks/acme",
        content=BODY,
        headers={
            "Content-Type": "application/json",
            "X-Acme-Timestamp": TS,
            "X-Acme-Signature": sign(TS, BODY),
        },
    )
    assert response.status_code == 204
    assert response.content == b""


def test_missing_signature_header_fails_closed(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr("webhook_app.now_seconds", lambda: 1_700_000_000)
    response = client.post(
        "/webhooks/acme",
        content=BODY,
        headers={"X-Acme-Timestamp": TS},
    )
    assert response.status_code == 401


@pytest.mark.parametrize(
    "skew", [MAX_SKEW_SECONDS + 1, -(MAX_SKEW_SECONDS + 1)]
)
def test_timestamp_outside_skew_fails_closed(
    client: TestClient, monkeypatch: pytest.MonkeyPatch, skew: int
) -> None:
    monkeypatch.setattr("webhook_app.now_seconds", lambda: 1_700_000_000 + skew)
    response = client.post(
        "/webhooks/acme",
        content=BODY,
        headers={
            "X-Acme-Timestamp": TS,
            "X-Acme-Signature": sign(TS, BODY),
        },
    )
    assert response.status_code == 401


def test_reserialized_body_does_not_match_signature(
    client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setattr("webhook_app.now_seconds", lambda: 1_700_000_000)
    mutated = b'{"ok": false}'
    response = client.post(
        "/webhooks/acme",
        content=mutated,
        headers={
            "X-Acme-Timestamp": TS,
            "X-Acme-Signature": sign(TS, BODY),
        },
    )
    assert response.status_code == 401

Generate only against failing tests

When you want a model to draft webhook_app.py, keep the prompt narrow and point it at the failing tests. Disclosure: This article was prepared as part of MonkeyCode's product outreach. MonkeyCode provides free model access and a free server option you can use after the contract file already exists. Free model access is enough to draft the verify path; the free server option only matters when a provider sandbox must reach a public callback. Do not paste the production secret into any prompt; inject WEBHOOK_SECRET from the environment at process start.

Proposed local commands, labeled as an unexecuted example:

python -m venv .venv
. .venv/bin/activate
pip install fastapi httpx pytest uvicorn
export WEBHOOK_SECRET='test-secret-not-for-production'
pytest -q tests/test_webhook_replay.py

If pytest cannot import webhook_app, add the stub below and keep every test red until the real module lands. You want import success with assertion failure, not a green suite that never executed HMAC code.

# webhook_app.py — stub so pytest can collect before generation
from fastapi import FastAPI

MAX_SKEW_SECONDS = 300
app = FastAPI()

def now_seconds() -> int:
    raise NotImplementedError

A handler that should pass the frozen table

The FastAPI module below is a proposed example, not production-hardened code. Review the compare_digest call, the raw body path, and the integer timestamp parse before you expose it. Reject non-digit timestamps before HMAC so the handler does not raise five-hundreds on garbage headers.

# webhook_app.py — proposed example that matches tests/test_webhook_replay.py
from __future__ import annotations

import hashlib
import hmac
import os
import time

from fastapi import FastAPI, Header, HTTPException, Request, Response

MAX_SKEW_SECONDS = 300
app = FastAPI()


def now_seconds() -> int:
    return int(time.time())


def _secret() -> bytes:
    value = os.environ.get("WEBHOOK_SECRET", "")
    if not value:
        raise RuntimeError("WEBHOOK_SECRET is missing")
    return value.encode("ascii")


@app.post("/webhooks/acme")
async def acme_webhook(
    request: Request,
    x_acme_timestamp: str | None = Header(default=None),
    x_acme_signature: str | None = Header(default=None),
) -> Response:
    if not x_acme_timestamp or not x_acme_timestamp.isdigit():
        raise HTTPException(status_code=401, detail="invalid timestamp")
    if not x_acme_signature or not x_acme_signature.startswith("sha256="):
        raise HTTPException(status_code=401, detail="invalid signature")

    skew = abs(now_seconds() - int(x_acme_timestamp))
    if skew > MAX_SKEW_SECONDS:
        raise HTTPException(status_code=401, detail="replay window")

    body = await request.body()
    expected = "sha256=" + hmac.new(
        _secret(),
        x_acme_timestamp.encode("ascii") + b"." + body,
        hashlib.sha256,
    ).hexdigest()
    if not hmac.compare_digest(expected, x_acme_signature):
        raise HTTPException(status_code=401, detail="bad mac")

    return Response(status_code=204)

Results

This section does not report production metrics, incident rates, or latency numbers from a live webhook fleet. It records the failure classes the frozen suite is designed to catch when an agent ignores the table. Run the proposed session on the example handler after generation, and treat any extra passing tests as untrusted. Extra tests the agent adds often restate the implementation instead of protecting the canonical byte string.

Expected outcomes when the contract is frozen first:

1. An agent that HMAC-signs json.dumps(payload) fails test_reserialized_body_does_not_match_signature.
2. An agent that reads X-Hub-Signature-256 fails test_missing_signature_header_fails_closed and the valid case together.
3. An agent that skips timestamp checks fails test_timestamp_outside_skew_fails_closed for both directions.
4. An agent that returns 200 with a JSON error body fails the 401 and 204 status assertions.

That is the practical result of freezing replay rules: you see the invented protocol in red pytest output instead of in a missed incident. Keep the first green run in CI so later refactors cannot quietly switch back to parsed JSON.

Limitations and who should not use this

This approach is a teaching contract for one HMAC header pair and a fixed skew window. It is not a full webhook platform, and it will not replace provider SDKs that already document a different canonical form. Clock skew tests assume one injected now_seconds fixture, so they will not catch a handler that samples the clock twice around a slow body read. Rotate secrets, cap body size, and put network authentication in front of any public callback before you handle money or personal data.

Do not use this pattern when any of the following is true:

• You terminate TLS at a proxy that rewrites bodies or collapses duplicate headers.
• Your provider signs a different canonical form, such as a hex timestamp or an ordered header list.
• You still need idempotency keys, at-least-once deduplication, or ordered event processing.
• You cannot keep the shared secret out of prompts, logs, and the git repository.

Lessons learned

Freeze the replay window in pytest before you generate a single route, including the stub that only exists so collection works. Header names, canonical bytes, and skew belong in assertions, not in chat history that an agent can quietly rewrite. A public callback is optional infrastructure after the contract exists, never a substitute for fail-closed tests. If you skip the test file, the agent will still write a handler that looks careful and still accepts a replayed POST.