Case Study: Freeze Webhook Skew and Replay Rules Before an Agent Writes HMAC Verification

You should freeze clock-skew, nonce replay, and signature-base encoding before a coding agent writes HMAC webhook verification. Those three choices decide whether a delayed retry is valid or whether an attacker can replay a captured body. If you leave them implicit, the generated handler will look correct while accepting the wrong events. This case study walks through a small inbound webhook service so you can copy the freeze-first workflow.

Background: one inbound endpoint, three silent forks

Imagine you are adding POST /webhooks/provider to a tiny order-status service that already stores shipments. The provider signs each body with HMAC-SHA256 and sends a Unix timestamp plus a hex digest. Your agent can produce a verifier in one pass, and the happy-path test will pass. The trouble starts when the provider retries after four minutes, or when two workers see the same event.

Most generated handlers pick defaults you never discussed in the prompt. Some parse the timestamp as milliseconds, which rejects every real event from the provider. Others skip replay storage because the sample request used a unique body. You need those choices written down before any model emits a route.

The project in this walkthrough stays deliberately small and local. You will freeze a decision table, encode it as failing tests, and only then allow an agent to fill verify_webhook.py. You will not start from a prompt that says "implement Stripe-like webhooks" and hope library defaults match your provider.

Goal: accept late retries, refuse replays, never guess encoding

You want three outcomes that look similar in logs and become expensive when mixed. A request with a valid signature and a timestamp inside the skew window must reach the business handler exactly once. A second request with the same signature must return success without shipping the order again. A request outside the skew window, or with a mutated body, must fail closed before any database write.

Those outcomes are not coding style preferences you can bikeshed later. They are product rules about inventory movement, customer email, and provider retry storms. An agent that invents a thirty-second skew will drop the provider's legitimate five-minute retry. An agent that stores only an event id from parsed JSON will miss a replay that changes insignificant whitespace.

Label the following constants as case-study choices, not as universal vendor law. You can change the numbers, but you must change the tests in the same commit. If a reviewer cannot find the number in both places, the rule is not frozen.

Frozen rules for this case study

• Timestamp header X-Webhook-Timestamp is Unix seconds, UTC, and an integer with no fraction.
• Signature header X-Webhook-Signature is lowercase hex HMAC-SHA256 of the signed payload.
• Signed payload is f"{timestamp}.{raw_body}" using the exact request bytes and no extra newline.
• Compare digests with hmac.compare_digest, and reject closed on any parse error.
• Skew window is ±300 seconds against the verifier clock, which you treat as UTC.
• Replay key is SHA-256 of the signature hex, and the cache keeps it for 600 seconds.
• First valid event returns HTTP 204 after side effects; a duplicate valid event returns HTTP 204 with no side effects.
• Invalid signature, skew, or malformed headers return HTTP 401 and trigger no side effects.

Decision table you freeze before generation

Paste this table into the repository next to the tests, not into a slide deck. Agents follow a nearby table more reliably than a paragraph buried in a README. If a later prompt asks for "more resilient retries," you update the table first so the new behavior is explicit.

Case	Timestamp relative to verifier	Signature	Already in replay cache	HTTP	Side effects
A	+0s	Valid hex HMAC	No	204	Yes, once
B	+0s	Valid hex HMAC	Yes	204	No
C	+240s (inside ±300)	Valid hex HMAC	No	204	Yes, once
D	+301s (outside window)	Valid hex HMAC	No	401	No
E	+0s	Body mutated after signing	No	401	No
F	missing / not an integer	Valid hex HMAC	No	401	No
G	+0s	Header missing or not hex	No	401	No
H	−120s (inside window)	Valid hex HMAC	No	204	Yes, once

Case B is the one agents most often get wrong. They return 409, which makes some providers retry forever, or they return 401, which hides that the signature was actually fine. You freeze 204-without-side-effects so the provider stops and your shipment table stays unique.

Failing tests as the only contract the agent may satisfy

Create a folder that contains rules and tests, not an implementation yet. The commands below assume Python 3.12 and pytest, and they are a labeled walkthrough you run locally. They are not a claim about production traffic on anyone's cluster.

