import hmac
import hashlib
import time
from fastapi import Request, HTTPException, status
def parse_signature_header(header: str) -> tuple[int, list[str]]:
timestamp: int | None = None
signatures: list[str] = []
try:
for part in header.split(","):
key, _, value = part.strip().partition("=")
if key == "t":
timestamp = int(value)
elif key == "v1":
signatures.append(value)
except ValueError:
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Malformed signature header")
if timestamp is None or not signatures:
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Missing signature parts")
return timestamp, signatures
def verify_signature(secret: str, body: bytes, timestamp: int, signatures: list[str]) -> None:
signed_payload = f"{timestamp}.".encode() + body
expected = hmac.new(secret.encode(), signed_payload, hashlib.sha256).hexdigest()
if not any(hmac.compare_digest(expected, s) for s in signatures):
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Signature verification failed")
class RequireWebhookSignature:
def __init__(self, secret: str, header_name: str = "Stripe-Signature", tolerance_seconds: int = 300):
self.secret = secret
self.header_name = header_name
self.tolerance_seconds = tolerance_seconds
async def __call__(self, request: Request) -> bytes:
header = request.headers.get(self.header_name)
if not header:
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Missing signature header")
timestamp, signatures = parse_signature_header(header)
if abs(time.time() - timestamp) > self.tolerance_seconds:
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Signature timestamp outside tolerance")
body = await request.body()
verify_signature(self.secret, body, timestamp, signatures)
return body
from functools import lru_cache
from pydantic_settings import BaseSettings
from .webhook_security import RequireWebhookSignature
class Settings(BaseSettings):
stripe_webhook_secret: str
github_webhook_secret: str = ""
class Config:
env_file = ".env"
@lru_cache
def get_settings() -> Settings:
return Settings()
stripe_signature = RequireWebhookSignature(
secret=get_settings().stripe_webhook_secret,
header_name="Stripe-Signature",
tolerance_seconds=300,
)
import json
from fastapi import APIRouter, Depends, HTTPException, status
from .dependencies import stripe_signature
from .tasks import enqueue_event_processing
router = APIRouter(prefix="/webhooks", tags=["webhooks"])
@router.post("/stripe", status_code=status.HTTP_202_ACCEPTED)
async def stripe_webhook(raw_body: bytes = Depends(stripe_signature)):
try:
event = json.loads(raw_body)
except json.JSONDecodeError:
raise HTTPException(status.HTTP_400_BAD_REQUEST, "Invalid JSON payload")
event_id = event.get("id")
event_type = event.get("type")
if not event_id or not event_type:
raise HTTPException(status.HTTP_400_BAD_REQUEST, "Missing event id or type")
# dedupe_key lets the worker skip retried deliveries of the same event
dedupe_key = f"stripe:{event_id}"
await enqueue_event_processing(
dedupe_key=dedupe_key,
event_type=event_type,
payload=event.get("data", {}),
)
return {"received": True, "id": event_id}
Webhooks arrive over the public internet, so an endpoint must prove that a payload really came from the provider before acting on it. The standard defense is an HMAC signature: the sender hashes the raw request body with a shared secret, and the receiver recomputes the same hash and compares. This snippet packages that verification into a single FastAPI dependency so every webhook route gets the same protection without duplicating crypto code.
The webhook_security module holds the core logic. parse_signature_header splits a Stripe-style t=...,v1=... header into a timestamp and one or more candidate signatures, tolerating malformed input by raising a 401. verify_signature reconstructs the signed payload as "{timestamp}.{body}", computes an HMAC-SHA256 with the endpoint secret, and compares it to each supplied signature using hmac.compare_digest. That constant-time comparison matters: a naive == leaks timing information that can let an attacker recover the signature byte by byte.
Two edge cases are handled explicitly. The timestamp is checked against tolerance_seconds so a captured request cannot be replayed hours later — this is the replay-attack guard. And the dependency reads await request.body() rather than a parsed model, because signatures are computed over the exact bytes on the wire; re-serializing parsed JSON would change whitespace or key order and break verification.
The RequireWebhookSignature class is a small callable configured with a secret and tolerance, which makes it reusable across providers with different secrets. FastAPI treats an instance as a dependency, and Depends injects the verified raw body into the handler.
In webhooks router, the stripe_webhook route declares Depends(stripe_signature) and receives the trusted raw_body. Only after verification does it parse JSON and enqueue work, keeping the signature check strictly ahead of any business logic. The dedupe_key derived from the event id supports idempotent processing, since providers retry deliveries and the same event may arrive twice.
This pattern centralizes a security-critical check, keeps route handlers thin, and makes it trivial to add new verified webhook endpoints by injecting another configured verifier.
Related snips
payload = {
sub: user.id,
iss: 'https://auth.example.com',
aud: 'codesnips-api',
exp: 15.minutes.from_now.to_i,
iat: Time.now.to_i,JWT issuance and verification without common footguns
timestamp = request.headers.fetch('X-Signature-Timestamp')
signature = request.headers.fetch('X-Signature')
payload = request.raw_post
data = "#{timestamp}.#{payload}"
expected = OpenSSL::HMAC.hexdigest('SHA256', ENV.fetch('WEBHOOK_SECRET'), data)HMAC signed API requests for webhook and partner integrity
package com.example.myapp
import android.app.Application
import dagger.hilt.android.HiltAndroidApp
@HiltAndroidAppDependency injection with Hilt
#!/usr/bin/env bash
set -euo pipefail
export VAULT_ADDR="https://vault.internal:8200"
export VAULT_TOKEN="${VAULT_TOKEN:?missing VAULT_TOKEN}"Secrets management with environment isolation and Vault
<?php
namespace App\Providers;
use App\Contracts\PaymentGateway;
use App\Services\StripePaymentGateway;Laravel service container and dependency injection
import crypto from 'node:crypto';
import jwt from 'jsonwebtoken';
export function signAccessToken(payload: object) {
return jwt.sign(payload, process.env.JWT_SECRET!, { expiresIn: '15m' });
}JWT access + refresh token rotation (conceptual)
Share this code
Here's the card — post it anywhere.
