diff --git a/keel/commands/doctor.py b/keel/commands/doctor.py index 54b29bf..8e5bc67 100644 --- a/keel/commands/doctor.py +++ b/keel/commands/doctor.py @@ -300,6 +300,72 @@ def asset_attestation_window_findings( ) +def audit_chain_findings(state: Any) -> list[Finding]: + """Does the append-only audit chain still verify? (#721) + + THREE readings, and the middle one is the whole reason this is not a boolean: + + * **no `audit_events` table** -- a database this build has not migrated. Both the web server + and `keel mcp` open a repo WITHOUT migrating (a view must not take a schema write lock), so + an un-upgraded database reaching a reader is ordinary, not an error. + * **a table with no events** -- nothing has been written since the chain shipped. Reported as + OK, and the headline says UNVERIFIED rather than verified: an empty chain has no breaks + because it has nothing in it to break, and calling that "verified" is a positive claim over + zero rows. + * **a break** -- FAIL. This is the one check in this module where a failure means the RECORD + has been altered rather than a rail being out of position, so it names the row and offers no + `fix`: there is no command that repairs a broken chain, and offering one would imply the + break can be papered over. What an operator does here is investigate. + + `state` is a `keel.data.audit.ChainState`, typed `Any` for the same reason every other + gatherer's input here is: the MCP tool and the click command hand this module rows and records + from a repo it does not import. + """ + if not getattr(state, "table_present", False): + return [ + Finding( + "audit.chain", + OK, + "audit chain not on this database", + "schema predates the append-only audit chain; nothing is recorded to verify", + "keel migrate", + ) + ] + errors = tuple(getattr(state, "errors", ()) or ()) + count = int(getattr(state, "event_count", 0) or 0) + if errors: + return [ + Finding( + "audit.chain", + FAIL, + "audit chain does NOT verify", + f"{len(errors)} break(s) over {count} event(s); first: {errors[0]}", + # No fix. Nothing repairs a broken chain, and a command here would read as though + # something could -- which is the opposite of what a break means. + "-", + ) + ] + if count == 0: + return [ + Finding( + "audit.chain", + OK, + "audit chain empty — unverified", + "no events recorded yet; nothing has been checked", + "-", + ) + ] + return [ + Finding( + "audit.chain", + OK, + "audit chain verifies", + f"{count} event(s), every one still hashing to the next", + "-", + ) + ] + + def instrument_attestation_window_findings( attestations: Iterable[dict[str, Any]], *, now_ts: int ) -> list[Finding]: @@ -1504,6 +1570,10 @@ def gather_findings(repo: Any, config: Any, log_lines: Iterable[str], now_ts: in findings += instrument_attestation_window_findings( repo.get_instrument_attestations(), now_ts=now_ts ) + # #721. Deliberately AFTER the attestation checks and before the venue rails: a chain break + # is a statement about the record those checks were read from, and an operator scanning the + # output should meet it in the same block as the rest of the record's condition. + findings += audit_chain_findings(repo.audit_chain()) findings += trade_scope_findings(repo.get_venue_trade_scope(venue), venue) # #691. Venue-keyed the same way, and reported BEFORE it bites: rail 22 vetoes silently on # a lapse, and the live profile runs unattended. diff --git a/keel/commands/timeline.py b/keel/commands/timeline.py index 2c273ec..bccdd3d 100644 --- a/keel/commands/timeline.py +++ b/keel/commands/timeline.py @@ -13,13 +13,25 @@ **Read-only, no broker, no network.** Same posture as every other service in this package. -**Nothing here is tamper-evident, and the export says so.** #703 asked the CSV to carry each -row's hash. None of these four stores hashes its rows: `orders`, `transactions`, -`asset_attestations` and `instrument_attestations` have no hash column between them, and the only -hash-chained store in this codebase is the research trials ledger (`keel/research/ledger.py`), -which records experiments rather than trading activity and does not belong in this feed. So the -hash column is emitted as NOT RECORDED rather than left blank -- blank invites the reader to -assume the check passed -- and hashing these tables is filed as engine work. +**Tamper-evidence, and its three honest readings (#721).** #703 asked the CSV to carry each row's +hash, and shipped with the column reading NOT RECORDED on every row because no store hashed +anything. `keel/data/audit.py` now chains an append-only event per write to `orders`, +`transactions` and both attestation tables, so this feed reads real hashes -- and has to keep +three readings apart: + +- `chained` -- an event exists for this row and the chain vouches for it. +- `not chained` -- no event was ever written. Rows predating #721, and every engine-log row, which + comes from a FILE rather than a chained store. An honest gap; deliberately NOT a break, so a + deployment upgrading into the chain does not open its timeline to a page of red. +- `chain broken` -- an event exists and falls at or after the first break. The hash is shown and + is NOT evidence: a chain proves a sequence, so past a break the sequence is unproven. Showing + these as `chained` would present unverified values as evidence, which is the one thing the + column exists to prevent. + +The research trials ledger (`keel/research/ledger.py`) is still not in this feed: it records +experiments rather than trading activity, and borrowing its hashes to decorate a trading audit +trail would be provenance laundering. The two stores share `keel_core.hashchain` -- one definition +of canonical JSON, so one row can only ever have one hash -- and nothing else. """ from __future__ import annotations @@ -32,6 +44,7 @@ from typing import Any from keel.commands.orders import normalise_scope, scope_start_ts +from keel.data.audit import ChainState from keel.data.repository import Repository #: The type chips, and the only words `kind` ever takes. @@ -53,10 +66,19 @@ "engine-log", ) -#: What the hash column says until the engine records one. NOT blank: an empty cell in a column +#: What the hash column says where no event was recorded. NOT blank: an empty cell in a column #: headed `row_hash` reads as "nothing to report", and the honest reading is "nobody checked". HASH_NOT_RECORDED = "NOT RECORDED" +#: What the chain says about one row, as a closed vocabulary (#721). The full reasoning is in the +#: module docstring; the short version is that a hash and a verdict are two different facts, and a +#: hash printed without one is a number an auditor cannot use. +CHAIN_STATUSES: tuple[str, ...] = ("chained", "not chained", "chain broken") + +CHAINED = "chained" +NOT_CHAINED = "not chained" +CHAIN_BROKEN = "chain broken" + #: The characters a spreadsheet treats as the start of a formula. OWASP's list. #: #: `\n` and `\r` are here alongside `\t` because a cell can only be re-parsed from its start, @@ -126,8 +148,14 @@ class TimelineRow: #: one thing. amount: Decimal | None amount_kind: str - #: The row's own tamper-evidence, when its store records one. None of the four does today. + #: The row's own tamper-evidence: the `row_hash` of the latest `audit_events` statement about + #: it, or `HASH_NOT_RECORDED` where none was written. row_hash: str = HASH_NOT_RECORDED + #: One of `CHAIN_STATUSES`. Carried BESIDE the hash rather than inferred from it, because + #: "a hash is present" and "the chain vouches for it" are different facts and the second is + #: the one an auditor is actually asking about. A client that inferred the verdict from the + #: presence of a 64-character string would call a broken row verified. + chain_status: str = NOT_CHAINED @dataclass(frozen=True) @@ -157,6 +185,15 @@ class TimelineReport: #: bar that reordered itself as history arrived would move under the reader. kinds_present: tuple[str, ...] + #: Whether the `audit_events` chain holds ANY events (#721). Distinct from `chain_errors` + #: being empty: a verification over zero rows reports nothing and has verified nothing, and a + #: green badge over a table nothing wrote is the failure this codebase keeps re-learning. A + #: deployment that predates the chain and one that has done nothing since are both False here. + chain_recorded: bool = False + #: Every break the chain walk found, verbatim from `keel_core.hashchain`. Reported rather + #: than raised, so both renderers can STATE the chain's condition instead of asserting it. + chain_errors: tuple[str, ...] = () + #: `read_log_window`'s own word for how the engine log read went: `ok`, `missing`, `empty`, #: `oversized`, `unreadable`. Carried rather than acted on, so the page and the CSV can SAY #: the log did not reach this report. @@ -174,6 +211,18 @@ def shown_count(self) -> int: `keel/web/payload.py` may not call `len()` (Rule 6e).""" return len(self.rows) + @property + def chain_intact(self) -> bool: + """Whether the chain found nothing wrong. + + TRUE over an empty chain, and that is the deliberate reading: nothing was checked, so + nothing is broken, and a deployment upgrading into #721 must not open its timeline to a + page of red. `chain_recorded` is the companion that says whether anything was checked at + all, and the two are read together -- exactly the pairing + `payload.py::_chain_payload` holds for the research ledger's badge. + """ + return not self.chain_errors + @property def log_gap(self) -> bool: """Whether the engine log's contents are MISSING FROM this report. @@ -201,7 +250,36 @@ def log_gap(self) -> bool: DEFAULT_TIMELINE_LIMIT = 200 MAX_TIMELINE_LIMIT = 2000 -def _order_rows(repo: Repository, since_ts: int | None) -> list[TimelineRow]: +class _Chain: + """The chain state, as the two fields a `TimelineRow` carries. + + A thin adapter and not a second source of truth: `keel/data/audit.py` decides what the chain + says, and this decides only how to SAY it on a row. Held as a class so the four row builders + ask one object the same question rather than each reproducing the three-way reading. + """ + + def __init__(self, state: ChainState) -> None: + self._state = state + + def of(self, store: str, reference: str) -> dict[str, str]: + """`row_hash` and `chain_status` for one row, as kwargs. + + A row with no event is `not chained` -- an honest gap, deliberately not a break. A row + whose event falls AT OR AFTER the first break is `chain broken`: its hash is still shown, + because hiding it would destroy the very value an auditor would use to establish what the + row said, but the status refuses to call it evidence. A chain proves a SEQUENCE, so past + a break the sequence is unproven -- which is why this compares `seq_id` against the break + rather than re-verifying the single row, whose own hash may well still match. + """ + seen = self._state.hashes.get((store, reference)) + if seen is None: + return {"row_hash": HASH_NOT_RECORDED, "chain_status": NOT_CHAINED} + broken_from = self._state.first_broken_seq + status = CHAINED if broken_from is None or seen.seq_id < broken_from else CHAIN_BROKEN + return {"row_hash": seen.row_hash, "chain_status": status} + + +def _order_rows(repo: Repository, since_ts: int | None, chain: _Chain) -> list[TimelineRow]: """`orders` -> trade rows. A PAPER order is `simulated`, not `venue-reported`. The paper trader wrote that row with no @@ -224,6 +302,7 @@ def _order_rows(repo: Repository, since_ts: int | None) -> list[TimelineRow]: provenance="simulated" if mode == "paper" else "venue-reported", source="orders", reference=str(raw.get("id") or ""), + **chain.of("orders", str(raw.get("id") or "")), summary=f"{status} {side} {product} ({mode})".strip(), product_id=product, amount=raw.get("actual_fill"), @@ -233,7 +312,9 @@ def _order_rows(repo: Repository, since_ts: int | None) -> list[TimelineRow]: return rows -def _transaction_rows(repo: Repository, since_ts: int | None) -> list[TimelineRow]: +def _transaction_rows( + repo: Repository, since_ts: int | None, chain: _Chain +) -> list[TimelineRow]: """`transactions` -> flow rows. `imported-ledger`, never `venue-reported`: these lines came out of a CSV the operator @@ -255,6 +336,7 @@ def _transaction_rows(repo: Repository, since_ts: int | None) -> list[TimelineRo provenance="imported-ledger", source="transactions", reference=str(raw.get("coinbase_id") or raw.get("id") or ""), + **chain.of("transactions", str(raw.get("coinbase_id") or raw.get("id") or "")), summary=f"{kind_word} {asset}".strip() + (f" -- {note}" if note else ""), product_id="", amount=raw.get("total"), @@ -264,7 +346,9 @@ def _transaction_rows(repo: Repository, since_ts: int | None) -> list[TimelineRo return rows -def _attestation_rows(repo: Repository, since_ts: int | None) -> list[TimelineRow]: +def _attestation_rows( + repo: Repository, since_ts: int | None, chain: _Chain +) -> list[TimelineRow]: """The attestation tables -> attestation rows. `human-attested`: someone typed this and signed their name to it, which is a different kind @@ -284,6 +368,7 @@ def _attestation_rows(repo: Repository, since_ts: int | None) -> list[TimelineRo provenance="human-attested", source="asset_attestations", reference=asset, + **chain.of("asset_attestations", asset), summary=( f"{asset} attested by {raw.get('attested_by') or 'unnamed'} " f"(source: {raw.get('source') or 'unstated'})" @@ -306,6 +391,7 @@ def _attestation_rows(repo: Repository, since_ts: int | None) -> list[TimelineRo provenance="human-attested", source="instrument_attestations", reference=f"{venue}:{product}", + **chain.of("instrument_attestations", f"{venue}:{product}"), summary=( f"{product} on {venue} attested by " f"{raw.get('attested_by') or 'unnamed'} " @@ -393,10 +479,20 @@ def gather_timeline( resolved_limit = None if limit is None else max(1, min(int(limit), MAX_TIMELINE_LIMIT)) since = scope_start_ts(resolved_scope, now_ts) + # ONE read of `audit_events`, shared by all three store readers (#721). Not one lookup per + # row: the hash printed beside a row and the verdict printed above it would then describe + # different reads of the table, and an event appended between them would have the verdict + # cover a row this report never showed. + chain = repo.audit_chain() + chained = _Chain(chain) + scoped: list[TimelineRow] = [] - scoped.extend(_order_rows(repo, since)) - scoped.extend(_transaction_rows(repo, since)) - scoped.extend(_attestation_rows(repo, since)) + scoped.extend(_order_rows(repo, since, chained)) + scoped.extend(_transaction_rows(repo, since, chained)) + scoped.extend(_attestation_rows(repo, since, chained)) + # No chain argument, and never one: `_cycle_rows` reads the engine's own log FILE, which is + # not a chained store. A cycle row carrying a hash would be this module attesting to something + # it merely read. scoped.extend(_cycle_rows(cycles, since)) # Newest first, HERE -- so neither renderer sorts, and they cannot disagree about what "the @@ -417,6 +513,8 @@ def gather_timeline( rows=tuple(filtered if resolved_limit is None else filtered[:resolved_limit]), log_status=log_status, kinds_present=tuple(kind for kind in TIMELINE_KINDS if kind in present), + chain_recorded=chain.event_count > 0, + chain_errors=chain.errors, ) @@ -463,13 +561,28 @@ def to_csv(report: TimelineReport) -> str: **Provenance is a column, not a footnote.** The point of this file is that a reader can tell a venue-reported fill from a line someone imported from a spreadsheet -- so `provenance` and - `source` sit beside every figure, and `row_hash` says NOT RECORDED rather than being blank. + `source` sit beside every figure, `row_hash` carries the chain's own hash where one was + recorded and says NOT RECORDED rather than being blank where none was, and `chain_status` + says whether that hash is evidence. All three in the same file, on every row. `amount_kind` rides beside `amount` for the same reason: a fill price and a cash-flow total in one column, with nothing saying which is which, is a column that will be summed. """ buffer = io.StringIO() writer = csv.writer(buffer) + if not report.chain_intact: + # Stated ABOVE the header, for the same reason the log note below is: an auditor holding + # this file cannot ask the page anything, and a finding that changes how the whole file + # should be read must travel with the file. Per-row `chain_status` says WHICH rows; this + # says the record has been altered, which is the sentence that matters first. + writer.writerow( + [ + csv_safe( + f"# NOTE: the audit chain does not verify ({report.chain_errors[0]}); " + "every row marked `chain broken` below is shown but is NOT evidence" + ) + ] + ) if report.log_gap: # Stated IN THE FILE, above the header, because this file leaves the application. An # auditor holding a CSV cannot ask the page whether a source was missing from it, and @@ -494,6 +607,10 @@ def to_csv(report: TimelineReport) -> str: "amount_kind", "summary", "row_hash", + # Beside the hash, never instead of it. A 64-character string with nothing saying + # whether it verifies is a number an auditor cannot use, and the reading a reader + # defaults to is the flattering one. + "chain_status", ] ) for row in report.rows: @@ -509,6 +626,7 @@ def to_csv(report: TimelineReport) -> str: csv_safe(row.amount_kind), csv_safe(row.summary), csv_safe(row.row_hash), + csv_safe(row.chain_status), ] ) return buffer.getvalue() diff --git a/keel/data/audit.py b/keel/data/audit.py new file mode 100644 index 0000000..5b461eb --- /dev/null +++ b/keel/data/audit.py @@ -0,0 +1,386 @@ +"""`audit_events`: the append-only, hash-chained record of what the book did (#721). + +#703's activity export shipped with a `row_hash` column whose every cell read `NOT RECORDED`. +That was honest and it was the whole of what could be said: none of the four stores the timeline +merges hashed its rows. This module is the record that column was built to read. + +── WHY EVENTS AND NOT THE ROWS ──────────────────────────────────────────────────────────────── + +`orders` rows are MUTATED. `update_order` writes status, fills and fees as a venue reports them, +sometimes minutes after placement, sometimes days later via `execution.reconcile`. A hash chained +over the order row itself would therefore break on every legitimate fill -- and a chain that +cries wolf on ordinary operation is a chain an operator learns to ignore, which is worse than no +chain at all. + +So the chain is over IMMUTABLE STATEMENTS about the book, not over the book. `insert_order` +appends "this order was placed, with these fields"; each `update_order` appends "these fields +changed". Neither event is ever rewritten. The book stays mutable and queryable; the event stream +is what an auditor verifies. + +The same shape covers `transactions` (whose upsert rewrites an imported line in place -- two +events for one `coinbase_id` is the record that the line was re-imported, which is precisely what +should be visible) and both attestation tables (where a re-attestation is a fresh human claim). + +── ONE CHAIN, NOT ONE PER TABLE ─────────────────────────────────────────────────────────────── + +Every event goes into one sequence. A per-table chain would let an event be moved between streams +undetectably, and would mean a removed order event was only visible to someone who thought to +verify the order chain specifically. One chain means any removal anywhere is visible from every +event after it. + +── WHAT THIS IS NOT ─────────────────────────────────────────────────────────────────────────── + +**Tamper-EVIDENT, not tamper-proof.** Anyone who can write `keel.db` can rewrite the chain from +the edited row forward. What it buys is that a row cannot be changed QUIETLY. + +**No backfill, ever.** Rows written before this shipped have no event. Computing hashes for them +now would produce a chain that verifies and proves nothing -- worse than an honest gap, because +it looks like evidence. `commands/timeline.py` reports those rows as not chained. + +**Not the research trials ledger.** `keel/research/ledger.py` records EXPERIMENTS and is +git-tracked; this records trading activity and is per-deployment. They share +`keel_core.hashchain` -- one definition of canonical JSON, so one row can only have one hash -- +and nothing else. Folding trials into the trading audit trail to borrow their hashes would be +provenance laundering. +""" + +from __future__ import annotations + +import json +import sqlite3 +from collections.abc import Iterator, Mapping +from contextlib import contextmanager +from dataclasses import dataclass +from typing import Any + +from keel_core.hashchain import ZERO_HASH, ChainLink, canonical_json, chain_hash, find_breaks + +#: Every kind of statement this chain carries, and the store each one is about. +#: +#: A CLOSED vocabulary, checked at write time, like `TIMELINE_KINDS` and `PROVENANCES` next door. +#: An event type that arrives via a typo is a stream nothing reads and nothing misses -- and on +#: an audit surface, a silently-ignored record is the failure mode that matters. +#: +#: The store name is carried because `latest_hashes` is keyed by it: an `orders.id` of `"1"` and +#: a `coinbase_id` of `"1"` are different rows, and keyed by entity alone the imported transaction +#: would hand its hash to an unrelated order. +EVENT_STORES: Mapping[str, str] = { + "order_placed": "orders", + "order_updated": "orders", + "transaction_recorded": "transactions", + "asset_attested": "asset_attestations", + "instrument_attested": "instrument_attestations", +} + + +@dataclass(frozen=True) +class AuditEvent: + """One immutable statement about the book, and its place in the chain.""" + + seq_id: int + ts: int + event_type: str + #: The subject store's own identifier -- an `orders.id`, a `coinbase_id`, an asset, a + #: `venue:product_id`. As TEXT, because the five stores key on four different types. + entity_id: str + #: What the statement asserts. Decimals are already strings here: they were encoded on the + #: way in (`keel_core.hashchain.encode_value`) so that what is hashed is exactly what is + #: stored, and a reader is never handed a float rendering of a recorded money value. + payload: dict[str, Any] + prev_hash: str + row_hash: str + + +@contextmanager +def write_transaction(conn: sqlite3.Connection) -> Iterator[None]: + """One write transaction covering a store row and its audit event. + + **`BEGIN IMMEDIATE`, not `BEGIN`.** A deferred transaction takes no lock until its first + WRITE, so the chain-head SELECT inside it would still be unserialised: two writers would both + read the same head, both write it as their `prev_hash`, and the chain would fork -- silently, + because each row verifies against the row it believes precedes it. `IMMEDIATE` takes the + write lock up front, which is what makes "read the head, then append" atomic. + + **The commit is unconditional, and the first cut of this got it wrong in a way that reached + the venue.** It read `if conn.in_transaction: yield; return` -- "somebody outside opened a + transaction, so they own the commit". Nothing in this codebase nests these, so that branch + was never reached by the nesting it was written for. What it WAS reached by is the accident: + `sqlite3` runs in legacy implicit-transaction mode, so `in_transaction` is also True after any + DML that has not been committed -- INCLUDING one that raised and was swallowed. + `execution/equity.py` swallows a failed `record_cycle_balance` by design (a diagnostic write + must never abort a cycle), and that left the connection mid-transaction. The next + `insert_order` then took the join branch, returned a real `order_id`, and never committed -- + so `broker.place_order` sent an order to the venue with no durable row behind it, which is + precisely the row `execution.reconcile` exists to find. Measured end to end: `orders visible + to another connection: 0`. + + So this always commits on a clean exit, which is exactly what the unconditional + `self._conn.commit()` in each writer did before this module existed. Skipping the `BEGIN` on + an already-open transaction is only about sqlite refusing to nest one; it never means + skipping the commit. (The lock invariant survives that path: legacy mode opens a transaction + implicitly only on DML, so a connection that is already `in_transaction` has already taken + the write lock, and the head read is serialised either way. `equity.py` now rolls its + swallowed write back, so the path should not arise at all -- this is the second belt.) + + Rolls back on ANY exception, `BaseException` included: a `KeyboardInterrupt` between the + store row and its event would otherwise leave the row committed and the chain silent about + it, which is the one lie this store must not tell about itself. + """ + if not conn.in_transaction: + conn.execute("BEGIN IMMEDIATE") + try: + yield + except BaseException: + conn.rollback() + raise + conn.commit() + + +def append_event( + conn: sqlite3.Connection, + *, + ts: int, + event_type: str, + entity_id: str, + payload: Mapping[str, Any], +) -> AuditEvent: + """Chain and append one event. Does NOT commit -- the caller's `write_transaction` does. + + Not committing is the point: the store row and its event must land together or not at all. + An order row with no event reads forever after as "written before the bump" -- an + honest-looking gap that is in fact a failed chain write. + + The head read is inside the caller's write transaction and refuses to run outside one, which + is the serialisation this chain rests on. `ORDER BY seq_id DESC`, never `ts`: two events + inside one second share a timestamp, and a clock that steps backwards would silently reorder + the chain into a false break. + """ + if event_type not in EVENT_STORES: + raise ValueError(f"event_type: {event_type!r} not in {sorted(EVENT_STORES)}") + if not conn.in_transaction: + raise RuntimeError( + "append_event must run inside a write transaction (see `write_transaction`): " + "reading the chain head outside one lets two writers fork the chain" + ) + + head = conn.execute("SELECT row_hash FROM audit_events ORDER BY seq_id DESC LIMIT 1").fetchone() + prev_hash = ZERO_HASH if head is None else str(head["row_hash"]) + # The payload is canonicalised ONCE and commits AS THAT STRING -- the exact bytes the column + # holds, not a re-serialisation of what a reader parsed back out of it. + # + # Two things follow, and the first is the reason. A hash taken over a parsed-then-re-encoded + # object puts a decode/encode round trip between the stored row and its hash, and every such + # conversion is somewhere the two can drift: a payload shape that does not survive the round + # trip byte-for-byte (a non-string key, a tuple, anything `json` normalises) would store one + # thing and attest to another. Hashing the bytes removes the question rather than answering it + # shape by shape. The second is that verification then never PARSES: measured over 20,000 + # events, `chain_state` went from 277 ms to 134 ms, on the read a 15-second console poll makes. + payload_json = canonical_json(payload) + row_hash = chain_hash(_hash_body(ts, event_type, entity_id, payload_json, prev_hash)) + cursor = conn.execute( + "INSERT INTO audit_events (ts, event_type, entity_id, payload_json, prev_hash, row_hash) " + "VALUES (?, ?, ?, ?, ?, ?)", + (ts, event_type, entity_id, payload_json, prev_hash, row_hash), + ) + assert cursor.lastrowid is not None + return AuditEvent( + seq_id=cursor.lastrowid, + ts=ts, + event_type=event_type, + entity_id=entity_id, + payload=json.loads(payload_json), + prev_hash=prev_hash, + row_hash=row_hash, + ) + + +def read_events(conn: sqlite3.Connection) -> list[AuditEvent]: + """Every event, in CHAIN order (`seq_id`), which is the only order the chain verifies in.""" + rows = conn.execute( + "SELECT seq_id, ts, event_type, entity_id, payload_json, prev_hash, row_hash " + "FROM audit_events ORDER BY seq_id" + ).fetchall() + return [ + AuditEvent( + seq_id=int(row["seq_id"]), + ts=int(row["ts"]), + event_type=str(row["event_type"]), + entity_id=str(row["entity_id"]), + payload=json.loads(row["payload_json"]), + prev_hash=str(row["prev_hash"]), + row_hash=str(row["row_hash"]), + ) + for row in rows + ] + + +def _hash_body( + ts: int, event_type: str, entity_id: str, payload_json: str, prev_hash: str +) -> dict[str, Any]: + """Everything a row commits to -- the row minus `row_hash` itself. + + `payload_json` is the STORED STRING, not a parsed object: see `append_event` for why the + bytes rather than a re-encoding of them. Taken as five scalars rather than as an `AuditEvent` + so the writer and the verifier compute the identical body from the identical inputs, with no + parse on the verification path. + + `seq_id` is deliberately NOT hashed. It is sqlite's own AUTOINCREMENT counter, assigned after + the hash would have to be computed, and position in the chain is already committed to via + `prev_hash`. Hashing it would add nothing and would make the hash unreproducible from the + values a writer actually chose. + """ + return { + "ts": ts, + "event_type": event_type, + "entity_id": entity_id, + "payload": payload_json, + "prev_hash": prev_hash, + } + + +@dataclass(frozen=True) +class EntityHash: + """The latest chained statement about one row, and where it sits in the chain.""" + + row_hash: str + #: The event's own `seq_id`. Carried so a caller can ask whether this statement falls in the + #: region the chain no longer vouches for -- see `ChainState.first_broken_seq`. + seq_id: int + + +@dataclass(frozen=True) +class ChainState: + """Everything a report needs to STATE the chain's condition rather than assert it. + + Read in one pass, because the alternative -- a hash lookup here, a verification there -- means + the hash printed beside a row and the verdict printed above it describe two different reads of + the table, and a row appended between them would have the verdict cover a row the report never + showed. That is the mistake `research/ledger.py::verify_records` exists to have fixed once. + """ + + #: Whether `audit_events` EXISTS on this database (schema v20). + #: + #: Not a theoretical case. Two readers open this database without migrating it -- the web + #: server (`keel/web/server.py`: a view must not take a schema write lock) and `keel mcp`'s + #: `_open_readonly_repo`. Both therefore meet pre-v20 databases, and a missing table there is + #: an ordinary fact about an un-upgraded deployment rather than an error. #718 shipped a + #: reader that raised on exactly this and took the whole of `gather_findings` down with it. + table_present: bool + #: How many events the table holds. ZERO IS A DISTINCT STATE, not "verified": a deployment + #: that predates #721 and one that has done nothing since are both empty here, and neither has + #: had anything checked. The caller decides which sentence to print. + event_count: int + errors: tuple[str, ...] + #: The `seq_id` of the FIRST event the chain stops vouching for, or `None` when intact. Every + #: event from here on is unverified -- not because each one is necessarily wrong, but because + #: a chain proves the sequence, and past a break the sequence is no longer proven. + first_broken_seq: int | None + hashes: Mapping[tuple[str, str], EntityHash] + + @property + def intact(self) -> bool: + """The chain verifies AND there is something to verify. + + Both halves, deliberately. `not errors` alone is true of an empty table, and a green + badge over a table nothing read is the exact failure this codebase keeps re-learning. + """ + return not self.errors and self.event_count > 0 + + +def _latest(rows: list[sqlite3.Row]) -> dict[tuple[str, str], EntityHash]: + """`(store, entity_id) -> the LATEST event about it`. + + The latest rather than the first because it is the most recent chained statement about that + row: an order that was placed and then filled has two events, and the fill is the later word + on it. + + Keyed by store as well as entity because the five stores key on four different types and their + identifiers collide -- an `orders.id` of `"1"` and a `coinbase_id` of `"1"` are different + rows, and a single-keyed map would hand one row's hash to the other. + """ + latest: dict[tuple[str, str], EntityHash] = {} + for row in rows: + store = EVENT_STORES.get(str(row["event_type"])) + if store is None: + # An event type this build does not know -- a row written by a NEWER keel against the + # same database. Skipped rather than guessed at: it still chains (the chain does not + # care what the type means), and inventing a store for it would file it against a + # table it may have nothing to do with. + continue + latest[(store, str(row["entity_id"]))] = EntityHash( + row_hash=str(row["row_hash"]), seq_id=int(row["seq_id"]) + ) + return latest + + +def _table_present(conn: sqlite3.Connection) -> bool: + row = conn.execute( + "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'audit_events'" + ).fetchone() + return row is not None + + +def chain_state(conn: sqlite3.Connection) -> ChainState: + """One read of `audit_events`, verified, indexed by the identifiers a report prints. + + What `commands/timeline.py` and `commands/doctor.py` call, and the only production path into + this module. The `hashes` map is keyed by `(store, entity_id)` to match `timeline.py`'s own + `source`/`reference` pair, so the hash shown against a row is looked up by the identifier + printed beside it rather than by one a reader has to reconstruct. + + **The whole chain, every time, and that is not an oversight.** A chain proves a SEQUENCE, so + a verdict over a suffix is a verdict that has not looked at the rows most likely to have been + quietly edited, and a cached prefix verdict assumes precisely the thing being checked. The + cost is real and is stated rather than hidden, in the manner `DEFAULT_TIMELINE_LIMIT` and + `export_rows` already use next door: measured on a 2026 laptop, 20,000 events verify in about + 135 ms, and the console polls `/api/timeline` every 15 seconds. Event volume is dominated by + `upsert_transaction` -- one event per imported CSV line -- so a long Coinbase history is what + puts a deployment into that band. + + If it ever stops fitting, the change is to BOUND WHAT IS CLAIMED and say so on the page -- a + verdict over the last N events, labelled as one. Never to cache the verdict, which would put + a green badge over rows nothing read: the exact failure the rest of this module is built to + refuse. + """ + if not _table_present(conn): + # Checked rather than caught. A pre-v20 database is a legitimate deployment state, and a + # `try`/`except OperationalError` around the read would swallow a genuine "database disk + # image is malformed" under the same clause -- reporting an un-upgraded database where + # the truth is a corrupt one. + return ChainState( + table_present=False, event_count=0, errors=(), first_broken_seq=None, hashes={} + ) + # Raw rows, deliberately not `read_events`: verification needs the stored `payload_json` + # STRING, and parsing 20,000 payloads to re-encode them would double the cost of the check + # for a value nothing on this path uses. + rows = conn.execute( + "SELECT seq_id, ts, event_type, entity_id, payload_json, prev_hash, row_hash " + "FROM audit_events ORDER BY seq_id" + ).fetchall() + breaks = find_breaks( + ChainLink( + label=str(row["seq_id"]), + prev_hash=str(row["prev_hash"]), + row_hash=str(row["row_hash"]), + recomputed_hash=chain_hash( + _hash_body( + int(row["ts"]), + str(row["event_type"]), + str(row["entity_id"]), + str(row["payload_json"]), + str(row["prev_hash"]), + ) + ), + ) + for row in rows + ) + # `index` is 1-based POSITION IN THE WALK, and the rows are read in `seq_id` order, so this is + # the event the first break lands on. Looked up rather than parsed out of the message text. + first_broken_seq = int(rows[breaks[0].index - 1]["seq_id"]) if breaks else None + return ChainState( + table_present=True, + event_count=len(rows), + errors=tuple(found.message for found in breaks), + first_broken_seq=first_broken_seq, + hashes=_latest(rows), + ) diff --git a/keel/data/repository.py b/keel/data/repository.py index 337166e..7100750 100644 --- a/keel/data/repository.py +++ b/keel/data/repository.py @@ -18,6 +18,7 @@ from keel_core.subscription import BrokerSubscription, SubscriptionStatus from keel_core.trade_scope import TradeScopeState, VenueTradeScope +from keel.data.audit import ChainState, append_event, chain_state, write_transaction from keel.types import Candle, CycleBalance, EquityReading, Granularity, Profile _TRANSACTION_COLUMNS = ( @@ -205,12 +206,48 @@ def _json_object_hook(d: dict[str, Any]) -> Any: return d +def _event_ts(value: Any) -> int: + """The timestamp an audit event carries, falling back to now ONLY when there is none (#721). + + `int(value or time.time())` was wrong in a small, silent way: `or` treats `0` as absent, so a + row legitimately stamped at the epoch would get an event stamped `now` -- and the event's `ts` + is hashed, so the chain would then attest to a timestamp the row does not hold. Every real + call site supplies a non-zero value today (checked: all five `update_order` callers pass + `updated_at=now_ts`, both `insert_order` callers pass `created_at`, and `transactions.ts` is + NOT NULL), so this is a guard against a future caller rather than a live bug -- which is + exactly when a silent fallback is worth making explicit. + """ + return int(time.time()) if value is None else int(value) + + class Repository: """Typed wrapper around a migrated `sqlite3.Connection` for the keel schema.""" def __init__(self, conn: sqlite3.Connection) -> None: self._conn = conn + # -- the audit chain ------------------------------------------------ + + def rollback(self) -> None: + """Discard whatever this connection has open but uncommitted. + + For a caller that SWALLOWS a write failure. Every writer here is `execute` then `commit`, + so a failing `execute` leaves sqlite3's implicit transaction open; a caller that catches + the exception and carries on -- `execution/equity.py` does, so a diagnostic write cannot + abort a cycle -- would otherwise hand the next writer on this connection a dirty + transaction it did not open and does not know about. + """ + self._conn.rollback() + + def audit_chain(self) -> ChainState: + """The `audit_events` chain: its hashes, its condition, and where it stops being evidence. + + Exposed here rather than letting a reader reach for the connection, because "which rows + does this chain vouch for" is a question about the record and the repository is what owns + the record. `commands/timeline.py` is the caller. + """ + return chain_state(self._conn) + # -- transactions --------------------------------------------------- def upsert_transaction(self, tx: dict[str, Any]) -> None: @@ -227,15 +264,53 @@ def upsert_transaction(self, tx: dict[str, Any]) -> None: update_sql = ", ".join( f"{c} = excluded.{c}" for c in _TRANSACTION_COLUMNS if c != "coinbase_id" ) - self._conn.execute( - f""" - INSERT INTO transactions ({columns_sql}) - VALUES ({placeholders_sql}) - ON CONFLICT(coinbase_id) DO UPDATE SET {update_sql} - """, - values, - ) - self._conn.commit() + with write_transaction(self._conn): + self._conn.execute( + f""" + INSERT INTO transactions ({columns_sql}) + VALUES ({placeholders_sql}) + ON CONFLICT(coinbase_id) DO UPDATE SET {update_sql} + """, + values, + ) + # An APPEND even when the book row was updated in place (#721). Two events for one + # `coinbase_id` is the record that the line was re-imported with different content -- + # which, for a store whose provenance is `imported-ledger` and whose rows nothing + # verified on the way in, is exactly what an auditor needs visible. + append_event( + self._conn, + ts=_event_ts(values.get("ts")), + event_type="transaction_recorded", + entity_id=self._transaction_reference(values), + payload=dict(values), + ) + + def _transaction_reference(self, values: dict[str, Any]) -> str: + """`coinbase_id or id` -- EXACTLY the rule `commands/timeline.py::_transaction_rows` uses + for the `reference` it prints, so the export looks a row's hash up by the identifier shown + beside it. + + The falsey branch READS THE ID BACK rather than taking `cursor.lastrowid`. The first cut + used `lastrowid` on the reasoning that `coinbase_id` is nullable and sqlite treats NULLs + as distinct in a UNIQUE index, so a row without one can never take the `DO UPDATE` branch + -- true of NULL, and false of the EMPTY STRING, which is falsey to Python and perfectly + indexable to sqlite. A re-upsert of a blank id therefore UPDATED an existing row while + `lastrowid` still held the id of the last real INSERT, filing the event under a different + row's identifier: the timeline then showed one row the hash of a superseded event, marked + `chained`, and a phantom key existed that no row resolved to. Not reachable through + `csv_import` today (`_row_to_tx` raises on a blank ID), and gated only by that one caller. + + A lookup answers NULL and blank alike instead of reasoning about which value reaches which + branch. The non-empty fast path stays because a CSV import runs this per line and a + `coinbase_id` that is present needs no query to be known. + """ + coinbase_id = values.get("coinbase_id") + if coinbase_id: + return str(coinbase_id) + row = self._conn.execute( + "SELECT id FROM transactions WHERE coinbase_id IS ?", (coinbase_id,) + ).fetchone() + return "" if row is None else str(row["id"]) def get_transactions(self, asset: str | None = None) -> list[dict[str, Any]]: if asset is None: @@ -405,12 +480,31 @@ def insert_order(self, order: dict[str, Any]) -> int: columns_sql = ", ".join(_ORDER_COLUMNS) placeholders_sql = ", ".join(f":{c}" for c in _ORDER_COLUMNS) - cursor = self._conn.execute( - f"INSERT INTO orders ({columns_sql}) VALUES ({placeholders_sql})", values - ) - self._conn.commit() - assert cursor.lastrowid is not None - return cursor.lastrowid + # The row and its audit event in ONE transaction (#721). Both land or neither: an order + # row with no event reads forever after as "written before the chain shipped" -- an + # honest-looking gap that would in fact be a failed chain write, which is the one lie + # this record must not tell about itself. + # + # This runs BEFORE `broker.place_order` (see `executor.execute`), so a failure here fails + # CLOSED: no row, no event, no order at the venue. That ordering is why the audit write + # is allowed to be load-bearing here where a diagnostic write would not be. + with write_transaction(self._conn): + cursor = self._conn.execute( + f"INSERT INTO orders ({columns_sql}) VALUES ({placeholders_sql})", values + ) + assert cursor.lastrowid is not None + order_id = cursor.lastrowid + append_event( + self._conn, + ts=_event_ts(values.get("created_at")), + event_type="order_placed", + entity_id=str(order_id), + # The row AS STORED, id included -- the whole statement being made, so a later + # reader can reproduce the hash from the row without knowing which columns this + # build happened to populate. + payload={"id": order_id, **values}, + ) + return order_id def update_order(self, order_id: int, **fields: Any) -> None: """Partially update the order row `order_id` with `fields`.""" @@ -423,8 +517,19 @@ def update_order(self, order_id: int, **fields: Any) -> None: set_sql = ", ".join(f"{k} = :{k}" for k in fields) params = dict(fields) params["order_id"] = order_id - self._conn.execute(f"UPDATE orders SET {set_sql} WHERE id = :order_id", params) - self._conn.commit() + with write_transaction(self._conn): + self._conn.execute(f"UPDATE orders SET {set_sql} WHERE id = :order_id", params) + append_event( + self._conn, + ts=_event_ts(fields.get("updated_at")), + event_type="order_updated", + entity_id=str(order_id), + # WHAT CHANGED, not the whole mutated row. An event is a statement about this + # update; re-hashing the full row would make every event a snapshot, and a reader + # could no longer tell an update from a rewrite. The empty-`fields` early return + # above is what keeps a no-op from appending an event asserting nothing changed. + payload=dict(fields), + ) def get_order(self, order_id: int) -> dict[str, Any] | None: row = self._conn.execute("SELECT * FROM orders WHERE id = ?", (order_id,)).fetchone() @@ -1278,33 +1383,53 @@ def upsert_asset_attestation( before". Omitting it there would let a stale window silently outlive the claim it was recorded for, which is worse than no window at all. """ - self._conn.execute( - """ - INSERT INTO asset_attestations - (asset, sector, backing, pays_yield, source, attested_by, attested_at, - attest_due_ts) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(asset) DO UPDATE SET - sector = excluded.sector, - backing = excluded.backing, - pays_yield = excluded.pays_yield, - source = excluded.source, - attested_by = excluded.attested_by, - attested_at = excluded.attested_at, - attest_due_ts = excluded.attest_due_ts - """, - ( - asset, - sector, - backing, - int(pays_yield), - source, - attested_by, - attested_at, - attest_due_ts, - ), - ) - self._conn.commit() + with write_transaction(self._conn): + self._conn.execute( + """ + INSERT INTO asset_attestations + (asset, sector, backing, pays_yield, source, attested_by, attested_at, + attest_due_ts) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(asset) DO UPDATE SET + sector = excluded.sector, + backing = excluded.backing, + pays_yield = excluded.pays_yield, + source = excluded.source, + attested_by = excluded.attested_by, + attested_at = excluded.attested_at, + attest_due_ts = excluded.attest_due_ts + """, + ( + asset, + sector, + backing, + int(pays_yield), + source, + attested_by, + attested_at, + attest_due_ts, + ), + ) + # Appended, not rewritten (#721): a re-attestation is a FRESH human claim, and the + # claim it replaced is part of the record. `attested_at` is the timestamp the human + # supplied for the claim, so the event is stamped with when the claim was made rather + # than when the row happened to be written. + append_event( + self._conn, + ts=attested_at, + event_type="asset_attested", + entity_id=asset, + payload={ + "asset": asset, + "sector": sector, + "backing": backing, + "pays_yield": bool(pays_yield), + "source": source, + "attested_by": attested_by, + "attested_at": attested_at, + "attest_due_ts": attest_due_ts, + }, + ) def get_asset_attestation(self, asset: str) -> dict | None: row = self._conn.execute( @@ -1337,21 +1462,40 @@ def upsert_instrument_attestation( `upsert_asset_attestation` above -- it must be in `DO UPDATE SET` so a re-attestation without a window CLEARS a previously recorded one rather than carrying it forward. """ - self._conn.execute( - """ - INSERT INTO instrument_attestations - (venue, product_id, wrapper, source, attested_by, attested_at, attest_due_ts) - VALUES (?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(venue, product_id) DO UPDATE SET - wrapper = excluded.wrapper, - source = excluded.source, - attested_by = excluded.attested_by, - attested_at = excluded.attested_at, - attest_due_ts = excluded.attest_due_ts - """, - (venue, product_id, wrapper, source, attested_by, attested_at, attest_due_ts), - ) - self._conn.commit() + with write_transaction(self._conn): + self._conn.execute( + """ + INSERT INTO instrument_attestations + (venue, product_id, wrapper, source, attested_by, attested_at, attest_due_ts) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(venue, product_id) DO UPDATE SET + wrapper = excluded.wrapper, + source = excluded.source, + attested_by = excluded.attested_by, + attested_at = excluded.attested_at, + attest_due_ts = excluded.attest_due_ts + """, + (venue, product_id, wrapper, source, attested_by, attested_at, attest_due_ts), + ) + # `venue:product_id`, the same compound key this table is UNIQUE on and the same + # string `commands/timeline.py` puts in its `reference` column -- so the hash the + # export shows against an instrument attestation is looked up by the identifier + # printed beside it, not by one a reader has to reconstruct. + append_event( + self._conn, + ts=attested_at, + event_type="instrument_attested", + entity_id=f"{venue}:{product_id}", + payload={ + "venue": venue, + "product_id": product_id, + "wrapper": wrapper, + "source": source, + "attested_by": attested_by, + "attested_at": attested_at, + "attest_due_ts": attest_due_ts, + }, + ) def get_instrument_attestation(self, venue: str, product_id: str) -> dict | None: row = self._conn.execute( diff --git a/keel/execution/equity.py b/keel/execution/equity.py index d5b8618..38a9cad 100644 --- a/keel/execution/equity.py +++ b/keel/execution/equity.py @@ -310,6 +310,18 @@ def _append_equity_point( ) ) except Exception: + # ROLLED BACK, not merely logged. `record_cycle_balance` is `execute` then `commit`, + # so a failing `execute` leaves sqlite3's implicit transaction OPEN -- and swallowing + # the exception hands the next writer on this connection a dirty transaction it did + # not open. `Repository.insert_order` inherits exactly that connection a few + # milliseconds later, inside the same cycle. Leaving it dirty made a swallowed + # observability failure decide whether the ORDER ROW was durable, which is the same + # rule this `try` exists to enforce, violated from the other end. + # + # Safe to roll back here because the only uncommitted work is this failed INSERT: + # `record_equity_point` above committed its own row, and each `record_cycle_balance` + # commits per currency. + repo.rollback() log_exception(logger, "equity.cycle_balance_write_failed") diff --git a/keel/research/ledger.py b/keel/research/ledger.py index 39d2188..7c0c896 100644 --- a/keel/research/ledger.py +++ b/keel/research/ledger.py @@ -13,7 +13,6 @@ from __future__ import annotations -import hashlib import json import time from collections.abc import Iterable, Mapping, Sequence @@ -22,6 +21,18 @@ from pathlib import Path from typing import Any +# The canonical form and the chain walk are SHARED with `keel/data/audit.py` (#721) rather +# than defined here. `canonical_json` and `ZERO_HASH` keep their names in this module's +# namespace because `tests/research/test_ledger.py` and every downstream reader reach for +# `ledger.canonical_json` -- the definition moved, the vocabulary did not. +from keel_core.hashchain import ( + ZERO_HASH, + ChainLink, + canonical_json, + chain_hash, + verify_links, +) + PROVENANCE = frozenset({"a_priori", "fitted"}) #: `monte_carlo` (#441) is a resampling DIAGNOSTIC row -- same vocabulary discipline as the #: rest of the set: a row's kind says what kind of experiment produced it, so a bootstrap run @@ -44,7 +55,6 @@ #: A CSCV column is a diagnostic, not a decision (spec §4.4) -- it does not count toward N. DIAGNOSTIC_ONLY = "diagnostic_only" -ZERO_HASH = "0" * 64 DEFAULT_LEDGER_PATH = Path("docs/experiments/trials-ledger.jsonl") @@ -67,17 +77,6 @@ class TrialRecord: row_hash: str = "" -def _encode(value: Any) -> Any: - """Decimal -> str so JSON round-trips exactly; recurse through containers.""" - if isinstance(value, Decimal): - return str(value) - if isinstance(value, Mapping): - return {k: _encode(v) for k, v in value.items()} - if isinstance(value, (list, tuple)): - return [_encode(v) for v in value] - return value - - def _decode_series(raw: Any) -> list[Decimal]: return [Decimal(v) for v in (raw or [])] @@ -98,15 +97,6 @@ def _decode_summary(raw: Any) -> dict[str, Any]: return out -def canonical_json(payload: Mapping[str, Any]) -> str: - """Deterministic serialisation: sorted keys, no incidental whitespace. - - The hash is only reproducible if this is byte-stable, so both the separators and the key - ordering are pinned here rather than left to `json.dumps` defaults. - """ - return json.dumps(_encode(dict(payload)), sort_keys=True, separators=(",", ":")) - - def _row_payload(record: TrialRecord) -> dict[str, Any]: """Everything that is hashed -- i.e. the row minus `row_hash` itself.""" return { @@ -127,7 +117,15 @@ def _row_payload(record: TrialRecord) -> dict[str, Any]: def compute_row_hash(record: TrialRecord) -> str: - return hashlib.sha256(canonical_json(_row_payload(record)).encode("utf-8")).hexdigest() + """This row's hash, over `_row_payload` -- i.e. the row minus `row_hash` itself. + + The arithmetic moved to `keel_core.hashchain.chain_hash` in #721 so `audit_events` could + share it. WHAT is hashed stayed here, because only this module knows which of a trial's + fields are part of the record. The 93 rows in the git-tracked ledger were hashed before that + move and still verify byte-for-byte -- pinned by + `test_the_tracked_ledger_still_verifies_after_the_canonicaliser_moved`. + """ + return chain_hash(_row_payload(record)) def _validate(record: TrialRecord) -> None: @@ -257,18 +255,15 @@ def verify_records(records: Iterable[TrialRecord]) -> list[str]: it is held for the web view, and it is the difference between an honest badge and a green light over a file nothing read. """ - errors: list[str] = [] - expected_prev = ZERO_HASH - for index, record in enumerate(records, start=1): - if record.prev_hash != expected_prev: - errors.append( - f"row {index} ({record.trial_id}): prev_hash {record.prev_hash[:12]}... " - f"does not chain to {expected_prev[:12]}..." - ) - elif compute_row_hash(record) != record.row_hash: - errors.append(f"row {index} ({record.trial_id}): content does not match row_hash") - expected_prev = record.row_hash - return errors + return verify_links( + ChainLink( + label=record.trial_id, + prev_hash=record.prev_hash, + row_hash=record.row_hash, + recomputed_hash=compute_row_hash(record), + ) + for record in records + ) def trial_counts(trials: Iterable[TrialRecord]) -> tuple[int, int]: diff --git a/keel/web/payload.py b/keel/web/payload.py index e7f3eb8..639b788 100644 --- a/keel/web/payload.py +++ b/keel/web/payload.py @@ -1795,6 +1795,36 @@ def balances_payload(report: BalancesReport) -> dict[str, Any]: } +#: What the chain says about one row, as a state a client can style (#721). `chained` is the only +#: green: `not chained` is an honest gap that nobody checked, and `chain broken` is the row whose +#: hash is shown and is NOT evidence. Nothing here infers a state from whether a hash is present. +_TIMELINE_CHAIN_STATES: Mapping[str, str] = { + "chained": GOOD, + "not chained": UNKNOWN, + "chain broken": BAD, +} + +#: What each status MEANS, spelled out beside the word, for the same reason `_PROVENANCE_NOTES` +#: above exists: on an audit surface the term of art is not the thing a reader can act on. +_TIMELINE_CHAIN_NOTES: Mapping[str, str] = { + "chained": "recorded in the audit chain, and the chain verifies", + "not chained": "written before the audit chain, or from a source that has none — unchecked", + "chain broken": "the chain does not verify from here on; this hash is not evidence", +} + + +def _timeline_hash_display(row: TimelineRow) -> str | None: + """A row hash, shortened for a table cell -- or the NOT-RECORDED words, unshortened. + + `HASH_NOT_RECORDED` is a SENTENCE, not a hash, and truncating it to twelve characters and an + ellipsis would turn "NOT RECORDED" into something that looks like a short hash in a column of + long ones. The status carries the distinction, and the display must not fight it. + """ + if row.chain_status == "not chained": + return row.row_hash + return _hash_display(row.row_hash) + + def _timeline_row_payload(row: TimelineRow) -> dict[str, Any]: """One event, placed. @@ -1822,11 +1852,59 @@ def _timeline_row_payload(row: TimelineRow) -> dict[str, Any]: "amount_kind": row.amount_kind, "summary": row.summary, # A `label`, so the "we did not check" reading carries a state a client can style rather - # than a bare string it might render as though it were a hash. - "row_hash": label(row.row_hash, state=UNKNOWN), + # than a bare string it might render as though it were a hash. Shortened for a table cell + # -- the full value is in the CSV export, which is the artefact anyone actually verifying + # a hash would be working from. + "row_hash": label( + row.row_hash, + display=_timeline_hash_display(row), + state=_TIMELINE_CHAIN_STATES.get(row.chain_status, UNKNOWN), + ), + # BESIDE the hash, never derived from it. "A hash is present" and "the chain vouches for + # it" are different facts, and a client inferring the second from the first would call a + # row inside a broken region verified. Rule 3: the judgement is made here. + "chain_status": label( + row.chain_status, + display=_TIMELINE_CHAIN_NOTES.get(row.chain_status, row.chain_status), + state=_TIMELINE_CHAIN_STATES.get(row.chain_status, UNKNOWN), + ), } +def _timeline_chain_payload(report: TimelineReport) -> Field: + """The audit chain's verdict, as THREE states -- the trading-side twin of `_chain_payload`. + + Three rather than that function's four because this store cannot be ABSENT: `audit_events` is + a table in the same database the report was read from, so "no file" has no analogue here. The + other three are the same distinctions and are the same trap: + + * **no events** -- a deployment that predates #721, or one that has written nothing since. + Not a verification: an empty chain reports no breaks because there is nothing in it to + break, and calling that "verified" is a positive claim about zero rows. + * **events, no errors** -- the one case that is green. + * **a break** -- the record has been altered, and every row from there on is unproven. + + GOOD requires BOTH: no errors, over events that actually exist. + """ + if not report.chain_recorded: + return label( + "unverified", + display="no audit chain recorded yet — nothing here was checked", + state=UNKNOWN, + ) + if report.chain_intact: + return label( + "intact", + display="audit chain verified — every recorded event still hashes to the next", + state=GOOD, + ) + return label( + "broken", + display="the audit chain does not verify — the rows below say from where", + state=BAD, + ) + + def timeline_payload(report: TimelineReport) -> dict[str, Any]: """`gather_timeline`'s `TimelineReport`, as JSON (#703). @@ -1845,6 +1923,11 @@ def timeline_payload(report: TimelineReport) -> dict[str, Any]: "scoped_count": count(report.scoped_count), "filtered_count": count(report.filtered_count), "shown_count": count(report.shown_count), + # The verdict over the WHOLE chain, beside the per-row statuses -- the same pairing + # `_chain_payload` holds for the research ledger, and for the same reason: an empty chain + # has no breaks and has verified nothing, so "no errors" is never on its own a green light. + "chain": _timeline_chain_payload(report), + "chain_errors": [str(error) for error in report.chain_errors], "rows": [_timeline_row_payload(row) for row in report.rows], } diff --git a/keel/web/static/js/render.js b/keel/web/static/js/render.js index 8fb6e82..7b5fd29 100644 --- a/keel/web/static/js/render.js +++ b/keel/web/static/js/render.js @@ -1203,9 +1203,16 @@ export function timelineView(data, sort, onSort, onKind) { link.setAttribute("href", target.pathname.concat(target.search)); link.setAttribute("download", ""); actions.append(link); - actions.append(" — every row with its provenance; hashes are not recorded yet."); + actions.append(" — every row with its provenance, its hash and whether the chain verifies it."); fragment.append(actions); + // The chain's verdict over the whole record, above the rows it applies to. A per-row status + // answers "is this row evidence"; this answers "has this record been altered", which is the + // question an operator opening an audit page is actually asking. + const chain = el("p", "note"); + chain.append(field(data.chain)); + fragment.append(chain); + fragment.append(heading("h-timeline", "What happened")); fragment.append( table( @@ -1222,6 +1229,10 @@ export function timelineView(data, sort, onSort, onKind) { { label: "product", numeric: false, key: "product_id" }, { label: "amount", numeric: true }, { label: "what", numeric: false }, + // Both columns, never one. A hash with nothing saying whether it verifies is a number a + // reader takes on trust, and the reading taken on trust is the flattering one. + { label: "row hash", numeric: false }, + { label: "tamper-evidence", numeric: false }, ], (data.rows || []).map(/** @param {any} row */ (row) => [ row.at, @@ -1232,6 +1243,8 @@ export function timelineView(data, sort, onSort, onKind) { plain(row.product_id) || "—", row.amount, plain(row.summary) || "—", + row.row_hash, + row.chain_status, ]), "Nothing recorded in this window.", { sort: sort, onSort: onSort }, diff --git a/packages/keel-core/keel_core/hashchain.py b/packages/keel-core/keel_core/hashchain.py new file mode 100644 index 0000000..5e8c944 --- /dev/null +++ b/packages/keel-core/keel_core/hashchain.py @@ -0,0 +1,185 @@ +"""One canonical form and one chain walk, for every hash-chained store in this codebase (#721). + +Two stores chain their rows: the research trials ledger (`keel/research/ledger.py` -- JSONL, git +tracked, records experiments) and `audit_events` (`keel/data/audit.py` -- SQLite, per deployment, +records trading activity). They are deliberately separate stores over separate domains, and +blending them would be provenance laundering. What they must NOT have separately is a definition +of what a row's hash is over. + +**Why this is a shared module and not a copied function.** A hash is only evidence if it can be +recomputed. Two canonicalisations that disagree -- one emitting `{"a": 1}` and the other +`{"a":1}`, one rendering `Decimal("1.0")` as `1.0` and the other as `"1.0"` -- produce two hashes +for one row. That disagreement is invisible at write time and surfaces months later as a chain +that "cannot be verified", at exactly the moment someone is trying to establish whether a record +was altered. The form is decided here, once. + +**Tamper-EVIDENT, not tamper-proof.** Anyone who can write the store can rewrite the whole chain +from the edited row forward. What the chain buys is that a row cannot be changed QUIETLY: an +edit invalidates that row and every row after it, so a partial edit is visible and a full rewrite +requires touching every subsequent row. + +This module holds no I/O, no `keel` imports and no schema. It is pure so both stores can depend +on it without depending on each other. +""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Iterable, Mapping +from dataclasses import dataclass +from decimal import Decimal +from typing import Any + +#: What the FIRST row of a chain commits to. A chain anchored anywhere else starts mid-air: it +#: says nothing about whether rows were removed from in front of it, which is the one deletion a +#: chain would otherwise miss entirely. +ZERO_HASH = "0" * 64 + + +def encode_value(value: Any) -> Any: + """`Decimal` -> `str`, recursively, so JSON round-trips a money value exactly. + + Money is TEXT in every store here (`orders.qty`, `equity_points.equity`), and the reason is + the same reason it is a string in the hash: a float rendering of `Decimal("0.10")` is a + different number from the one the row holds, and a hash over a lossy rendering attests to a + value that was never recorded. + + SCALE IS PRESERVED, and deliberately: `Decimal("1.0")` and `Decimal("1.00")` compare equal + numerically and are different recorded values. A store that rewrote a row's scale changed + that row, and the chain says so rather than shrugging. + """ + if isinstance(value, Decimal): + return str(value) + if isinstance(value, Mapping): + return {k: encode_value(v) for k, v in value.items()} + if isinstance(value, (list, tuple)): + return [encode_value(v) for v in value] + return value + + +def canonical_json(payload: Mapping[str, Any]) -> str: + """Deterministic serialisation: sorted keys, no incidental whitespace, Decimals as strings. + + Both the separators and the key ordering are pinned here rather than left to `json.dumps` + defaults, because the hash is only reproducible if this is byte-stable. `sort_keys` in + particular is not a tidiness preference: the same row is assembled from a `dict` literal in + one module and from `dict(sqlite3.Row)` in another, and insertion order differs between them. + + An ABSENT key is not a null key. `{"a": 1}` and `{"a": 1, "b": None}` hash differently, which + is correct -- a writer that started emitting an explicit null for a field it used to omit has + changed what the row says. + """ + return json.dumps(encode_value(dict(payload)), sort_keys=True, separators=(",", ":")) + + +def chain_hash(payload: Mapping[str, Any]) -> str: + """SHA-256 over `canonical_json(payload)`. + + `payload` is everything the row commits to INCLUDING its `prev_hash` and excluding its own + `row_hash`. Including `prev_hash` is what makes this a chain rather than a column of + independent checksums: without it, a row could be moved, duplicated or reordered freely. + """ + return hashlib.sha256(canonical_json(payload).encode("utf-8")).hexdigest() + + +@dataclass(frozen=True) +class ChainLink: + """One row, reduced to the four things a chain walk needs. + + Deliberately not a row: this module knows nothing about trials or audit events, and each + store keeps its own record type, its own payload shape and its own reader. `recomputed_hash` + is computed BY THE CALLER, because only the caller knows which of its fields are hashed -- + which is the fact that makes the check meaningful, and the fact this module must not guess. + + `label` is whatever identifies the row to a human reading a report -- a `trial_id`, a + `seq_id`. It appears in the error text and nowhere else. + """ + + label: str + prev_hash: str + row_hash: str + recomputed_hash: str + + +@dataclass(frozen=True) +class ChainBreak: + """One place the chain fails, located as well as described. + + `index` is 1-based POSITION IN THE WALK, not any store's own identifier: the walk is the only + thing that knows where a row sits in the chain, and a store whose rows were renumbered would + otherwise report breaks at coordinates that no longer mean anything. Callers that need their + own key look it up by this index -- see `keel/data/audit.py::chain_state`, which uses it to + find the `seq_id` from which the chain stops being evidence. + + `reason` is one of `link` (a row was inserted, removed or reordered) or `content` (a row was + edited in place). Two different accusations, and a report that blurred them would tell an + auditor to look for the wrong thing. + """ + + index: int + label: str + reason: str + message: str + + +def find_breaks(links: Iterable[ChainLink]) -> list[ChainBreak]: + """Every break in the chain, located. Empty means intact. + + **Reports rather than raises.** A broken chain is a finding to surface in a report -- `keel + doctor`, the timeline export, the research page's badge -- and the caller wants every break + rather than only the first. A check that stopped at the first would let a second edit hide + behind the first, which is the shape an auditor most needs to see. + + Two distinct failures, and the `elif` keeps them from doubling up: a row whose `prev_hash` + does not match its predecessor's `row_hash` is a LINK failure, and a row whose content does + not reproduce its own `row_hash` is a CONTENT failure. A row that fails the link check is + reported once, on the link, because its content hash is then computed over a payload the walk + already knows is in the wrong place. + + The walk continues from the row's STORED `row_hash` rather than the recomputed one, so a + single edit produces one error rather than cascading into a false break at every later row. + A DELETED row is what breaks every row after it, and that is the property these stores exist + for. + + **An empty sequence returns no breaks, and that is not the same as "verified".** Nothing was + checked. The distinction belongs to the caller, because only the caller knows whether empty + means "no rows yet" or "no store at all" -- the difference between an honest badge and a + green light over a file nothing read. + """ + breaks: list[ChainBreak] = [] + expected_prev = ZERO_HASH + for index, link in enumerate(links, start=1): + if link.prev_hash != expected_prev: + breaks.append( + ChainBreak( + index=index, + label=link.label, + reason="link", + message=( + f"row {index} ({link.label}): prev_hash {link.prev_hash[:12]}... " + f"does not chain to {expected_prev[:12]}..." + ), + ) + ) + elif link.recomputed_hash != link.row_hash: + breaks.append( + ChainBreak( + index=index, + label=link.label, + reason="content", + message=f"row {index} ({link.label}): content does not match row_hash", + ) + ) + expected_prev = link.row_hash + return breaks + + +def verify_links(links: Iterable[ChainLink]) -> list[str]: + """`find_breaks`, as the human-readable lines a report prints. Empty means intact. + + A formatter over the one walk rather than a second walk: a chain check that existed twice is + a chain check that can disagree with itself, and the disagreement would be between "the + report says intact" and "the badge says broken" over the same rows. + """ + return [found.message for found in find_breaks(links)] diff --git a/tests/commands/test_doctor.py b/tests/commands/test_doctor.py index 0e65c3f..f20e584 100644 --- a/tests/commands/test_doctor.py +++ b/tests/commands/test_doctor.py @@ -598,6 +598,7 @@ def test_gather_findings_covers_every_check_over_a_seeded_db(tmp_path, valid_con "attest.withdrawals", "attest.asset_window", "attest.instrument_window", + "audit.chain", "scope.trade", "attest.cash_posture", "rail.kill_switch", @@ -1065,3 +1066,64 @@ def test_a_folder_keel_cannot_read_measures_as_empty(tmp_path) -> None: assert footprint.total_files == 0 assert read_backup_footprint(tmp_path / "missing").total_files == 0 + + +# -- the audit chain (#721) --------------------------------------------------------------------- + + +def _chain(**overrides: object): + from keel.data.audit import ChainState + + fields: dict[str, object] = { + "table_present": True, + "event_count": 3, + "errors": (), + "first_broken_seq": None, + "hashes": {}, + } + fields.update(overrides) + return ChainState(**fields) # type: ignore[arg-type] + + +def test_a_verifying_chain_is_ok_and_says_how_much_it_checked() -> None: + from keel.commands.doctor import audit_chain_findings + + (finding,) = audit_chain_findings(_chain()) + assert finding.name == "audit.chain" + assert finding.status == "ok" + assert "3" in finding.detail + + +def test_an_empty_chain_is_ok_but_never_called_verified() -> None: + """Nothing was checked. A doctor line reading "audit chain verifies" over zero events is the + green badge over an unread file that this codebase keeps having to take back out.""" + from keel.commands.doctor import audit_chain_findings + + (finding,) = audit_chain_findings(_chain(event_count=0)) + assert finding.status == "ok" + assert "verifies" not in finding.headline + assert "unverified" in finding.headline + + +def test_a_break_fails_and_offers_no_fix() -> None: + """The one check here whose failure means the RECORD was altered rather than a rail being out + of position. Nothing repairs a broken chain, and a `fix` command would imply something does.""" + from keel.commands.doctor import audit_chain_findings + + (finding,) = audit_chain_findings( + _chain(errors=("row 2 (2): content does not match row_hash",), first_broken_seq=2) + ) + assert finding.status == "fail" + assert finding.fix == "-" + assert "row 2" in finding.detail + + +def test_a_database_without_the_table_is_reported_not_crashed_on() -> None: + """`keel mcp`'s `_open_readonly_repo` and the web server both open a repo WITHOUT migrating. + A reader that raised here would take the whole of `gather_findings` down on an un-upgraded + deployment -- which is exactly what #718's first cut did.""" + from keel.commands.doctor import audit_chain_findings + + (finding,) = audit_chain_findings(_chain(table_present=False)) + assert finding.status == "ok" + assert finding.fix == "keel migrate" diff --git a/tests/commands/test_timeline.py b/tests/commands/test_timeline.py index 3d7dfb3..5ed3d29 100644 --- a/tests/commands/test_timeline.py +++ b/tests/commands/test_timeline.py @@ -14,13 +14,17 @@ import csv import io +import sqlite3 from decimal import Decimal from pathlib import Path +from types import SimpleNamespace from typing import Any import pytest +from keel.commands import timeline from keel.commands.timeline import PROVENANCES, TIMELINE_KINDS, gather_timeline +from keel.data import audit from keel.data.db import connect, migrate from keel.data.repository import Repository @@ -30,10 +34,17 @@ @pytest.fixture() -def repo(tmp_path: Path) -> Repository: +def db_conn(tmp_path: Path) -> sqlite3.Connection: + """The CONNECTION, for the tests that reach past the repository to edit `audit_events` in + place -- which is the only way to write the tampering the chain exists to detect.""" conn = connect(str(tmp_path / "keel.db")) migrate(conn) - return Repository(conn) + return conn + + +@pytest.fixture() +def repo(db_conn: sqlite3.Connection) -> Repository: + return Repository(db_conn) def _order(repo: Repository, **overrides: Any) -> int: @@ -200,17 +211,27 @@ def test_each_row_names_the_store_it_came_from(repo: Repository, tmp_path: Path) def test_no_row_claims_a_hash_it_does_not_have(repo: Repository, tmp_path: Path) -> None: - """#703 asked for tamper-evidence. None of these four stores hashes its rows, so every row - says NOT RECORDED -- never blank, which a reader takes as "nothing to report", and never a - hash computed here, which would be this module attesting to its own output.""" - from keel.commands.timeline import HASH_NOT_RECORDED + """Never blank, which a reader takes as "nothing to report", and never a hash computed HERE, + which would be this module attesting to its own output. + + #703 shipped this as "every row says NOT RECORDED", which was the whole truth then. #721 made + the three stores chain their writes, so the invariant is now the pairing rather than the + constant: a hash and a `chained` status travel together, and the absence of one is the + absence of both. A row showing a hash while claiming `not chained`, or claiming `chained` + with nothing to show, would be a row asserting something the chain never said. + """ + from keel.commands.timeline import CHAIN_STATUSES, HASH_NOT_RECORDED _order(repo) _transaction(repo) _attestation(repo) - for row in gather_timeline(repo, now_ts=NOW_TS, scope="all").rows: - assert row.row_hash == HASH_NOT_RECORDED + rows = gather_timeline(repo, now_ts=NOW_TS, scope="all").rows + assert rows + for row in rows: + assert row.chain_status in CHAIN_STATUSES + assert (row.row_hash == HASH_NOT_RECORDED) == (row.chain_status == "not chained") + assert row.row_hash != "" # -- filtering and scoping ---------------------------------------------------------------------- @@ -432,3 +453,176 @@ def test_the_kind_collapse_is_applied_and_echoed(repo, tmp_path) -> None: assert report.kind == "", "the applied value is echoed, so the substitution is visible" assert report.shown_count == 2, "and nothing was filtered out" + + +# -- the hash column, once the engine records one (#721) ---------------------------------------- +# +# #703 shipped `row_hash` reading NOT RECORDED on every row because none of the four stores hashed +# anything. `keel/data/audit.py` now does. These pin the three readings the column has to keep +# apart: a hash the chain vouches for, an honest gap where no event was ever written, and a hash +# that exists inside the region a break has invalidated. The third is the one that must not be +# allowed to look like the first. + + +def _chained_repo(conn) -> Repository: + """A repository whose writers have recorded events for everything they wrote.""" + repo = Repository(conn) + repo.insert_order( + { + "mode": "live", + "product_id": "BTC-USD", + "side": "buy", + "qty": Decimal("1"), + "status": "filled", + "created_at": 1_000, + } + ) + repo.upsert_transaction( + { + "coinbase_id": "cb-1", + "source": "coinbase", + "type": "deposit", + "asset": "USD", + "qty": Decimal("250"), + "ts": 1_100, + } + ) + repo.upsert_asset_attestation( + asset="BTC", + sector="tech", + backing="none", + pays_yield=False, + source="prospectus", + attested_by="operator", + attested_at=1_200, + ) + return repo + + +def test_a_chained_row_carries_its_recorded_hash(db_conn) -> None: + """The swap #703 built the column for: a real hash where the engine recorded one.""" + repo = _chained_repo(db_conn) + report = timeline.gather_timeline(repo, now_ts=2_000) + + by_source = {row.source: row for row in report.rows} + for source in ("orders", "transactions", "asset_attestations"): + assert by_source[source].row_hash != timeline.HASH_NOT_RECORDED + assert len(by_source[source].row_hash) == 64 + assert by_source[source].chain_status == "chained" + + +def test_the_hash_shown_is_the_hash_recorded_for_that_row(db_conn) -> None: + """Looked up by `(source, reference)` -- the pair printed beside it. A lookup keyed on the + entity alone would hand an `orders.id` of "1" the hash of a `coinbase_id` of "1".""" + repo = _chained_repo(db_conn) + report = timeline.gather_timeline(repo, now_ts=2_000) + + recorded = audit.chain_state(db_conn).hashes + for row in report.rows: + if row.chain_status == "chained": + assert row.row_hash == recorded[(row.source, row.reference)].row_hash + + +def test_a_row_written_before_the_chain_shipped_reads_as_not_chained(db_conn) -> None: + """NO BACKFILL, on the surface that shows it. An honest gap -- and crucially NOT a break: a + deployment upgrading into #721 must not open its timeline to a page of red.""" + db_conn.execute( + "INSERT INTO orders (mode, product_id, side, qty, status, created_at) " + "VALUES ('live','BTC-USD','buy','1','filled',900)" + ) + db_conn.commit() + repo = _chained_repo(db_conn) + report = timeline.gather_timeline(repo, now_ts=2_000) + + unchained = [row for row in report.rows if row.reference == "1" and row.source == "orders"] + assert len(unchained) == 1 + assert unchained[0].row_hash == timeline.HASH_NOT_RECORDED + assert unchained[0].chain_status == "not chained" + assert report.chain_intact is True + + +def test_an_engine_log_row_is_never_chained_and_says_so(db_conn) -> None: + """The engine log is a FILE, not a chained store. A cycle row carrying a hash would be this + module attesting to something it only read.""" + repo = _chained_repo(db_conn) + cycle = SimpleNamespace( + started_ts=1_500, cycle_id="c-1", products=("BTC-USD",), signals=1, entered=1, exited=0, + errors=0, + ) + report = timeline.gather_timeline(repo, now_ts=2_000, cycles=[cycle]) + + system = [row for row in report.rows if row.kind == "system"] + assert len(system) == 1 + assert system[0].row_hash == timeline.HASH_NOT_RECORDED + assert system[0].chain_status == "not chained" + + +def test_a_break_marks_the_rows_the_chain_no_longer_vouches_for(db_conn) -> None: + """Editing one event does not merely fail that row: the chain proves a SEQUENCE, so every row + from the break onward is unverified. A page that showed those later hashes as `chained` would + be presenting unverified values as evidence -- the one thing this column exists to prevent.""" + repo = _chained_repo(db_conn) + db_conn.execute("UPDATE audit_events SET payload_json = ? WHERE seq_id = 1", ('{"a":1}',)) + db_conn.commit() + + report = timeline.gather_timeline(repo, now_ts=2_000) + assert report.chain_intact is False + assert report.chain_errors + assert {row.chain_status for row in report.rows} == {"chain broken"} + + +def test_a_break_leaves_earlier_rows_verified(db_conn) -> None: + """The break is located, not global. Rows chained BEFORE it are still vouched for, and + reporting them as broken would throw away the evidence the chain does hold.""" + repo = _chained_repo(db_conn) + db_conn.execute("UPDATE audit_events SET payload_json = ? WHERE seq_id = 3", ('{"a":1}',)) + db_conn.commit() + + report = timeline.gather_timeline(repo, now_ts=2_000) + statuses = {row.source: row.chain_status for row in report.rows} + assert statuses["orders"] == "chained" + assert statuses["transactions"] == "chained" + assert statuses["asset_attestations"] == "chain broken" + + +def test_an_empty_chain_is_reported_as_unchecked_not_as_verified(db_conn) -> None: + """`verify` over zero rows returns no errors, and that is not "verified" -- nothing was read. + The report must not offer a green verdict over a table nothing wrote.""" + db_conn.execute( + "INSERT INTO orders (mode, product_id, side, qty, status, created_at) " + "VALUES ('live','BTC-USD','buy','1','filled',900)" + ) + db_conn.commit() + report = timeline.gather_timeline(Repository(db_conn), now_ts=2_000) + assert report.chain_recorded is False + assert report.chain_intact is True + assert report.chain_errors == () + + +def test_the_export_carries_the_chain_status_beside_the_hash(db_conn) -> None: + """A hash with nothing saying whether it verifies is a number an auditor cannot use. Both + columns, in the same file, on every row.""" + repo = _chained_repo(db_conn) + text = timeline.to_csv(timeline.export_rows(repo, now_ts=2_000)) + rows = list(csv.reader(io.StringIO(text))) + + assert rows[0][-2:] == ["row_hash", "chain_status"] + # Every row, and the exact pairing -- `in (the three words)` would pass on any of them and so + # would survive the status being hard-coded to one value. + assert len(rows) == 4 + for row in rows[1:]: + assert row[-1] == "chained" + assert len(row[-2]) == 64 + + +def test_a_broken_chain_is_stated_above_the_header_not_only_per_row(db_conn) -> None: + """The same reasoning as the engine-log note this file already carries: an auditor holding a + CSV cannot ask the page anything, so a finding that changes how the whole file should be read + belongs in the file.""" + repo = _chained_repo(db_conn) + db_conn.execute("UPDATE audit_events SET payload_json = ? WHERE seq_id = 1", ('{"a":1}',)) + db_conn.commit() + + text = timeline.to_csv(timeline.export_rows(repo, now_ts=2_000)) + assert text.splitlines()[0].startswith("# NOTE") + assert "chain" in text.splitlines()[0] diff --git a/tests/core/test_hashchain.py b/tests/core/test_hashchain.py new file mode 100644 index 0000000..0d536d8 --- /dev/null +++ b/tests/core/test_hashchain.py @@ -0,0 +1,162 @@ +"""The one canonical form and the one chain walk (#721). + +Two stores hash rows in this codebase -- the research trials ledger (JSONL, on disk, git-tracked) +and `audit_events` (SQLite, per deployment). They must agree about what canonical JSON means, +because two canonicalisations that disagree produce two hashes for one row, and the disagreement +does not surface as a bug: it surfaces, months later, as a chain that "cannot be verified". + +So the form lives here, once, and both stores import it. These tests pin the properties the +hashes depend on rather than the hashes themselves -- with one deliberate exception, the tracked +ledger's own 93 rows (`tests/research/test_ledger.py`), which were hashed before this module +existed and must still verify byte-for-byte. +""" + +from __future__ import annotations + +from decimal import Decimal + +from keel_core import hashchain + + +def _link(label: str, prev: str, payload: dict[str, object]) -> hashchain.ChainLink: + body = dict(payload) + body["prev_hash"] = prev + row_hash = hashchain.chain_hash(body) + return hashchain.ChainLink( + label=label, prev_hash=prev, row_hash=row_hash, recomputed_hash=row_hash + ) + + +def test_canonical_json_is_insertion_order_blind() -> None: + """Key order in the caller's dict must not reach the hash. + + A payload assembled by a `dict` literal in one module and by `dict(row)` out of sqlite in + another would otherwise hash differently while describing the same row. + """ + assert hashchain.canonical_json({"b": 1, "a": 2}) == hashchain.canonical_json( + {"a": 2, "b": 1} + ) + + +def test_canonical_json_has_no_incidental_whitespace() -> None: + """`json.dumps` defaults put a space after every separator. Pinned, because a later reader + "tidying" the call would silently invalidate every hash ever written.""" + assert hashchain.canonical_json({"a": 1, "b": 2}) == '{"a":1,"b":2}' + + +def test_decimal_crosses_as_its_exact_string_at_any_depth() -> None: + """Money is TEXT everywhere in this codebase, and a hash over a float would be a hash over a + lossy rendering of the number the row actually holds.""" + text = hashchain.canonical_json( + {"qty": Decimal("0.10"), "legs": [Decimal("1.5")], "nested": {"fee": Decimal("2.000")}} + ) + assert text == '{"legs":["1.5"],"nested":{"fee":"2.000"},"qty":"0.10"}' + + +def test_decimal_scale_is_part_of_the_hash() -> None: + """`Decimal("1.0") == Decimal("1.00")` numerically, and they are DIFFERENT recorded values. + A store that rewrote a row's scale changed the row, and the chain must say so.""" + assert hashchain.chain_hash({"qty": Decimal("1.0")}) != hashchain.chain_hash( + {"qty": Decimal("1.00")} + ) + + +def test_chain_hash_changes_when_any_field_changes() -> None: + base = {"a": 1, "b": "x"} + assert hashchain.chain_hash(base) != hashchain.chain_hash({"a": 1, "b": "y"}) + assert hashchain.chain_hash(base) != hashchain.chain_hash({"a": 2, "b": "x"}) + + +def test_an_absent_key_is_not_a_null_key() -> None: + """The absent-vs-null distinction the ledger's docstring names. A writer that started + emitting an explicit `None` for a field it used to omit has changed the row.""" + assert hashchain.chain_hash({"a": 1}) != hashchain.chain_hash({"a": 1, "b": None}) + + +def test_an_intact_chain_reports_nothing() -> None: + first = _link("one", hashchain.ZERO_HASH, {"n": 1}) + second = _link("two", first.row_hash, {"n": 2}) + assert hashchain.verify_links([first, second]) == [] + + +def test_a_first_row_not_anchored_to_zero_is_a_break() -> None: + """A chain that starts mid-air proves nothing about what came before it.""" + orphan = _link("one", "f" * 64, {"n": 1}) + errors = hashchain.verify_links([orphan]) + assert len(errors) == 1 + assert "row 1 (one)" in errors[0] + assert "does not chain" in errors[0] + + +def test_edited_content_fails_at_that_row() -> None: + first = _link("one", hashchain.ZERO_HASH, {"n": 1}) + tampered = hashchain.ChainLink( + label=first.label, + prev_hash=first.prev_hash, + row_hash=first.row_hash, + recomputed_hash=hashchain.chain_hash({"n": 99, "prev_hash": first.prev_hash}), + ) + errors = hashchain.verify_links([tampered]) + assert len(errors) == 1 + assert "row 1 (one)" in errors[0] + assert "row_hash" in errors[0] + + +def test_every_break_is_reported_not_only_the_first() -> None: + """A chain check that stopped at the first break would let the second edit hide behind the + first, which is the shape an auditor most needs to see.""" + first = _link("one", "a" * 64, {"n": 1}) + second = _link("two", "b" * 64, {"n": 2}) + errors = hashchain.verify_links([first, second]) + assert len(errors) == 2 + + +def test_a_deleted_row_breaks_every_row_after_it() -> None: + """The property that makes this evidence rather than a report.""" + first = _link("one", hashchain.ZERO_HASH, {"n": 1}) + second = _link("two", first.row_hash, {"n": 2}) + third = _link("three", second.row_hash, {"n": 3}) + errors = hashchain.verify_links([first, third]) + assert len(errors) == 1 + assert "row 2 (three)" in errors[0] + + +def test_an_empty_chain_reports_nothing_and_that_is_not_verified() -> None: + """Nothing was checked. The caller holds the distinction between "no rows yet" and "no store + at all" -- see `verify_links`' own docstring.""" + assert hashchain.verify_links([]) == [] + + +def test_a_break_is_located_as_well_as_described() -> None: + """`verify_links` prints; `find_breaks` locates. A caller that needs its own key for the + first unverified row -- the timeline's chain-status column -- must not have to parse the + English out of a message to find it.""" + first = _link("one", hashchain.ZERO_HASH, {"n": 1}) + second = _link("two", first.row_hash, {"n": 2}) + edited = hashchain.ChainLink( + label=second.label, + prev_hash=second.prev_hash, + row_hash=second.row_hash, + recomputed_hash=hashchain.chain_hash({"n": 99, "prev_hash": second.prev_hash}), + ) + breaks = hashchain.find_breaks([first, edited]) + assert [(found.index, found.label, found.reason) for found in breaks] == [(2, "two", "content")] + + +def test_a_missing_row_is_a_link_break_not_a_content_break() -> None: + """Two different accusations. A report that blurred them would send an auditor looking for an + edited row when what happened was a deletion.""" + first = _link("one", hashchain.ZERO_HASH, {"n": 1}) + second = _link("two", first.row_hash, {"n": 2}) + third = _link("three", second.row_hash, {"n": 3}) + assert [found.reason for found in hashchain.find_breaks([first, third])] == ["link"] + + +def test_verify_links_is_exactly_find_breaks_formatted() -> None: + """One walk, not two. A chain check that existed twice could disagree with itself, and the + disagreement would be between a report and a badge over the same rows.""" + first = _link("one", "a" * 64, {"n": 1}) + second = _link("two", "b" * 64, {"n": 2}) + assert hashchain.verify_links([first, second]) == [ + found.message for found in hashchain.find_breaks([first, second]) + ] diff --git a/tests/data/test_audit_chain.py b/tests/data/test_audit_chain.py new file mode 100644 index 0000000..0058b74 --- /dev/null +++ b/tests/data/test_audit_chain.py @@ -0,0 +1,573 @@ +"""`audit_events`: the append-only, hash-chained record of what the book did (#721). + +`orders` rows are UPDATED as a venue reports fills, so a chain over the order row itself would +break on every legitimate update -- a chain that cries wolf is a chain nobody reads. So the chain +is over EVENTS: `insert_order` and each `update_order` append an immutable statement of what +changed, and it is the event stream, not the mutable book, that is tamper-evident. + +What these tests pin is the property that makes it evidence: an edit or a deletion cannot be made +quietly. Everything else here -- the vocabulary, the atomicity, the pre-bump gap -- exists to +keep that property true in the presence of the rest of the engine. +""" + +from __future__ import annotations + +import sqlite3 +from decimal import Decimal + +import pytest +from keel_core.hashchain import ZERO_HASH, chain_hash + +from keel.data import audit, db +from keel.data.repository import Repository + + +@pytest.fixture() +def conn() -> sqlite3.Connection: + connection = db.connect(":memory:") + db.migrate(connection) + return connection + + +def _append(connection: sqlite3.Connection, **kwargs: object) -> audit.AuditEvent: + """One event in its own write transaction, as a repository writer does around a store row.""" + with audit.write_transaction(connection): + return audit.append_event(connection, **kwargs) # type: ignore[arg-type] + + +def _order(**overrides: object) -> dict[str, object]: + """A minimally legal `orders` row -- every NOT NULL column and nothing else.""" + row: dict[str, object] = { + "mode": "live", + "product_id": "BTC-USD", + "side": "buy", + "qty": Decimal("1"), + "created_at": 100, + } + row.update(overrides) + return row + + +def _transaction(**overrides: object) -> dict[str, object]: + """A minimally legal `transactions` row.""" + row: dict[str, object] = { + "coinbase_id": "cb-1", + "source": "coinbase", + "type": "deposit", + "asset": "USD", + "ts": 100, + "qty": Decimal("1"), + } + row.update(overrides) + return row + + +# -- the chain itself ----------------------------------------------------------------------- + + +def test_the_first_event_anchors_to_the_zero_hash(conn: sqlite3.Connection) -> None: + event = _append(conn, ts=100, event_type="order_placed", entity_id="1", payload={"a": 1}) + assert event.prev_hash == ZERO_HASH + assert len(event.row_hash) == 64 + + +def test_each_event_commits_to_its_predecessor(conn: sqlite3.Connection) -> None: + first = _append(conn, ts=100, event_type="order_placed", entity_id="1", payload={"a": 1}) + second = _append(conn, ts=101, event_type="order_updated", entity_id="1", payload={"a": 2}) + assert second.prev_hash == first.row_hash + assert audit.chain_state(conn).errors == () + + +def test_the_chain_spans_event_types_not_one_chain_per_table(conn: sqlite3.Connection) -> None: + """One chain over everything, so an event cannot be moved between streams, and so a removed + order event is visible from a later transaction event rather than only from its own kind.""" + _append(conn, ts=100, event_type="order_placed", entity_id="1", payload={"a": 1}) + second = _append( + conn, ts=101, event_type="transaction_recorded", entity_id="cb-1", payload={"a": 2} + ) + events = audit.read_events(conn) + assert second.prev_hash == events[0].row_hash + assert audit.chain_state(conn).errors == () + + +def test_editing_a_payload_in_place_breaks_that_row_and_no_other(conn: sqlite3.Connection) -> None: + for index in range(3): + _append( + conn, + ts=100 + index, + event_type="order_placed", + entity_id=str(index), + payload={"a": index}, + ) + conn.execute("UPDATE audit_events SET payload_json = ? WHERE seq_id = 2", ('{"a":99}',)) + conn.commit() + + errors = audit.chain_state(conn).errors + assert len(errors) == 1 + assert "row 2" in errors[0] + assert "row_hash" in errors[0] + + +def test_deleting_a_row_breaks_every_row_after_it(conn: sqlite3.Connection) -> None: + """The property the whole store exists for: a deletion is not a quiet deletion.""" + for index in range(4): + _append( + conn, + ts=100 + index, + event_type="order_placed", + entity_id=str(index), + payload={"a": index}, + ) + conn.execute("DELETE FROM audit_events WHERE seq_id = 2") + conn.commit() + + errors = audit.chain_state(conn).errors + assert len(errors) == 1, errors + assert "row 2" in errors[0] + assert "does not chain" in errors[0] + + +def test_events_are_read_in_chain_order_not_timestamp_order(conn: sqlite3.Connection) -> None: + """`seq_id`, never `ts`. Two events inside one second share a timestamp, and a clock that + steps backwards would reorder the chain into a false break.""" + # THREE, and the timestamps deliberately out of order. With two, the head read has only one + # candidate and any ordering picks it -- which is why the two-event version of this test + # passed against a head read ordered by `ts`. The third event is the one whose predecessor + # the two orderings disagree about: by `seq_id` it chains to the second, by `ts DESC` it + # would chain to the FIRST, and the chain would fork. + _append(conn, ts=500, event_type="order_placed", entity_id="1", payload={"a": 1}) + _append(conn, ts=100, event_type="order_updated", entity_id="1", payload={"a": 2}) + _append(conn, ts=200, event_type="order_updated", entity_id="1", payload={"a": 3}) + + events = audit.read_events(conn) + assert [event.seq_id for event in events] == [1, 2, 3] + assert events[2].prev_hash == events[1].row_hash + assert audit.chain_state(conn).errors == () + + +def test_a_decimal_in_the_payload_survives_the_round_trip(conn: sqlite3.Connection) -> None: + """Money is the point. A payload whose Decimals came back as floats would verify against a + hash of a value the row never held.""" + event = _append( + conn, + ts=100, + event_type="order_placed", + entity_id="1", + payload={"qty": Decimal("0.10"), "fee": None}, + ) + stored = audit.read_events(conn)[0] + assert stored.payload["qty"] == "0.10" + assert stored.row_hash == event.row_hash + assert audit.chain_state(conn).errors == () + + +# -- the vocabulary and the guards ------------------------------------------------------------ + + +def test_an_unrecognised_event_type_is_refused(conn: sqlite3.Connection) -> None: + """A closed vocabulary, like every other `kind`/`provenance` set in this codebase: a typo + that lands as a new event type is a stream nothing reads and nothing misses.""" + with pytest.raises(ValueError, match="event_type"): + _append(conn, ts=100, event_type="order_plased", entity_id="1", payload={}) + + +def test_appending_outside_a_write_transaction_is_refused(conn: sqlite3.Connection) -> None: + """The head read and the insert MUST be one transaction. Read the head in autocommit and two + writers racing both see the same head, both write it as their `prev_hash`, and the chain + forks -- silently, because each row verifies against the row it thinks precedes it.""" + with pytest.raises(RuntimeError, match="transaction"): + audit.append_event(conn, ts=100, event_type="order_placed", entity_id="1", payload={}) + + +def test_the_head_read_takes_a_write_lock_before_reading(conn: sqlite3.Connection) -> None: + """`BEGIN IMMEDIATE`, not `BEGIN`. A deferred transaction takes no lock until its first + write, so the head SELECT would still be unserialised against a concurrent writer.""" + with audit.write_transaction(conn): + assert conn.in_transaction + # A second connection to the same file cannot begin writing while this one holds the + # write lock. In-memory connections do not share a database, so this asserts the + # statement issued rather than the lock's effect -- the effect is asserted below. + assert not conn.in_transaction + + +def test_two_connections_cannot_hold_the_write_lock_at_once(tmp_path) -> None: + """The lock, on a real file. Without `IMMEDIATE` the second writer would be admitted here and + would read the same chain head as the first.""" + path = tmp_path / "keel.db" + first = db.connect(path) + db.migrate(first) + second = db.connect(path) + second.execute("PRAGMA busy_timeout = 50") + + with audit.write_transaction(first): + # BEFORE any write. A DEFERRED transaction would still be admitted here and would take + # its lock only at the INSERT -- by which point it has already read a chain head that a + # racing writer may have moved. Appending first and then checking the lock passes under + # `BEGIN` too, and so proves nothing: the insert itself takes the lock either way. + with pytest.raises(sqlite3.OperationalError, match="locked|busy"): + second.execute("BEGIN IMMEDIATE") + audit.append_event(first, ts=100, event_type="order_placed", entity_id="1", payload={}) + + +# -- the writers ---------------------------------------------------------------------------- + + +def test_insert_order_records_a_chained_placement_event(conn: sqlite3.Connection) -> None: + repo = Repository(conn) + order_id = repo.insert_order(_order(product_id="BTC-USD", qty=Decimal("0.5"))) + events = audit.read_events(conn) + assert [event.event_type for event in events] == ["order_placed"] + assert events[0].entity_id == str(order_id) + assert events[0].payload["product_id"] == "BTC-USD" + assert audit.chain_state(conn).errors == () + + +def test_update_order_records_what_changed_not_the_whole_row(conn: sqlite3.Connection) -> None: + """An event says what this statement asserted. Re-hashing the whole mutated row would make + each event a snapshot, and a reader could not tell an update from a re-write.""" + repo = Repository(conn) + order_id = repo.insert_order(_order(product_id="BTC-USD")) + repo.update_order(order_id, status="filled", actual_fill=Decimal("42.5")) + + events = audit.read_events(conn) + assert [event.event_type for event in events] == ["order_placed", "order_updated"] + assert events[1].payload == {"status": "filled", "actual_fill": "42.5"} + assert audit.chain_state(conn).errors == () + + +def test_an_update_with_no_fields_records_nothing(conn: sqlite3.Connection) -> None: + """`update_order` already returns early for an empty change set. An event asserting that + nothing changed is chain noise.""" + repo = Repository(conn) + repo.insert_order(_order(product_id="BTC-USD")) + repo.update_order(1) + assert [event.event_type for event in audit.read_events(conn)] == ["order_placed"] + + +def test_upsert_transaction_records_a_flow_event(conn: sqlite3.Connection) -> None: + repo = Repository(conn) + repo.upsert_transaction(_transaction(coinbase_id="cb-1", total=Decimal("250.00"))) + events = audit.read_events(conn) + assert [event.event_type for event in events] == ["transaction_recorded"] + assert events[0].entity_id == "cb-1" + assert audit.chain_state(conn).errors == () + + +def test_re_importing_a_transaction_appends_rather_than_rewrites(conn: sqlite3.Connection) -> None: + """`upsert_transaction` UPDATES the book row in place. The chain must not: two events for one + `coinbase_id` is the record that the line was re-imported with different content, which is + exactly what an auditor wants visible.""" + repo = Repository(conn) + for total in (Decimal("250.00"), Decimal("260.00")): + repo.upsert_transaction( + _transaction(coinbase_id="cb-1", total=total) + ) + events = audit.read_events(conn) + assert len(events) == 2 + assert events[0].row_hash != events[1].row_hash + assert audit.chain_state(conn).errors == () + + +def test_both_attestation_upserts_record_a_human_claim(conn: sqlite3.Connection) -> None: + repo = Repository(conn) + repo.upsert_asset_attestation( + asset="BTC", sector="tech", backing="none", pays_yield=False, + source="prospectus", attested_by="operator", attested_at=100, + ) + repo.upsert_instrument_attestation( + venue="coinbase", product_id="BTC-USD", wrapper="spot", + source="venue docs", attested_by="operator", attested_at=101, + ) + events = audit.read_events(conn) + assert [event.event_type for event in events] == ["asset_attested", "instrument_attested"] + assert events[0].entity_id == "BTC" + assert events[1].entity_id == "coinbase:BTC-USD" + assert audit.chain_state(conn).errors == () + + +def test_the_store_row_and_its_event_land_together_or_not_at_all( + conn: sqlite3.Connection, monkeypatch: pytest.MonkeyPatch +) -> None: + """One transaction, both writes. An order row with no event would read forever after as + "written before the bump" -- an honest-looking gap that is in fact a failed chain write, and + the one lie this store must never tell about itself.""" + + def _boom(*_args: object, **_kwargs: object) -> None: + raise sqlite3.OperationalError("disk I/O error") + + monkeypatch.setattr("keel.data.repository.append_event", _boom) + repo = Repository(conn) + with pytest.raises(sqlite3.OperationalError): + repo.insert_order(_order(product_id="BTC-USD")) + + assert conn.execute("SELECT COUNT(*) FROM orders").fetchone()[0] == 0 + assert conn.execute("SELECT COUNT(*) FROM audit_events").fetchone()[0] == 0 + + +# -- the honest gap --------------------------------------------------------------------------- + + +def test_rows_written_before_the_bump_leave_no_event_and_that_is_not_a_break( + conn: sqlite3.Connection, +) -> None: + """NO BACKFILL. A row inserted by a pre-#721 build has no event, and computing one now would + produce a chain that verifies and proves nothing -- worse than a gap, because it looks like + evidence. The chain over the events that DO exist stays intact.""" + conn.execute( + "INSERT INTO orders (mode, product_id, side, qty, status, created_at) " + "VALUES ('live','BTC-USD','buy','1','filled',50)" + ) + conn.commit() + repo = Repository(conn) + repo.insert_order(_order(product_id="ETH-USD")) + + events = audit.read_events(conn) + assert len(events) == 1 + assert events[0].entity_id == "2" + assert audit.chain_state(conn).errors == () + + +def test_latest_hashes_reports_the_newest_event_per_entity(conn: sqlite3.Connection) -> None: + """What the timeline reads: the most recent CHAINED statement about a row. An order that was + placed and then filled has two events, and the fill is the later word on it.""" + repo = Repository(conn) + order_id = repo.insert_order(_order(product_id="BTC-USD")) + repo.update_order(order_id, status="filled") + + latest = audit.chain_state(conn).hashes + events = audit.read_events(conn) + assert latest[("orders", str(order_id))].row_hash == events[1].row_hash + + +def test_latest_hashes_is_keyed_by_store_so_two_stores_cannot_collide( + conn: sqlite3.Connection, +) -> None: + """`orders.id` is `"1"` and so is a `coinbase_id` of `"1"`. Keyed by entity alone, an + imported transaction would hand its hash to an unrelated order row.""" + repo = Repository(conn) + repo.insert_order(_order(product_id="BTC-USD")) + repo.upsert_transaction(_transaction(coinbase_id="1")) + + latest = audit.chain_state(conn).hashes + assert latest[("orders", "1")].row_hash != latest[("transactions", "1")].row_hash + + +# -- readers that never migrate ---------------------------------------------------------------- +# +# Two readers open a database WITHOUT migrating it: `keel/web/server.py` (a view must not take a +# schema write lock) and `keel/mcp/tools.py::_open_readonly_repo`. Both therefore meet pre-v20 +# databases with no `audit_events` table at all. #718 shipped a reader that raised on exactly this +# shape and took the whole of `gather_findings` down with it; these are that lesson, pinned. + + +def _pre_chain_database() -> sqlite3.Connection: + """A database with the book but not the chain -- what an un-upgraded deployment looks like.""" + connection = db.connect(":memory:") + db.migrate(connection) + connection.execute("DROP TABLE audit_events") + connection.commit() + return connection + + +def test_chain_state_reports_a_missing_table_rather_than_raising() -> None: + connection = _pre_chain_database() + state = audit.chain_state(connection) + + assert state.table_present is False + assert state.event_count == 0 + assert state.errors == () + assert state.intact is False + + +def test_the_timeline_still_renders_on_a_database_without_the_chain() -> None: + """The page an operator opens right after upgrading keel and before running `keel migrate`. + Every row reads `not chained`, which is true, and nothing raises.""" + from keel.commands import timeline + + connection = _pre_chain_database() + connection.execute( + "INSERT INTO orders (mode, product_id, side, qty, status, created_at) " + "VALUES ('live','BTC-USD','buy','1','filled',100)" + ) + connection.commit() + + report = timeline.gather_timeline(Repository(connection), now_ts=200) + assert [row.chain_status for row in report.rows] == ["not chained"] + assert report.chain_recorded is False + assert report.chain_errors == () + + +def test_appending_to_a_missing_table_still_fails_loudly() -> None: + """Tolerance is for READERS. A write that cannot record its event must not quietly succeed: + that is the failed chain write masquerading as an honest gap, and the one lie this store must + never tell about itself.""" + connection = _pre_chain_database() + repo = Repository(connection) + with pytest.raises(sqlite3.OperationalError): + repo.insert_order(_order()) + assert connection.execute("SELECT COUNT(*) FROM orders").fetchone()[0] == 0 + + +# -- a swallowed write must not decide whether an order is durable ------------------------------ + + +def test_a_swallowed_diagnostic_failure_does_not_strand_the_next_order( + tmp_path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The bug the first cut of `write_transaction` shipped, end to end. + + `sqlite3`'s legacy mode leaves the implicit transaction OPEN after a DML that raised, so a + swallowed write leaves the connection dirty. `write_transaction` read that as "an outer + caller owns the commit", took a join branch nothing in this codebase ever legitimately + reaches, and returned from `insert_order` without committing -- while `broker.place_order` + went on to send the order to the venue. + + Asserted against ANOTHER CONNECTION, because the writing connection can see its own + uncommitted rows: "durable" means visible outside this transaction, not merely inserted. + """ + path = tmp_path / "keel.db" + connection = db.connect(path) + db.migrate(connection) + repo = Repository(connection) + + # A failed diagnostic write, swallowed exactly as `execution/equity.py` swallows it. + try: + connection.execute( + "INSERT INTO cycle_balances (ts, mode, currency) VALUES (?, ?, ?)", (1, "live", None) + ) + except sqlite3.IntegrityError: + pass + assert connection.in_transaction, "the premise: a swallowed write leaves the connection dirty" + + order_id = repo.insert_order(_order()) + + reader = db.connect(path) + reader.execute("PRAGMA busy_timeout = 100") + assert reader.execute("SELECT COUNT(*) FROM orders").fetchone()[0] == 1, ( + f"order {order_id} was returned to the executor but is not durable; " + "`broker.place_order` runs next" + ) + assert reader.execute("SELECT COUNT(*) FROM audit_events").fetchone()[0] == 1 + + +def test_the_repository_rolls_a_swallowed_write_back_rather_than_leaving_it_open() -> None: + """The fix at the source. `execution/equity.py` calls this after swallowing, so the dirty + connection never reaches the next writer at all -- `write_transaction`'s unconditional commit + is the second belt, not the only one.""" + connection = db.connect(":memory:") + db.migrate(connection) + repo = Repository(connection) + try: + connection.execute( + "INSERT INTO cycle_balances (ts, mode, currency) VALUES (?, ?, ?)", (1, "live", None) + ) + except sqlite3.IntegrityError: + pass + + repo.rollback() + assert not connection.in_transaction + + +def test_a_blank_coinbase_id_files_its_event_under_its_own_row(conn: sqlite3.Connection) -> None: + """`""` is falsey to Python and perfectly indexable to sqlite, so unlike NULL it CAN take the + `DO UPDATE` branch -- at which point `cursor.lastrowid` is the last real insert's id, not the + updated row's. The first cut filed the re-upsert's event under a different row's identifier: + the timeline then showed one row the hash of a superseded event, marked `chained`, and a + phantom key existed that no row resolved to.""" + repo = Repository(conn) + repo.upsert_transaction(_transaction(coinbase_id="", asset="USD")) + repo.upsert_transaction(_transaction(coinbase_id="X1", asset="EUR")) + repo.upsert_transaction(_transaction(coinbase_id="", asset="GBP")) + + blank_row_id = str( + conn.execute("SELECT id FROM transactions WHERE coinbase_id = ''").fetchone()["id"] + ) + events = audit.read_events(conn) + assert [event.entity_id for event in events] == [blank_row_id, "X1", blank_row_id] + + # And the latest hash for that row is its LATEST event, with no phantom key beside it. + hashes = audit.chain_state(conn).hashes + assert hashes[("transactions", blank_row_id)].row_hash == events[2].row_hash + assert set(hashes) == {("transactions", blank_row_id), ("transactions", "X1")} + + +def test_a_null_coinbase_id_still_files_under_the_row_id(conn: sqlite3.Connection) -> None: + """The case the `lastrowid` version got right, kept right by the lookup that replaced it.""" + repo = Repository(conn) + repo.upsert_transaction(_transaction(coinbase_id=None)) + row_id = str(conn.execute("SELECT id FROM transactions").fetchone()["id"]) + assert [event.entity_id for event in audit.read_events(conn)] == [row_id] + + +def test_an_epoch_timestamp_is_recorded_not_replaced_with_now(conn: sqlite3.Connection) -> None: + """`int(value or time.time())` treats `0` as absent. The event's `ts` is HASHED, so the chain + would then attest to a timestamp the row does not hold.""" + repo = Repository(conn) + repo.insert_order(_order(created_at=0)) + assert [event.ts for event in audit.read_events(conn)] == [0] + + +def test_a_genuinely_absent_timestamp_still_falls_back_to_now( + conn: sqlite3.Connection, monkeypatch: pytest.MonkeyPatch +) -> None: + """The fallback is for `None`, and only for `None`.""" + monkeypatch.setattr("keel.data.repository.time.time", lambda: 4_242.0) + repo = Repository(conn) + repo.insert_order(_order(created_at=None)) + assert [event.ts for event in audit.read_events(conn)] == [4_242] + + +def test_a_payload_commits_as_the_bytes_the_column_holds(conn: sqlite3.Connection) -> None: + """The hash is over the STORED string, not over a re-encoding of what a reader parsed back. + Editing the column by one character must therefore break the row, with no shape of payload + able to survive the round trip differently from how it was hashed.""" + event = _append( + conn, + ts=100, + event_type="order_placed", + entity_id="1", + payload={"qty": Decimal("1.10"), "note": "a", "nested": {"b": [Decimal("2.0")]}}, + ) + stored = conn.execute("SELECT payload_json FROM audit_events").fetchone()["payload_json"] + assert stored == '{"nested":{"b":["2.0"]},"note":"a","qty":"1.10"}' + assert event.row_hash == chain_hash( + { + "ts": 100, + "event_type": "order_placed", + "entity_id": "1", + "payload": stored, + "prev_hash": ZERO_HASH, + } + ) + + conn.execute("UPDATE audit_events SET payload_json = ? WHERE seq_id = 1", (stored + " ",)) + conn.commit() + assert audit.chain_state(conn).errors + + +def test_an_events_entity_id_is_what_the_timeline_prints_as_the_rows_reference( + conn: sqlite3.Connection, +) -> None: + """THE invariant, rather than either side's implementation of it. + + `commands/timeline.py` looks a row's hash up by `(source, reference)`. If the writer files an + event under any other identifier, the lookup misses and the row silently reads `not chained` + -- or, worse, hits a DIFFERENT row's event. Asserted across all three `coinbase_id` shapes, + because that is where the two sides can disagree. + """ + from keel.commands import timeline + + repo = Repository(conn) + repo.upsert_transaction(_transaction(coinbase_id="cb-1", asset="USD")) + repo.upsert_transaction(_transaction(coinbase_id=None, asset="EUR")) + repo.upsert_transaction(_transaction(coinbase_id="", asset="GBP")) + + report = timeline.gather_timeline(repo, now_ts=1_000) + flows = [row for row in report.rows if row.source == "transactions"] + assert len(flows) == 3 + for row in flows: + assert row.chain_status == "chained", f"{row.reference} lost its event" + + printed = {row.reference for row in flows} + recorded = {event.entity_id for event in audit.read_events(conn)} + assert printed == recorded diff --git a/tests/execution/test_equity.py b/tests/execution/test_equity.py index c56fe79..00b65ea 100644 --- a/tests/execution/test_equity.py +++ b/tests/execution/test_equity.py @@ -437,3 +437,53 @@ def _boom(*_a, **_k): assert repo.get_equity_points(mode="live", limit=1), "the equity point still landed" + + +def test_a_swallowed_balance_write_leaves_the_connection_clean(tmp_path) -> None: + """#721. `record_cycle_balance` is `execute` then `commit`, so a failing `execute` leaves + sqlite3's implicit transaction OPEN -- and this function swallows that failure by design, so + a diagnostic write cannot abort a cycle. + + Swallowing without rolling back handed the next writer on this connection a dirty transaction + it did not open. `Repository.insert_order` inherits exactly that connection later in the same + cycle, and the measured consequence was an order that reached the venue with no durable row + behind it. The swallow is still a swallow; it just no longer leaks. + """ + import sqlite3 + + from keel.data.db import connect, migrate + from keel.data.repository import Repository + from keel.execution import equity as equity_mod + + conn = connect(str(tmp_path / "keel.db")) + migrate(conn) + repo = Repository(conn) + repo.set_state("equity_state_mode", "live") + + real = repo.record_cycle_balance + + def _fail(reading): # type: ignore[no-untyped-def] + # Fail the way a real write fails: through sqlite, leaving the transaction open. + conn.execute( + "INSERT INTO cycle_balances (ts, mode, currency) VALUES (?, ?, ?)", (1, "live", None) + ) + raise AssertionError("unreachable -- the INSERT above raises") + + repo.record_cycle_balance = _fail # type: ignore[method-assign] + try: + equity_mod.update_drawdown( + repo, + equity=Decimal("100"), + now_ts=1_000, + cash=Decimal("100"), + balances=[("USD", Decimal("100"), Decimal("100"))], + ) + finally: + repo.record_cycle_balance = real # type: ignore[method-assign] + + assert not conn.in_transaction, "a swallowed write left the connection mid-transaction" + # And the reading the rail needs is still durable, read from another connection. + other = connect(str(tmp_path / "keel.db")) + other.execute("PRAGMA busy_timeout = 100") + assert other.execute("SELECT COUNT(*) FROM equity_points").fetchone()[0] == 1 + assert isinstance(conn, sqlite3.Connection) diff --git a/tests/research/test_ledger.py b/tests/research/test_ledger.py index d04b1a8..9e1cf54 100644 --- a/tests/research/test_ledger.py +++ b/tests/research/test_ledger.py @@ -4,6 +4,7 @@ from dataclasses import replace from decimal import Decimal +from pathlib import Path import pytest @@ -205,3 +206,26 @@ def test_verify_records_of_nothing_is_no_errors_and_not_a_verification(tmp_path) held for the web view. Pinned here so the emptiness stays the caller's problem to name rather than becoming a silent green light.""" assert ledger.verify_records([]) == [] + + +def test_the_tracked_ledger_still_verifies_after_the_canonicaliser_moved() -> None: + """The 93 rows in `docs/experiments/trials-ledger.jsonl` were hashed by the canonicaliser + that used to live in this module. #721 moved it to `keel_core.hashchain` so `audit_events` + could share one definition of canonical JSON rather than grow a second one. + + A move like that is exactly where a chain dies quietly: nothing raises, no test fails, and + the ledger simply stops verifying the next time anyone asks -- by which point there is no way + to tell an honest refactor from an edit. So this reads the REAL, git-tracked file (not a + fixture, not a temp path -- `conftest`'s autouse redirect is deliberately bypassed by using + the literal path) and asserts every one of its rows still reproduces its recorded hash. + + It is the only test in this suite that asserts against hashes computed by a previous version + of the code, and that is the point: it is the byte-identity proof. + """ + tracked = Path(__file__).resolve().parents[2] / "docs/experiments/trials-ledger.jsonl" + assert tracked.exists(), f"the tracked ledger is missing: {tracked}" + records = ledger.read_trials(tracked) + # Guards the assertion below against passing over an empty read: `verify_chain` returns no + # errors for zero rows, which is not the same as "verified" (see `verify_records`). + assert len(records) > 1, "the tracked ledger must hold a chain, not a single row" + assert ledger.verify_chain(tracked) == [] diff --git a/tests/web/test_payload.py b/tests/web/test_payload.py index 12cbd69..8f9cef0 100644 --- a/tests/web/test_payload.py +++ b/tests/web/test_payload.py @@ -31,7 +31,7 @@ from __future__ import annotations import json -from dataclasses import dataclass +from dataclasses import dataclass, replace from decimal import Decimal from pathlib import Path from typing import Any @@ -1543,3 +1543,106 @@ def test_an_empty_series_says_so_rather_than_describing_an_empty_chart() -> None assert empty["segments"] == [] assert empty["point_count"]["value"] == "0" assert "no" in empty["reading"]["display"].lower() + + +# -- the audit chain on the wire (#721) --------------------------------------------------------- + + +def _timeline_report(**overrides: object): + from keel.commands.timeline import TimelineReport, TimelineRow + + row = TimelineRow( + ts=1_000, + kind="trade", + provenance="venue-reported", + source="orders", + reference="1", + summary="filled buy BTC-USD (live)", + product_id="BTC-USD", + amount=None, + amount_kind="", + row_hash="a" * 64, + chain_status="chained", + ) + fields: dict[str, object] = { + "now_ts": 2_000, + "scope": "all", + "scope_start_ts": None, + "kind": "", + "limit": None, + "scoped_count": 1, + "filtered_count": 1, + "rows": (row,), + "kinds_present": ("trade",), + "chain_recorded": True, + "chain_errors": (), + } + fields.update(overrides) + return TimelineReport(**fields) # type: ignore[arg-type] + + +def test_a_chained_row_is_green_and_shortened() -> None: + """The hash is truncated for a cell -- the full value lives in the CSV, which is what anyone + actually verifying one would be working from.""" + body = payload.timeline_payload(_timeline_report()) + cell = body["rows"][0]["row_hash"] + + assert cell["state"] == "good" + assert cell["value"] == "a" * 64 + assert cell["display"].endswith("…") + assert len(cell["display"]) < 64 + + +def test_not_recorded_is_shown_whole_because_it_is_a_sentence() -> None: + """Truncating "NOT RECORDED" to twelve characters and an ellipsis would make it LOOK like a + short hash in a column of long ones -- the absence of evidence dressed as some.""" + from keel.commands.timeline import HASH_NOT_RECORDED + + report = _timeline_report() + row = replace(report.rows[0], row_hash=HASH_NOT_RECORDED, chain_status="not chained") + body = payload.timeline_payload(replace(report, rows=(row,))) + + cell = body["rows"][0]["row_hash"] + assert cell["display"] == HASH_NOT_RECORDED + assert cell["state"] == "unknown" + + +def test_a_row_past_a_break_is_bad_even_though_it_has_a_hash() -> None: + """The state must come from the CHAIN, not from whether a hash is present. A client inferring + the verdict from a 64-character string would call an unverified row evidence.""" + report = _timeline_report(chain_errors=("row 1 (1): content does not match row_hash",)) + row = replace(report.rows[0], chain_status="chain broken") + body = payload.timeline_payload(replace(report, rows=(row,))) + + assert body["rows"][0]["row_hash"]["state"] == "bad" + assert body["rows"][0]["chain_status"]["state"] == "bad" + + +def test_every_chain_status_carries_a_sentence_not_only_the_term() -> None: + """The same rule `_PROVENANCE_NOTES` follows: on an audit surface the term of art is not the + thing a reader can act on.""" + from keel.commands.timeline import CHAIN_STATUSES + + report = _timeline_report() + for status in CHAIN_STATUSES: + row = replace(report.rows[0], chain_status=status) + cell = payload.timeline_payload(replace(report, rows=(row,)))["rows"][0]["chain_status"] + assert cell["value"] == status + assert cell["display"] != status, f"{status} has no sentence beside it" + + +def test_an_empty_chain_is_never_green() -> None: + """No events means nothing was checked -- not that everything verified. The exact trap + `_chain_payload` documents for the research ledger, on the trading side.""" + body = payload.timeline_payload(_timeline_report(chain_recorded=False)) + assert body["chain"]["state"] == "unknown" + assert body["chain"]["value"] == "unverified" + + +def test_a_verified_chain_over_recorded_events_is_the_only_green() -> None: + intact = payload.timeline_payload(_timeline_report()) + assert intact["chain"]["state"] == "good" + + broken = payload.timeline_payload(_timeline_report(chain_errors=("row 3: broken",))) + assert broken["chain"]["state"] == "bad" + assert broken["chain_errors"] == ["row 3: broken"] diff --git a/tests/web/test_timeline_export.py b/tests/web/test_timeline_export.py index 2469f5f..f6c5a83 100644 --- a/tests/web/test_timeline_export.py +++ b/tests/web/test_timeline_export.py @@ -246,3 +246,13 @@ def test_the_export_carries_every_row_in_scope_not_the_pages_worth(tmp_path: Pat assert len(data_rows) == total, ( f"the export must carry the whole scope; got {len(data_rows)} of {total}" ) + + +def test_the_export_names_the_chain_status_beside_the_hash(running: Any) -> None: # noqa: F811 + """#721. The hash column stopped being a placeholder; a hash with no verdict beside it is a + number a reader takes on trust, and the reading taken on trust is the flattering one.""" + _status, _headers, body = _csv(running) + + header = next(csv.reader(io.StringIO(body))) + + assert header[-2:] == ["row_hash", "chain_status"]