Audit trail¶

FastAPIAudit, its entries and events, sinks, and the trail's verification. In fastapi_alcazaba.audit.

In the guide: Audit trail. In an app: the example.

fastapi_alcazaba.audit.FastAPIAudit ¶

FastAPIAudit(sinks: Iterable[AuditSink] | None = None, *, signing_key: Secrets | None = None, redact: Collection[str] = (), timeout: float = 5.0, metrics: Metrics | None = None)

An audit trail: action() records what endpoints do, record() and log() the rest.

Parameters:

Name	Type	Description	Default
`sinks`	`Iterable[AuditSink] \| None`	where the events go; a LoggingAuditSink by default.	`None`
`signing_key`	`Secrets \| None`	signs and chains the events (see `verify_audit_trail`); a secret of at least 32 bytes, or several to rotate it (the first signs). The same in every process.	`None`
`redact`	`Collection[str]`	words that mark secrets in the keys of `details`, besides `SENSITIVE_WORDS`.	`()`
`timeout`	`float`	seconds each sink has to write an event.	`5.0`
`metrics`	`Metrics \| None`	where to count the events (`fastapi_alcazaba.audit`) and those not written (`fastapi_alcazaba.audit.undelivered`); FastAPICache's by default.	`None`

action ¶

action(name: str, *, resource: str | None = None) -> Callable[[HTTPConnection], AsyncIterator[AuditEntry]]

A dependency that records the action name when the endpoint ends, and gives the endpoint its AuditEntry to fill in (actor, details...). resource is a template of the path parameters, such as "invoice:{invoice_id}", filled in before anything runs: the requests refused are recorded with it too.

record `async` ¶

record(action: str, *, actor: str | int | UUID | None = None, resource: str | int | UUID | None = None, details: Mapping[str, Any] | None = None, request: HTTPConnection | None = None) -> AsyncIterator[AuditEntry]

Records the block as the action action, when it ends: outside endpoints (jobs, tasks, commands), or around a part of one. request adds the request's method, route and client.

log `async` ¶

log(action: str, *, outcome: Outcome = 'success', actor: str | int | UUID | None = None, resource: str | int | UUID | None = None, reason: str | None = None, details: Mapping[str, Any] | None = None, request: HTTPConnection | None = None) -> AuditEvent

Writes one event and returns it, e.g. for a failed login. request adds the request's method, route and client.

fastapi_alcazaba.audit.AuditEntry ¶

AuditEntry(action: str, *, actor: str | int | UUID | None = None, resource: str | int | UUID | None = None, details: Mapping[str, Any] | None = None)

The event an action writes when it ends: the endpoint fills in what it learns.

outcome is decided by how the action ends; set it ("denied", "failure") when the endpoint returns an error response instead of raising one. reason says why it failed or was denied, the exception's type by default.

actor `property` `writable` ¶

actor: str | None

Who does it, e.g. the user's ID.

resource `property` `writable` ¶

resource: str | None

What it's done to, e.g. "invoice:42".

fastapi_alcazaba.audit.AuditEvent `dataclass` ¶

AuditEvent(id: str, time: datetime, action: str, outcome: Outcome, actor: str | None = None, resource: str | None = None, reason: str | None = None, status: int | None = None, request_id: str | None = None, trace_id: str | None = None, method: str | None = None, route: str | None = None, client: str | None = None, details: Mapping[str, Any] = dict(), stream: str = '', sequence: int = 0, previous: str | None = None, signature: str | None = None)

An event of the trail, as the sinks get it; to_dict() and to_json() give its JSON form.

Parameters:

Name	Type	Description	Default
`id`	`str`	random, unique.	required
`time`	`datetime`	when it was written (UTC).	required
`action`	`str`	what was done, e.g. "invoice.delete".	required
`outcome`	`Outcome`	success, failure or denied.	required
`actor`	`str \| None`	who did it (a user's ID), if known.	`None`
`resource`	`str \| None`	what it was done to, e.g. "invoice:42".	`None`
`reason`	`str \| None`	why it failed or was denied: the exception's type, or the reason given.	`None`
`status`	`int \| None`	the HTTP status of the exception that ended it.	`None`
`request_id`	`str \| None`	the request's ID (`RequestIDMiddleware`).	`None`
`trace_id`	`str \| None`	the OpenTelemetry trace it belongs to.	`None`
`method`	`str \| None`	the request's HTTP method.	`None`
`route`	`str \| None`	the template of the route that matched, e.g. "/invoices/{invoice_id}".	`None`
`client`	`str \| None`	the client's address, as the server reports it.	`None`
`details`	`Mapping[str, Any]`	anything else, redacted and bounded.	`dict()`
`stream`	`str`	the writer's stream: one per process and FastAPIAudit.	`''`
`sequence`	`int`	its number in the stream, from 1.	`0`
`previous`	`str \| None`	the signature of the stream's previous event (with a `signing_key`).	`None`
`signature`	`str \| None`	with a `signing_key`, HMAC-SHA256 in hex of `fastapi_alcazaba/audit2`, a NUL byte, and the RFC 8785 canonical JSON of `to_dict()` without `signature`: a verifier in any language computes the same.	`None`

to_json ¶

to_json() -> str

One line of JSON, ASCII only: nothing in it can split a log line.

fastapi_alcazaba.audit.AuditSink ¶

Bases: Protocol

Where events go. write gets each one, redacted and signed, when its action ends, within the FastAPIAudit's timeout: queue it if writing takes longer. What it raises is logged.

fastapi_alcazaba.audit.LoggingAuditSink ¶

LoggingAuditSink(log: str | Logger = 'fastapi_alcazaba.audit.events', level: int = logging.INFO)

Writes each event as one line of JSON to the logger log (a name or a Logger), at level; handlers that format their own JSON find the event in the record's audit attribute too. When the logger outputs nothing at that level (no handler, or a higher level), it says so once, as a warning.

fastapi_alcazaba.audit.MemoryAuditSink ¶

MemoryAuditSink(maxlen: int | None = 1000)

Keeps the last maxlen events in events: for tests, and to look at them while developing.

fastapi_alcazaba.audit.verify_audit_trail ¶

verify_audit_trail(events: Iterable[AuditEvent | Mapping[str, Any]], signing_key: Secrets) -> list[str]

What's wrong with a trail written with signing_key (or one of its keys, when it was rotated): an empty list when nothing is. events are AuditEvents or their to_dict() (the JSON lines read back), of one stream or several, in any order; the same event given twice counts once. Each stream's chain is checked from its oldest event given, since older ones may have been rotated out of the logs.

fastapi_alcazaba.audit.SENSITIVE_WORDS `module-attribute` ¶

SENSITIVE_WORDS: frozenset[str] = frozenset({'accesskey', 'apikey', 'auth', 'authorization', 'cardnumber', 'cookie', 'credential', 'cvc', 'cvv', 'otp', 'passphrase', 'passwd', 'password', 'pin', 'privatekey', 'pwd', 'secret', 'secretkey', 'session', 'signature', 'token'})