middleware.audit

Page actions AI-ready formats and sharing
Open LLM text
Share with AI
Ask Claude Ask ChatGPT Ask Gemini Ask Copilot

HTTP audit middleware — opt-in per-request who/what/when/status trail.

AuditMiddlewareemits one structured event per (audited) request through the existingemit_security_event() sink, under an http.requestnamespace — so general request auditing and…

HTTP audit middleware — opt-in per-request who/what/when/status trail.

AuditMiddlewareemits one structured event per (audited) request through the existingemit_security_event() sink, under an http.requestnamespace — so general request auditing and auth/CSRF telemetry flow through one pipeline (one sink, one SIEM forwarder).

It is opt-in and off by default (explicit-over-magic, like the auth leg): nothing is emitted unlessAuditConfig.level is raised above "none". Wire it viasecure_stack(config, audit=AuditConfig(level="metadata"))or app.add_middleware(AuditMiddleware(AuditConfig(...))).

Verbosity is tiered (NONE < METADATA < REQUEST < REQUEST_RESPONSE):

  • none— disabled; emit nothing.
  • metadata— who/what/when/status only (method, path, status_code, source_ip, user_agent, user_id). No body capture.
  • request— metadata plus a byte-capped, redacted snapshot of the request body for audited methods.
  • request_response — reserved; treated as requestfor body capture (response-body capture is intentionally not implemented to avoid buffering large/streaming responses).

Hypermedia correctness (the critical safety property): the middleware runs afternext()has resolved the response, then branches on the response type. ForStreamingResponse / SSEResponse / FileResponse (Chirp'sStream / Suspense / EventStreamreturn types resolve to these) it downgrades to metadata-only and never touches request.body() / request.form()— draining a streaming request would buffer unbounded or break the stream, and a request body already consumed by the handler is read from cache only when present.

Source IP comes fromtrusted_client_ip— the trusted-proxy-corrected client, never a re-parsed (spoofable)X-Forwarded-For.

User identity comes fromrequest.user.idand is only non-anonymous when AuthMiddlewareis wired upstream; without it the trail records an anonymous user (it never crashes).

middleware.audit

Name Type Default Description
type
qualified_name
element_type
description
source_file
line_number
is_autodoc
autodoc_element
_autodoc_template
_autodoc_url_path
_autodoc_page_type
title
doc_content_hash

Symbols on this page

_default_audited_methods
function
def _default_audited_methods() -> frozenset[str]

Load the canonical mutating-method set without an import-time cycle.

No parameters.

AuditConfig
class

Configuration forAuditMiddleware(frozen, thread-safe).

AuditMiddleware
class

Opt-in per-request audit trail over the security-event sink.

Usage::

from chirp.middleware.audit import AuditConfig, AuditMiddleware

app.add_middleware(AuditMiddleware(AuditConfig(level="metadata")))

Or as the outermost leg of the secure-by-default stack::

for mw in secure_stack(app.config, audit=AuditConfig(level="request")):
    app.add_middleware(mw)

Holds only the immutableAuditConfig in __slots__(no shared mutable state); the event sink is lock-guarded inchirp.security.audit.

_status_of
function
def _status_of(response: AnyResponse) -> int

Best-effort status code for any resolved response type.

Parameters

Name Type Default Description
response AnyResponse
_user_id_of
function
def _user_id_of(request: Request) -> str | None

Resolve the audited user id (anonymous when AuthMiddleware is absent).

Parameters

Name Type Default Description
request Request

View source · /home/runner/work/chirp/chirp/site/../src/chirp/middleware/audit.py:1