Webhooks

Webhooks are HTTP endpoints that receive data from external services. Anyone who discovers your webhook URL can send requests to it. That’s a problem. Here’s how to secure them properly. The Threat Model Your webhook endpoint faces several threats: Spoofing: Attacker sends fake payloads pretending to be Stripe/GitHub/etc. Replay attacks: Attacker captures a legitimate request and resends it Tampering: Attacker intercepts and modifies payloads in transit Enumeration: Attacker discovers your webhook URLs through guessing Denial of service: Attacker floods your endpoint with requests Signature Verification Most webhook providers sign their payloads. Always verify: ...

Webhooks flip the API model: instead of polling, the service calls you. Here’s how to handle them without losing events. Basic Webhook Handler 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 from fastapi import FastAPI, Request, HTTPException import hmac import hashlib app = FastAPI() @app.post("/webhooks/stripe") async def stripe_webhook(request: Request): payload = await request.body() sig_header = request.headers.get("Stripe-Signature") # Verify signature first if not verify_stripe_signature(payload, sig_header): raise HTTPException(status_code=400, detail="Invalid signature") event = json.loads(payload) # Process event if event["type"] == "payment_intent.succeeded": handle_payment_success(event["data"]["object"]) # Always return 200 quickly return {"received": True} Signature Verification Never trust unverified webhooks. ...

Webhooks are the internet’s way of saying “hey, something happened.” Simple in concept, surprisingly tricky in practice. The challenge: HTTP is unreliable, servers go down, networks flake out, and your webhook payload might arrive zero times, once, or five times. Building reliable webhook infrastructure means handling all of these gracefully. The Sender Side Retry with Exponential Backoff First attempt fails? Try again. But not immediately. 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 RETRY_DELAYS = [60, 300, 900, 3600, 14400, 43200] # seconds def deliver_webhook(event, attempt=0): try: response = requests.post( event.webhook_url, json=event.payload, timeout=30, headers={ "Content-Type": "application/json", "X-Webhook-ID": event.id, "X-Webhook-Timestamp": str(int(time.time())), "X-Webhook-Signature": sign_payload(event.payload) } ) if response.status_code >= 200 and response.status_code < 300: mark_delivered(event) return if response.status_code >= 500: # Server error, retry schedule_retry(event, attempt) else: # Client error (4xx), don't retry mark_failed(event, response.status_code) except requests.Timeout: schedule_retry(event, attempt) except requests.ConnectionError: schedule_retry(event, attempt) def schedule_retry(event, attempt): if attempt >= len(RETRY_DELAYS): mark_failed(event, "max_retries") return delay = RETRY_DELAYS[attempt] # Add jitter to prevent thundering herd jitter = random.uniform(0, delay * 0.1) queue_at = time.time() + delay + jitter enqueue(event, attempt + 1, queue_at) Key decisions: ...

Webhooks flip the API model: instead of polling for changes, services push events to you. GitHub notifies you of commits. Stripe tells you about payments. Twilio delivers SMS receipts. The concept is simple. Production-grade implementation requires care. Basic Webhook Receiver 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 from fastapi import FastAPI, Request, HTTPException app = FastAPI() @app.post("/webhooks/stripe") async def stripe_webhook(request: Request): payload = await request.json() event_type = payload.get("type") if event_type == "payment_intent.succeeded": handle_payment_success(payload["data"]["object"]) elif event_type == "payment_intent.failed": handle_payment_failure(payload["data"]["object"]) return {"status": "received"} This works for demos. Production needs more. ...

Webhooks seem simple: receive an HTTP POST, do something. In practice, they’re a minefield of security issues, reliability problems, and edge cases. Let’s build webhooks that actually work in production. Receiving Webhooks Basic Receiver 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 from fastapi import FastAPI, Request, HTTPException, Header from typing import Optional import hmac import hashlib app = FastAPI() WEBHOOK_SECRET = "your-webhook-secret" @app.post("/webhooks/stripe") async def stripe_webhook( request: Request, stripe_signature: Optional[str] = Header(None, alias="Stripe-Signature") ): payload = await request.body() # Verify signature if not verify_stripe_signature(payload, stripe_signature, WEBHOOK_SECRET): raise HTTPException(status_code=401, detail="Invalid signature") event = await request.json() # Process event event_type = event.get("type") if event_type == "payment_intent.succeeded": await handle_payment_success(event["data"]["object"]) elif event_type == "customer.subscription.deleted": await handle_subscription_cancelled(event["data"]["object"]) return {"received": True} def verify_stripe_signature(payload: bytes, signature: str, secret: str) -> bool: """Verify Stripe webhook signature.""" if not signature: return False # Parse signature header elements = dict(item.split("=") for item in signature.split(",")) timestamp = elements.get("t") expected_sig = elements.get("v1") # Compute expected signature signed_payload = f"{timestamp}.{payload.decode()}" computed_sig = hmac.new( secret.encode(), signed_payload.encode(), hashlib.sha256 ).hexdigest() return hmac.compare_digest(computed_sig, expected_sig) Idempotent Processing Webhooks can be delivered multiple times. Make your handlers idempotent: ...