Audit Logging and Tamper-Evident Event Chains

WORM storage proves that a single DVIR record was not altered. It does not, by itself, prove that the full set of records is complete and in order — a deleted event, a quietly reordered pair, or an inserted back-dated entry can leave every surviving object individually intact. A tamper-evident event chain closes that gap: each entry embeds the SHA-256 audit_hash of the entry before it as prev_hash, so any deletion, insertion, or reorder breaks the linkage at every following step and is detected by re-walking the chain. This guide builds that append-only log and its verifier, the tamper-evidence half of the Immutable Audit Trail and WORM Retention design within the Core DVIR Architecture & FMCSA Compliance Mapping pipeline.

The stakes are concrete. When a DOT auditor asks the carrier to demonstrate that the inspection history for a vehicle is complete, “trust our database” is not an answer. A verifiable chain lets the carrier — and the auditor — recompute the linkage from the first entry to a securely-held head hash and show that no event between them could have been removed or altered without detection. This is what turns a mutable event table into evidence.

• Python 3.10+ — X | Y unions, match, and modern typing.
• hashlib and json (stdlib) — SHA-256 over canonical JSON, using the same canonicalization as the rest of the pipeline.
• Pydantic 2.x — immutable (frozen=True) chain-entry models.
• An append-only store with a single writer per chain — a partitioned Kafka topic, an append-only table, or a lock-guarded writer; concurrent appends to one chain are the primary correctness hazard.
• A secured head-hash register — a small, separately-protected store (an HSM value, a locked DynamoDB item) holding the current chain head, so the chain cannot be re-forged wholesale.
• The audit_hash (SHA-256 over canonical JSON) convention and the record_id / severity_score / compliance_guardrail field names from the parent section.

Each entry is frozen once built. It carries a gap-free seq, the prev_hash of the entry before it, and its own audit_hash. The genesis entry uses an all-zero prev_hash so the chain has a well-defined start.

import hashlib
import json

from pydantic import BaseModel

GENESIS = "0" * 64


class ChainEntry(BaseModel):
    model_config = {"frozen": True}

    seq: int
    record_id: str
    payload_hash: str   # audit_hash of the sealed DVIR record this entry attests
    prev_hash: str      # audit_hash of entry seq-1, or GENESIS
    audit_hash: str     # SHA-256 over this entry's own canonical fields


def _entry_digest(seq: int, record_id: str, payload_hash: str, prev_hash: str) -> str:
    canonical = json.dumps(
        {"seq": seq, "record_id": record_id,
         "payload_hash": payload_hash, "prev_hash": prev_hash},
        sort_keys=True, separators=(",", ":"), ensure_ascii=False,
    ).encode("utf-8")
    return hashlib.sha256(canonical).hexdigest()

Appending reads only the current head hash and the next sequence number — it does not re-scan the log — so it is O(1). The write must be single-writer per chain so two appends cannot both claim the same prev_hash.

def append(head: str | None, next_seq: int,
           record_id: str, payload_hash: str) -> ChainEntry:
    """O(1): link to the current head, compute this entry's digest, return it.
    Persist the entry, then advance the head register to entry.audit_hash."""
    prev = head or GENESIS
    digest = _entry_digest(next_seq, record_id, payload_hash, prev)
    return ChainEntry(
        seq=next_seq,
        record_id=record_id,
        payload_hash=payload_hash,
        prev_hash=prev,
        audit_hash=digest,
    )

Verification walks the entries in seq order. At each step it recomputes the entry’s digest and confirms two things: the recomputed digest equals the stored audit_hash (the entry is intact), and the entry’s prev_hash equals the previous entry’s audit_hash (the linkage is unbroken). It ends by confirming the last entry’s hash equals the securely-held head. Any failure names the exact seq where the chain broke.

import hmac


class ChainBreak(Exception):
    def __init__(self, seq: int, reason: str):
        super().__init__(f"chain broken at seq={seq}: {reason}")
        self.seq, self.reason = seq, reason


