Verify incoming webhooks in FastAPI

Approach

Read the raw body (await request.body()), verify the HMAC signature in constant time with a timestamp tolerance, insert the event ID into a table with a unique constraint to drop duplicates, return 2xx quickly and process asynchronously.

Solution

import hashlib, hmac, time
from fastapi import HTTPException, Request
from sqlalchemy.dialects.postgresql import insert


def verify_signature(secret: str, body: bytes, header: str, tolerance: int = 300) -> None:
    parts = dict(p.split("=", 1) for p in header.split(",") if "=" in p)
    ts = int(parts.get("t", "0"))
    expected = hmac.new(secret.encode(), f"{ts}.".encode() + body, hashlib.sha256).hexdigest()
    if abs(time.time() - ts) > tolerance or not hmac.compare_digest(expected, parts.get("v1", "")):
        raise HTTPException(400, "Invalid signature")


@app.post("/webhooks/payments", status_code=200)
async def payments_webhook(request: Request, db: DB, settings: SettingsDep):
    body = await request.body()
    verify_signature(settings.webhook_secret.get_secret_value(), body, request.headers.get("x-signature", ""))
    event = json.loads(body)
    result = await db.execute(
        insert(WebhookEvent).values(event_id=event["id"], type=event["type"], payload=event)
        .on_conflict_do_nothing(index_elements=["event_id"])
    )
    await db.commit()
    if result.rowcount:                       # new event → process in the background
        await request.app.state.arq.enqueue_job("process_payment_event", event["id"])
    return {"received": True}

Pitfalls

• Parsing JSON first and verifying a re-serialized body.
• Comparing signatures with ==.
• Slow processing inside the webhook request causing provider retries.

Variations

• Use the provider SDK's verification helper (e.g., stripe.Webhook.construct_event).

Further reading: Stripe: Verify webhook events