Building a Hash-Chained Audit Ledger in Python

Within the broader Cryptographic Audit Chains specification — and the wider Audit Logging & Compliance Evidence framework it sits in — this page is the concrete build: an append-only ledger, in plain Python, that records every state transition of a data subject request (DSR) as a link in a tamper-evident chain. The engineers who need it are privacy and platform teams who have decided that a mutable log index is not defensible evidence and want a small, dependency-light structure they can drop onto the critical path of a DSR pipeline. The failure it prevents is concrete: an audit record that a regulator, or opposing counsel, could plausibly claim was edited after the fact. Once you can hand over a head digest and say “recompute it,” that claim evaporates. This page builds the append path; its sibling, Verifying Audit Chain Integrity After Tampering, builds the detection path that reads what you write here.

The ledger has three moving parts — a validated event model, an append operation that computes each link, and a durable JSONL persistence layer — wired together as shown below:

Prerequisites #

The ledger targets Python 3.11+ for tuple[...] / list[...] generics without from __future__ friction and for datetime UTC handling that avoids the deprecated utcnow(). It uses only one third-party package plus the standard library:

• pydantic (v2) — strict validation of every event before it enters the chain. The code uses v2 idioms exclusively (model_config = ConfigDict(...), field_validator, model_validate); v1 patterns (class Config, @validator, .dict()) will not run.
• hashlib (stdlib) — the SHA-256 implementation. SHA-256 is a NIST-approved function under NIST SP 800-107, which is why it is the default here rather than a faster non-approved alternative.
• json, os, pathlib (stdlib) — canonical serialization and durable, flushed appends to a JSONL file.

Install the single dependency with pip install "pydantic>=2.6". No database, broker, or cloud client is required to build and test the ledger; durable object storage enters only at the WORM handoff, which is out of scope here. The event schema and canonicalization rules below are the same ones specified in the parent Cryptographic Audit Chains page.

Step-by-step implementation #

Step 1 — Model the audit event with a canonical serialization #

Every link is only as reproducible as the bytes it hashes, so the event model is where determinism is enforced. The model is frozen (immutable after construction), strict (no silent type coercion), and closed to extra fields (schema drift fails loudly). Its one job beyond validation is to emit canonical_bytes(): a single, stable byte string produced with sorted keys, no insignificant whitespace, and a UTC-normalized timestamp, so the same logical event hashes identically on any host and at any future date a regulator audits it.

from __future__ import annotations

import hashlib
import json
import os
from datetime import datetime, timezone
from enum import Enum
from pathlib import Path

from pydantic import BaseModel, ConfigDict, Field, field_validator


class DSRStage(str, Enum):
    """The ordered state transitions a DSR ledger records."""

    INTAKE = "intake"
    VERIFIED = "verified"
    ROUTED = "routed"
    DISCOVERED = "discovered"
    REDACTED = "redacted"
    DELIVERED = "delivered"
    CLOSED = "closed"


class AuditEvent(BaseModel):
    """One immutable, deterministically serializable audit record.

    ``frozen=True`` blocks post-construction mutation; ``strict=True`` refuses
    coercions that would silently change the hashed bytes; ``extra="forbid"``
    turns schema drift into a validation error rather than an unhashed field.
    """

    model_config = ConfigDict(strict=True, frozen=True, extra="forbid")

    request_id: str = Field(pattern=r"^dsr_[0-9a-f]{32}$")
    stage: DSRStage
    legal_basis: str = Field(min_length=3, max_length=64)
    actor: str = Field(min_length=1, max_length=128)
    at: datetime

    @field_validator("at")
    @classmethod
    def must_be_timezone_aware(cls, value: datetime) -> datetime:
        """Reject naive datetimes so the canonical timestamp never drifts."""
        if value.tzinfo is None:
            raise ValueError("at must be timezone-aware (UTC)")
        return value

    def canonical_bytes(self) -> bytes:
        """Return the one true byte string used to hash this event."""
        payload = {
            "request_id": self.request_id,
            "stage": self.stage.value,
            "legal_basis": self.legal_basis,
            "actor": self.actor,
            "at": self.at.astimezone(timezone.utc).isoformat(),
        }
        return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()

