Designing Append-Only Fare Ledgers with Hash Chaining

This page implements a hash-chained, append-only fare ledger you can drop into a settlement worker: every entry carries the hash of the one before it, a verify() walk recomputes the whole chain and pinpoints any tampering, and the only way to reverse money is a compensating entry — never an edit. It is the concrete build behind the Ledger & Audit Trail Design component inside the Revenue Reconciliation & Settlement pipeline, written for the revenue analysts and Python developers who have to prove to an auditor that a fare total was not quietly changed after the books closed. Every amount is a Decimal in minor units; a single float here is a restated quarter later.

Hash-Chain Flow

Each append seals an entry by hashing its canonical body together with the previous entry’s hash, then the new hash becomes the head that the next append binds to. A verifier re-walks the same links: if any body was altered, the recomputed hash diverges from the stored one at exactly that entry and every entry after it. The flow below traces one append and the verify walk that validates it:

Step 1 — Canonicalize and Hash a Single Entry

The chain is only reproducible if every entry serializes to the exact same bytes on any machine. Sort the keys, fix the JSON separators, and render every Decimal through str() — which is lossless — rather than float(). The hash mixes the previous hash in first, separated from the body by a byte that can never appear inside the JSON, so a prev_hash boundary can never be forged by shifting bytes between the two.

from __future__ import annotations

import hashlib
import json
from dataclasses import dataclass, field
from datetime import datetime, timezone
from decimal import Decimal

GENESIS_HASH = "0" * 64
_SEP = b"\x1e"  # record separator, never present in JSON output


@dataclass(frozen=True)
class FareEntry:
    entry_key: str
    account: str
    amount_minor: Decimal      # signed minor units; credit +, debit -
    kind: str                  # 'fare' | 'refund' | 'compensation'
    occurred_utc: datetime
    prev_hash: str
    corrects: str | None = None
    meta: dict = field(default_factory=dict)

    def canonical_body(self) -> bytes:
        payload = {
            "entry_key": self.entry_key,
            "account": self.account,
            "amount_minor": str(self.amount_minor),
            "kind": self.kind,
            "occurred_utc": self.occurred_utc.astimezone(timezone.utc).isoformat(),
            "corrects": self.corrects,
            "meta": self.meta,
        }
        return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")

    def seal(self) -> str:
        """Hash prev_hash together with the canonical body."""
        digest = hashlib.sha256()
        digest.update(self.prev_hash.encode("ascii"))
        digest.update(_SEP)
        digest.update(self.canonical_body())
        return digest.hexdigest()

The separator matters more than it looks: without it, an attacker could move characters across the prev_hash/body seam and produce a colliding preimage. A byte that the serializer provably never emits closes that gap.

Step 2 — Build the Append-Only Ledger

The ledger holds the entries, the set of seen entry_keys for idempotency, and the current head. append refuses float, refuses naive timestamps, refuses a duplicate key, seals the entry against the head, and advances the head. There is no update method by design.

import logging
from typing import Iterator

logger = logging.getLogger("transit.fare_ledger")


class ChainBroken(Exception):
    """Raised by verify() when the chain no longer validates."""


class SealedFareEntry(FareEntry):
    """A FareEntry that has been hashed and committed."""
    entry_hash: str


class FareLedger:
    def __init__(self) -> None:
        self._entries: list[FareEntry] = []
        self._hashes: list[str] = []
        self._keys: set[str] = set()
        self._head: str = GENESIS_HASH

    @staticmethod
    def entry_key(source: str, natural_id: str) -> str:
        return hashlib.sha256(f"{source}:{natural_id}".encode()).hexdigest()

    def append(self, *, entry_key: str, account: str, amount_minor: Decimal,
               kind: str, occurred_utc: datetime, corrects: str | None = None,
               meta: dict | None = None) -> str:
        if not isinstance(amount_minor, Decimal):
            raise TypeError("amount_minor must be Decimal")
        if occurred_utc.tzinfo is None:
            raise ValueError("occurred_utc must be timezone-aware")
        if entry_key in self._keys:
            logger.info("idempotent skip for entry_key=%s", entry_key[:12])
            return self._hashes[self._index_of(entry_key)]

        entry = FareEntry(
            entry_key=entry_key, account=account, amount_minor=amount_minor,
            kind=kind, occurred_utc=occurred_utc, prev_hash=self._head,
            corrects=corrects, meta=meta or {},
        )
        entry_hash = entry.seal()
        self._entries.append(entry)
        self._hashes.append(entry_hash)
        self._keys.add(entry_key)
        self._head = entry_hash
        return entry_hash

    def compensate(self, target_key: str, reason: str) -> str:
        """Reverse a booked entry with its signed inverse; never edit."""
        idx = self._index_of(target_key)
        target = self._entries[idx]
        return self.append(
            entry_key=self.entry_key("compensation", target_key),
            account=target.account,
            amount_minor=-target.amount_minor,
            kind="compensation",
            occurred_utc=datetime.now(timezone.utc),
            corrects=target_key,
            meta={"reason": reason},
        )

    def balance(self, account: str) -> Decimal:
        return sum((e.amount_minor for e in self._entries if e.account == account),
                   Decimal("0"))

    def _index_of(self, entry_key: str) -> int:
        for i, e in enumerate(self._entries):
            if e.entry_key == entry_key:
                return i
        raise KeyError(entry_key)

    def __iter__(self) -> Iterator[tuple[FareEntry, str]]:
        return zip(self._entries, self._hashes)

