mobius-auditlog-py 1.0.2

CloudEvents-style audit event model, builder, integrity signing and Kafka publishing for Mobius Python (FastAPI) services.

mobius-auditlog-py

Build, sign, and publish CloudEvents-style audit events from Mobius Python (FastAPI / Starlette) services.

• PyPI: pip install mobius-auditlog-py
• Import package: mobius_auditlog
• Python: 3.10+

Install

pip install mobius-auditlog-py

pydantic, starlette, and py-kafka-producer-client are installed automatically.

Quick start (auto-capture middleware)

from fastapi import FastAPI
from kafka_producer_client import KafkaProducerConfig
from mobius_auditlog import setup_auditlog

app = FastAPI()

setup_auditlog(
    app,
    topic="audit-logs",
    signing_key="your-hmac-key",                # optional signature
    kafka_config=KafkaProducerConfig(bootstrap_servers="broker:9092"),
    capture_payloads=False,                      # set True to buffer req/resp bodies
    compliance_category="GDPR",
)

The middleware emits one audit event per request: action (route, method, status, severity), network (client IP + user-agent), objectref (from the path), and the envelope/subject from the request context. It skips health/metrics/swagger paths.

The event schema

AuditLogEvent — flat envelope + nested sections, serialized with empty values omitted and keys alphabetically ordered:

Section	Fields
envelope	id, specversion, timestamp, traceid, transactionid, tenantid
subject	userid, type, groups[]
kubernetes	namespacename, podname, containername, nodename
objectref	resource, resourceid, apiversion
action	name, method, severity, status
network	sourceip, useragent
eventdata	requestpayload{}, responsepayload{}, metadata{issensitive, compliancecategory}
security	hash, signature

event.to_dict()    # pruned dict (empties dropped), ready to publish
event.to_json()    # canonical JSON: pruned + sorted keys (used for hashing)

Building events manually

from mobius_auditlog import AuditLogBuilder

event = (
    AuditLogBuilder()
    .from_context()                              # tenant/trace/txn/subject from context
    .object_ref(resource="order", resourceid="order-789", apiversion="v1.0")
    .action(name="order.create", method="POST", severity="INFO", status="SUCCESS")
    .network(sourceip="1.2.3.4", useragent="curl/8")
    .event_data(requestpayload={"amount": 42}, issensitive=True, compliancecategory="PCI")
    .build()
)

from_context() pulls tenantid/traceid/transactionid/subject from the mobius-tracer-py request context when that package is installed, and kubernetes from the downward-API env vars (POD_NAMESPACE, POD_NAME, CONTAINER_NAME, NODE_NAME).

Integrity: hash + signature

from mobius_auditlog import seal, verify, compute_hash

seal(event, signing_key="key")     # sets security.hash (+ signature)
verify(event, signing_key="key")   # True if hash + signature match

• hash = SHA-256 over the canonical JSON excluding the security block.
• signature = HMAC-SHA256 of the hash with your key (omitted if no key).

The publisher seals events automatically before sending.

Publishing

The publisher depends only on a small protocol — no hard Kafka coupling:

class LogEventSender(Protocol):
    def send(self, value: dict, *, topic: str) -> Any: ...

setup_auditlog resolves a publisher from (in order) kafka_sender → kafka_producer → kafka_config → the configured py-kafka-producer-client singleton. Or use it directly:

from mobius_auditlog import AuditLogPublisher, PyKafkaProducerClientSender, set_publisher

publisher = AuditLogPublisher(PyKafkaProducerClientSender(client),
                              topic="audit-logs", signing_key="key")
set_publisher(publisher)
publisher.publish(event)           # seals + sends, fire-and-forget

Publishing runs on a background thread and never raises into the request path.

Payload capture

With capture_payloads=True, the middleware buffers request and response bodies (JSON-parsed, size-capped at 64 KB) into eventdata.requestpayload / responsepayload. Off by default for privacy and overhead. Non-JSON bodies are stored as {"_raw": "..."}.

Development

python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"
pytest