Validating at emission serves GDPR Art. 30 record-of-processing completeness directly: an event missing its legal_basis cannot enter the ledger as an unprovable gap, because construction fails before append() is ever reached.

Step 2 — Define the ledger record #

A record is what actually gets persisted: the event, the digest of the predecessor it chains from, and its own computed link digest. Modeling it as its own frozen Pydantic type (rather than a loose dict) means a corrupted or hand-edited JSONL line fails model_validate on load instead of silently verifying. The stored prev_hash and entry_hash are what a verifier recomputes.

class LedgerRecord(BaseModel):
    """A persisted chain link: the event plus its chaining digests."""

    model_config = ConfigDict(strict=True, frozen=True, extra="forbid")

    seq: int = Field(ge=0)
    prev_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
    entry_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
    event: AuditEvent

Step 3 — Build the append-only ledger #

The AppendOnlyLedger holds exactly one piece of mutable state — the head digest — and exposes a single mutating operation, append(). On construction the head is the fixed genesis anchor (sixty-four zeros), which commits the first real record to something an attacker cannot forge into a shorter valid chain. Each append() computes sha256(head + canonical_bytes), makes that digest the record’s entry_hash and the ledger’s new head, persists the record, and only then returns — so the head always reflects durable state, never a record that failed to write.

GENESIS: str = "0" * 64


class AppendOnlyLedger:
    """A durable, hash-chained, append-only DSR audit ledger.

    State is a single head digest. Records are persisted as JSON Lines,
    one record per line, flushed and fsync'd before ``append`` returns so
    the head never leads the durable file.
    """

    def __init__(self, path: str | Path) -> None:
        self._path = Path(path)
        self._head = GENESIS
        self._seq = 0
        if self._path.exists():
            self._restore()

    @property
    def head(self) -> str:
        """The digest fingerprinting the entire chain so far (32 bytes hex)."""
        return self._head

    @staticmethod
    def _link(prev_hash: str, event: AuditEvent) -> str:
        """Compute the tamper-evident digest for ``event`` after ``prev_hash``."""
        hasher = hashlib.sha256()
        hasher.update(prev_hash.encode("ascii"))
        hasher.update(event.canonical_bytes())
        return hasher.hexdigest()

    def append(self, event: AuditEvent) -> LedgerRecord:
        """Chain ``event`` onto the head and durably persist it."""
        entry_hash = self._link(self._head, event)
        record = LedgerRecord(
            seq=self._seq,
            prev_hash=self._head,
            entry_hash=entry_hash,
            event=event,
        )
        self._persist(record)
        # Advance in-memory state only after the write is durable.
        self._head = entry_hash
        self._seq += 1
        return record

    def _persist(self, record: LedgerRecord) -> None:
        """Append one record as a JSONL line, flushed and fsync'd to disk."""
        line = record.model_dump_json() + "\n"
        with self._path.open("a", encoding="utf-8") as handle:
            handle.write(line)
            handle.flush()
            os.fsync(handle.fileno())

    def _restore(self) -> None:
        """Rebuild head and sequence from an existing ledger file on startup."""
        with self._path.open("r", encoding="utf-8") as handle:
            for line in handle:
                line = line.strip()
                if not line:
                    continue
                record = LedgerRecord.model_validate_json(line)
                self._head = record.entry_hash
                self._seq = record.seq + 1

Because append() fsyncs before advancing the head, a crash mid-write leaves the durable file and the in-memory head consistent on the next _restore() — the partially written tail line simply fails model_validate_json and is the point from which you resume, never a silently accepted half-record.

Step 4 — Emit an event on the critical path #

The ledger is only evidence if the write happens before the action it records is allowed to proceed. Wiring it into a pipeline transition therefore means: build the event, append it, and let the failure of the append block the transition rather than letting the work run unlogged.

