# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/through-what-append-only-and-hash-chained-ledger-invariants-must-an-instance-record-audit-and-state-transitions-to-prevent-silent-retroactive-mutation.md # Remnant Warrant: https://corpusfetch.com/c/through-what-append-only-and-hash-chained-ledger-invariants-must-an-instance-record-audit-and-state-transitions-to-prevent-silent-retroactive-mutation.warrant.json # Markdown SHA-256: ff214ea45ae0b59565d9c2b447e0d978ed4c2544e779c284b0508b36bc63d5df # Code AST SHA-256: 80a44e45df05a41231e85086ed6b588d8254bfd0113fc9de72f7321bc51da205 # Verified Classes: LedgerIntegrityError, LedgerEntry, TamperEvidentLedger # Defensive Guards: 4 fail-closed invariants (zero external deps) import hashlib import json import sqlite3 from dataclasses import dataclass from datetime import datetime, timezone from typing import Any, Mapping GENESIS_HASH = "0" * 64 class LedgerIntegrityError(RuntimeError): '''Raised when a hash-chain discontinuity, sequence gap, or mutation is detected.''' pass @dataclass(frozen=True) class LedgerEntry: stream_id: str seq_no: int recorded_at_utc: str entry_type: str amount_delta_cents: int metadata_json: str prev_hash: str entry_hash: str class TamperEvidentLedger: ''' An append-only, SHA-256 hash-chained ledger backed by SQLite with engine-enforced triggers that block UPDATE and DELETE operations. ''' def __init__(self, conn: sqlite3.Connection) -> None: self.conn = conn self._init_schema() def _init_schema(self) -> None: with self.conn: self.conn.execute( "CREATE TABLE IF NOT EXISTS ledger_entries (" " stream_id TEXT NOT NULL," " seq_no INTEGER NOT NULL," " recorded_at_utc TEXT NOT NULL," " entry_type TEXT NOT NULL," " amount_delta_cents INTEGER NOT NULL," " metadata_json TEXT NOT NULL," " prev_hash TEXT NOT NULL," " entry_hash TEXT NOT NULL," " PRIMARY KEY (stream_id, seq_no)" ")" ) # Gate 1: Database-level triggers strictly forbidding UPDATE or DELETE self.conn.execute( "CREATE TRIGGER IF NOT EXISTS forbid_ledger_update " "BEFORE UPDATE ON ledger_entries " "BEGIN " " SELECT RAISE(ABORT, 'Immutable ledger violation: UPDATE is forbidden'); " "END;" ) self.conn.execute( "CREATE TRIGGER IF NOT EXISTS forbid_ledger_delete " "BEFORE DELETE ON ledger_entries " "BEGIN " " SELECT RAISE(ABORT, 'Immutable ledger violation: DELETE is forbidden'); " "END;" ) @staticmethod def _compute_entry_hash( stream_id: str, seq_no: int, recorded_at_utc: str, entry_type: str, amount_delta_cents: int, metadata_json: str, prev_hash: str, ) -> str: canonical_record = "|".join([ stream_id, str(seq_no), recorded_at_utc, entry_type, str(amount_delta_cents), metadata_json, prev_hash, ]) return hashlib.sha256(canonical_record.encode("utf-8")).hexdigest() def append_entry( self, stream_id: str, entry_type: str, amount_delta_cents: int, metadata: Mapping[str, Any] | None = None, ) -> LedgerEntry: ''' Appends a new immutable event to `stream_id`, chained cryptographically to the previous entry's SHA-256 hash. ''' if isinstance(amount_delta_cents, bool) or not isinstance(amount_delta_cents, int): raise TypeError("amount_delta_cents must be an exact integer") meta_str = json.dumps(metadata or {}, sort_keys=True, separators=(",", ":")) now_utc = datetime.now(timezone.utc).isoformat() with self.conn: row = self.conn.execute( "SELECT seq_no, entry_hash FROM ledger_entries " "WHERE stream_id = ? ORDER BY seq_no DESC LIMIT 1", (stream_id,), ).fetchone() if row is None: next_seq = 1 prev_hash = GENESIS_HASH else: next_seq = int(row[0]) + 1 prev_hash = str(row[1]) entry_hash = self._compute_entry_hash( stream_id, next_seq, now_utc, entry_type, amount_delta_cents, meta_str, prev_hash, ) self.conn.execute( "INSERT INTO ledger_entries (" " stream_id, seq_no, recorded_at_utc, entry_type, " " amount_delta_cents, metadata_json, prev_hash, entry_hash" ") VALUES (?, ?, ?, ?, ?, ?, ?, ?)", ( stream_id, next_seq, now_utc, entry_type, amount_delta_cents, meta_str, prev_hash, entry_hash, ), ) return LedgerEntry( stream_id=stream_id, seq_no=next_seq, recorded_at_utc=now_utc, entry_type=entry_type, amount_delta_cents=amount_delta_cents, metadata_json=meta_str, prev_hash=prev_hash, entry_hash=entry_hash, ) def verify_stream_integrity(self, stream_id: str) -> int: ''' Recomputes the entire SHA-256 hash chain and verifies gapless monotonic sequence numbers from seq_no=1 to the head of the stream. Returns the verified net balance in cents. ''' rows = self.conn.execute( "SELECT seq_no, recorded_at_utc, entry_type, amount_delta_cents, " " metadata_json, prev_hash, entry_hash " "FROM ledger_entries WHERE stream_id = ? ORDER BY seq_no ASC", (stream_id,), ).fetchall() expected_seq = 1 expected_prev_hash = GENESIS_HASH net_balance_cents = 0 for row in rows: seq_no, ts, etype, delta, meta, prev_h, actual_h = ( int(row[0]), str(row[1]), str(row[2]), int(row[3]), str(row[4]), str(row[5]), str(row[6]), ) if seq_no != expected_seq: raise LedgerIntegrityError( f"Sequence gap detected in {stream_id!r}: expected seq {expected_seq}, got {seq_no}" ) if prev_h != expected_prev_hash: raise LedgerIntegrityError( f"Broken hash chain link at {stream_id!r} seq {seq_no}: " f"prev_hash={prev_h} != expected={expected_prev_hash}" ) recomputed = self._compute_entry_hash( stream_id, seq_no, ts, etype, delta, meta, prev_h ) if recomputed != actual_h: raise LedgerIntegrityError( f"Tampered payload at {stream_id!r} seq {seq_no}: " f"recomputed hash {recomputed} != stored {actual_h}" ) net_balance_cents += delta expected_seq += 1 expected_prev_hash = actual_h return net_balance_cents