python -m venv .venv
source .venv/bin/activate
pip install pytest
mkdir -p webhook_case && cd webhook_case

# test_verify_webhook.py
# Example walkthrough tests. Adjust names if your provider uses different headers.
from __future__ import annotations

import hashlib
import hmac
import time

import pytest

from verify_webhook import ReplayCache, VerifyResult, verify_webhook

SECRET = b"case-study-secret"
SKEW = 300


def sign(ts: int, body: bytes) -> str:
    payload = f"{ts}.".encode("ascii") + body
    digest = hmac.new(SECRET, payload, hashlib.sha256).hexdigest()
    return digest


@pytest.fixture
def cache() -> ReplayCache:
    return ReplayCache(ttl_seconds=600)


def test_case_a_first_valid_event_runs_side_effects(cache: ReplayCache) -> None:
    ts = int(time.time())
    body = b'{"order_id":"ord_1","status":"shipped"}'
    result = verify_webhook(
        timestamp_header=str(ts),
        signature_header=sign(ts, body),
        raw_body=body,
        secret=SECRET,
        now=ts,
        cache=cache,
        skew_seconds=SKEW,
    )
    assert result == VerifyResult(status=204, apply_side_effects=True)


def test_case_b_duplicate_signature_is_ack_without_side_effects(cache: ReplayCache) -> None:
    ts = int(time.time())
    body = b'{"order_id":"ord_1","status":"shipped"}'
    sig = sign(ts, body)
    first = verify_webhook(str(ts), sig, body, SECRET, ts, cache, SKEW)
    second = verify_webhook(str(ts), sig, body, SECRET, ts, cache, SKEW)
    assert first.apply_side_effects is True
    assert second == VerifyResult(status=204, apply_side_effects=False)


def test_case_d_outside_skew_fails_closed(cache: ReplayCache) -> None:
    now = int(time.time())
    ts = now - 301
    body = b'{"order_id":"ord_2"}'
    result = verify_webhook(str(ts), sign(ts, body), body, SECRET, now, cache, SKEW)
    assert result == VerifyResult(status=401, apply_side_effects=False)


def test_case_e_mutated_body_fails_closed(cache: ReplayCache) -> None:
    ts = int(time.time())
    signed = b'{"order_id":"ord_3"}'
    received = b'{"order_id":"ord_3","status":"cancelled"}'
    result = verify_webhook(str(ts), sign(ts, signed), received, SECRET, ts, cache, SKEW)
    assert result == VerifyResult(status=401, apply_side_effects=False)

Notice that the tests never import a framework router. You are freezing crypto and replay behavior, not Flask versus FastAPI. After the tests fail for a missing verify_webhook module, you may let an agent write that module and nothing else.

Run the contract before any generated file exists:

pytest -q test_verify_webhook.py
# Expected on a clean tree: collection error or import error for verify_webhook

Implementation the agent is allowed to write

Keep the allowed surface tiny so the model cannot "helpfully" invent headers. The agent may create verify_webhook.py and a memory-backed ReplayCache. It may not invent a nonce header your provider does not send. It may not switch the signed payload to canonical JSON, because that would invalidate every existing test without a table change.

# verify_webhook.py
# Example implementation that satisfies the frozen table. Review before production use.
from __future__ import annotations

from dataclasses import dataclass
import hashlib
import hmac
import time
from typing import Dict


@dataclass(frozen=True)
class VerifyResult:
    status: int
    apply_side_effects: bool


class ReplayCache:
    def __init__(self, ttl_seconds: int) -> None:
        self.ttl_seconds = ttl_seconds
        self._entries: Dict[str, int] = {}

    def seen(self, key: str, now: int) -> bool:
        expires_at = self._entries.get(key)
        if expires_at is None:
            return False
        if expires_at <= now:
            del self._entries[key]
            return False
        return True

    def remember(self, key: str, now: int) -> None:
        self._entries[key] = now + self.ttl_seconds


