Building an Append-Only Hash-Chained Audit Log
A billing audit trail is only trustworthy if altering a past entry is detectable, not merely discouraged. Permissions and updated_at columns prove nothing: a privileged insider — or a bug — can rewrite a row and leave no trace. The fix is a hash chain, where each entry commits to the hash of the entry before it, so that changing any historical byte invalidates every hash from that point to the head. This page sits under GASB audit-trail compliance patterns, part of the broader Billing Reconciliation & Audit Logging subsystem, and builds the one component that page’s Step 2 only sketched: a focused AuditLog class whose append computes a canonical-JSON SHA3-256 hash chained to its predecessor, and whose verify_chain walks the trail and returns the exact index of any tamper.
Minimal Prerequisites
- Python 3.11+ — for
zoneinfo,StrEnum, and timezone-awaredatetime. Every entry’s timestamp is a UTC instant; a naivedatetimeis never chained. hashlib(standard library) — the chain usessha3_256; no third-party dependency, so the verifier is reproducible on any conformant interpreter.- Deterministic
jsonserialization — hashing is overjson.dumps(..., sort_keys=True, separators=(",", ":"), ensure_ascii=False). The same event must always produce the same bytes. decimal.Decimalfor amounts, serialized as strings — afloatin a payload will not survive a byte-exact replay.
Data assumptions: events arrive already free of personally identifiable information, referencing subjects by an opaque surrogate key. The store here is an in-process list to keep the mechanism legible; in production the same append-only contract is backed by object-lock storage or an INSERT-only table, as covered in the parent page.
Annotated Implementation
The class owns three things: the ordered list of committed entries, the running head hash, and a lock that serializes appends so two writers cannot fork the chain. Each entry stores its own seq, the prev hash it was built on, its logged_utc, the event record, and the resulting hash. Verification never trusts a stored hash; it recomputes from the other stored fields, which is what makes tampering detectable.
from __future__ import annotations
import hashlib
import json
import threading
from datetime import datetime, timezone
from decimal import Decimal
GENESIS_PREV = "0" * 64 # fixed anchor: the "previous hash" of the first real entry
def _canonical(frame: dict) -> bytes:
"""Deterministic bytes for hashing: sorted keys, no incidental whitespace."""
return json.dumps(
frame, sort_keys=True, separators=(",", ":"), ensure_ascii=False
).encode("utf-8")
def _entry_hash(record: dict, prev: str, seq: int, logged_utc: str) -> str:
# Bind position (seq) and log time into the hash, not just the payload, so an
# entry cannot be reordered or backdated without breaking every later hash.
frame = {"seq": seq, "prev": prev, "logged_utc": logged_utc, "record": record}
return hashlib.sha3_256(_canonical(frame)).hexdigest()
def _normalize(record: dict) -> dict:
"""Serialize Decimal amounts to strings so exact values survive replay."""
return {k: (str(v) if isinstance(v, Decimal) else v) for k, v in record.items()}
class AuditLog:
"""Append-only, hash-chained log of billing events."""
def __init__(self) -> None:
self._entries: list[dict] = []
self._lock = threading.Lock()
@property
def head_hash(self) -> str:
# The genesis anchor when empty; otherwise the last entry's hash.
return self._entries[-1]["hash"] if self._entries else GENESIS_PREV
def append(self, record: dict) -> dict:
"""Chain one event to the current head and commit it. Thread-safe."""
norm = _normalize(record)
# Serialize the whole read-modify-write so concurrent callers cannot both
# build on the same head and fork the chain.
with self._lock:
seq = len(self._entries)
prev = self.head_hash
logged_utc = datetime.now(timezone.utc).isoformat()
digest = _entry_hash(norm, prev, seq, logged_utc)
entry = {"seq": seq, "prev": prev, "logged_utc": logged_utc,
"record": norm, "hash": digest}
self._entries.append(entry) # append only; never reassign an index
return entry
def verify_chain(self) -> tuple[bool, int | None]:
"""Walk genesis → head, recomputing each hash.
Returns (True, None) if intact, else (False, index_of_first_break).
"""
prev = GENESIS_PREV
for i, e in enumerate(self._entries):
expected = _entry_hash(e["record"], prev, e["seq"], e["logged_utc"])
# Three independent invariants must all hold at position i.
if e["seq"] != i or e["prev"] != prev or e["hash"] != expected:
return (False, i)
prev = e["hash"]
return (True, None)
def entries(self) -> list[dict]:
return list(self._entries) # defensive copy; callers cannot mutate in place
Two design choices carry the guarantee. First, the hash covers a frame that includes seq and logged_utc, not just the event record — so an attacker who swaps two entries or backdates one invalidates the recomputation even if the payloads are individually valid. Second, append performs the read of head_hash and the write of the new entry inside a single lock, which is what stops two concurrent workers from both chaining to the same predecessor and producing a fork. The genesis case needs no special branch: an empty log reports head_hash as GENESIS_PREV, so the first real entry chains to the fixed anchor exactly like every later entry chains to its predecessor.
Edge Cases and Billing Gotchas
Chaining is simple until the awkward realities of a production billing run arrive. Each of these has a specific, tested answer.
- Ordering vs. clock skew. The chain orders by
seq(append order), never by timestamp. Keep the event’s business time in the payload (occurred_utc) separate from the entry’slogged_utc; a late-arriving event is appended at the head with an earlier business time, and that gap is auditable rather than an error. Never sort committed entries by time — that reordersseqand breaks every hash. - Canonicalization drift. If two callers serialize the same event differently — a stray space, unsorted keys, or
ensure_asciiflipping the encoding of a non-ASCII service address code — their hashes differ and verification fails spuriously. Route every hash through the single_canonicalhelper and pin its options; treat a serialization change as a breaking version bump. - The genesis entry. The first real entry must chain to a fixed, well-known anchor, not to “nothing.”
head_hashreturnsGENESIS_PREVwhen the log is empty, so the firstappenduses the anchor as itsprevwith no special-case branch — andverify_chainstarts its walk from the same anchor. - Concurrent appends. Two workers that both read the head and append will fork the chain (two entries claiming the same
prev). The lock around the read-modify-write inappendserializes them; in a multi-process deployment, replace the in-process lock with aUNIQUE(seq)database constraint or a single-writer queue so exactly one entry can claim each position.
Verification Snippet
Prove both properties directly: an intact chain verifies, and a single altered byte is caught at the right index.
def test_chain_detects_tampering() -> None:
log = AuditLog()
log.append({"event_type": "bill_issued", "subject_ref": "acct:9f3c",
"amount": Decimal("41.07")})
log.append({"event_type": "adjustment_posted", "subject_ref": "acct:9f3c",
"amount": Decimal("-5.00")})
log.append({"event_type": "payment_applied", "subject_ref": "acct:9f3c",
"amount": Decimal("36.07")})
# A clean chain verifies end to end.
assert log.verify_chain() == (True, None)
# Amounts round-trip exactly as Decimal-derived strings.
assert log.entries()[0]["record"]["amount"] == "41.07"
# Tamper with entry 1's stored record, leaving its hash untouched.
log.entries() # copy for inspection; now mutate the internal list directly
log._entries[1]["record"]["amount"] = "-500.00"
intact, break_at = log.verify_chain()
assert intact is False
assert break_at == 1 # first break localized to the edited entry
Running this in continuous integration turns the tamper-evidence guarantee into an executed test rather than a claim. Extend it with a fixture that inserts or deletes an entry (breaking seq) to confirm the same localization behavior for structural tampering, not just content edits.
FAQ
Related Topics
- GASB Audit-Trail Compliance Patterns — the parent design: event taxonomy, write-once storage, retention, and PII separation around this class.
- Reconciling a Billing Run Against the General Ledger — the reconciliation whose sign-off is appended to this log.
- Hash-Total Validation Controls — the same hashing discipline applied to batch totals rather than to an event chain.
- Data Governance & Privacy Compliance — why the events chained here must stay free of customer PII.
Up: GASB Audit-Trail Compliance Patterns · Billing Reconciliation & Audit Logging