Tests inbound webhook receivers for replay-attack resistance: capture incoming webhook payloads + headers, replay against the receiver under test, validate the Standard Webhooks signature scheme (svix-id + svix-timestamp + svix-signature, HMAC-SHA256 over `{id}.{timestamp}.{payload}`), svix-id idempotency dedup, and 5-minute timestamp-window enforcement by signing fixtures at runtime. Does NOT cover outbound delivery, retry-on-5xx, or failure-event exhaustion - those belong to an outbound webhook delivery harness. Use when testing the receiving side of a webhook integration.
74
93%
Does it follow best practices?
Run evals on this skill
Adds up to 20 points to the overall score
View guide
High
Do not use without reviewing
Per the Standard Webhooks spec, "every webhook implementation needs to protect themselves and their users from SSRF, spoofing, and replay attacks." Replay tests verify the receiver enforces the signature, the timestamp window, and idempotency-key dedup.
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 helpersPer the Standard Webhooks spec, required headers:
| Header | Meaning |
|---|---|
svix-id | Unique webhook identifier |
svix-timestamp | Unix timestamp (seconds) |
svix-signature | v1,<base64-hmac-sha256> (one or more, space-separated) |
Signature input: HMAC-SHA256 over {id}.{timestamp}.{payload} with
the shared secret as key.
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 == 200Per the Standard Webhooks spec, a 5-minute window prevents replay:
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)Per the Standard Webhooks spec, "idempotency-key behavior" handles
duplicates gracefully. The receiver should treat a second delivery
of the same svix-id as a no-op (or a 200 with no side effects).
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_secondPer the Standard Webhooks spec, svix-signature can carry
multiple versions/keys (space-separated v1,...). Receiver
accepts if any key validates.
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 == 200For captured payloads, sanitize before committing:
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)| Anti-pattern | Why it fails | Fix |
|---|---|---|
| Skip signature test in dev (mock the verifier) | Prod-only signature bug ships | Use the same verifier in test as prod (Step 4) |
| Hard-code timestamps in fixtures | Old fixtures fail Step 5 windowed-replay protection | Sign at test runtime (Step 3) |
| No idempotency test | Duplicate webhook deliveries cause double-charges | Step 7 mandatory |
| Commit raw production payloads | PII leak in repo | Sanitize before commit (Step 9) |
| Use single key, no rotation path | Forced re-signing at rotation; outage risk | Multi-key acceptance (Step 8) |
webhook-delivery-tester for delivery
webhook-delivery-tester (in the qa-notifications plugin) -
sister skill for delivery + retry on the sender sidemqtt-tests,
websocket-tests - alternative
realtime transports