def verify_webhook(
    timestamp_header: str,
    signature_header: str,
    raw_body: bytes,
    secret: bytes,
    now: int,
    cache: ReplayCache,
    skew_seconds: int,
) -> VerifyResult:
    try:
        timestamp = int(timestamp_header)
    except (TypeError, ValueError):
        return VerifyResult(status=401, apply_side_effects=False)

    if abs(now - timestamp) > skew_seconds:
        return VerifyResult(status=401, apply_side_effects=False)

    if not signature_header or any(c not in "0123456789abcdef" for c in signature_header):
        return VerifyResult(status=401, apply_side_effects=False)

    payload = f"{timestamp}.".encode("ascii") + raw_body
    expected = hmac.new(secret, payload, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, signature_header):
        return VerifyResult(status=401, apply_side_effects=False)

    replay_key = hashlib.sha256(signature_header.encode("ascii")).hexdigest()
    if cache.seen(replay_key, now):
        return VerifyResult(status=204, apply_side_effects=False)

    cache.remember(replay_key, now)
    return VerifyResult(status=204, apply_side_effects=True)

Run the same pytest command until it is green. If the agent loosens skew to one day "to be safe," the table and the tests should make that diff obvious in review. You are not asking the model for architecture opinions until the eight cases above stay red or green for the right reasons.

A useful extra check is a property that the signed payload uses raw bytes. Add it if your provider might send UTF-8 bodies with non-ASCII characters.

def test_raw_bytes_are_signed_not_redecoded_json(cache: ReplayCache) -> None:
    ts = int(time.time())
    body = b'{"note":"caf\xc3\xa9"}'  # UTF-8 é, not a Python str round-trip
    result = verify_webhook(str(ts), sign(ts, body), body, SECRET, ts, cache, SKEW)
    assert result.apply_side_effects is True

Where a disposable coding loop helps

When you iterate on agent-written crypto, you want a throwaway shell rather than your laptop's production credentials. Disclosure: This article was prepared as part of MonkeyCode's product outreach. If you already have a MonkeyCode workspace, its free model access and free server option can host this freeze-test-generate loop while you keep secrets off the prompt. The workflow itself does not depend on that product; any isolated runner with pytest is enough.

Results of this walkthrough

After the tests pass, you have a verifier whose retry and replay behavior is reviewable in one table. You did not measure production traffic here, and you should not treat the ±300 second window as a published benchmark. What you can reproduce is local: the eight pytest cases covering skew, replay, malformed headers, and encoding.

If you later connect a real provider, you add a captured fixture with that vendor's actual header names and test vectors. You do not ask an agent to "make it compatible" without a new failing row in the table. Compatibility without a fixture is just a guess with extra confidence.

A second result is process, not code. Reviewers can reject a pull request that changes Case B to 409 without touching the table. That is cheaper than debugging duplicate shipments after a weekend retry storm.

Limitations: who should not copy this blindly

• Do not treat this snippet as a PCI, banking, or healthcare auditor's control.
• Do not replace Stripe, GitHub, Slack, or similar official verification SDKs with this toy HMAC.
• Do not run a process-local memory cache across several workers and call the replay window durable.
• Do not let an agent add ordinary == on hex strings because it looks more readable in a tutorial.
• Skip this workflow if your provider already documents a required library, header format, and test vectors you must use unchanged.
• Skip it if you need millisecond timestamps, multiple rotating secrets, or signature version prefixes; freeze those as new table rows first.

The example also uses time.time() in tests that inject now. That is safe here because production code receives now as an argument. If you later read the clock inside the function, you will make skew tests flaky, and an agent will "fix" them by widening the window.

Lessons learned

1. Signature encoding is a product rule. Timestamp units, hex versus base64, and whether a newline is included will not be inferred reliably from a prose prompt.
2. Replay is not the same as business idempotency. You can ack a duplicate HMAC and still need a unique constraint on order_id plus event type.
3. HTTP status on duplicates trains the provider. 204 without side effects stops retries; 409 often extends them; 401 conceals that the signature was valid.
4. Freeze the table before the prompt. The agent is allowed to write verify_webhook.py only after pytest already describes the forks you care about.

You can reuse this shape on the next agent-written endpoint. Write the silent forks as rows, turn the rows into failing tests, and keep the generated surface small enough that a loosened default shows up as a red case instead of a quiet production incident.