Signed webhooks¶

A webhook sender signs each request with a secret you share: an HMAC-SHA256 of the raw body, sent in a header. HMACSignatureAuth checks it before your route runs:

import os

from fastapi import FastAPI, Security
from pydantic import BaseModel

from fastapi_multiauth import HMACSignatureAuth

github = HMACSignatureAuth(
    secret_key=os.environ["GITHUB_WEBHOOK_SECRET"],
    header="X-Hub-Signature-256",
    prefix="sha256=",
)


class Delivery(BaseModel):
    action: str


app = FastAPI()


@app.post("/webhooks/github", dependencies=[Security(github)])
async def github_event(delivery: Delivery):
    return {"action": delivery.action}

The signature covers the exact bytes received, so the check runs on the raw body. Your route can still declare a JSON body parameter: FastAPI parses it from the same bytes. To take the verified bytes instead, write body: bytes = Security(github).

Form bodies are the exception: FastAPI consumes the raw body to parse Form() and File() parameters, before the signature can be checked. Take the bytes from the source and parse them with urllib.parse.parse_qs, as the Slack route below does.

A missing, malformed or wrong signature gets a 401. Signatures are compared in constant time, and the secret must be at least 16 random bytes.

Senders¶

Senders differ in where they put the signature and what exactly they sign. Describe yours with the constructor's arguments. Stripe and Slack, for instance:

import json
import os
from urllib.parse import parse_qs

from fastapi import FastAPI, Security

from fastapi_multiauth import HMACSignatureAuth

stripe = HMACSignatureAuth(
    secret_key=os.environ["STRIPE_WEBHOOK_SECRET"],
    header="Stripe-Signature",
    signature_key="v1",
    timestamp_key="t",
)
slack = HMACSignatureAuth(
    secret_key=os.environ["SLACK_SIGNING_SECRET"],
    header="X-Slack-Signature",
    prefix="v0=",
    timestamp_header="X-Slack-Request-Timestamp",
    signed_message="v0:{timestamp}:{body}",
)

app = FastAPI()


@app.post("/webhooks/stripe")
async def stripe_event(body: bytes = Security(stripe)):
    return {"type": json.loads(body)["type"]}


@app.post("/webhooks/slack")
async def slack_command(body: bytes = Security(slack)):
    form = parse_qs(body.decode())  # a form body: parse the verified bytes
    return {"command": form["command"][0]}

A timestamp must be part of signed_message, or anyone could replay a request with a fresh one: the constructor refuses that configuration.

Replays¶

A signature proves who sent the body, not that it is the first time you see it. With a timestamp, a request older than tolerance (300 seconds by default) is refused, but a copy replayed within that window still passes. GitHub sends no timestamp at all. Make your handler idempotent, or deduplicate on the event id the sender provides, such as X-GitHub-Delivery.

With a timestamp, replay_check records each verified signature so it cannot be sent twice. Make it atomic across your workers, e.g. Redis SET NX with the expiry it receives. Without a timestamp the constructor refuses it: signatures would have to be kept forever.

Rotating a secret¶

Pass a list: secret_key=[new_secret, old_secret] accepts either. Remove the old one once the sender only signs with the new one.