Stripe Webhook Signature Verification in Python (FastAPI)

Stripe Webhook Signature Verification in Python FastAPI: Production Recipe

⚡ Quick Answer (The FastAPI Rule)

To verify Stripe webhooks in FastAPI, you must read the raw unparsed byte payload via await request.body() before any JSON serialization, verify it against the Stripe-Signature header using stripe.Webhook.construct_event(), and immediately return HTTP 200 within 200ms.

from fastapi import FastAPI, Request, HTTPException, status
import stripe
import os
import redis.asyncio as redis

app = FastAPI()
redis_client = redis.from_url(os.getenv("REDIS_URL", "redis://localhost:6379/0"))
WEBHOOK_SECRET = os.getenv("STRIPE_WEBHOOK_SECRET")

@app.post("/webhooks/stripe")
async def handle_stripe_webhook(request: Request):
    # 1. Read raw byte payload (CRITICAL: Never use request.json())
    payload = await request.body()
    sig_header = request.headers.get("stripe-signature")

    if not sig_header:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Missing signature")

    # 2. Cryptographic signature check with timestamp tolerance (300s)
    try:
        event = stripe.Webhook.construct_event(
            payload=payload,
            sig_header=sig_header,
            secret=WEBHOOK_SECRET,
            tolerance=300
        )
    except (ValueError, stripe.error.SignatureVerificationError):
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid signature")

    # 3. Idempotency Check: Prevent duplicate processing of retried events
    event_id = event["id"]
    is_new = await redis_client.set(f"webhook:stripe:{event_id}", "locked", ex=86400, nx=True)
    if not is_new:
        return {"status": "already_processed"}

    # 4. Offload heavy processing to async worker (Celery/BullMQ/SQS)
    # await task_queue.enqueue(event["type"], event["data"])

    # 5. Acknowledge within 200ms
    return {"status": "success"}

1. The Raw Byte Parsing Trap

The most common bug in modern Python frameworks (FastAPI, Flask, Starlette) occurs when developers attempt to access request.json() or pass a Pydantic model directly into the route handler. Stripe’s HMAC-SHA256 signature is calculated against the exact, byte-for-byte serialized string generated on Stripe’s servers.

If your Python framework parses the JSON and then dumps it back to a string, key ordering differences, floating-point representations, or whitespace variations will mutate the hash, triggering constant SignatureVerificationError exceptions.

2. Idempotency & Replay Attack Defense

Because Stripe guarantees at-least-once delivery, network timeouts or transient 504 gateway errors will cause Stripe to retry the exact same event. Without an atomic idempotency lock (e.g. Redis SETNX with a 24-hour TTL), your billing listener could accidentally grant duplicate credits, send multiple invoice receipts, or trigger duplicate shipping labels.