rust-py-audit

🌐 rust-py-audit.vercel.app

Event audit logging library for Python applications, with a Rust core.

Records audit events (who did what, when, on which resource) in a fast, structured way, and chains each event to the previous one with SHA-256 — any later edit, deletion, or reordering of the log file is detectable with verify().

Features

• AuditLogger — simple API: log(...), verify(), last_hash()
• Hash chain (SHA-256) — each event embeds the hash of the previous event; altering any recorded event breaks the chain in a detectable way
• JSONL storage — one event per line, append-only, no database required
• Thread-safe — a single AuditLogger can be shared across threads (e.g. a multi-threaded WSGI server, or one middleware instance serving concurrent requests); the hash chain stays linear under concurrency
• Free-form metadata — any JSON-serializable dict (IP, reason, request_id, etc.)
• FastAPI middleware — automatically logs state-changing requests (POST/PUT/PATCH/DELETE)
• Django middleware — same idea, supports WSGI and ASGI
• ImmutableLog integration — local/remote/hybrid modes, automatic retry, pending queue, and flush_pending() (see dedicated section)
• Rust core — hash generation, serialization, I/O, and the ImmutableLog HTTP client all run in Rust via PyO3; the Python API stays simple

Requirements

• Python 3.10+
• No required runtime dependencies

Optional, installed separately:

• fastapi + starlette — for rust_py_audit.fastapi.AuditMiddleware
• django — for rust_py_audit.django.AuditMiddleware

Installation

pip install rust-py-audit

With optional extras:

pip install "rust-py-audit[fastapi]"
pip install "rust-py-audit[django]"

Quick Start

from rust_py_audit import AuditLogger

audit = AuditLogger(app_name="billing-api", file_path="./audit.jsonl")

event = audit.log(
    actor_id="user_123",
    action="DELETE_INVOICE",
    resource="invoice",
    resource_id="inv_987",
    metadata={"ip": "192.168.0.10", "reason": "duplicate invoice"},
)

print(event["id"])     # uuid v4
print(event["hash"])   # sha256, 64 hex characters

print(audit.last_hash())  # hash of the last recorded event

result = audit.verify()
print(result)
# {"valid": True, "total_events": 1, "last_hash": "..."}

Chain integrity

Each event records the hash of the previous event (previous_hash) and its own hash (hash), computed from the event's content + previous_hash. The first event in the chain has previous_hash = null.

{"id":"evt_123","timestamp":"2026-06-17T10:00:00Z","app_name":"billing-api","actor_id":"user_123","action":"DELETE_INVOICE","resource":"invoice","resource_id":"inv_987","metadata":{"ip":"192.168.0.10"},"previous_hash":null,"hash":"abc123..."}

verify() re-reads the file from scratch and recomputes everything — it never trusts any in-memory cache:

result = audit.verify()

If the chain is intact:

{"valid": True, "total_events": 10, "last_hash": "..."}

If any event was edited, removed, or reordered:

{"valid": False, "total_events": 10, "error_index": 4, "reason": "hash_mismatch"}
# or "reason": "broken_chain" (removed/reordered/forged event)

FastAPI

from fastapi import FastAPI
from rust_py_audit.fastapi import AuditMiddleware

app = FastAPI()
app.add_middleware(AuditMiddleware, app_name="billing-api", file_path="./audit.jsonl")


@app.delete("/invoices/{invoice_id}")
async def delete_invoice(invoice_id: str):
    return {"deleted": invoice_id}

By default, only POST/PUT/PATCH/DELETE requests are logged. actor_id comes from the X-User-Id header (adjustable via actor_header=); falls back to "anonymous" if absent.

See the full example in examples/fastapi_app.py.

Django

# settings.py
MIDDLEWARE = [
    "rust_py_audit.django.AuditMiddleware",
    # ... other middlewares ...
]

# Optional:
RUST_PY_AUDIT_APP_NAME = "my-django-app"
RUST_PY_AUDIT_FILE_PATH = "./audit.jsonl"
RUST_PY_AUDIT_METHODS = {"POST", "PUT", "PATCH", "DELETE"}

actor_id comes from request.user.pk when there's an authenticated user (via django.contrib.auth); falls back to "anonymous" otherwise. The middleware supports both WSGI and ASGI applications automatically.

See the full example in examples/django_example/.

ImmutableLog Integration

rust_py_audit can send each audit event to ImmutableLog (documentation), in addition to — or instead of — writing locally.

Operating modes

mode="local" is the default — existing code calling AuditLogger(app_name, file_path) keeps working unchanged.

Basic example

from rust_py_audit import AuditLogger

