Audit trail

What it's for: a record of who did what to which resource, allowed or not, that nobody can alter without it showing.

An admin API whose sensitive actions leave an audit trail: who changed which account's role or deleted it, whether they were allowed to, and the request it came from. FastAPIAudit writes each event when the endpoint ends, as a line of JSON on standard output, signed and chained to the previous one, and verify_audit_trail checks that none was altered or removed.

Run it

From a clone of the repository:

make deps                 # .venv, with every extra
source .venv/bin/activate
uvicorn examples.audit.main:app

The events are signed: set AUDIT_SIGNING_KEY (secrets.token_urlsafe(32)), the same in every process, and keep it from the people whose actions are recorded. The example has three accounts: u1 (Ada, an admin, token ada-token), u2 (Sam, support, token sam-token) and u3 (Vic, a viewer).

Try it

Ada makes Vic support:

$ curl -i -X PUT http://127.0.0.1:8000/accounts/u3/role -H "Authorization: Bearer ada-token" \
    -H "Content-Type: application/json" -d '{"role": "support"}'
HTTP/1.1 200 OK
x-request-id: 2b78a9c373054722779a26e2904a6b0a

{"id":"u3","email":"vic@example.com","role":"support"}

and the server's standard output gets the event, the request's ID included:

{"id":"eec01d5969ce42deb8787aae473acfa1","time":"2026-10-26T15:44:27.175075+00:00","action":"account.role.change","outcome":"success","actor":"u1","resource":"account:u3","reason":null,"status":null,"request_id":"2b78a9c373054722779a26e2904a6b0a","trace_id":null,"method":"PUT","route":"/accounts/{account_id}/role","client":"127.0.0.1","details":{"from":"viewer","to":"support"},"stream":"1c33125463a71ecb","sequence":1,"previous":null,"signature":"13c0da20bf320784d6a1a4698cd91574b2b97b8c8e818668ef104aa9cc65fb34"}

Sam, who isn't an admin, tries to delete Vic's account; then someone without a token does:

$ curl -i -X DELETE http://127.0.0.1:8000/accounts/u3 -H "Authorization: Bearer sam-token"
HTTP/1.1 403 Forbidden

{"detail":"Only admins can do this"}

$ curl -i -X DELETE http://127.0.0.1:8000/accounts/u3
HTTP/1.1 401 Unauthorized
www-authenticate: Bearer

{"detail":"Not authenticated"}

Both are recorded as denied: Sam's with who he is and why, the anonymous one with the account it was after. Each event carries the signature of the one before:

{"id":"2221174595b04a82b269e441227a0933","time":"2026-10-26T15:44:27.178638+00:00","action":"account.delete","outcome":"denied","actor":"u2","resource":"account:u3","reason":"support, not admin","status":403,"request_id":"14500c6bd67ff72fd772d584cdb9fee5","trace_id":null,"method":"DELETE","route":"/accounts/{account_id}","client":"127.0.0.1","details":{},"stream":"1c33125463a71ecb","sequence":2,"previous":"13c0da20bf320784d6a1a4698cd91574b2b97b8c8e818668ef104aa9cc65fb34","signature":"41ea75d99c58cd8618d1f3e5bad140e52f270929beb26cc811a53c9bee452aa6"}
{"id":"41a26fffb4bd41df9d8c370da2d9cc7d","time":"2026-10-26T15:44:27.182414+00:00","action":"account.delete","outcome":"denied","actor":null,"resource":"account:u3","reason":"fastapi.exceptions.HTTPException","status":401,"request_id":"028f8c3e03768e47f6f85152ba18cf48","trace_id":null,"method":"DELETE","route":"/accounts/{account_id}","client":"127.0.0.1","details":{},"stream":"1c33125463a71ecb","sequence":3,"previous":"41ea75d99c58cd8618d1f3e5bad140e52f270929beb26cc811a53c9bee452aa6","signature":"be3d2ff3bb03a6642e81027d5ac0edc75356ab880f9434b2ba88a089ff91eda5"}

A sign-in with a wrong password records which account was tried, never the password:

$ curl -i -X POST http://127.0.0.1:8000/sessions -H "Content-Type: application/json" \
    -d '{"email": "ada@example.com", "password": "hunter2"}'
HTTP/1.1 401 Unauthorized
x-request-id: b544c2688f7bd5b8cc1c58784a44ca68

{"detail":"Wrong e-mail or password"}

