Fix Insecure Webhooks in FastAPI

Webhooks are a massive attack vector if left unauthenticated. An attacker can spoof events, bypass business logic, or trigger unauthorized actions by sending forged POST requests to your endpoint. In FastAPI, the most common mistake is trusting the payload without cryptographic verification. To secure them, you must implement HMAC signature verification using a shared secret to ensure the payload hasn't been tampered with and originated from a trusted source.

The Vulnerable Pattern

from fastapi import FastAPI, Request
app = FastAPI()
@app.post(“/webhook”)
async def insecure_webhook(request: Request):
# VULNERABILITY: No authentication or signature check.
# Anyone can POST any JSON to this endpoint.
data = await request.json()
process_order(data[‘order_id’])
return {“status”: “received”}

The Secure Implementation

The secure implementation relies on HMAC (Hash-based Message Authentication Code). 1. The Raw Body: We use 'request.body()' instead of 'request.json()' because the signature is calculated against the raw bytes; any whitespace difference in JSON parsing would break the hash. 2. HMAC-SHA256: We generate a local hash using a shared secret and the incoming body. 3. Constant-Time Comparison: We use 'hmac.compare_digest' rather than a standard '==' operator. This mitigates timing attacks where an attacker could measure the time it takes to reject a signature to guess it byte-by-byte. 4. FastAPI Dependencies: We abstract the logic into a dependency to ensure the endpoint logic only executes if the signature is valid.

import hmac
import hashlib
from fastapi import FastAPI, Request, HTTPException, Header, Depends
app = FastAPI()
WEBHOOK_SECRET = b”your_ultra_secure_shared_secret”
async def verify_signature(request: Request, x_signature: str = Header(None)):
if not x_signature:
raise HTTPException(status_code=400, detail=“Missing signature header”)
body = await request.body()
# Calculate HMAC SHA256 signature
expected_signature = hmac.new(WEBHOOK_SECRET, body, hashlib.sha256).hexdigest()

# Use compare_digest to prevent timing attacks
if not hmac.compare_digest(expected_signature, x_signature):
    raise HTTPException(status_code=401, detail="Invalid signature")
@app.post(“/webhook”, dependencies=[Depends(verify_signature)])
async def secure_webhook(request: Request):
data = await request.json()
return {“status”: “verified_and_processed”}