def record_transition(
    ledger: AppendOnlyLedger,
    request_id: str,
    stage: DSRStage,
    legal_basis: str,
    actor: str,
) -> str:
    """Append a transition event and return the ledger's new head digest.

    Raises before the caller proceeds if the event is invalid or the append
    cannot be persisted, so no DSR action outruns its own evidence.
    """
    event = AuditEvent(
        request_id=request_id,
        stage=stage,
        legal_basis=legal_basis,
        actor=actor,
        at=datetime.now(timezone.utc),
    )
    record = ledger.append(event)
    return record.entry_hash

Configuration reference #

Verification #

The property to prove is that a freshly built chain verifies end to end and that its head is reproducible. This unit test appends a short DSR lifecycle, walks the chain from genesis recomputing every link, and asserts both integrity and a stable, non-genesis head.

from datetime import datetime, timezone


def _fixed_event(ledger_stage: DSRStage) -> AuditEvent:
    return AuditEvent(
        request_id="dsr_" + "a" * 32,
        stage=ledger_stage,
        legal_basis="art6-1-c",
        actor="svc:pipeline",
        at=datetime(2026, 7, 16, 12, 0, tzinfo=timezone.utc),
    )


def test_chain_verifies_end_to_end(tmp_path) -> None:
    ledger = AppendOnlyLedger(tmp_path / "ledger.jsonl")
    for stage in (DSRStage.INTAKE, DSRStage.VERIFIED, DSRStage.CLOSED):
        ledger.append(_fixed_event(stage))

    # Re-read the persisted records and recompute the chain from genesis.
    records = [
        LedgerRecord.model_validate_json(line)
        for line in (tmp_path / "ledger.jsonl").read_text().splitlines()
    ]
    prev = GENESIS
    for record in records:
        assert record.prev_hash == prev
        assert record.entry_hash == AppendOnlyLedger._link(prev, record.event)
        prev = record.entry_hash

    assert ledger.head == prev          # in-memory head matches durable chain
    assert ledger.head != GENESIS       # a non-empty chain moved off the anchor

Running under pytest should report the test passing. A second run against the same tmp_path exercises _restore(): reopen the ledger and assert head matches the value from the first run, confirming crash-recovery rebuilds identical state.

Troubleshooting #

Non-deterministic head across machines — Root cause: serialization that is not canonical, usually unsorted keys, a locale-dependent float, or a naive timestamp. Fix: hash only canonical_bytes() (sorted keys, separators=(",", ":"), UTC ISO-8601) and keep the field_validator that rejects naive datetimes, so the same event always yields the same bytes.

ValidationError on load of an old ledger — Root cause: a schema change (a renamed or added field) makes historical JSONL lines fail model_validate_json. Fix: never mutate the event schema in place; version it and keep a reader for each version, since sealed records are immutable by design and must remain loadable exactly as written.

Head advances but the file is missing the last line — Root cause: the head was updated before the write was durable (for example fsync removed or exceptions swallowed). Fix: keep the ordering in append() — persist and fsync first, advance _head and _seq only after — so the in-memory head can never lead the durable file.

Forked chain: two records share a prev_hash — Root cause: concurrent append() calls read the same head before either persisted. Fix: serialize appends behind a single-writer lock (or an atomic compare-and-set on the head); a single ledger instance must have exactly one writer.

Duplicate events inflate the chain — Root cause: a retried pipeline transition appends the same logical event twice. Fix: make transitions idempotent upstream on request_id plus stage, so a retry re-reads the existing head rather than chaining a redundant link. The sibling Verifying Audit Chain Integrity After Tampering page shows how to walk this ledger and prove no such anomaly was introduced maliciously.

Frequently Asked Questions #

Related #

• Cryptographic Audit Chains — the parent specification for canonicalization, link construction, and chain granularity.
• Verifying Audit Chain Integrity After Tampering — the detection path that walks the ledger this page writes and localizes the first broken link.
• WORM Storage for DSR Evidence — anchoring the JSONL ledger in write-once storage so it is immutable, not merely tamper-evident.
• Closure Manifests & Evidence Packaging — sealing the ledger’s head digest and events into a regulator-verifiable artifact.
• Audit Logging & Compliance Evidence — the parent section this ledger serves.