rust-py-audit 0.2.4

Fast, tamper-evident event auditing (hash chain) for Python apps, with a Rust core.

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	Writes local JSONL	Sends to ImmutableLog	Typical use
"local" (default)	✅	❌	The library's original behavior, no external dependency
"remote"	❌ (except for pending entries, see below)	✅	ImmutableLog is the single source of truth; a delivery failure raises an exception
"hybrid"	✅	✅	Local chain + remote receipt; a delivery failure NEVER raises — it becomes delivery_status="pending"

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, ...} if delivery failed (hybrid mode)

# Retries delivery of every event still marked "pending":
print(audit.flush_pending())
# {"flushed": 1, "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", the same scenario marks the event as delivery_status="pending", writes it to audit.pending.jsonl, and never raises — call audit.flush_pending() (manually, or from a cron/worker) to retry later.

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)

Parameter	Type	Description
app_name	str	Application name, recorded on every event
file_path	str	Path to the JSONL file. If it already exists, the chain resumes from the last recorded hash
mode	str | None	"local" (default) / "remote" / "hybrid". None falls back to RUST_PY_AUDIT_MODE, and finally to "local"
immutablelog_url	str | None	ImmutableLog base URL. None falls back to IMMUTABLELOG_URL. Required (one way or another) in mode="remote"/"hybrid"
immutablelog_api_key	str | None	API key (Bearer). None falls back to IMMUTABLELOG_API_KEY. Same requirement as immutablelog_url
timeout_ms	int	HTTP request timeout to ImmutableLog, in milliseconds
retry_enabled	bool	If True, retries retryable errors (5xx, timeout) up to max_retries times
max_retries	int	Maximum number of retries (in addition to the initial attempt)
immutablelog_env	str | None	Logical environment (meta.env, e.g. "production"). None falls back to IMMUTABLELOG_ENV; if neither is set, the field is omitted

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.

Event field	Type	Description
id	str	UUID v4
timestamp	str	RFC3339 / UTC, e.g.: 2026-06-17T10:00:00Z
app_name	str	Comes from the AuditLogger
actor_id	str	Who performed the action
action	str	E.g.: DELETE_INVOICE
resource	str	E.g.: invoice
resource_id	str	E.g.: inv_987
metadata	dict	Free-form — any JSON-serializable value
previous_hash	str | None	Hash of the previous event in the chain
hash	str	SHA-256 (64 hex chars) of the event + previous_hash
severity	str | absent	Only present if passed to log(). Becomes meta.type on ImmutableLog
immutable_trail	str | absent	Only present if passed to log() (and not empty after sanitization). Becomes meta.immutable_trail
immutablelog	dict | absent	Only present in mode="remote"/"hybrid". status is "delivered" or "pending"; other fields (tx_id, payload_hash, duplicate, request_id, ...) come from the ImmutableLog response

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, "still_pending": 0, "total": 1}

On success, updates event["immutablelog"] in audit.jsonl (without changing hash) and removes the event from the queue. Events that fail again stay in the queue for the next call.

---

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.