From 1e2cae95faa859e83fde52aabcad1c20cf43eb23 Mon Sep 17 00:00:00 2001 From: arpan Date: Mon, 14 Sep 2026 22:59:13 +0530 Subject: [PATCH 1/3] Seven defects the independent review found, each fixed and pinned MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit §4.6's laundering hole: verify_anchors built its anchored-checkpoint set from held | cached, and cached is the local table its own docstring calls a cache and never the record. One INSERT bought supersession, with a hash that was not a hash of anything, and the row was never checked against the provider. Supersession now comes from the provider alone and the pair must match, not merely the seq. On SQLite the prune held no lock through its destructive half: put_anchor and put_checkpoint use 'with connection:', which commits, so BEGIN IMMEDIATE ended at the first of them. Postgres had the guard from the start; SQLite did not, because the defect was found on Postgres and the fix was applied where it was found. SQLite is the default backend. A prune that failed after the checkpoint left the row behind and the reader announced a gap in an intact chain. A hold placed while a prune was in flight was ignored and its receipts deleted. The holds table does not contend with the receipt_chain row lock, so consulting inside the transaction closed nothing; put_hold now takes the same lock. The prune's bound came from the receipt_chain row, which is the row §2.1 assumes an attacker rewrites. One UPDATE turned --through 8 into a delete of every receipt, after which both readers reported clean. The rule-2 simulation filtered to Receipt and re-derived the head, so one unreadable row cost the whole retention feature and a damaged head row refused honest prunes for a break the store already had. The checkpoint could name a (seq, hash) pair that never existed, and anchor it through the provider, which corrupts the external record §4.6 rests on. A refused prune and a successful one left byte-identical receipts, and --older-than was not in the record at all. Signed-off-by: arpan --- src/ctrlrun/anchor.py | 22 ++- src/ctrlrun/cli/main.py | 51 ++++-- src/ctrlrun/postgres.py | 19 +++ src/ctrlrun/retention.py | 76 +++++++-- src/ctrlrun/state.py | 46 ++++- tests/test_retention.py | 353 +++++++++++++++++++++++++++++++++++++++ 6 files changed, 537 insertions(+), 30 deletions(-) diff --git a/src/ctrlrun/anchor.py b/src/ctrlrun/anchor.py index cd6f1f15..554e9cd4 100644 --- a/src/ctrlrun/anchor.py +++ b/src/ctrlrun/anchor.py @@ -464,8 +464,26 @@ def verify_anchors(store: AnchorSource, provider: AnchorProvider) -> AnchorRepor by_seq = _chain_hashes(store) checkpoint = store.checkpoint() checkpoint_seq = None if checkpoint is None else checkpoint[0] - anchored_checkpoints = {anchor.seq for anchor in held if anchor.kind == CHECKPOINT} | { - anchor.seq for anchor in cached.values() if anchor.kind == CHECKPOINT + # §4.6: the set of checkpoints that may supersede an anchor comes from the **provider alone**. + # + # **An earlier version unioned the local table into this, and an independent review bought + # supersession with one `INSERT` into it.** That is the laundering hole §4.6 exists to close, + # reopened by the same confusion §3.3 spends a subsection on: the local table is a cache the + # writer under suspicion can write, so a rule that reads it takes its answer from the side + # that cannot be trusted. The forged row did not even need a real hash -- only `(seq, kind)` + # was read -- and it was never checked against the provider, because the walk iterates what + # the provider holds. + # + # **And the pair must match**, not merely the `seq`. An anchored checkpoint supersedes only + # if the provider vouches for the hash the store's checkpoint row actually names; otherwise + # an attacker anchors any checkpoint at that `seq` and rewrites the row underneath it. + anchored_checkpoints = { + anchor.seq + for anchor in held + if anchor.kind == CHECKPOINT + and checkpoint is not None + and anchor.seq == checkpoint[0] + and anchor.hash == checkpoint[1] } checked = 0 diff --git a/src/ctrlrun/cli/main.py b/src/ctrlrun/cli/main.py index e5e459df..e6db036e 100644 --- a/src/ctrlrun/cli/main.py +++ b/src/ctrlrun/cli/main.py @@ -745,15 +745,25 @@ def prune_command( # # The receipt records an **intent**, so a refused prune leaves one saying DENIED rather than # one asserting an erasure that never happened. - intent = _prune_receipt(through=through, by=by, reason=reason, now=now) + intent = _prune_receipt( + through=through, older_than=older_than, by=by, reason=reason, now=now, stage="proposed" + ) try: store.put_receipt(intent) except CTRLRunError as exc: raise _fail(exc) from exc - try: - result = prune(store, through=through, older_than=window, anchor=provider, now=now) - except CTRLRunError as exc: + def _outcome(refusal: str | None = None) -> None: + """Record what became of the intent above, against the same `action_id`. + + **Every prune leaves exactly two receipts, and they are distinguishable**, which an + independent review found they were not: a refused prune left an `allow`/`committed` + receipt beside the `deny` one, and a successful prune left an identical + `allow`/`committed` receipt, so the record could not tell an erasure that happened from + one that was refused. §4.2 says an operator deleting records should leave one, and a + receipt that over-states what happened is worse than none. + """ + stage = "refused" if refusal is not None else "completed" with suppress(CTRLRunError): store.put_receipt( replace( @@ -762,13 +772,20 @@ def prune_command( seq=None, prev_hash=None, hash=None, - decision=Decision.DENY, - decision_reason="refused", - result=ReceiptResult.DENIED, - error=str(exc), + arguments={**dict(intent.arguments), "stage": stage}, + decision=Decision.DENY if refusal is not None else Decision.ALLOW, + decision_reason="refused" if refusal is not None else intent.decision_reason, + result=ReceiptResult.DENIED if refusal is not None else ReceiptResult.COMMITTED, + error=refusal or "", ) ) + + try: + result = prune(store, through=through, older_than=window, anchor=provider, now=now) + except CTRLRunError as exc: + _outcome(str(exc)) raise _fail(exc) from exc + _outcome() if as_json: click.echo(json.dumps(result.to_dict(), ensure_ascii=False, separators=(",", ":"))) @@ -892,7 +909,9 @@ def _duration(text: str) -> timedelta: return timedelta(**{units[text[-1]]: int(text[:-1])}) -def _prune_receipt(*, through: int, by: str, reason: str, now: datetime) -> Receipt: +def _prune_receipt( + *, through: int, older_than: str, by: str, reason: str, now: datetime, stage: str +) -> Receipt: """The receipt a prune writes before it takes the lock (§4.2, §4.5). **Not routed through `Control.execute`**, and §4.2 is why: the gate is @@ -904,9 +923,18 @@ def _prune_receipt(*, through: int, by: str, reason: str, now: datetime) -> Rece **And the receipt is not what the walk trusts.** A receipt naming itself a checkpoint is a string in a document; the checkpoint row is what `verify_chain` reads. This is for a human. """ + # **`older_than` is in the record**, and an independent review found it was not. It is the + # single input that decides whether the prune destroyed budget ledger rows, and therefore + # whether authority was handed back: a receipt that omits it cannot answer the one question + # somebody reading it afterwards would ask. action = Action( name=PRUNE_ACTION, - arguments={"through": through, "reason": reason}, + arguments={ + "through": through, + "older_than": older_than, + "reason": reason, + "stage": stage, + }, principal=Principal(agent=by), ) return Receipt( @@ -920,7 +948,8 @@ def _prune_receipt(*, through: int, by: str, reason: str, now: datetime) -> Rece environment=action.environment, decision=Decision.ALLOW, decision_reason="an operator's act at the CLI; policy does not mediate shell access", - result=ReceiptResult.COMMITTED, + # The intent is **proposed**, not committed: what happened is on the second receipt. + result=ReceiptResult.COMMITTED if stage != "proposed" else ReceiptResult.BLOCKED, started_at=now, finished_at=now, ) diff --git a/src/ctrlrun/postgres.py b/src/ctrlrun/postgres.py index 7848b09c..097f77aa 100644 --- a/src/ctrlrun/postgres.py +++ b/src/ctrlrun/postgres.py @@ -2283,9 +2283,28 @@ def put_checkpoint(self, checkpoint: Checkpoint) -> None: self._commit(connection) def put_hold(self, hold: Hold) -> None: + """Place a hold. **It takes the prune's lock**, and an independent review is why. + + §4.5 says a hold is consulted *inside* the prune's transaction so that one placed between + the consult and the delete is not missed by both. That closes nothing on this backend: + `holds` does not contend with `SELECT seq FROM receipt_chain ... FOR UPDATE`, and the + prune's snapshot is READ COMMITTED. A review ran it multi-process and the prune deleted + three receipts a hold had been placed over mid-flight:: + + prune: holds consulted, []; now pausing where the operator's hold lands + CHILD placing hold 1..3 + CHILD hold committed; store now holds [('litigation', 1, 3, True)] + prune COMPLETED: receipts deleted 3 + holds in the store now: [('litigation', 1, 3, True)] <- live, over nothing + + Taking the same row lock here is what makes the prune's single consult authoritative: a + hold cannot land while a prune holds it, and a prune cannot start while a hold is landing. + SQLite needs nothing extra, because `BEGIN IMMEDIATE` admits one writer. + """ connection = self._connection() try: with connection.cursor() as cursor: + cursor.execute(f"SELECT seq FROM {self._q}.receipt_chain WHERE id = 1 FOR UPDATE") cursor.execute( f"INSERT INTO {self._q}.holds " "(hold_id, from_seq, to_seq, reason, placed_by, placed_at) " diff --git a/src/ctrlrun/retention.py b/src/ctrlrun/retention.py index 72a1bb01..650a1626 100644 --- a/src/ctrlrun/retention.py +++ b/src/ctrlrun/retention.py @@ -225,14 +225,31 @@ def _pairs(report: ChainReport) -> set[tuple[str, int | None]]: return {(item.name, item.seq) for item in report.breaks} -def _after_prune(receipts: Sequence[Receipt], through: int, checkpoint: Checkpoint) -> ChainReport: +def _after_prune( + receipts: Sequence[Any], through: int, checkpoint: Checkpoint, head: tuple[int, str] | None +) -> ChainReport: """What `verify_chain` would report on this store after the prune, without doing it. The prune is validated against this rather than against its own arithmetic, because rule 2 is a statement about what the **reader** says and the reader is the thing an operator runs. + + **It must be the store and not a tidier version of it**, and an independent review found two + ways it was not. It filtered to `Receipt`, so every row `from_dict` refuses vanished from the + simulation and the prune refused honest prunes naming breaks that would not occur:: + + prune(through=3) REFUSED, claiming: ... would leave the chain reporting missing at seq 6 + what the store ACTUALLY reports after that same prune: [('content_altered', 6), + ('link_broken', 7)] + breaks the prune would really have caused: none + + One tampered row cost the whole retention feature, against rule 3. And it re-derived the head + from the last kept receipt, where `delete_prefix` never touches `receipt_chain`, so a store + whose head row was damaged was refused for a `head_mismatch` it already had. """ - kept = tuple(item for item in receipts if item.seq is None or item.seq > through) - return verify_chain(_PrunedChain(kept, checkpoint)) + kept = tuple( + item for item in receipts if getattr(item, "seq", None) is None or item.seq > through + ) + return verify_chain(_PrunedChain(kept, checkpoint, head)) @dataclass(frozen=True) @@ -250,17 +267,16 @@ class _PrunedChain: -> ok: False breaks: [('missing', 1)] """ - _receipts: tuple[Receipt, ...] + _receipts: tuple[Any, ...] _checkpoint: Checkpoint + #: The store's **own** head row, unchanged: `delete_prefix` never writes `receipt_chain`. + _head: tuple[int, str] | None - def receipts(self) -> tuple[Receipt, ...]: + def receipts(self) -> tuple[Any, ...]: return self._receipts def chain_head(self) -> tuple[int, str] | None: - if not self._receipts: - return (self._checkpoint.seq, self._checkpoint.hash) - last = self._receipts[-1] - return None if last.seq is None or last.hash is None else (last.seq, last.hash) + return self._head def checkpoint(self) -> tuple[int, str] | None: return (self._checkpoint.seq, self._checkpoint.hash) @@ -402,10 +418,29 @@ def _prune_locked( now: datetime, ) -> PruneResult: """Everything a prune does while it holds the receipt-write lock.""" + receipts = tuple(item for item in store.receipts() if isinstance(item, Receipt)) + positions = [item.seq for item in receipts if item.seq is not None] + if not positions: + raise InvalidArgument("this store holds no chained receipt; there is nothing to prune") + head = store.chain_head() if head is None: raise InvalidArgument("this store has no chain head; there is nothing to prune") - head_seq, _ = head + # **The bound is the highest chained receipt, not the head row**, and an independent review + # is why. `chain_head()` reads `receipt_chain`, which is the row `SPEC-v0.11.md` §2.1 already + # assumes an attacker rewrites -- it is the whole reason the anchor exists. Deciding the + # prune's limit from it meant one `UPDATE receipt_chain SET seq = 99` turned + # `prune --through 8` into a delete of every receipt in the store, after which both + # `verify_chain` and `verify_anchors` reported clean: + # + # after UPDATE receipt_chain SET seq=99, prune --through 8 COMPLETED, deleted 8 + # end state: receipts=0 verify_chain ok=True verify_anchors ok=True + # + # The rule-2 delta permitted it because `head_mismatch` at 99 pre-existed, which is rule 2 + # read literally producing total erasure. The receipts are the thing being deleted, so they + # are what bounds the deletion; the head is checked **as well**, below, because a prune that + # leaves the head naming a row it just deleted is §10's refusal too. + head_seq = max(positions) if through >= head_seq: # §10: a prune through the head leaves no chained receipt for the head to name, so the # store would report `head_mismatch` about a chain nothing is wrong with. @@ -430,7 +465,6 @@ def _prune_locked( f"hold {held.hold_id!r} covers receipts this prune would delete: {held.reason}" ) - receipts = tuple(item for item in store.receipts() if isinstance(item, Receipt)) prefix = [item for item in receipts if item.seq is not None and item.seq <= through] if not prefix: raise InvalidArgument(f"no chained receipt at or below seq {through}; nothing to prune") @@ -441,8 +475,24 @@ def _prune_locked( f"the receipt at seq {through} has no stored hash, so a checkpoint over it would " "name a hash nobody can compare against" ) + # **The checkpoint names the pair that exists, not the number the operator typed**, and an + # independent review is why. This took `seq=through` with the hash of whatever readable + # receipt happened to be highest at or below it, so on a chain whose seq 3 had already been + # deleted by somebody else, `prune --through 3` wrote a checkpoint asserting `(3, hash@2)` -- + # a pair that never existed -- and **that fabricated pair is what went to the provider**:: + # + # checkpoint written : seq=3 hash=sha256:874c40cb... + # the real hash at seq 3 was : sha256:09694471... + # the hash at seq 2 is : sha256:874c40cb... + # + # §4.6's whole argument rests on the anchored checkpoint being a claim an operator can check + # against the chain, so a checkpoint that names a hash the chain never had corrupts exactly + # the external record the anchor exists to provide. The deletion still takes the operator's + # `through`; only the claim is narrowed to a row that was really there. + boundary_seq = boundary_receipt.seq + assert boundary_seq is not None # `prefix` filtered on it checkpoint = Checkpoint( - seq=through, + seq=boundary_seq, hash=boundary_receipt.hash, # The version current when it was written, because a store pruned today and read in two # years is the case this milestone is about (§4.2). @@ -456,7 +506,7 @@ def _prune_locked( raise InvalidArgument("this prune was refused: " + "; ".join(refusals)) before = _pairs(verify_chain(store)) - after = _pairs(_after_prune(receipts, through, checkpoint)) + after = _pairs(_after_prune(store.receipts(), through, checkpoint, head)) caused = after - before if caused: # Rule 2, as a **delta**: `unchained` is a pre-existing condition on any store migrated diff --git a/src/ctrlrun/state.py b/src/ctrlrun/state.py index a7b51407..0a5f4c22 100644 --- a/src/ctrlrun/state.py +++ b/src/ctrlrun/state.py @@ -1647,6 +1647,10 @@ def __init__( self._local = threading.local() self._open: weakref.WeakSet[_HeldConnection] = weakref.WeakSet() self._open_lock = threading.Lock() + #: True while `pruning()` holds `BEGIN IMMEDIATE`. Inner writes must not commit through + #: it: `with connection:` commits, and committing there releases the receipt-write lock + #: in the middle of a prune (SPEC-v0.11 §4.5). + self._pruning = False self._path.parent.mkdir(parents=True, exist_ok=True) # SPEC-v0.6 §3. The store's admission check: classify, then migrate or refuse. It runs # before any other table is read, and there is no argument, keyword or environment @@ -1811,6 +1815,21 @@ def chain_head(self) -> tuple[int, str] | None: # --- anchors (SPEC-v0.11 §3.3) ---------------------------------------------------- + @contextmanager + def _writing(self) -> Iterator[Any]: + """The connection, committed on exit **unless a prune holds the transaction** (§4.5). + + `with connection:` commits, which is right for a standalone write and wrong for one + inside `pruning()`: committing there releases the receipt-write lock in the middle of a + prune. Every write that a prune calls goes through here instead. + """ + connection = self._connection() + if self._pruning: + yield connection + return + with connection: + yield connection + def put_anchor(self, anchor: Anchor) -> None: """Cache one anchor the provider made. **A cache, never the record** (§3.3). @@ -1822,8 +1841,7 @@ def put_anchor(self, anchor: Anchor) -> None: §3.2 orders the two kinds separately, and the token is the one value a provider promises to recognise again. """ - connection = self._connection() - with connection: + with self._writing() as connection: connection.execute( "INSERT INTO anchors (token, seq, hash, kind, at) VALUES (?, ?, ?, ?, ?) " "ON CONFLICT(token) DO NOTHING", @@ -1872,8 +1890,7 @@ def put_checkpoint(self, checkpoint: Checkpoint) -> None: racing prunes are each individually valid under §10, and the second overwriting the first's row is what a review measured leaving `[('missing', 4), ('link_broken', 6)]`. """ - connection = self._connection() - with connection: + with self._writing() as connection: connection.execute( "INSERT INTO prune_checkpoint (id, seq, hash, schema, at) VALUES (1, ?, ?, ?, ?) " "ON CONFLICT(id) DO UPDATE SET seq = excluded.seq, hash = excluded.hash, " @@ -1960,11 +1977,32 @@ def pruning(self) -> Iterator[None]: """ connection = self._connection() connection.execute("BEGIN IMMEDIATE") + # **Inner writes must not commit through this, and an independent review found they + # did.** `put_anchor` and `put_checkpoint` use `with connection:`, whose `__exit__` calls + # `commit()`, and a prune calls both -- so the transaction opened above ended at the + # first of them and the whole destructive half ran with no lock at all. Probed from a + # second OS process at each step: + # + # before put_anchor in_transaction=True CHILD blocked + # after put_anchor in_transaction=False CHILD took BEGIN IMMEDIATE + # before delete_prefix in_transaction=False CHILD took BEGIN IMMEDIATE + # + # Worse than the missing exclusion: `pruning()`'s own `commit()` and its `rollback()` + # were then no-ops on a connection with no open transaction, so a prune that failed + # after writing the checkpoint left the row behind and the store reported + # `[('missing', 4), ('link_broken', 1)]` on a chain that was completely intact. + # + # Postgres had this guard in `_commit` from the start (`postgres.py`). SQLite did not, + # because the defect was found on Postgres and the fix was applied where it was found. + # SQLite is the **default** backend. + self._pruning = True try: yield except BaseException: + self._pruning = False connection.rollback() raise + self._pruning = False connection.commit() def delete_prefix(self, through: int, effect_keys: Sequence[str]) -> tuple[int, int]: diff --git a/tests/test_retention.py b/tests/test_retention.py index db959aea..a5904b70 100644 --- a/tests/test_retention.py +++ b/tests/test_retention.py @@ -891,3 +891,356 @@ def test_T549b_a_prune_and_a_receipt_write_exclude_each_other_on_postgres() -> N assert "pruning()" in prune_source, ( "prune() does not hold the store's prune lock, so its refusals are checked outside it" ) + + +# --- T550 to T556: the independent review's findings, each pinned ------------------------------ + + +def test_T550_supersession_comes_from_the_provider_and_not_from_the_local_cache(tmp_path) -> None: + """**§4.6's laundering hole, found by the required independent review.** + + `verify_anchors` built its set of anchored checkpoints from `held | cached`, and `cached` is + the local table its own docstring calls *"a cache, never the record"*. One `INSERT` bought + supersession, and the forged row did not even need a real hash: only `(seq, kind)` was read, + and the row was never checked against the provider because the walk iterates what the + provider holds. + + `T547b` passed over this, because it writes the checkpoint row and **no** anchors row. The + statement it does not run is the obvious second one for anyone who can run the first. + """ + database = tmp_path / "state.db" + store = a_chain(database) + provider = Provider() + rows = store.receipts() + low = rows[2] + assert low.seq == 3 and low.hash is not None + make_anchor(store, provider, at=(low.seq, low.hash)) + boundary = rows[4] + assert boundary.seq == 5 and boundary.hash is not None + store.close() + + # The attack: erase a prefix, write a checkpoint row, and forge the local anchor that would + # make it supersede. The hash is deliberately not a hash of anything. + connection = sqlite3.connect(database) + connection.execute("DELETE FROM receipts WHERE seq <= 5") + connection.execute( + "INSERT INTO prune_checkpoint (id, seq, hash, schema, at) VALUES (1, ?, ?, ?, ?)", + (5, boundary.hash, "ctrlrun.receipt/v7", NOW.isoformat()), + ) + connection.execute( + "INSERT INTO anchors (token, seq, hash, kind, at) VALUES (?, ?, ?, ?, ?)", + ("forged", 5, "sha256:not-a-hash-at-all", CHECKPOINT, NOW.isoformat()), + ) + connection.commit() + connection.close() + + reopened = SQLiteStateStore(database, clock=lambda: T0) + report = verify_anchors(reopened, provider) + cached = [(item.seq, item.kind) for item in reopened.anchors()] + reopened.close() + + assert (5, CHECKPOINT) in cached, "the forged row is not there, so this proves nothing" + assert not report.ok, ( + "a forged row in the local anchors cache bought supersession. The provider never saw " + f"that checkpoint: {report}" + ) + assert report.superseded == 0 + assert ("anchor_broken", 3) in [(item.name, item.seq) for item in report.breaks] + + +def test_T550b_a_checkpoint_the_provider_anchored_at_a_DIFFERENT_hash_does_not_supersede( + tmp_path, +) -> None: + """The pair must match, not merely the `seq`. + + Otherwise an attacker anchors any checkpoint at that `seq` through the provider and then + rewrites the store's checkpoint row underneath it. + """ + database = tmp_path / "state.db" + store = a_chain(database) + provider = Provider() + rows = store.receipts() + make_anchor(store, provider, at=(rows[2].seq, rows[2].hash)) + # The provider anchors a checkpoint at seq 5, with an honest hash. + make_anchor(store, provider, kind=CHECKPOINT, at=(rows[4].seq, rows[4].hash)) + store.close() + + connection = sqlite3.connect(database) + connection.execute("DELETE FROM receipts WHERE seq <= 5") + # ...and the checkpoint ROW names a different hash at the same seq. + connection.execute( + "INSERT INTO prune_checkpoint (id, seq, hash, schema, at) VALUES (1, ?, ?, ?, ?)", + (5, "sha256:" + "ab" * 32, "ctrlrun.receipt/v7", NOW.isoformat()), + ) + connection.commit() + connection.close() + + reopened = SQLiteStateStore(database, clock=lambda: T0) + report = verify_anchors(reopened, provider) + reopened.close() + assert not report.ok and report.superseded == 0, report + + +def test_T551_the_sqlite_prune_holds_its_lock_through_every_write(tmp_path) -> None: + """**§4.5 on the default backend**, which no test asserted and which did not hold. + + `put_anchor` and `put_checkpoint` use `with connection:`, whose `__exit__` commits, and a + prune calls both, so `BEGIN IMMEDIATE` ended at the first of them and the whole destructive + half ran unlocked. Postgres had the guard from the start; SQLite did not, because the defect + was found on Postgres and the fix was applied where it was found. **SQLite is the default.** + """ + database = tmp_path / "state.db" + store = a_chain(database) + provider = Provider() + + seen: list[bool] = [] + real_put_anchor = store.put_anchor + real_put_checkpoint = store.put_checkpoint + real_delete = store.delete_prefix + + def watch(label: str) -> None: + seen.append(store._connection().in_transaction) + + class _Watched: + def put_anchor(self, anchor): + watch("anchor") + real_put_anchor(anchor) + watch("anchor") + + def put_checkpoint(self, checkpoint): + watch("checkpoint") + real_put_checkpoint(checkpoint) + watch("checkpoint") + + def delete_prefix(self, through, keys): + watch("delete") + return real_delete(through, keys) + + def __getattr__(self, name): + return getattr(store, name) + + prune(_Watched(), through=3, older_than=DAY, anchor=provider, now=NOW) + store.close() + + assert seen, "the prune did not reach the writes this is about" + assert all(seen), ( + "the SQLite prune dropped its transaction partway through, so the destructive half ran " + f"with no lock: in_transaction at each write was {seen}" + ) + + +def test_T551b_a_prune_that_fails_after_the_checkpoint_leaves_the_store_as_it_was(tmp_path): + """The consequence of `T551`, and the one that loses evidence. + + A prune killed between the checkpoint and the delete used to leave the checkpoint row behind + and the store reporting `[('missing', 4), ('link_broken', 1)]` on a chain that was completely + intact. §4.5's crash-window argument does not cover it: an orphaned checkpoint row is not an + over-report, it makes the reader announce a gap in a chain with no gap. + """ + database = tmp_path / "state.db" + store = a_chain(database) + before = {(item.name, item.seq) for item in verify_chain(store).breaks} + + class _Dies: + def delete_prefix(self, through, keys): + raise RuntimeError("killed between the checkpoint and the delete") + + def __getattr__(self, name): + return getattr(store, name) + + with pytest.raises(RuntimeError): + prune(_Dies(), through=3, older_than=DAY, anchor=Provider(), now=NOW) + + after = {(item.name, item.seq) for item in verify_chain(store).breaks} + checkpoint = store.checkpoint() + receipts = len(store.receipts()) + store.close() + + assert checkpoint is None, f"a failed prune left a checkpoint row behind: {checkpoint}" + assert receipts == 8, "a failed prune deleted receipts" + assert not (after - before), ( + f"a failed prune left new breaks: {sorted(after - before, key=str)}" + ) + + +def test_T552_the_prune_bound_comes_from_the_receipts_and_not_the_head_row(tmp_path) -> None: + """**§10's head refusal, decided by the row §2.1 assumes is rewritten.** + + One `UPDATE receipt_chain SET seq = 99` turned `prune --through 8` into a delete of every + receipt in the store, after which both readers reported clean. The delta rule permitted it + because `head_mismatch` at 99 pre-existed: rule 2 read literally, producing total erasure. + """ + database = tmp_path / "state.db" + store = a_chain(database) + store.close() + + connection = sqlite3.connect(database) + connection.execute("UPDATE receipt_chain SET seq = 99 WHERE id = 1") + connection.commit() + connection.close() + + reopened = SQLiteStateStore(database, clock=lambda: T0) + with pytest.raises(InvalidArgument) as refused: + prune(reopened, through=8, older_than=DAY, anchor=Provider(), now=NOW) + remaining = len(reopened.receipts()) + reopened.close() + + assert "would take the chain's head" in str(refused.value) + assert remaining == 8, "a rewritten head row let the prune delete the whole chain" + + +def test_T553_one_unreadable_row_does_not_cost_the_whole_retention_feature(tmp_path) -> None: + """**Rule 3 against rule 2.** The simulation filtered to `Receipt`, so every row the reader + refuses vanished from it, and the prune refused honest prunes naming breaks that would not + occur:: + + prune(through=3) REFUSED, claiming: ... would leave the chain reporting missing at seq 6 + what the store ACTUALLY reports after that same prune: + [('content_altered', 6), ('link_broken', 7)] + breaks the prune would really have caused: none + + One tampered row cost the whole feature on that store, which is exactly what item 1 exists + to prevent one surface out. + """ + database = tmp_path / "state.db" + store = a_chain(database) + store.close() + + connection = sqlite3.connect(database) + connection.execute("UPDATE receipts SET json = 'not json at all' WHERE seq = 6") + connection.commit() + connection.close() + + reopened = SQLiteStateStore(database, clock=lambda: T0) + before = {(item.name, item.seq) for item in verify_chain(reopened).breaks} + result = prune(reopened, through=3, older_than=DAY, anchor=Provider(), now=NOW) + after = {(item.name, item.seq) for item in verify_chain(reopened).breaks} + reopened.close() + + assert result.receipts_deleted == 3, "an unreadable row above the prune point refused it" + assert not (after - before), sorted(after - before, key=str) + + +def test_T554_the_checkpoint_names_a_pair_that_existed(tmp_path) -> None: + """**And that pair is what goes to the provider**, which is why it matters. + + This took `seq=through` with the hash of whatever readable receipt was highest at or below + it, so on a chain whose seq 3 had already been deleted, `prune --through 3` wrote and + anchored `(3, hash@2)` -- a pair that never existed. §4.6's argument rests on the anchored + checkpoint being a claim an operator can check against the chain. + """ + database = tmp_path / "state.db" + store = a_chain(database) + rows = {item.seq: item.hash for item in store.receipts()} + store.close() + + connection = sqlite3.connect(database) + connection.execute("DELETE FROM receipts WHERE seq = 3") + connection.commit() + connection.close() + + reopened = SQLiteStateStore(database, clock=lambda: T0) + provider = Provider() + result = prune(reopened, through=3, older_than=DAY, anchor=provider, now=NOW) + reopened.close() + + assert result.checkpoint is not None + assert result.checkpoint.seq == 2, ( + f"the checkpoint names seq {result.checkpoint.seq}, and seq 3 is not in this chain" + ) + assert result.checkpoint.hash == rows[2] + anchored = [(item.seq, item.hash) for item in provider.held.values()] + assert (2, rows[2]) in anchored, ( + f"the provider was asked to vouch for a pair the chain never had: {anchored}" + ) + + +def test_T555_a_prune_leaves_two_receipts_that_say_which_is_which(tmp_path) -> None: + """**§4.2's record, which over-stated what happened.** + + A refused prune left an `allow`/`committed` receipt beside the `deny` one, and a successful + prune left an identical `allow`/`committed` receipt, so the evidence could not tell an + erasure that happened from one that was refused. And `--older-than` was absent entirely: it + is the single input deciding whether the prune destroyed ledger rows, and therefore whether + authority was handed back. + """ + import os + + from click.testing import CliRunner + + from ctrlrun.cli.main import main + + workspace = tmp_path / "work" + workspace.mkdir() + (workspace / "ctrlrun.yaml").write_text(ALLOW, encoding="utf-8") + database = tmp_path / "state.db" + store = a_chain(database) + store.put_hold(Hold("legal", 1, 5, "subpoena", "cli:alice", T0)) + store.close() + + provider_module = workspace / "prov.py" + provider_module.write_text( + "from datetime import UTC, datetime, timedelta\n" + "from ctrlrun.anchor import Anchor\n" + "class P:\n" + " def __init__(self):\n" + " self.held = {}\n" + " self.at = datetime(2026, 1, 1, tzinfo=UTC)\n" + " def make(self, seq, hash, kind):\n" + " self.at += timedelta(minutes=1)\n" + " token = f'tok-{kind}-{seq}'\n" + " self.held[token] = Anchor(\n" + " seq=seq, hash=hash, token=token, kind=kind, at=self.at\n" + " )\n" + " return token, self.at\n" + " def check(self, s, h, t): return t in self.held\n" + " def latest(self): return None\n" + " def since(self, s): return ()\n" + "provider = P()\n", + encoding="utf-8", + ) + + cwd = os.getcwd() + os.chdir(workspace) + try: + import sys + + sys.path.insert(0, str(workspace)) + arguments = [ + "prune", + "--through", + "3", + "--older-than", + "90d", + "--provider", + "prov:provider", + "--by", + "ops@example.com", + "--reason", + "retention", + "--store-url", + f"sqlite:///{database}", + ] + refused = CliRunner().invoke(main, arguments) + finally: + sys.path.remove(str(workspace)) + os.chdir(cwd) + + assert refused.exit_code == 1, refused.output + + reopened = SQLiteStateStore(database, clock=lambda: T0) + prunes = [item for item in reopened.receipts() if item.action == PRUNE_ACTION] + reopened.close() + + assert len(prunes) == 2, f"a refused prune left {len(prunes)} receipts" + stages = [item.arguments.get("stage") for item in prunes] + assert stages == ["proposed", "refused"], stages + assert [str(item.result) for item in prunes] == ["blocked", "denied"], ( + "the intent receipt claims the prune committed, so a refused erasure is indistinguishable " + "from one that happened" + ) + for item in prunes: + assert item.arguments.get("older_than") == "90d", ( + "the receipt omits --older-than, which is the input deciding whether the prune " + f"destroyed ledger rows: {dict(item.arguments)}" + ) From e73d6c4c807418f8e19cffb6f97dcccd28c6fdc7 Mon Sep 17 00:00:00 2001 From: arpan Date: Mon, 14 Sep 2026 23:02:54 +0530 Subject: [PATCH 2/3] Six mutations survived the review fixes; each asked for a behaviour test T546 asserted a constant rather than the branch that reads it, T544's forced refusal stopped reaching the refusal once the simulation became faithful, three Postgres guards had no test at all, and T550's forged row carried a hash that the pair check alone refuses, so the source check never fired. Signed-off-by: arpan --- tests/test_retention.py | 242 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 242 insertions(+) diff --git a/tests/test_retention.py b/tests/test_retention.py index a5904b70..486a4286 100644 --- a/tests/test_retention.py +++ b/tests/test_retention.py @@ -1244,3 +1244,245 @@ def test_T555_a_prune_leaves_two_receipts_that_say_which_is_which(tmp_path) -> N "the receipt omits --older-than, which is the input deciding whether the prune " f"destroyed ledger rows: {dict(item.arguments)}" ) + + +# --- T556 to T559: what the second mutation round found still unasserted ----------------------- + + +def test_T556_a_prune_that_would_cause_a_break_is_refused(tmp_path) -> None: + """Rule 2's refusal, reached the way an operator would reach it. + + **`T544` no longer exercises this path.** It forced the refusal with a store whose checkpoint + write was a no-op, and once `_after_prune` began reading the store's real head and keeping + rows the reader refuses, that construction stopped producing a caused break. A mutation + deleting the refusal entirely then survived the whole file. + + A prune whose checkpoint would be **unusable** is the honest way in: here the row at the + prune point carries no stored hash, so the chain after the prune could not be seeded from it. + """ + database = tmp_path / "state.db" + store = a_chain(database) + store.close() + + connection = sqlite3.connect(database) + connection.execute("UPDATE receipts SET hash = NULL WHERE seq = 3") + connection.commit() + connection.close() + + reopened = SQLiteStateStore(database, clock=lambda: T0) + before = {(item.name, item.seq) for item in verify_chain(reopened).breaks} + with pytest.raises(InvalidArgument) as refused: + prune(reopened, through=3, older_than=DAY, anchor=Provider(), now=NOW) + after = {(item.name, item.seq) for item in verify_chain(reopened).breaks} + remaining = len(reopened.receipts()) + reopened.close() + + assert "no stored hash" in str(refused.value) or "would leave the chain" in str(refused.value) + assert remaining == 8, "a refused prune deleted receipts" + assert after == before + + +def test_T556b_a_prune_whose_simulation_shows_a_new_break_is_refused(tmp_path) -> None: + """The refusal itself, forced by a store that reports differently after the delete. + + A prune through a `seq` whose receipt is present but whose **successor's** link would then be + unseedable is the shape rule 2 exists for. This drives the comparison directly rather than + through a store double, so the branch is exercised rather than described. + """ + from ctrlrun.retention import Checkpoint as _Checkpoint + from ctrlrun.retention import _after_prune + + database = tmp_path / "state.db" + store = a_chain(database) + rows = store.receipts() + head = store.chain_head() + before = {(item.name, item.seq) for item in verify_chain(store).breaks} + + # A checkpoint naming a hash that is not the one at that seq: the walk cannot seed from it, + # and every row after it reports. This is what "the prune caused it" looks like. + wrong = _Checkpoint(seq=3, hash="sha256:" + "cd" * 32, schema="ctrlrun.receipt/v7", at=NOW) + simulated = _after_prune(rows, 3, wrong, head) + after = {(item.name, item.seq) for item in simulated.breaks} + store.close() + + assert after - before, ( + "a checkpoint naming the wrong hash produced no new break in the simulation, so rule 2's " + "comparison has nothing to refuse and the guard cannot fire" + ) + + +def test_T557_a_ledger_row_whose_effect_is_still_held_refuses_the_prune(tmp_path) -> None: + """**`T546` asserts a constant, not a behaviour**, and a mutation deleting the held-state + branch survived it: `set(HELD_EFFECT_STATES) == {...}` stays true however the branch that + reads it is written. This drives `_ledger_refusals` over each held state.""" + from ctrlrun.effect import EffectState + from ctrlrun.retention import HELD_EFFECT_STATES, _ledger_refusals + + class _Row: + effect_key = "refund:p0" + consumed_at = NOW - timedelta(days=400) + released_at = None + + class _Store: + def __init__(self, state): + self._state = state + + def consumptions(self): + return (_Row(),) + + def get_effect(self, key): + return type("E", (), {"state": self._state})() + + for state in HELD_EFFECT_STATES: + refusals = _ledger_refusals(_Store(state), ["refund:p0"], now=NOW, older_than=DAY) + assert refusals, f"a {state} effect's ledger row was prunable" + assert "still holds its charge" in refusals[0], refusals + + # And the two terminal ones, well outside the window, are not refused. + for state in (EffectState.COMMITTED, EffectState.FAILED): + assert _ledger_refusals(_Store(state), ["refund:p0"], now=NOW, older_than=DAY) == [] + + +@postgres +def test_T558_the_postgres_prune_lock_is_a_real_transaction(tmp_path) -> None: + """**Three Postgres guards no test asserted**, each found by a mutation surviving. + + `pruning()` drops its `BEGIN`; `_commit` stops suppressing inside a prune; `put_hold` stops + taking the lock. All three were demonstrated defects, and all three are invisible to an + outcome test because the connection is `autocommit=True` and the damage is a window rather + than a result. + """ + import ast + + import ctrlrun.postgres as module + + tree = ast.parse(Path(module.__file__).read_text(encoding="utf-8")) + named = { + node.name: ast.unparse(node) for node in ast.walk(tree) if isinstance(node, ast.FunctionDef) + } + + assert "BEGIN" in named["pruning"], ( + "pruning() does not open a transaction. The connection is autocommit=True with every " + "write taking an explicit BEGIN, so a bare SELECT ... FOR UPDATE commits the instant it " + "returns and holds no lock at all (SPEC-v0.11 §4.5)" + ) + assert "FOR UPDATE" in named["pruning"] + assert "self._pruning" in named["_commit"], ( + "_commit no longer suppresses inside a prune, so put_anchor or put_checkpoint ends the " + "transaction pruning() opened and releases the lock mid-prune" + ) + assert "FOR UPDATE" in named["put_hold"], ( + "put_hold does not take the prune lock. `holds` does not contend with the receipt_chain " + "row lock, so a hold placed while a prune is in flight is missed by both (§4.5)" + ) + + +@postgres +def test_T558b_a_hold_cannot_land_while_a_prune_holds_the_lock() -> None: + """The behaviour `T558` asserts statically, run against a real server. + + Measured: the child blocked 1.85s and placed its hold the moment the prune released. + """ + import subprocess + import sys + import time + + from ctrlrun.postgres import PostgresStateStore + + schema = f"holdlock_{uuid.uuid4().hex[:10]}" + PostgresStateStore.create_schema(POSTGRES_URL, schema) + child_source = ( + "import time\n" + "from datetime import UTC, datetime\n" + "from ctrlrun.postgres import PostgresStateStore\n" + "from ctrlrun.retention import Hold\n" + f"s = PostgresStateStore({POSTGRES_URL!r}, schema={schema!r})\n" + "start = time.time()\n" + "s.put_hold(Hold('late', 1, 3, 'litigation', 'cli:bob', datetime.now(UTC)))\n" + "print('%.3f' % (time.time() - start))\n" + "s.close()\n" + ) + try: + store = PostgresStateStore(POSTGRES_URL, schema=schema, clock=lambda: T0) + control = Control(Policy.from_yaml(ALLOW), store, clock=lambda: T0) + for index in range(6): + control.execute( + an_action(f"p{index}"), lambda: {"ok": True}, f"refund:p{index}", lease=LEASE + ) + store.close() + + parent = PostgresStateStore(POSTGRES_URL, schema=schema, clock=lambda: T0) + with parent.pruning(): + child = subprocess.Popen( + [sys.executable, "-c", child_source], + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + time.sleep(1.5) + still_waiting = child.poll() is None + output, _ = child.communicate(timeout=60) + parent.close() + + assert still_waiting, ( + "a hold landed while a prune held the lock, so it can be placed over receipts the " + f"prune is about to delete: {output}" + ) + # The child's own measurement starts after its interpreter and imports, so it is + # necessarily less than the parent's sleep. `still_waiting` above is the real proof; + # this is the coarse floor that separates blocking from an unblocked call, which takes + # single-digit milliseconds. + waited = float(output.strip().splitlines()[-1]) + assert waited >= 0.25, f"the child did not block on the lock: waited {waited}s" + finally: + PostgresStateStore.drop_schema(POSTGRES_URL, schema) + + +def test_T559_a_forged_local_anchor_matching_the_checkpoint_still_does_not_supersede( + tmp_path, +) -> None: + """`T550`'s forged row carries a hash that is not a hash of anything, so the pair check alone + refuses it and the **source** check is never the thing that fires. + + This forges a local row whose hash matches the checkpoint row exactly, so only *the provider + never saw it* can refuse it. Without that, a mutation restoring `held | cached` survives. + """ + database = tmp_path / "state.db" + store = a_chain(database) + provider = Provider() + rows = store.receipts() + make_anchor(store, provider, at=(rows[2].seq, rows[2].hash)) + boundary = rows[4] + assert boundary.seq == 5 and boundary.hash is not None + store.close() + + connection = sqlite3.connect(database) + connection.execute("DELETE FROM receipts WHERE seq <= 5") + connection.execute( + "INSERT INTO prune_checkpoint (id, seq, hash, schema, at) VALUES (1, ?, ?, ?, ?)", + (5, boundary.hash, "ctrlrun.receipt/v7", NOW.isoformat()), + ) + # The forged local row agrees with the checkpoint in every field. Only the provider's own + # record can tell that no such anchor was ever made. + connection.execute( + "INSERT INTO anchors (token, seq, hash, kind, at) VALUES (?, ?, ?, ?, ?)", + ("forged", 5, boundary.hash, CHECKPOINT, NOW.isoformat()), + ) + connection.commit() + connection.close() + + reopened = SQLiteStateStore(database, clock=lambda: T0) + report = verify_anchors(reopened, provider) + reopened.close() + + assert ("checkpoint", 5) in [ + (item.kind, item.seq) for item in [*provider.since(0)] + ] or True # the provider holds no such checkpoint; stated for the reader + assert not any(item.kind == CHECKPOINT and item.seq == 5 for item in provider.since(0)), ( + "the provider holds the checkpoint, so this is not testing the forgery" + ) + assert not report.ok, ( + "a forged local anchor agreeing with the checkpoint row bought supersession. Only the " + f"provider's record can refuse it, and it was not consulted: {report}" + ) + assert report.superseded == 0 From 735f98b5dfee317ffe09e91502bb138e5ae87961 Mon Sep 17 00:00:00 2001 From: arpan Date: Mon, 14 Sep 2026 23:12:50 +0530 Subject: [PATCH 3/3] The one guard no store state can reach, reached R1 was the last surviving mutation in item 3's table: delete rule 2's comparison entirely, `if caused:` to `if False:`, and all 34 tests in `test_retention.py` still passed. The reason is worth stating rather than papering over. Once `T554` made the checkpoint name the pair that really exists, and `T553` made the simulation the store rather than a tidier version of it, a prefix prune can no longer introduce a break. Every construction that would is caught before rule 2 runs: by the head bound, by the forward-only checkpoint, or by the missing-hash refusal. `T556` looked like it covered this and did not, because its assertion is an `or` that the earlier hash refusal satisfies. So the comparison is a backstop, and `T556c` keeps it honest by feeding it a simulation reporting one more break than the store does, then asserting the refusal names the `seq` and that nothing was deleted and no checkpoint left behind. A guard nothing can trigger is still a guard somebody will edit. Mutation table for item 3: 22 of 22 caught. Signed-off-by: arpan --- tests/test_retention.py | 42 ++++++++++++++++++++++++++++++++++++++++- 1 file changed, 41 insertions(+), 1 deletion(-) diff --git a/tests/test_retention.py b/tests/test_retention.py index 486a4286..690f11e5 100644 --- a/tests/test_retention.py +++ b/tests/test_retention.py @@ -24,6 +24,7 @@ import os import sqlite3 import uuid +from dataclasses import replace from datetime import UTC, datetime, timedelta from pathlib import Path @@ -33,7 +34,7 @@ from ctrlrun.action import Action, Principal from ctrlrun.anchor import CHECKPOINT, Anchor, make_anchor, verify_anchors from ctrlrun.errors import InvalidArgument -from ctrlrun.receipt import verify_chain +from ctrlrun.receipt import ChainBreak, verify_chain from ctrlrun.retention import PRUNE_ACTION, Hold, prune T0 = datetime(2026, 1, 1, 12, 0, tzinfo=UTC) @@ -1311,6 +1312,45 @@ def test_T556b_a_prune_whose_simulation_shows_a_new_break_is_refused(tmp_path) - ) +def test_T556c_the_rule_2_comparison_refuses_when_the_simulation_shows_a_new_break( + tmp_path, monkeypatch +) -> None: + """Rule 2's guard, driven directly, because **no store state reaches it any more.** + + That is worth stating rather than hiding: once the checkpoint names the pair that really + exists (`T554`) and the simulation is the store rather than a tidier version of it (`T553`), + a prefix prune cannot introduce a break. Every way of constructing one is caught earlier -- + by the head bound, the forward-only checkpoint, or the missing-hash refusal -- so a mutation + deleting this comparison survived the whole file. + + It stays, as a backstop, and this is what keeps it honest: the comparison is fed a simulation + that reports one more break than the store does, and the prune must refuse with the `seq` + named. A guard nothing can trigger is still a guard somebody will edit. + """ + import ctrlrun.retention as module + + database = tmp_path / "state.db" + store = a_chain(database) + real = module._after_prune + + def _one_more(receipts, through, checkpoint, head): + report = real(receipts, through, checkpoint, head) + return replace(report, ok=False, breaks=[*report.breaks, ChainBreak("missing", 99, "x")]) + + monkeypatch.setattr(module, "_after_prune", _one_more) + + with pytest.raises(InvalidArgument) as refused: + prune(store, through=3, older_than=DAY, anchor=Provider(), now=NOW) + remaining = len(store.receipts()) + checkpoint = store.checkpoint() + store.close() + + assert "missing at seq 99" in str(refused.value), str(refused.value) + assert "no flag that admits it" in str(refused.value) + assert remaining == 8, "a prune refused by rule 2 deleted receipts anyway" + assert checkpoint is None, "a prune refused by rule 2 left a checkpoint behind" + + def test_T557_a_ledger_row_whose_effect_is_still_held_refuses_the_prune(tmp_path) -> None: """**`T546` asserts a constant, not a behaviour**, and a mutation deleting the held-state branch survived it: `set(HELD_EFFECT_STATES) == {...}` stays true however the branch that