audit = AuditLogger(
    app_name="billing-api",
    file_path="./audit.jsonl",
    mode="hybrid",
    immutablelog_url="https://api.immutablelog.com",
    immutablelog_api_key="iml_live_xxxxx",
    timeout_ms=500,
    retry_enabled=True,
    max_retries=3,
)

event = audit.log(
    actor_id="user_123",
    action="DELETE_INVOICE",
    resource="invoice",
    resource_id="inv_987",
    metadata={"ip": "192.168.0.10", "reason": "duplicate invoice"},
)

print(event["immutablelog"])
# {"status": "delivered", "tx_id": "tx_...", "payload_hash": "...", ...}
# or {"status": "pending", "tx_id": None, ...} on a transient failure (hybrid mode)
# or {"status": "failed", "tx_id": None, ...} on a permanent failure (hybrid mode)

# Retries delivery of every event still marked "pending":
print(audit.flush_pending())
# {"flushed": 1, "failed": 0, "still_pending": 0, "total": 1}

Environment variables

mode, immutablelog_url, and immutablelog_api_key accept None (the default) to fall back to an environment variable — handy for not hardcoding credentials:

export RUST_PY_AUDIT_MODE=hybrid
export IMMUTABLELOG_URL=https://api.immutablelog.com
export IMMUTABLELOG_API_KEY=iml_live_xxxxx

# Without passing mode/immutablelog_url/immutablelog_api_key explicitly,
# they come from the environment variables above:
audit = AuditLogger(app_name="billing-api", file_path="./audit.jsonl")

An explicit parameter always takes priority over the environment variable. mode="remote"/"hybrid" without immutablelog_url/immutablelog_api_key (neither as a parameter nor as an env var) raises ValueError when the AuditLogger is created — failing fast instead of only on the first log() call.

Severity, immutable_trail, and env

audit.log(...) accepts two optional parameters that only affect what gets sent to ImmutableLog (they never enter the hash):

event = audit.log(
    actor_id="user_123",
    action="DELETE_INVOICE",
    resource="invoice",
    resource_id="inv_987",
    severity="error",                    # meta.type — defaults to "info" if omitted
    immutable_trail="order-2026-00441",  # meta.immutable_trail — groups related events
)

• severity must be one of "error", "warning", "info", "success" — any other value raises ValueError, in any mode (even "local", where severity is just stored without being used).
• immutable_trail is sanitized automatically (trimmed, : replaced with -, truncated at 256 chars); if it ends up empty after that, the field is omitted instead of being sent broken.
• Both are preserved locally (without affecting hash) precisely so that flush_pending() can resend later with the same original classification.
• immutablelog_env (on the AuditLogger constructor, falling back to the IMMUTABLELOG_ENV env var) sets meta.env — useful for telling staging/production apart in ImmutableLog.

FastAPI

from fastapi import FastAPI
from rust_py_audit.fastapi import AuditMiddleware

app = FastAPI()
app.add_middleware(
    AuditMiddleware,
    app_name="billing-api",
    file_path="./audit.jsonl",
    mode="hybrid",
    immutablelog_url="https://api.immutablelog.com",
    immutablelog_api_key="iml_live_xxxxx",
    immutablelog_env="production",
    trail_header="X-Audit-Trail",  # default — read from the request, becomes meta.immutable_trail
)

In mode="remote"/"hybrid", the middleware computes severity automatically from the response's status_code (>=400 → "error", 300-399 → "info", 200-299 → "success"). In mode="remote", if delivery fails the middleware logs a logging.warning(...) and moves on — an audit failure never takes down the actual response already computed by the application.

Django

# settings.py
MIDDLEWARE = [
    "rust_py_audit.django.AuditMiddleware",
    # ... other middlewares ...
]

RUST_PY_AUDIT_MODE = "hybrid"
RUST_PY_AUDIT_FILE_PATH = "./audit.jsonl"
RUST_PY_AUDIT_IMMUTABLELOG_URL = "https://api.immutablelog.com"
RUST_PY_AUDIT_IMMUTABLELOG_API_KEY = "iml_live_xxxxx"
RUST_PY_AUDIT_IMMUTABLELOG_ENV = "production"
RUST_PY_AUDIT_TRAIL_HEADER = "X-Audit-Trail"  # default — read from the request, becomes meta.immutable_trail

Same behavior as FastAPI: severity computed from status_code, and delivery failures logged via logging.warning(...) without affecting the response.

Retry and idempotency

• retry_enabled/max_retries control how many times a retryable failure is retried (the same Idempotency-Key is used on every attempt — never creates duplicate events on ImmutableLog).
• Retryable: 5xx and timeouts.
• Permanent (never retried): 400, 401, 403, 429, and any other client error.
• In mode="remote", exhausting retries (or a permanent error) raises RuntimeError.
• In mode="hybrid", delivery never raises:
  • a retryable failure (5xx/timeout) marks the event status="pending" and queues it in audit.pending.jsonl — call audit.flush_pending() (manually, or from a cron/worker) to retry later;
  • a permanent failure (4xx) marks the event status="failed" and does not queue it (retrying would never succeed, so it stays out of the queue instead of getting stuck there forever). The event is still recorded locally and the chain stays valid — "failed" is just operational metadata.

