Audit Logging in Python Applications

An audit log answers a narrow question with a high standard of proof: who did what, to which resource, when, and with what result. It is evidence rather than diagnostics, and its requirements are nearly the opposite of operational logging — nothing may be sampled or dropped, records must be attributable and trustworthy, and retention is set by someone outside engineering. This page covers building that path in a Python service: the event schema, a dedicated logger, explicit actors, tamper evidence and durable writes. It is a task article under logging security and compliance, part of the Python logging fundamentals and structured data section.

Prerequisites

pip install "python-json-logger>=2.0.7,<4.0.0"

Implementation

Step 1 — Define the events and their fields first. An audit log is only as useful as its schema. Before writing code, list the actions that must be recorded — permission changes, data exports, credential rotations, approvals, account deletions — and fix the fields every record carries. Consistency matters more than richness: an auditor querying "every action by this actor last quarter" needs the actor field to have the same name and meaning everywhere.

from dataclasses import dataclass, asdict, field
from datetime import datetime, timezone

@dataclass(frozen=True)
class AuditEvent:
    action: str                       # "permission.grant", "data.export", …
    actor_id: str                     # who — a user or a service identity
    actor_type: str                   # "user" | "service"
    target_type: str                  # "user", "api_key", "dataset", …
    target_id: str
    outcome: str                      # "success" | "denied" | "error"
    reason: str | None = None         # required for some actions
    source_ip: str | None = None
    request_id: str | None = None
    at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())

Step 2 — Emit through a dedicated logger that does not propagate. A separate logger with its own handler and propagation disabled keeps audit events out of the operational stream entirely. They cannot be sampled away by a noisy-logger filter, raised above the configured level by an operator quietening things down, or deleted after seven days by the operational retention policy.

import logging

audit_logger = logging.getLogger("audit")
audit_logger.setLevel(logging.INFO)
audit_logger.propagate = False        # 1. never into the operational handlers

Step 3 — Pass the actor explicitly. Ambient context — a context variable set by authentication middleware — is convenient and is also the most common source of misattributed audit records. A record written from a background thread without the context propagated names no actor, or the wrong one. Requiring the actor as an argument makes a missing actor a programming error at the call site rather than a silent gap in the evidence.

def record(event: AuditEvent) -> None:
    if not event.actor_id:
        raise ValueError("audit events must name an actor")
    audit_logger.info(event.action, extra={"audit": asdict(event)})

record(AuditEvent(
    action="permission.grant", actor_id=admin.id, actor_type="user",
    target_type="user", target_id=target.id, outcome="success",
    reason="ticket SUP-4812", source_ip=request.client_ip,
    request_id=request.id))

Step 4 — Chain records so alteration is detectable. Each record includes a hash of its own canonical content and the previous record's hash. Altering, inserting or deleting any record breaks every subsequent hash, which a verification job detects. The chain lives per writer — per process or per stream — so concurrent writers do not interleave and invalidate each other's chains.

import hashlib
import json
import os
import threading

class ChainingAuditHandler(logging.Handler):
    """Writes one JSON line per event, each linked to the previous by hash."""

    def __init__(self, path: str, stream_id: str):
        super().__init__()
        self._fh = open(path, "a", buffering=1, encoding="utf-8")
        self._stream_id = stream_id
        self._prev = "0" * 64
        self._lock = threading.Lock()

    def emit(self, record: logging.LogRecord) -> None:
        body = dict(record.audit, stream=self._stream_id)
        with self._lock:
            canonical = json.dumps(body, sort_keys=True, separators=(",", ":"))
            digest = hashlib.sha256((self._prev + canonical).encode()).hexdigest()
            line = json.dumps({**body, "prev_hash": self._prev, "hash": digest},
                              sort_keys=True)
            self._fh.write(line + "\n")
            self._fh.flush()
            os.fsync(self._fh.fileno())        # 2. durable before the action returns
            self._prev = digest

Expected Output: records linked by hash.

{"action": "permission.grant", "actor_id": "u_12", "actor_type": "user", "at": "2026-09-18T10:21:07.412Z", "hash": "9c1f…a2", "outcome": "success", "prev_hash": "4e07…d1", "reason": "ticket SUP-4812", "stream": "api-3", "target_id": "u_88", "target_type": "user"}

