How-to: receiving webhooks and verifying HMAC-SHA256 signatures

A pattern for receiving webhooks from external services such as GitHub or Stripe and verifying their HMAC-SHA256 signatures.

---

1. Basic pattern (GitHub style)

GitHub sends the signature in an X-Hub-Signature-256: sha256=<hex> header.

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

from nene2.security import verify_hmac_signature

WEBHOOK_SECRET = "your-secret-key"

app = FastAPI()

@app.post("/webhooks/github")
async def github_webhook(request: Request) -> JSONResponse:
    signature = request.headers.get("X-Hub-Signature-256", "")
    body = await request.body()

    if not signature:
        return JSONResponse({"error": "Missing signature"}, status_code=400)

    if not verify_hmac_signature(body, WEBHOOK_SECRET, signature, prefix="sha256="):
        return JSONResponse({"error": "Invalid signature"}, status_code=401)

    payload = await request.json()
    event = request.headers.get("X-GitHub-Event", "unknown")
    # ... process the event
    return JSONResponse({"status": "received", "event": event})

---

2. Stripe style (signature with a timestamp)

Stripe sends Stripe-Signature: t=<timestamp>,v1=<hex>. You HMAC the timestamp + body.

import hashlib
import hmac
import time

@app.post("/webhooks/stripe")
async def stripe_webhook(request: Request) -> JSONResponse:
    stripe_sig = request.headers.get("Stripe-Signature", "")
    body = await request.body()

    if not stripe_sig:
        return JSONResponse({"error": "Missing Stripe-Signature"}, status_code=400)

    parts = dict(item.split("=", 1) for item in stripe_sig.split(",") if "=" in item)
    timestamp = parts.get("t", "")
    v1_sig = parts.get("v1", "")

    # Stripe style: HMAC of "timestamp." + body
    signed_payload = f"{timestamp}.".encode() + body
    if not verify_hmac_signature(signed_payload, WEBHOOK_SECRET, v1_sig):
        return JSONResponse({"error": "Invalid signature"}, status_code=401)

    payload = await request.json()
    return JSONResponse({"status": "received", "type": payload.get("type")})

---

3. Double reading: await request.body() → await request.json()

Signature verification needs the raw bytes (body()), but you also want to parse it as JSON afterward. FastAPI caches the body internally, so you can call both.

@app.post("/webhooks/example")
async def handler(request: Request) -> JSONResponse:
    # ✅ json() still works even after calling body() first
    body = await request.body()        # raw bytes (for signature verification)
    payload = await request.json()     # JSON parse (uses the internal cache)
    return JSONResponse({"size": len(body), "action": payload.get("action")})

json.loads(body) works too, but await request.json() is more consistent with Pydantic model conversion.

---

4. The verify_hmac_signature() API

from nene2.security import verify_hmac_signature

verify_hmac_signature(
    body: bytes,       # the bytes to verify
    secret: str,       # the shared secret
    signature: str,    # the signature string to verify (prefix allowed)
    *,
    prefix: str = "",  # the signature prefix (e.g. "sha256=")
) -> bool

Protected against timing attacks via hmac.compare_digest(). Do not use == to compare signatures.

---

5. When to use this vs. BearerTokenMiddleware

Pattern	Auth method	nene2 support
API client auth	Authorization: Bearer <token>	BearerTokenMiddleware
Webhook signature verification	request body + secret	verify_hmac_signature()

Add webhook endpoints to BearerTokenMiddleware's exclude_paths and do your own signature verification. Because the middleware reads the raw body, BearerTokenMiddleware can't be used on it.

from nene2.middleware import BearerTokenMiddleware

app.add_middleware(
    BearerTokenMiddleware,
    verifier=token_verifier,
    exclude_paths=["/webhooks/"],  # exclude webhook endpoints
)

---

6. Testing

import hashlib
import hmac

def make_github_sig(body: bytes, secret: str) -> str:
    return "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()

def test_webhook_valid() -> None:
    payload = b'{"action": "opened"}'
    r = client.post(
        "/webhooks/github",
        content=payload,
        headers={
            "Content-Type": "application/json",
            "X-Hub-Signature-256": make_github_sig(payload, "your-secret-key"),
            "X-GitHub-Event": "issues",
        },
    )
    assert r.status_code == 200