In mode="hybrid", the event is appended to the local JSONL once, already carrying its final receipt (an O(1) append per event, not a full-file rewrite). A network failure still records the event locally (as pending/failed), so it is never lost to a failed delivery; the only loss window is the process being killed mid-request, and even then a delivery that did reach ImmutableLog is preserved in the remote store.

Integrity guarantee

The local hash (event["hash"]) is computed before any delivery attempt and never includes the immutablelog field — the remote receipt is operational metadata, attached afterward, and never invalidates verify():

audit.log(...)        # hash computed, event already recorded/chained
audit.flush_pending()  # only updates event["immutablelog"]; event["hash"] doesn't change
audit.verify()         # still valid, even after flush_pending()

API Reference

AuditLogger(app_name, file_path="./audit.jsonl", mode=None, immutablelog_url=None, immutablelog_api_key=None, timeout_ms=500, retry_enabled=True, max_retries=3, immutablelog_env=None)

See ImmutableLog Integration for details on each mode.

audit.log(actor_id, action, resource, resource_id, metadata=None, severity=None, immutable_trail=None) → dict

Records an event and returns the full event (already with id, timestamp, hash, etc.) as a dict. severity/immutable_trail are optional and only affect delivery to ImmutableLog — see Severity, immutable_trail, and env.

In mode="remote", a permanent failure or exhausted retries raise RuntimeError instead of returning the dict.

audit.verify() → dict

Re-reads the file and revalidates the entire chain from scratch. See Chain integrity. Unaffected by the immutablelog field — only the hashed fields matter (see Integrity guarantee).

audit.last_hash() → str | None

Hash of the last recorded event (in-memory cache, O(1)) — None if no event has been recorded yet.

audit.flush_pending() → dict

Attempts to redeliver to ImmutableLog every event marked as pending (recorded in audit.pending.jsonl, derived from file_path). Only relevant in mode="hybrid" — other modes never populate this queue.

{"flushed": 1, "failed": 0, "still_pending": 0, "total": 1}

For each queued event:

• delivered → updates event["immutablelog"] in audit.jsonl to "delivered" (without changing hash) and removes it from the queue (counts toward flushed);
• permanent failure → marks it "failed" in audit.jsonl and removes it from the queue, so it doesn't stay stuck forever (counts toward failed);
• retryable failure → left in the queue for the next call (counts toward still_pending).

Roadmap

rust-py-audit is mature (v0.3.0): the SHA-256 hash chain, JSONL storage, FastAPI/Django middlewares, and the ImmutableLog local/remote/hybrid integration are shipped and stable. Directional ideas under consideration (log rotation, incremental verification, more delivery backends, an async client, a Flask middleware) are tracked in ROADMAP.md.

Building from Source

Requires Rust and maturin.

git clone https://github.com/robertolima-dev/rust-py-audit
cd rust-py-audit

python3 -m venv .venv
source .venv/bin/activate
pip install maturin

# Development build (installs into the current Python environment)
maturin develop

# Release wheel
maturin build --release

Running tests

# Rust unit tests
cargo test --no-default-features

# Python integration tests
pip install -e ".[dev]"
pytest tests/

Architecture

Python API (rust_py_audit)
    ├── AuditLogger(...)         ──► src/audit_logger.rs (PyO3 #[pyclass])
    │       ├── log()            ──► src/event.rs               (AuditEvent)
    │       │                    ──► src/hash.rs                (deterministic SHA-256)
    │       │                    ──► src/storage.rs             (append/update in JSONL)
    │       │                    ──► src/immutablelog_client.rs (POST /v1/events, via reqwest)
    │       │                    ──► src/retry.rs               (retry with Idempotency-Key)
    │       ├── verify()         ──► src/verifier.rs (revalidates the local chain)
    │       ├── flush_pending()  ──► redelivers audit.pending.jsonl
    │       └── last_hash()      ──► in-memory cache
    │
    ├── fastapi.AuditMiddleware ──► audit.log() on every mutating request
    └── django.AuditMiddleware  ──► same idea, WSGI/ASGI

src/immutablelog_config.rs holds AuditMode/ImmutableLogConfig; src/immutablelog_receipt.rs defines the ImmutableLogReceipt attached to each event.

The core is compiled into a native extension (.so/.pyd) by maturin and PyO3. The Python layer is thin — it just routes calls and provides the framework adapters.

License

MIT — see LICENSE.