v0.11 item 3, after review: seven defects #203 merged with

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

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,
     )

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) "

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

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]: