Webhooks in and out¶
A shop that takes webhooks from its payment provider and sends webhooks of its own to the URLs
its users register. FastAPIWebhooks accepts only the provider's
events signed with its secret, and each delivery once. When an order is paid, the shop signs an
order.paid event with each user's secret (WebhookSigner) and
sends it through ResilientTransport, for a deadline and a
circuit breaker per endpoint, around
SSRFGuardTransport, so that a user can't point the
shop at its own network.
Run it¶
From a clone of the repository:
make deps # .venv, with every extra
source .venv/bin/activate
export PAYMENTS_WEBHOOK_SECRET=$(python -c "from fastapi_alcazaba.webhooks import WebhookSigner; print(WebhookSigner.generate_secret())")
uvicorn examples.webhooks.main:app
In production, the provider gives the secret when you set up the endpoint. Without
PAYMENTS_WEBHOOK_SECRET the app makes a throwaway one, which nobody else can sign with.
Try it¶
A request without a signature:
$ BODY='{"type": "payment.succeeded", "data": {"order_id": "ord_1", "amount": 1999}}'
$ curl -i http://127.0.0.1:8000/webhooks/payments -d "$BODY"
HTTP/1.1 400 Bad Request
{"detail":"Invalid webhook"}
Sign it as the provider would, with the same secret (in the same shell as the export), and
keep the headers in a file for curl:
$ sign() { python -c 'import os, sys; from fastapi_alcazaba.webhooks import WebhookSigner; \
signer = WebhookSigner(os.environ["PAYMENTS_WEBHOOK_SECRET"]); \
headers = signer.sign(sys.argv[1], message_id="msg_1"); \
print("\n".join(f"{name}: {value}" for name, value in headers.items()))' "$BODY" > signed.txt; }
$ sign && cat signed.txt
webhook-id: msg_1
webhook-timestamp: 1793012100
webhook-signature: v1,r/kZwT7lk1Z+roS1OQd6pcxNinGLv8NUwlkJ964CYK0=
$ curl -i http://127.0.0.1:8000/webhooks/payments -H @signed.txt -d "$BODY"
HTTP/1.1 204 No Content
The same request again, as someone who captured it would send it, and the same signature on another body:
$ curl -i http://127.0.0.1:8000/webhooks/payments -H @signed.txt -d "$BODY"
HTTP/1.1 401 Unauthorized
{"detail":"Invalid webhook"}
$ curl -i http://127.0.0.1:8000/webhooks/payments -H @signed.txt -d "${BODY/1999/1}"
HTTP/1.1 401 Unauthorized
{"detail":"Invalid webhook"}
A retry of the provider has the same webhook-id and a new timestamp and signature. It goes
through, and the handler, which has seen msg_1 already, does nothing:
$ sign && curl -i http://127.0.0.1:8000/webhooks/payments -H @signed.txt -d "$BODY"
HTTP/1.1 204 No Content
Registering an endpoint sends it a signed ping. A URL that isn't on the public internet is
refused before anything is sent: the cloud's metadata service, which hands out credentials,
the server itself, or a port other than 80 and 443:
$ curl -i http://127.0.0.1:8000/endpoints -H 'content-type: application/json' -d '{"url": "http://169.254.169.254/latest/meta-data/iam/security-credentials/"}'
HTTP/1.1 422 Unprocessable Content
{"detail":"Not allowed: http://169.254.169.254: its address isn't allowed"}
$ curl -i http://127.0.0.1:8000/endpoints -H 'content-type: application/json' -d '{"url": "http://localhost/admin"}'
HTTP/1.1 422 Unprocessable Content
{"detail":"Not allowed: http://localhost: its address isn't allowed"}
$ curl -i http://127.0.0.1:8000/endpoints -H 'content-type: application/json' -d '{"url": "http://localhost:6379/"}'
HTTP/1.1 422 Unprocessable Content
{"detail":"Not allowed: http://localhost:6379: port 6379 isn't allowed"}
A public URL that answers the ping with a 2xx is registered, and the answer carries its secret
(whsec_...), shown once. Its deliveries are listed at /endpoints/{id}/deliveries.
What to look at¶
- The signature is checked before the body is parsed: the endpoint takes the
WebhookMessageand validatesmessage.bodyitself. A pydantic body parameter would make FastAPI read and parse the whole body before the dependency runs, signed or not. - Replays and retries are different things:
FastAPIWebhooksrefuses a delivery it has seen (same id and timestamp), in every process that shares the backend. A retry is a new delivery of the same message, so the handler is idempotent onmessage.id: in a database, a unique constraint on a row written in the same transaction as the order. - One secret per endpoint: each user gets their own, made by
WebhookSigner.generate_secret(), so one leaked secret doesn't forge events for the others. The exact bytes sent are the ones signed, and every attempt of an event keeps its id. - Checked when sent, not only when registered:
SSRFGuardTransportresolves the host on every call and connects to the address it checked, so a name that later points to10.0.0.7(or answers DNS differently the second time) gets ablockeddelivery, not a request to the private network. - The guard inside
ResilientTransport: the 10-second deadline covers the DNS lookup, and each origin (scheme, host and port) has its own breaker: after five failures in a row it's skipped for a minute instead of being waited for. Responses are read up to 64 KiB, and redirects aren't followed. - After the response: deliveries run as a background task, all endpoints at once, so the provider gets its 204 without waiting for the users' servers. A real app hands them to a task queue that retries failures.
The code¶
examples/webhooks/main.py
"""A shop that receives signed webhooks from its payment provider, and sends signed webhooks to
the URLs its users register.
export PAYMENTS_WEBHOOK_SECRET=whsec_... # the provider's secret for this endpoint
uvicorn examples.webhooks.main:app
- `POST /webhooks/payments`: the provider's events, signed, recent and each delivery once
(`FastAPIWebhooks`): a captured request sent again gets a 401. The provider's retries keep the
`webhook-id`, so the handler is idempotent on it.
- `POST /endpoints`: a user registers a URL and gets the secret to check the shop's events with.
A signed `ping` goes there first, through `SSRFGuardTransport`: a URL that isn't on the public
internet (`http://169.254.169.254/`, `http://localhost`, a private network) gets a 422.
- When an order is paid, every endpoint gets a signed `order.paid` event after the response,
through `ResilientTransport`: a deadline for each call and a circuit breaker for each endpoint.
Replays are remembered in the cache's backend: InMemoryBackend here, Redis or Valkey in
production, so that every process refuses them.
"""
import asyncio
import json
import secrets
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from typing import Annotated, Any
import httpx2 as httpx
import uvicorn
from fastapi import APIRouter, BackgroundTasks, Depends, FastAPI, HTTPException, Request, status
from pydantic import BaseModel, Field, HttpUrl, SecretStr, ValidationError
from pydantic_settings import BaseSettings, SettingsConfigDict
from fastapi_alcazaba import FastAPICache
from fastapi_alcazaba.backends.inmemory import InMemoryBackend
from fastapi_alcazaba.outbound import (
BlockedDestinationError,
CircuitOpenError,
ResilientTransport,
SSRFGuardTransport,
)
from fastapi_alcazaba.webhooks import FastAPIWebhooks, WebhookMessage, WebhookSigner
class Settings(BaseSettings):
"""The configuration, from environment variables or a .env file."""
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")
# The provider gives it when the endpoint is set up. Without it, a throwaway one for local
# development, which nobody else can sign with.
payments_webhook_secret: SecretStr = Field(
default_factory=lambda: SecretStr(WebhookSigner.generate_secret())
)
settings = Settings()
payment_webhooks = FastAPIWebhooks(settings.payments_webhook_secret)
class Payment(BaseModel):
order_id: str
amount: int # in cents
class PaymentEvent(BaseModel):
type: str # payment.succeeded, payment.failed...
data: Payment
class NewEndpoint(BaseModel):
url: HttpUrl
class CreatedEndpoint(BaseModel):
id: str
url: str
secret: str # shown once: the user checks the shop's signatures with it
class Delivery(BaseModel):
event_id: str
type: str
outcome: str
@dataclass
class Endpoint:
"""A user's URL. A real database keeps its secret encrypted."""
id: str
url: str
signer: WebhookSigner
deliveries: list[Delivery] = field(default_factory=list)
@dataclass
class Shop:
"""Stands in for the database."""
paid: dict[str, int] = field(default_factory=dict)
# The provider's messages handled already: a unique constraint in a database, on a row
# inserted in the same transaction as their effects.
handled: set[str] = field(default_factory=set)
endpoints: dict[str, Endpoint] = field(default_factory=dict)
def get_shop(request: Request) -> Shop:
shop: Shop = request.app.state.shop
return shop
def get_client(request: Request) -> httpx.AsyncClient:
client: httpx.AsyncClient = request.app.state.client
return client
ShopDep = Annotated[Shop, Depends(get_shop)]
ClientDep = Annotated[httpx.AsyncClient, Depends(get_client)]
router = APIRouter()
async def post_event(
client: httpx.AsyncClient,
endpoint: Endpoint,
event: dict[str, Any],
) -> httpx.Response:
body = json.dumps(event).encode() # signed as sent: json= would serialise it again
headers = endpoint.signer.sign(body, message_id=event["id"]) # the same id on every retry
headers["content-type"] = "application/json"
return await client.post(endpoint.url, content=body, headers=headers)
async def deliver(client: httpx.AsyncClient, endpoint: Endpoint, event: dict[str, Any]) -> None:
"""Send `event` and log what happened. A real app retries failed deliveries later, from a
task queue, with the same event id."""
try:
response = await post_event(client, endpoint, event)
outcome = "delivered" if response.is_success else f"HTTP {response.status_code}"
except CircuitOpenError:
outcome = "skipped" # it failed too often lately: not called for a minute
except BlockedDestinationError:
outcome = "blocked" # its name points to a private address now
except httpx.HTTPError as exc: # can't connect, no answer in time, answer too large...
outcome = type(exc).__name__
endpoint.deliveries.append(Delivery(event_id=event["id"], type=event["type"], outcome=outcome))
async def notify(
client: httpx.AsyncClient,
endpoints: list[Endpoint],
event: dict[str, Any],
) -> None:
"""Send `event` to every endpoint at once: they don't wait for each other."""
await asyncio.gather(*(deliver(client, endpoint, event) for endpoint in endpoints))
@router.post("/webhooks/payments", status_code=status.HTTP_204_NO_CONTENT)
async def payment_event(
message: Annotated[WebhookMessage, Depends(payment_webhooks)],
shop: ShopDep,
client: ClientDep,
background: BackgroundTasks,
) -> None:
"""The provider's events. Every attempt of a message has the same id: the first one does the
work, the others only get their 204."""
if message.id in shop.handled:
return
# Parsed once its signature is checked: FastAPI would parse a body parameter before.
try:
event = PaymentEvent.model_validate_json(message.body)
except ValidationError:
raise HTTPException(422, "Not a payment event") from None
shop.handled.add(message.id)
if event.type == "payment.succeeded":
shop.paid[event.data.order_id] = event.data.amount
order_paid = {"id": f"evt_{secrets.token_hex(12)}", "type": "order.paid"}
# After the response: the provider doesn't wait for the users' servers.
background.add_task(
notify, client, list(shop.endpoints.values()), order_paid | event.data.model_dump()
)
@router.post("/endpoints", status_code=status.HTTP_201_CREATED)
async def add_endpoint(new: NewEndpoint, shop: ShopDep, client: ClientDep) -> CreatedEndpoint:
"""Register a URL for the shop's events, if it accepts a signed ping. A real app keeps each
user's endpoints apart."""
secret = WebhookSigner.generate_secret()
endpoint = Endpoint(f"ep_{secrets.token_hex(8)}", str(new.url), WebhookSigner(secret))
ping = {"id": f"evt_{secrets.token_hex(12)}", "type": "ping"}
try:
response = await post_event(client, endpoint, ping)
except BlockedDestinationError as exc: # says which host, never the address it resolved to
raise HTTPException(422, f"Not allowed: {exc}") from None
except httpx.HTTPError as exc:
raise HTTPException(422, f"The ping failed: {type(exc).__name__}") from None
if not response.is_success:
raise HTTPException(422, f"The ping got HTTP {response.status_code}")
shop.endpoints[endpoint.id] = endpoint
return CreatedEndpoint(id=endpoint.id, url=endpoint.url, secret=secret)
@router.get("/endpoints/{endpoint_id}/deliveries")
async def deliveries(endpoint_id: str, shop: ShopDep) -> list[Delivery]:
endpoint = shop.endpoints.get(endpoint_id)
if endpoint is None:
raise HTTPException(404, "Endpoint not found")
return endpoint.deliveries
def create_app(transport: httpx.AsyncBaseTransport | None = None) -> FastAPI:
"""The application. `transport` sends the outgoing requests once they're checked: the network by
default, a mock in tests."""
@asynccontextmanager
async def lifespan(application: FastAPI) -> AsyncIterator[None]:
# Redis or Valkey in production, so that every process refuses a replay. The marks are
# counters, which a signing_key doesn't cover: keep the backend out of others' reach.
FastAPICache.init(InMemoryBackend(), prefix="shop")
# Only public addresses, on the web's ports, checked again on every call.
guard = SSRFGuardTransport(transport, ports=(80, 443))
outbound = ResilientTransport(
guard, timeout=10, failure_threshold=5, reset_timeout=60, max_response_size=65_536
)
async with httpx.AsyncClient(transport=outbound) as client: # it follows no redirects
application.state.client = client
application.state.shop = Shop()
yield
application = FastAPI(lifespan=lifespan)
application.include_router(router)
return application
app = create_app()
if __name__ == "__main__":
uvicorn.run("main:app", reload=True)