Step 5 — Decide which actions must fail without a record. For actions whose audit trail is a hard requirement, the write must succeed before the action completes: performing a permission change without recording it is precisely the gap an audit exists to prevent. Writing synchronously with fsync, as above, and letting a failure propagate to the caller enforces that. For lower-stakes events a durable asynchronous path is acceptable, and the choice should be made per action and written down.

Step 6 — Verify the chain continuously. A chain nobody checks is not evidence. A scheduled job that recomputes hashes over each stream and compares them with the stored values, alerting on any mismatch, turns tamper evidence into tamper detection.

What makes an audit trail trustworthy

The technical controls above are necessary and not sufficient. An audit log earns trust through four properties, and each has an organisational side as well as a technical one.

Completeness. Every action in scope produces a record, including failed and denied attempts. A trail that records successful permission grants but not denied ones cannot show that an attacker tried and failed. Recording outcomes other than success is easy to forget and essential.

Attribution. Every record names who acted, reliably. Service accounts acting on behalf of users need both identities recorded — the service and the user it acted for — or an automated action becomes unattributable. This is the reason step 3 insists on explicit actors.

Integrity. Records cannot be altered or removed without detection. Hash chaining provides detection within a stream; storing periodic checkpoints — the latest hash, every hour — in a location the writing service cannot modify prevents the whole chain from being rewritten consistently. Write-once object storage is the usual destination for both.

Separation of duties. The people and systems that perform audited actions should not be able to modify the audit trail. A service that writes audit records and can also delete them has an audit trail that depends on the service's good behaviour. Granting the writer append-only access, and restricting deletion to a lifecycle policy nobody can change unilaterally, removes that dependency.

An audit log that satisfies all four holds up to scrutiny; one that satisfies the technical properties but lets the writing service delete its own records does not, however sophisticated its hashing. The connection to broader retention decisions is covered in log retention and tiering strategy.

Multiple processes and multiple replicas

A real service runs many processes, and the chain design has to accommodate that without making writers contend for a single global sequence.

The approach above gives each writer its own stream identifier and its own chain. A prefork server with four workers across six replicas produces twenty-four independent chains, each internally ordered and each verifiable on its own. This avoids any coordination between writers — no shared lock, no central sequence number — which matters because audit writes happen on the request path and must not become a bottleneck.

The cost is that there is no single global order across streams. Establishing the order of two events recorded by different workers relies on their timestamps, which is adequate for almost every audit question — who changed this permission, and when — and inadequate only for questions about sub-millisecond ordering across processes, which audit trails rarely need to answer.

The stream identifier must be stable for the life of the writer and unique across the fleet, and the verification job must know the full set of streams so it can notice when one disappears entirely. A stream that stops producing records mid-chain is either a process that ended — which should coincide with a termination — or a chain that was truncated, and distinguishing the two is one more thing the verification job can check by comparing stream end times with process lifetimes.

Configuration options

Verification

The verification job recomputes each stream's chain and reports the first break.

import hashlib, json

def verify(path: str) -> int | None:
    prev = "0" * 64
    for lineno, line in enumerate(open(path, encoding="utf-8"), 1):
        rec = json.loads(line)
        stored_hash, stored_prev = rec.pop("hash"), rec.pop("prev_hash")
        canonical = json.dumps(rec, sort_keys=True, separators=(",", ":"))
        if stored_prev != prev or \
           hashlib.sha256((prev + canonical).encode()).hexdigest() != stored_hash:
            return lineno
        prev = stored_hash
    return None

print(verify("/var/log/audit/api-3.jsonl") or "chain intact")

Expected Output: an intact chain, or the line at which it breaks.

chain intact

14207

Common mistakes

Audit events in the operational stream. Error signature: audit records missing because a filter, a level change or a retention policy removed them. Root cause: shared path. Remediation: a dedicated logger with its own destination.

Recording only successes. Error signature: no trace of an attacker's denied attempts. Root cause: audit calls only on the success path. Remediation: record denied and failed outcomes too.

Actors from ambient context. Error signature: records with an empty or wrong actor from background work. Root cause: context not propagated. Remediation: require the actor as an argument.

The writer can delete its own records. Error signature: an audit trail whose integrity depends on the service behaving. Root cause: full write access to the destination. Remediation: append-only access for the writer, deletion by lifecycle policy only.

A chain nobody verifies. Error signature: tampering discovered months later, or never. Root cause: evidence without detection. Remediation: a scheduled verification job with an alert.

Frequently Asked Questions