webhook

webhook

Production patterns for receiving and sending webhooks. Covers security verification, idempotency, queue-backed processing, and local development tunnels. Language-agnostic with FastAPI and Express examples.

1) HMAC Signature Verification

Never process a webhook without verifying the signature. Reject before parsing the body.

# FastAPI — generic HMAC-SHA256
import hashlib, hmac, time
from fastapi import APIRouter, Request, HTTPException

router = APIRouter()

def verify_hmac(body: bytes, signature: str, secret: str, prefix: str = "sha256=") -> bool:
    expected = prefix + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected.encode(), signature.encode())

@router.post("/webhooks/github")
async def github_webhook(request: Request):
    body = await request.body()
    sig = request.headers.get("X-Hub-Signature-256", "")
    if not verify_hmac(body, sig, secret=settings.GITHUB_WEBHOOK_SECRET):
        raise HTTPException(status_code=401, detail="Invalid signature")
    payload = await request.json()
    # ... process

// Express — Stripe signature verification
import Stripe from 'stripe';

app.post('/webhooks/stripe', express.raw({ type: 'application/json' }), (req, res) => {
  const sig = req.headers['stripe-signature']!;
  let event: Stripe.Event;
  try {
    event = stripe.webhooks.constructEvent(req.body, sig, process.env.STRIPE_WEBHOOK_SECRET!);
  } catch (err) {
    return res.status(400).send(`Webhook error: ${err.message}`);
  }
  // process event...
  res.json({ received: true });
});

2) Replay Attack Guard (Timestamp Check)

# Slack, GitHub, and Stripe all include timestamps
def verify_slack(body: bytes, timestamp: str, signature: str, secret: str) -> bool:
    if abs(time.time() - int(timestamp)) > 300:   # reject if >5 min old
        return False
    base = f"v0:{timestamp}:{body.decode()}"
    expected = "v0=" + hmac.new(secret.encode(), base.encode(), hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)