Webhooks

You’ve deployed your webhook endpoint. The external service says it’s sending requests. But your logs show nothing. Here’s how to systematically debug incoming webhook issues. The Debugging Checklist Before diving deep, run through these quick checks: 1 2 3 4 5 6 7 8 9 10 11 # 1. Is your endpoint actually reachable? curl -X POST https://your-domain.com/webhook \ -H "Content-Type: application/json" \ -d '{"test": true}' \ -w "\nHTTP Status: %{http_code}\n" # 2. Is DNS resolving correctly? dig your-domain.com +short # 3. Is the port open? nc -zv your-domain.com 443 If your own curl request doesn’t reach the endpoint, the problem is infrastructure. If it does but the external service’s requests don’t arrive, the problem is somewhere in between. ...

Webhooks are deceptively simple: someone sends you HTTP requests, you process them. What could go wrong? Everything. Webhooks are inbound attack surface, and most implementations have gaps you could drive a truck through. The Obvious One: Signature Verification Most webhook providers sign their payloads. Stripe uses HMAC-SHA256. GitHub uses HMAC-SHA1 or SHA256. Slack uses its own signing scheme. You’ve probably implemented this: 1 2 3 4 5 6 7 8 9 10 11 import hmac import hashlib def verify_stripe_signature(payload: bytes, signature: str, secret: str) -> bool: expected = hmac.new( secret.encode(), payload, hashlib.sha256 ).hexdigest() return hmac.compare_digest(f"sha256={expected}", signature) Good. But this is table stakes. What else? ...

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: ...