def verify_chain(entries: list[ChainEntry], trusted_head: str) -> bool:
    expected_prev = GENESIS
    for i, e in enumerate(entries):
        if e.seq != i:
            raise ChainBreak(e.seq, f"sequence gap: expected {i}")
        recomputed = _entry_digest(e.seq, e.record_id, e.payload_hash, e.prev_hash)
        if not hmac.compare_digest(recomputed, e.audit_hash):
            raise ChainBreak(e.seq, "entry payload altered")
        if not hmac.compare_digest(e.prev_hash, expected_prev):
            raise ChainBreak(e.seq, "prev_hash does not link to prior entry")
        expected_prev = e.audit_hash
    if not hmac.compare_digest(expected_prev, trusted_head):
        raise ChainBreak(len(entries) - 1, "head mismatch: entries truncated or added")
    return True

An auditor does not re-derive your business logic; they check the chain’s shape. Give them a read-only export and a verifier that answers four questions: is seq gap-free from 0, does every entry’s recomputed digest match, does every prev_hash link to the prior entry, and does the final hash equal the head you committed out-of-band. A pass on all four is the evidence that the inspection history is complete and unaltered for the retention window required by 49 CFR § 396.3©.

Prove the two properties that matter: tampering with a middle entry is detected, and appends are O(1). The first is the whole point; the second guards against a verifier accidentally being run on the append path.

import pytest


def _build_chain(n: int) -> tuple[list[ChainEntry], str]:
    entries, head = [], None
    for i in range(n):
        e = append(head, i, record_id=f"r{i}", payload_hash=f"{i:064x}")
        entries.append(e)
        head = e.audit_hash
    return entries, head


def test_tampering_with_middle_entry_breaks_verification():
    entries, head = _build_chain(5)
    # forge the payload_hash of the middle entry, leaving its stored audit_hash
    entries[2] = entries[2].model_copy(update={"payload_hash": "deadbeef" * 8})
    with pytest.raises(ChainBreak) as exc:
        verify_chain(entries, head)
    assert exc.value.seq == 2


def test_deleting_an_entry_is_detected():
    entries, head = _build_chain(5)
    del entries[3]  # remove one and renumber nothing
    with pytest.raises(ChainBreak):
        verify_chain(entries, head)


def test_append_is_constant_time(benchmark):
    entries, head = _build_chain(10_000)
    # appending to a 10k chain touches only the head + next seq, not the log
    result = benchmark(lambda: append(head, len(entries), "rX", "0" * 64))
    assert result.seq == 10_000

• Concurrent appends and ordering. If two writers append to one chain at once, both may read the same head and produce two entries with the same prev_hash — a fork. Enforce a single writer per chain (partition by vehicle or terminal, or serialize through one consumer) and assign seq from a monotonic source guarded by the same lock that advances the head.
• Chain forks. A fork is a chain with two entries claiming the same predecessor; verification detects it because only one branch can match the committed head. If a fork ever occurs, do not silently pick a branch — quarantine both and reconcile manually, because choosing wrong discards a real record.
• Storing the head insecurely. If the head hash lives in the same store as the log and an attacker can write both, they can delete a range of entries, re-link the chain, and update the head to match — verification would pass. Hold the head out-of-band (an HSM, a separately-permissioned item) and, for high assurance, periodically anchor it externally (a timestamping authority or a notarized digest) so even a full-store compromise is detectable.
• Non-canonical entry hashing. Verification recomputes each entry’s digest, so the append path and the verify path must serialize the entry’s fields identically. Reuse one canonicalization function for both; a stray field-order difference makes a clean chain fail to verify and sends an auditor chasing a phantom break.

• Immutable Audit Trail and WORM Retention — the audit-store design this chain anchors.
• WORM Retention and SHA-256 Hashing for DVIR Records — how each attested record is sealed and its audit_hash computed.
• Compliance Boundary Enforcement in Cloud Workflows — the role boundary that keeps the head register and log write-separated.
• Certifying Repairs Under 396.11©(2) — repair-state transitions that become entries in this chain.

Back to Immutable Audit Trail and WORM Retention, part of Core DVIR Architecture & FMCSA Compliance Mapping.