Note append returns the existing hash on a duplicate key rather than raising: at this call site an idempotent replay is a normal, expected event, so returning the prior hash lets the caller proceed without special-casing redelivery.

Step 3 — Verify the Chain

Verification re-derives every hash from the genesis forward. Two things can be wrong: an entry’s stored prev_hash might not match the running hash (a broken link, meaning an entry was inserted or removed), or an entry’s recomputed hash might not match what was stored against it (a mutated body). Either raises with the offending index.

def verify(self) -> int:
    """Return the number of entries verified, or raise ChainBroken at the
    first divergence, naming the index so the tamper can be located."""
    prev = GENESIS_HASH
    for idx, (entry, stored_hash) in enumerate(zip(self._entries, self._hashes)):
        if entry.prev_hash != prev:
            raise ChainBroken(f"broken link at index {idx}: "
                              f"prev_hash {entry.prev_hash[:12]} != {prev[:12]}")
        recomputed = entry.seal()
        if recomputed != stored_hash:
            raise ChainBroken(f"tampered entry at index {idx}: "
                              f"body hash {recomputed[:12]} != stored {stored_hash[:12]}")
        prev = stored_hash
    return len(self._entries)


FareLedger.verify = verify

Because each hash feeds the next, a single altered amount at index 4 changes that entry’s hash, which no longer matches index 5’s prev_hash — so verify() stops at the earliest corruption and reports precisely where the chain first diverges.

Validation & Test Cases

The tests cover a clean chain that verifies, a correction booked as a compensating entry that keeps the balance right, and a deliberately mutated middle entry that the verifier catches.

from decimal import Decimal
from datetime import datetime, timezone

def _utc(y, mo, d, h, mi):
    return datetime(y, mo, d, h, mi, tzinfo=timezone.utc)

led = FareLedger()
led.append(entry_key=FareLedger.entry_key("tap", "T1"), account="OP_METRO",
           amount_minor=Decimal("290"), kind="fare", occurred_utc=_utc(2026, 7, 3, 8, 0))
led.append(entry_key=FareLedger.entry_key("tap", "T2"), account="OP_METRO",
           amount_minor=Decimal("290"), kind="fare", occurred_utc=_utc(2026, 7, 3, 9, 0))
led.append(entry_key=FareLedger.entry_key("tap", "T3"), account="OP_METRO",
           amount_minor=Decimal("175"), kind="fare", occurred_utc=_utc(2026, 7, 3, 9, 5))

# Normal case: a clean chain verifies and the balance is the exact sum.
assert led.verify() == 3
assert led.balance("OP_METRO") == Decimal("755")

# Idempotent redelivery: the same tap booked twice does not double-count.
h = led.append(entry_key=FareLedger.entry_key("tap", "T1"), account="OP_METRO",
               amount_minor=Decimal("290"), kind="fare", occurred_utc=_utc(2026, 7, 3, 8, 0))
assert led.balance("OP_METRO") == Decimal("755")

# Correction: reverse T3 with a compensating entry, never an edit.
led.compensate(FareLedger.entry_key("tap", "T3"), reason="disputed overcharge")
assert led.verify() == 4
assert led.balance("OP_METRO") == Decimal("580")  # 755 - 175

# Edge case: tamper with a middle entry's amount in place.
object.__setattr__(led._entries[1], "amount_minor", Decimal("999"))
try:
    led.verify()
    raise SystemExit("verify should have detected the tamper")
except ChainBroken as exc:
    assert "index 1" in str(exc)

The last block is the one that matters for an audit. Mutating amount_minor on the entry at index 1 changes its recomputed hash, so verify() raises ChainBroken naming index 1 — the tamper cannot hide, and it cannot be localized to a single row because every subsequent prev_hash is now wrong too. A legitimate correction, by contrast, appends and keeps verify() green.

Edge Cases & Gotchas

• Never mutate to correct. A refund, an overcharge reversal, or a settlement adjustment is always a new compensating entry. The object.__setattr__ in the test only exists to simulate an attacker; production code has no such path.
• Serialize appends. Two writers extending the same head fork the chain. One writer per partition, or a database row lock on the head, is mandatory.
• Freeze the canonical format. A library change to how json.dumps orders or spaces output would change every hash. Cover canonical_body with a golden-vector test.
• Keep amounts Decimal. append rejects float, but also forbid arithmetic that promotes to float upstream — Decimal("290") / 3 stays Decimal, 290 / 3 does not.

Integration Note

This ledger is the storage substrate the parent Ledger & Audit Trail Design topic specifies, and its verified balances are what the reporting layer reads. Its closest sibling is Generating NTD Revenue Reports from Fare Ledgers: that report aggregates exactly the entries this chain protects and asserts its period totals tie back to balance(), so the hash-chain guarantee here is what makes the regulatory filing defensible under audit.

FAQ

Related

• Ledger & Audit Trail Design — the parent component this chain implements.
• Generating NTD Revenue Reports from Fare Ledgers — the report that aggregates and ties back to these entries.
• Daily Variance Reconciliation — matches settlement files against the balances this chain protects.
• AFC System Security Boundaries — the authentication every entry clears before it is booked.

Part of Ledger & Audit Trail Design, within Revenue Reconciliation & Settlement.