{"id":"6c10e3d17e1d4e3d9442c181ebc6e525","time":"2026-10-26T15:44:27.185776+00:00","action":"session.create","outcome":"denied","actor":"u1","resource":null,"reason":"wrong password","status":null,"request_id":"b544c2688f7bd5b8cc1c58784a44ca68","trace_id":null,"method":"POST","route":"/sessions","client":"127.0.0.1","details":{},"stream":"1c33125463a71ecb","sequence":4,"previous":"be3d2ff3bb03a6642e81027d5ac0edc75356ab880f9434b2ba88a089ff91eda5","signature":"9124fbfed7d210bde4fe27842947404b9c711d86acd3b6e1469e0af09d97c7d0"}

Ada checks the trail (and that check is recorded too, as audit.read). An event altered, forged or removed would be listed in problems:

$ curl http://127.0.0.1:8000/audit/verify -H "Authorization: Bearer ada-token"
{"events":4,"problems":[]}

GET /audit/events lists the last events, for admins only.

What to look at

• The audit dependency comes first: declared before current_account, it's entered before the authentication runs, so the 401s that one answers are recorded too, with the account from the path (resource="account:{account_id}").
• Who acts goes in before the check: require_admin sets entry.actor and then refuses, so Sam's 403 is recorded with Sam, and entry.reason says why.
• How it ended: a 401 or a 403 is denied, any other exception a failure (a 404 for an account that doesn't exist), with its type and status but never its message.
• No secrets: the sign-in records the account and the reason, not the e-mail address typed (it may be a password typed in the wrong field) nor the password. Had they gone in details, the password would have been [redacted], by its key.
• A stream of its own: the events go to the logger fastapi_alcazaba.audit.events, which writes them to standard output without the app's log format, one line each, for the log pipeline to ship to storage that can't be rewritten (an S3 bucket with Object Lock, a SIEM).
• Signed and chained: each event carries an HMAC of its content and the signature of the previous one of its process. verify_audit_trail finds an event altered, forged or missing in the middle; it can't tell that the newest were removed, nor stop whoever has the key.
• Correlation: the x-request-id a user quotes leads to the event (request_id), and with OpenTelemetry the event's trace_id leads to the request's trace.

The code

examples/audit/main.py

"""An admin API whose sensitive actions leave an audit trail: who did what to which account,
whether it was allowed, and the request it came from.

    uvicorn examples.audit.main:app

- Changing a role and deleting an account are recorded when they end: `success`, `denied` (a 401
  without a token, a 403 for someone who isn't an admin) or `failure` (a 404 for an account that
  doesn't exist). Reading the trail is recorded too.
- Each event is a line of JSON on standard output, apart from the app's other logs, for the log
  pipeline to ship to storage that can't be rewritten. The events are signed and chained:
  `GET /audit/verify` says whether any was altered or removed.
- A sign-in records its outcome with `audit.log`, never the password, and the response carries
  the request ID that the event holds.

The accounts are a dict in memory here, a table in your database; the last events are kept in
memory too, for `/audit/events` and `/audit/verify`.
"""

import hmac
import logging
import secrets
import sys
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Annotated, Any, Literal

import uvicorn
from fastapi import Depends, FastAPI, HTTPException, Request
from pydantic import BaseModel, Field, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict
from starlette.requests import HTTPConnection

from fastapi_alcazaba import FastAPIAudit
from fastapi_alcazaba.audit import AuditEntry, LoggingAuditSink, MemoryAuditSink, verify_audit_trail
from fastapi_alcazaba.observability import RequestIDMiddleware


def throwaway_secret() -> SecretStr:
    """A random secret, for trying the example: it changes at every start and differs between
    processes."""
    return SecretStr(secrets.token_urlsafe(32))


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")

    # Signs the events: AUDIT_SIGNING_KEY (secrets.token_urlsafe(32)), the same in every process,
    # kept from the people whose actions are recorded: with it, they could rewrite the trail.
    audit_signing_key: SecretStr = Field(default_factory=throwaway_secret)


settings = Settings()

# The trail: one line of JSON per event on standard output, without the app's log format.
trail_log = logging.getLogger("fastapi_alcazaba.audit.events")
trail_log.addHandler(logging.StreamHandler(sys.stdout))
trail_log.setLevel(logging.INFO)
trail_log.propagate = False

recent = MemoryAuditSink(maxlen=1000)
audit = FastAPIAudit([LoggingAuditSink(trail_log), recent], signing_key=settings.audit_signing_key)

Role = Literal["admin", "support", "viewer"]


class Account(BaseModel):
    id: str
    email: str
    role: Role


class RoleChange(BaseModel):
    role: Role


class SignIn(BaseModel):
    email: str
    password: str


# Stands in for your users table, its password hashes and your access tokens.
PASSWORDS = {"ada@example.com": "correct horse", "sam@example.com": "battery staple"}
TOKENS = {"ada-token": "u1", "sam-token": "u2"}


@asynccontextmanager
async def lifespan(application: FastAPI) -> AsyncIterator[None]:
    application.state.accounts = {
        "u1": Account(id="u1", email="ada@example.com", role="admin"),
        "u2": Account(id="u2", email="sam@example.com", role="support"),
        "u3": Account(id="u3", email="vic@example.com", role="viewer"),
    }
    yield


app = FastAPI(lifespan=lifespan)
# The ID of each request goes in its events and in the response: support finds the event a
# user's complaint is about.
app.add_middleware(RequestIDMiddleware)


def accounts(request: Request) -> dict[str, Account]:
    return request.app.state.accounts  # type: ignore[no-any-return]


Accounts = Annotated[dict[str, Account], Depends(accounts)]


async def current_account(connection: HTTPConnection) -> Account:
    """Stands in for your authentication: the account whose bearer token the request carries."""
    scheme, _, token = connection.headers.get("authorization", "").partition(" ")
    account_id = TOKENS.get(token) if scheme.lower() == "bearer" else None
    account = connection.app.state.accounts.get(account_id)
    if account is None:
        raise HTTPException(401, "Not authenticated", headers={"WWW-Authenticate": "Bearer"})
    return account  # type: ignore[no-any-return]


Me = Annotated[Account, Depends(current_account)]


def require_admin(entry: AuditEntry, me: Account) -> None:
    """Who acts goes in the event before the check: a refusal is recorded with its actor."""
    entry.actor = me.id
    if me.role != "admin":
        entry.reason = f"{me.role}, not admin"
        raise HTTPException(403, "Only admins can do this")


def find(all_accounts: dict[str, Account], account_id: str) -> Account:
    account = all_accounts.get(account_id)
    if account is None:
        raise HTTPException(404, "No such account")
    return account


# The audit dependency goes before the authentication (Me): it records what that one refuses,
# a 401, with the account from the path.
changing_role = audit.action("account.role.change", resource="account:{account_id}")
deleting = audit.action("account.delete", resource="account:{account_id}")
reading_trail = audit.action("audit.read")


@app.put("/accounts/{account_id}/role")
async def change_role(
    account_id: str,
    change: RoleChange,
    entry: Annotated[AuditEntry, Depends(changing_role)],
    me: Me,
    all_accounts: Accounts,
) -> Account:
    require_admin(entry, me)
    account = find(all_accounts, account_id)
    entry.details.update({"from": account.role, "to": change.role})
    account.role = change.role
    return account


@app.delete("/accounts/{account_id}", status_code=204)
async def delete_account(
    account_id: str,
    entry: Annotated[AuditEntry, Depends(deleting)],
    me: Me,
    all_accounts: Accounts,
) -> None:
    require_admin(entry, me)
    entry.details["email"] = find(all_accounts, account_id).email
    del all_accounts[account_id]


@app.post("/sessions")
async def sign_in(form: SignIn, request: Request, all_accounts: Accounts) -> dict[str, str]:
    account = next((a for a in all_accounts.values() if a.email == form.email), None)
    expected = PASSWORDS.get(form.email, "")
    # Compared in constant time, and the same answer whatever was wrong.
    if account is None or not hmac.compare_digest(form.password.encode(), expected.encode()):
        await audit.log(
            "session.create",
            outcome="denied",
            actor=account.id if account else None,
            reason="unknown e-mail" if account is None else "wrong password",
            request=request,
        )
        raise HTTPException(401, "Wrong e-mail or password")
    await audit.log("session.create", actor=account.id, request=request)
    token = next(token for token, owner in TOKENS.items() if owner == account.id)
    return {"access_token": token, "token_type": "bearer"}


@app.get("/audit/events")
async def audit_events(
    entry: Annotated[AuditEntry, Depends(reading_trail)],
    me: Me,
    limit: int = 20,
) -> list[dict[str, Any]]:
    require_admin(entry, me)
    return [event.to_dict() for event in list(recent.events)[-limit:]]


@app.get("/audit/verify")
async def audit_verify(
    entry: Annotated[AuditEntry, Depends(reading_trail)],
    me: Me,
) -> dict[str, Any]:
    require_admin(entry, me)
    events = list(recent.events)
    return {
        "events": len(events),
        "problems": verify_audit_trail(events, settings.audit_signing_key),
    }


if __name__ == "__main__":
    uvicorn.run("main:app", reload=True)