Verify Signed Webhooks
The payment provider posts signed webhooks for every invoice event, and the billing service charges, emails, and writes the ledger on each one. The README has the provider's contract and the operational stakes. Build the receiver with the assistant, and judge each suggestion by what it would do in production as well as by the tests.
webhooks.py1 of 3
Files
webhooks.py
Python · Editableimport hashlib
import hmac
import json
class InvalidWebhook(Exception):
"""The request isn't a genuine, fresh delivery. Answer 400."""
def sign(secret, timestamp, body):
"""The provider's signature: hex HMAC-SHA256 of the timestamp, a dot, and the raw body."""
raise NotImplementedError
class WebhookReceiver:
"""Verifies the payment provider's webhooks and hands each event to handler once."""
def __init__(self, secret, handler, clock, tolerance_s=300):
self.secret = secret # bytes
self.handler = handler
self.clock = clock
self.tolerance_s = tolerance_s
self.seen = set() # ids of events already handled
def receive(self, headers, body):
"""Returns "handled" or "duplicate". Raises InvalidWebhook for a delivery to reject."""
timestamp = headers.get("Webhook-Timestamp", "")
signature = headers.get("Webhook-Signature", "")
expected = sign(self.secret, timestamp, body)
event = json.loads(body)
self.handler(event)
return "handled"
def respond(receiver, headers, body):
"""The HTTP status to answer a delivery with."""
raise NotImplementedError
import json
import pytest
from webhooks import InvalidWebhook, WebhookReceiver, respond, sign
SECRET = b"whsec_test"
NOW = 1_700_000_000
class FakeClock:
def __init__(self):
self.now = NOW
def __call__(self):
return self.now
def make():
handled = []
return WebhookReceiver(SECRET, handled.append, FakeClock()), handled
def delivery(event_id="evt_1", sent_at=NOW, secret=SECRET):
body = json.dumps({"id": event_id, "type": "invoice.paid"}).encode()
timestamp = str(sent_at)
return {"Webhook-Timestamp": timestamp, "Webhook-Signature": sign(secret, timestamp, body)}, body
def test_sign_matches_the_provider():
assert sign(SECRET, "1700000000", b"{}") == "35495024f4ef3f94e5a93e22221544c4b75e9a42300cd965ab81cb85cd994e91"
def test_genuine_delivery_is_handled():
receiver, handled = make()
assert receiver.receive(*delivery()) == "handled"
assert handled == [{"id": "evt_1", "type": "invoice.paid"}]
def test_rejects_a_bad_signature():
receiver, _ = make()
with pytest.raises(InvalidWebhook):
receiver.receive(*delivery(secret=b"guessed"))
def test_rejects_a_tampered_body():
receiver, _ = make()
headers, _ = delivery()
with pytest.raises(InvalidWebhook):
receiver.receive(headers, b'{"id": "evt_1", "type": "invoice.refunded"}')
def test_rejects_a_stale_timestamp():
receiver, _ = make()
with pytest.raises(InvalidWebhook):
receiver.receive(*delivery(sent_at=NOW - 301))
def test_rejects_a_future_timestamp():
receiver, _ = make()
with pytest.raises(InvalidWebhook):
receiver.receive(*delivery(sent_at=NOW + 301))
def test_accepts_clock_skew_inside_the_window():
receiver, _ = make()
assert receiver.receive(*delivery("evt_1", sent_at=NOW + 299)) == "handled"
assert receiver.receive(*delivery("evt_2", sent_at=NOW - 299)) == "handled"
def test_rejects_a_malformed_timestamp():
receiver, _ = make()
body = b'{"id": "evt_1", "type": "invoice.paid"}'
headers = {"Webhook-Timestamp": "soon", "Webhook-Signature": sign(SECRET, "soon", body)}
with pytest.raises(InvalidWebhook):
receiver.receive(headers, body)
def test_duplicate_runs_the_handler_once():
receiver, handled = make()
headers, body = delivery()
assert receiver.receive(headers, body) == "handled"
assert receiver.receive(headers, body) == "duplicate"
assert len(handled) == 1
def test_failed_handler_lets_the_retry_through():
calls = []
def flaky(event):
calls.append(event["id"])
if len(calls) == 1:
raise RuntimeError("ledger unavailable")
receiver = WebhookReceiver(SECRET, flaky, FakeClock())
headers, body = delivery()
with pytest.raises(RuntimeError):
receiver.receive(headers, body)
assert receiver.receive(headers, body) == "handled"
assert calls == ["evt_1", "evt_1"]
def test_respond_maps_each_outcome_to_a_status():
receiver, _ = make()
headers, body = delivery()
assert respond(receiver, headers, body) == 200
assert respond(receiver, headers, body) == 200
assert respond(receiver, *delivery("evt_2", secret=b"guessed")) == 400
assert respond(receiver, *delivery("evt_3", sent_at=NOW - 3600)) == 400
# Payment Webhooks
The payment provider posts an event whenever an invoice changes. WebhookReceiver (webhooks.py) verifies each delivery and hands the event to the billing handler, which charges cards, sends receipts, and writes the ledger.
## The Provider's Contract
- Webhook-Timestamp is the Unix time the delivery was signed. Webhook-Signature is the hex HMAC-SHA256 of the timestamp, a dot, and the raw body, keyed with our shared secret.
- Delivery is at least once. A delivery that doesn't get a 2xx is retried with the same event id for up to three days, and an endpoint that keeps failing is disabled.
## What to Build
- Reject a bad signature, and compare signatures in constant time.
- Reject a timestamp more than tolerance_s seconds from our clock in either direction, and one that isn't an integer.
- Run the handler once per event id. Record the id only after the handler succeeds, so the provider retries a failed delivery.
- respond answers 400 for an invalid delivery and 200 for a handled or duplicate one. Handler errors propagate, so the framework answers 500 and the provider retries.
## Operational Stakes
- tolerance_s trades replay exposure against clock skew. Too tight, and a host with a drifting clock rejects genuine deliveries until the endpoint is disabled. Too loose, and a captured request stays replayable for longer.
- A duplicate answered with anything but a 2xx comes back for three days. A handler failure answered with a 2xx is an event lost for good.
- seen lives in memory for this exercise. In production it belongs in a shared store, with the check and the record done atomically, keeping ids for the full retry window.
The tests in test_webhooks.py use a fake clock and sign deliveries with a test secret.
Test Output
No runs yet.