tests/webhook-replay/
├── fixtures/
│   ├── stripe-charge-succeeded.json    # full request body
│   ├── stripe-charge-succeeded.headers.json   # incl. svix-* headers
│   └── github-pr-opened.json
├── replay.py                            # replay loop
└── conftest.py                          # signing helpers

import hmac, hashlib, base64, time, json

def sign_webhook(secret_b64: str, msg_id: str, payload: bytes,
                  timestamp: int | None = None) -> dict[str, str]:
    timestamp = timestamp or int(time.time())
    secret = base64.b64decode(secret_b64.removeprefix("whsec_"))
    signed_payload = f"{msg_id}.{timestamp}.".encode() + payload
    sig = base64.b64encode(hmac.new(secret, signed_payload, hashlib.sha256).digest()).decode()
    return {
        "svix-id": msg_id,
        "svix-timestamp": str(timestamp),
        "svix-signature": f"v1,{sig}",
        "Content-Type": "application/json",
    }

import requests
from pathlib import Path

def test_valid_signed_webhook_accepted():
    payload = Path("fixtures/stripe-charge-succeeded.json").read_bytes()
    headers = sign_webhook(
        secret_b64="whsec_<test-secret>",
        msg_id="msg_test_1",
        payload=payload,
    )
    resp = requests.post("http://localhost:8080/webhooks/stripe",
                          data=payload, headers=headers)
    assert resp.status_code == 200

def test_old_timestamp_rejected():
    payload = b'{"event":"x"}'
    old_ts = int(time.time()) - 600  # 10 min ago, > 5 min window
    headers = sign_webhook(
        secret_b64="whsec_<test-secret>",
        msg_id="msg_test_old",
        payload=payload,
        timestamp=old_ts,
    )
    resp = requests.post("http://localhost:8080/webhooks/stripe",
                          data=payload, headers=headers)
    assert resp.status_code in (400, 401)

def test_future_timestamp_rejected():
    payload = b'{"event":"x"}'
    future_ts = int(time.time()) + 600
    headers = sign_webhook("whsec_<test-secret>", "msg_test_future",
                            payload, timestamp=future_ts)
    resp = requests.post("http://localhost:8080/webhooks/stripe",
                          data=payload, headers=headers)
    assert resp.status_code in (400, 401)

def test_tampered_payload_rejected():
    payload = b'{"amount":100}'
    headers = sign_webhook("whsec_<test-secret>", "msg_test_tamper", payload)

    # Tamper after signing
    tampered = b'{"amount":1000000}'
    resp = requests.post("http://localhost:8080/webhooks/stripe",
                          data=tampered, headers=headers)
    assert resp.status_code in (400, 401)

def test_duplicate_svix_id_does_not_duplicate_side_effect():
    payload = Path("fixtures/stripe-charge-succeeded.json").read_bytes()
    msg_id = "msg_dup_test_1"
    headers = sign_webhook("whsec_<test-secret>", msg_id, payload)

    # Fresh state
    reset_orders_table()

    # Deliver once
    r1 = requests.post("http://localhost:8080/webhooks/stripe",
                       data=payload, headers=headers)
    assert r1.status_code == 200
    orders_after_first = count_orders()

    # Deliver again with same svix-id
    r2 = requests.post("http://localhost:8080/webhooks/stripe",
                       data=payload, headers=headers)
    assert r2.status_code == 200  # accept, no-op
    orders_after_second = count_orders()

    assert orders_after_first == orders_after_second

def test_accepts_during_key_rotation():
    payload = b'{"event":"x"}'
    msg_id = "msg_rotate_1"
    ts = int(time.time())

    sig_old = compute_sig(secret_b64="whsec_OLD", msg_id=msg_id,
                            payload=payload, timestamp=ts)
    sig_new = compute_sig(secret_b64="whsec_NEW", msg_id=msg_id,
                            payload=payload, timestamp=ts)

    headers = {
        "svix-id": msg_id,
        "svix-timestamp": str(ts),
        "svix-signature": f"v1,{sig_old} v1,{sig_new}",
        "Content-Type": "application/json",
    }
    resp = requests.post("http://localhost:8080/webhooks/stripe",
                          data=payload, headers=headers)
    assert resp.status_code == 200

def sanitize_capture(payload: dict) -> dict:
    SENSITIVE_KEYS = {"email", "phone", "ssn", "card", "address"}
    def walk(node):
        if isinstance(node, dict):
            return {k: ("***" if k.lower() in SENSITIVE_KEYS else walk(v))
                     for k, v in node.items()}
        if isinstance(node, list):
            return [walk(x) for x in node]
        return node
    return walk(payload)