From d5dc4a255b6c8f717337f1c14c139ebbe4ad95c2 Mon Sep 17 00:00:00 2001 From: bitmaster162 Date: Sun, 20 Sep 2026 22:40:22 +0700 Subject: [PATCH] R15E: add offline monotonic boundary reconciliation --- continuityos/gate/monotonic_anchor.py | 332 +++++++++++++++++- ...est_r15_tpm2_monotonic_execution_anchor.py | 229 ++++++++++++ 2 files changed, 559 insertions(+), 2 deletions(-) diff --git a/continuityos/gate/monotonic_anchor.py b/continuityos/gate/monotonic_anchor.py index edcfbae..a2c90f5 100644 --- a/continuityos/gate/monotonic_anchor.py +++ b/continuityos/gate/monotonic_anchor.py @@ -17,6 +17,7 @@ DOMAIN = "continuityos.governance.execution-anchor.v1" COMMITMENT_SCHEMA = "continuityos.governance.execution-anchor.commitment.v1" RECEIPT_SCHEMA = "continuityos.governance.execution-anchor.receipt.v1" +RECONCILIATION_PLAN_SCHEMA = "continuityos.governance.execution-anchor.reconciliation-plan.v1" EVENT_KIND = "monotonic_anchor" PHASE_GENESIS = "GENESIS" PHASE_STARTED = "EXECUTION_STARTED" @@ -247,7 +248,8 @@ def _event_payload(row) -> dict[str, Any]: return value def validate_global( - self, ledger, *, _allow_pending_boundary: tuple[str, str] | None = None + self, ledger, *, _allow_pending_boundary: tuple[str, str] | None = None, + _allow_hardware_frontier_mismatch: bool = False, ) -> dict[str, Any]: verification = ledger.verify() if not verification.get("ok"): @@ -352,7 +354,10 @@ def validate_global( raise MonotonicAnchorError( "monotonic anchor hardware identity substitution detected" ) - if provider["observed_digest"] != latest["observed_anchor_digest"]: + if ( + provider["observed_digest"] != latest["observed_anchor_digest"] + and not _allow_hardware_frontier_mismatch + ): raise MonotonicAnchorError( "local governance state differs from monotonic hardware frontier" ) @@ -668,6 +673,329 @@ def _record_execution_terminal_locked( ) return self._extend_and_persist(ledger, receipt) + @staticmethod + def _reconciliation_plan_hash(plan: dict[str, Any]) -> str: + body = dict(plan) + body.pop("plan_hash", None) + return hashlib.sha256(_canonical(body)).hexdigest() + + @classmethod + def _finalize_reconciliation_plan(cls, plan: dict[str, Any]) -> dict[str, Any]: + result = dict(plan) + result["plan_hash"] = cls._reconciliation_plan_hash(result) + return result + + def plan_offline_reconciliation( + self, ledger, *, preflight_hash: str + ) -> dict[str, Any]: + """Read-only deterministic recovery plan under the external witness lock.""" + with self._require_r14_witness(ledger).locked(): + return self._plan_offline_reconciliation_locked( + ledger, preflight_hash=preflight_hash + ) + + def _plan_offline_reconciliation_locked( + self, ledger, *, preflight_hash: str + ) -> dict[str, Any]: + if not _is_nonzero_sha256(preflight_hash): + raise MonotonicAnchorError("reconciliation preflight hash is invalid") + row = ledger._attempt_row(preflight_hash) + if row is None: + raise MonotonicAnchorError("reconciliation execution attempt is missing") + try: + attempt = ledger._validate_attempt_row(row) + except Exception as exc: + raise MonotonicAnchorError( + "reconciliation execution attempt is invalid" + ) from exc + + phase = attempt["phase"] + if phase not in {"ATTEMPT_STARTED", "TERMINAL"}: + raise MonotonicAnchorError( + "execution attempt has no monotonic boundary to reconcile" + ) + + if phase == "ATTEMPT_STARTED": + boundary_phase = PHASE_STARTED + boundary_hash = attempt.get("execution_started_hash") + boundary_kind = "execution_started" + terminal_kind = None + else: + if not attempt.get("execution_started_hash"): + state = self.validate_global( + ledger, _allow_hardware_frontier_mismatch=True + ) + hardware = self._read_provider() + action = ( + "ALREADY_RECONCILED" + if hardware["observed_digest"] == state["observed_anchor_digest"] + else "HOLD_DIVERGED" + ) + return self._finalize_reconciliation_plan({ + "schema": RECONCILIATION_PLAN_SCHEMA, + "domain": DOMAIN, + "state_id": state["state_id"], + "preflight_hash": preflight_hash, + "binding_sha256": attempt["binding_sha256"], + "attempt_phase": phase, + "boundary_phase": None, + "boundary_kind": None, + "boundary_hash": None, + "terminal_kind": attempt.get("terminal_kind"), + "action": action, + "receipt": None, + "local_anchor_generation": state["anchor_generation"], + "local_anchor_digest": state["observed_anchor_digest"], + "hardware_observed_digest": hardware["observed_digest"], + "note": "terminal attempt crossed no subprocess boundary", + }) + boundary_phase = PHASE_TERMINAL + boundary_hash = attempt.get("terminal_hash") + boundary_kind = attempt.get("terminal_kind") + terminal_kind = attempt.get("terminal_kind") + + if not _is_nonzero_sha256(boundary_hash): + raise MonotonicAnchorError("reconciliation boundary hash is invalid") + + # First validate all durable local evidence while allowing the one exact + # boundary to be pending and allowing hardware to be one deterministic + # digest ahead. This remains read-only. + state = self.validate_global( + ledger, + _allow_pending_boundary=(boundary_phase, boundary_hash), + _allow_hardware_frontier_mismatch=True, + ) + hardware = self._read_provider() + anchored = state["request_receipts"].get(preflight_hash) or {} + existing = ( + anchored.get("started") + if boundary_phase == PHASE_STARTED + else anchored.get("terminal") + ) + + if existing is not None: + if hardware["observed_digest"] != state["observed_anchor_digest"]: + action = "HOLD_DIVERGED" + note = "hardware differs from the latest durable monotonic receipt" + elif ( + boundary_phase == PHASE_STARTED + and attempt["phase"] == "ATTEMPT_STARTED" + and anchored.get("terminal") is None + ): + action = "OUTCOME_EVIDENCE_REQUIRED" + note = ( + "started receipt is durable but no verified terminal outcome exists; " + "do not infer or fabricate subprocess outcome" + ) + else: + action = "ALREADY_RECONCILED" + note = "exact monotonic boundary receipt is already durable" + return self._finalize_reconciliation_plan({ + "schema": RECONCILIATION_PLAN_SCHEMA, + "domain": DOMAIN, + "state_id": state["state_id"], + "preflight_hash": preflight_hash, + "binding_sha256": attempt["binding_sha256"], + "attempt_phase": attempt["phase"], + "boundary_phase": boundary_phase, + "boundary_kind": boundary_kind, + "boundary_hash": boundary_hash, + "terminal_kind": terminal_kind, + "action": action, + "receipt": existing, + "local_anchor_generation": state["anchor_generation"], + "local_anchor_digest": state["observed_anchor_digest"], + "hardware_observed_digest": hardware["observed_digest"], + "note": note, + }) + + frontier = self._frontier(ledger) + if frontier["event_hash"] != boundary_hash: + raise MonotonicAnchorError( + "unreconciled monotonic boundary is not the current ledger frontier" + ) + event = ledger.event(boundary_hash) + if event is None or event.get("kind") != boundary_kind: + raise MonotonicAnchorError( + "reconciliation boundary does not identify the expected ledger event" + ) + event_payload = event.get("payload") + if not isinstance(event_payload, dict): + raise MonotonicAnchorError("reconciliation boundary payload is invalid") + if event_payload.get("preflight_hash") != preflight_hash: + raise MonotonicAnchorError("reconciliation boundary preflight mismatch") + if boundary_phase == PHASE_TERMINAL: + started = anchored.get("started") + if started is None: + raise MonotonicAnchorError( + "terminal reconciliation requires a durable started receipt" + ) + if started["binding_sha256"] != attempt["binding_sha256"]: + raise MonotonicAnchorError( + "terminal reconciliation started binding mismatch" + ) + + receipt = self._build_receipt( + state_id=state["state_id"], + generation=state["anchor_generation"] + 1, + phase=boundary_phase, + preflight_hash=preflight_hash, + binding_sha256=attempt["binding_sha256"], + terminal_kind=terminal_kind, + frontier=frontier, + previous_digest=state["observed_anchor_digest"], + hardware=state, + ) + previous_digest = receipt["previous_anchor_digest"] + expected_digest = receipt["observed_anchor_digest"] + observed = hardware["observed_digest"] + if observed == previous_digest: + action = "NEEDS_EXACTLY_ONE_EXTEND" + note = "hardware is at the previous durable digest" + elif observed == expected_digest: + action = "APPEND_EXACT_RECEIPT_ONLY" + note = ( + "hardware already reached the deterministic next digest; " + "a second extend is forbidden" + ) + else: + action = "HOLD_DIVERGED" + note = ( + "hardware is neither at the previous durable digest nor the " + "deterministic expected next digest" + ) + + return self._finalize_reconciliation_plan({ + "schema": RECONCILIATION_PLAN_SCHEMA, + "domain": DOMAIN, + "state_id": state["state_id"], + "preflight_hash": preflight_hash, + "binding_sha256": attempt["binding_sha256"], + "attempt_phase": attempt["phase"], + "boundary_phase": boundary_phase, + "boundary_kind": boundary_kind, + "boundary_hash": boundary_hash, + "terminal_kind": terminal_kind, + "action": action, + "receipt": receipt, + "local_anchor_generation": state["anchor_generation"], + "local_anchor_digest": state["observed_anchor_digest"], + "hardware_observed_digest": observed, + "note": note, + }) + + def apply_offline_reconciliation( + self, ledger, *, preflight_hash: str, plan_hash: str + ) -> dict[str, Any]: + """Apply one exact pre-planned recovery action; never retries automatically.""" + if not _is_nonzero_sha256(plan_hash): + raise MonotonicAnchorError("reconciliation plan hash is invalid") + with self._require_r14_witness(ledger).locked(): + plan = self._plan_offline_reconciliation_locked( + ledger, preflight_hash=preflight_hash + ) + if plan["plan_hash"] != plan_hash: + raise MonotonicAnchorError( + "reconciliation plan hash differs from current authority state" + ) + action = plan["action"] + if action == "ALREADY_RECONCILED": + return { + "status": action, + "plan_hash": plan_hash, + "receipt_hash": None, + "anchor_generation": plan["local_anchor_generation"], + "observed_anchor_digest": plan["local_anchor_digest"], + } + if action in {"HOLD_DIVERGED", "OUTCOME_EVIDENCE_REQUIRED"}: + raise MonotonicAnchorError( + f"reconciliation cannot apply while plan action is {action}" + ) + if action not in { + "NEEDS_EXACTLY_ONE_EXTEND", "APPEND_EXACT_RECEIPT_ONLY" + }: + raise MonotonicAnchorError("unsupported reconciliation action") + + receipt = plan.get("receipt") + receipt = self._require_receipt(receipt) + self._validate_receipt_crypto(receipt) + frontier = self._frontier(ledger) + if frontier["event_hash"] != receipt["ledger_event_hash"]: + raise MonotonicAnchorError( + "reconciliation boundary is no longer the current ledger frontier" + ) + before = self._read_provider() + if ( + before["nv_public_sha256"] != receipt["nv_public_sha256"] + or before["nv_name_sha256"] != receipt["nv_name_sha256"] + ): + raise MonotonicAnchorError( + "reconciliation hardware identity changed before apply" + ) + + if action == "NEEDS_EXACTLY_ONE_EXTEND": + if before["observed_digest"] != receipt["previous_anchor_digest"]: + raise MonotonicAnchorError( + "reconciliation hardware moved before the planned extend" + ) + try: + returned = _require_snapshot(self.provider.extend( + expected_previous_digest=receipt["previous_anchor_digest"], + commitment_sha256=receipt["commitment_sha256"], + )) + except MonotonicAnchorError: + raise + except Exception as exc: + raise MonotonicAnchorError( + f"reconciliation extend failed: {type(exc).__name__}: {exc}" + ) from exc + if ( + returned["nv_public_sha256"] != receipt["nv_public_sha256"] + or returned["nv_name_sha256"] != receipt["nv_name_sha256"] + or returned["observed_digest"] != receipt["observed_anchor_digest"] + ): + raise MonotonicAnchorError( + "reconciliation extend readback mismatch" + ) + fresh = self._read_provider() + if fresh != returned: + raise MonotonicAnchorError( + "reconciliation hardware changed after extend" + ) + else: + if before["observed_digest"] != receipt["observed_anchor_digest"]: + raise MonotonicAnchorError( + "append-only reconciliation hardware digest changed" + ) + + # No retry loop: if this durable append fails after hardware advance, + # the operator must generate a fresh plan. That fresh plan will + # deterministically classify APPEND_EXACT_RECEIPT_ONLY. + try: + receipt_hash = ledger.append(EVENT_KIND, receipt) + except Exception as exc: + raise MonotonicAnchorError( + "reconciliation receipt was not durably recorded" + ) from exc + + after = self._read_provider() + if ( + after["nv_public_sha256"] != receipt["nv_public_sha256"] + or after["nv_name_sha256"] != receipt["nv_name_sha256"] + or after["observed_digest"] != receipt["observed_anchor_digest"] + ): + raise MonotonicAnchorError( + "reconciliation hardware frontier changed after receipt" + ) + validated = self.validate_global(ledger) + return { + "status": "RECONCILED", + "plan_hash": plan_hash, + "receipt_hash": receipt_hash, + "anchor_generation": validated["anchor_generation"], + "observed_anchor_digest": validated["observed_anchor_digest"], + } + def require_terminal_receipt( self, ledger, *, preflight_hash: str, binding_sha256: str, attempt: dict[str, Any] ) -> dict[str, Any]: diff --git a/tests/test_r15_tpm2_monotonic_execution_anchor.py b/tests/test_r15_tpm2_monotonic_execution_anchor.py index a2c1539..6bc2ff1 100644 --- a/tests/test_r15_tpm2_monotonic_execution_anchor.py +++ b/tests/test_r15_tpm2_monotonic_execution_anchor.py @@ -1,6 +1,7 @@ """R15 hardware-free TPM2 monotonic execution-anchor acceptance tests.""" from __future__ import annotations +from pathlib import Path import hashlib import json import os @@ -806,3 +807,231 @@ def test_terminal_anchor_replay_is_rejected_without_second_extend(r15): ) assert provider.extend_calls == before assert (tmp_path / "effect.txt").read_text() == "x" + + +def test_r15e_started_fail_before_plans_one_extend_then_requires_outcome_evidence(r15): + broker = r15["broker"] + provider = r15["provider"] + tmp_path = r15["tmp_path"] + pre = _preflight(broker, tmp_path, "r15e-start-before") + provider.fail_before.add(2) + first = broker.execute_preflight("r15e-start-before") + assert first["state"] == "HELD", first + assert provider.extend_calls == 1 + assert not (tmp_path / "effect.txt").exists() + provider.fail_before.clear() + + with broker._ledger() as ledger: + plan = r15["anchor"].plan_offline_reconciliation( + ledger, preflight_hash=pre["preflight_hash"] + ) + assert plan["action"] == "NEEDS_EXACTLY_ONE_EXTEND" + result = r15["anchor"].apply_offline_reconciliation( + ledger, preflight_hash=pre["preflight_hash"], + plan_hash=plan["plan_hash"], + ) + assert result["status"] == "RECONCILED" + follow = r15["anchor"].plan_offline_reconciliation( + ledger, preflight_hash=pre["preflight_hash"] + ) + assert provider.extend_calls == 2 + assert follow["action"] == "OUTCOME_EVIDENCE_REQUIRED" + assert not (tmp_path / "effect.txt").exists() + + +def test_r15e_started_fail_after_appends_exact_receipt_without_second_extend(r15): + broker = r15["broker"] + provider = r15["provider"] + tmp_path = r15["tmp_path"] + pre = _preflight(broker, tmp_path, "r15e-start-after") + provider.fail_after.add(2) + first = broker.execute_preflight("r15e-start-after") + assert first["state"] == "HELD", first + assert provider.extend_calls == 2 + assert not (tmp_path / "effect.txt").exists() + + with broker._ledger() as ledger: + plan = r15["anchor"].plan_offline_reconciliation( + ledger, preflight_hash=pre["preflight_hash"] + ) + assert plan["action"] == "APPEND_EXACT_RECEIPT_ONLY" + result = r15["anchor"].apply_offline_reconciliation( + ledger, preflight_hash=pre["preflight_hash"], + plan_hash=plan["plan_hash"], + ) + follow = r15["anchor"].plan_offline_reconciliation( + ledger, preflight_hash=pre["preflight_hash"] + ) + assert result["status"] == "RECONCILED" + assert provider.extend_calls == 2 + assert follow["action"] == "OUTCOME_EVIDENCE_REQUIRED" + + +def test_r15e_terminal_fail_before_plans_exactly_one_extend(r15): + broker = r15["broker"] + provider = r15["provider"] + tmp_path = r15["tmp_path"] + pre = _preflight(broker, tmp_path, "r15e-terminal-before") + provider.fail_before.add(3) + first = broker.execute_preflight("r15e-terminal-before") + assert first["state"] == "HELD", first + assert provider.extend_calls == 2 + assert (tmp_path / "effect.txt").read_text() == "x" + provider.fail_before.clear() + + with broker._ledger() as ledger: + plan = r15["anchor"].plan_offline_reconciliation( + ledger, preflight_hash=pre["preflight_hash"] + ) + assert plan["action"] == "NEEDS_EXACTLY_ONE_EXTEND" + result = r15["anchor"].apply_offline_reconciliation( + ledger, preflight_hash=pre["preflight_hash"], + plan_hash=plan["plan_hash"], + ) + follow = r15["anchor"].plan_offline_reconciliation( + ledger, preflight_hash=pre["preflight_hash"] + ) + assert result["status"] == "RECONCILED" + assert provider.extend_calls == 3 + assert follow["action"] == "ALREADY_RECONCILED" + + +def test_r15e_terminal_fail_after_appends_only_without_second_extend(r15): + broker = r15["broker"] + provider = r15["provider"] + tmp_path = r15["tmp_path"] + pre = _preflight(broker, tmp_path, "r15e-terminal-after") + provider.fail_after.add(3) + first = broker.execute_preflight("r15e-terminal-after") + assert first["state"] == "HELD", first + assert provider.extend_calls == 3 + assert (tmp_path / "effect.txt").read_text() == "x" + + with broker._ledger() as ledger: + plan = r15["anchor"].plan_offline_reconciliation( + ledger, preflight_hash=pre["preflight_hash"] + ) + assert plan["action"] == "APPEND_EXACT_RECEIPT_ONLY" + result = r15["anchor"].apply_offline_reconciliation( + ledger, preflight_hash=pre["preflight_hash"], + plan_hash=plan["plan_hash"], + ) + assert result["status"] == "RECONCILED" + assert provider.extend_calls == 3 + + +def test_r15e_diverged_hardware_holds_without_extend_or_receipt(r15): + broker = r15["broker"] + provider = r15["provider"] + tmp_path = r15["tmp_path"] + pre = _preflight(broker, tmp_path, "r15e-diverged") + provider.fail_before.add(2) + first = broker.execute_preflight("r15e-diverged") + assert first["state"] == "HELD", first + provider.digest = _expected_digest(provider.digest, "f" * 64) + before_calls = provider.extend_calls + + with broker._ledger() as ledger: + plan = r15["anchor"].plan_offline_reconciliation( + ledger, preflight_hash=pre["preflight_hash"] + ) + assert plan["action"] == "HOLD_DIVERGED" + with pytest.raises(MonotonicAnchorError, match="HOLD_DIVERGED"): + r15["anchor"].apply_offline_reconciliation( + ledger, preflight_hash=pre["preflight_hash"], + plan_hash=plan["plan_hash"], + ) + assert ledger.con.execute( + "SELECT COUNT(*) FROM events WHERE kind=?", (EVENT_KIND,) + ).fetchone()[0] == 1 + assert provider.extend_calls == before_calls + + +def test_r15e_plan_hash_is_exact_and_duplicate_apply_never_extends_twice(r15): + broker = r15["broker"] + provider = r15["provider"] + tmp_path = r15["tmp_path"] + pre = _preflight(broker, tmp_path, "r15e-plan-hash") + provider.fail_before.add(3) + assert broker.execute_preflight("r15e-plan-hash")["state"] == "HELD" + provider.fail_before.clear() + + with broker._ledger() as ledger: + plan = r15["anchor"].plan_offline_reconciliation( + ledger, preflight_hash=pre["preflight_hash"] + ) + before_calls = provider.extend_calls + with pytest.raises(MonotonicAnchorError, match="plan hash differs"): + r15["anchor"].apply_offline_reconciliation( + ledger, preflight_hash=pre["preflight_hash"], + plan_hash="0" * 63 + "1", + ) + assert provider.extend_calls == before_calls + result = r15["anchor"].apply_offline_reconciliation( + ledger, preflight_hash=pre["preflight_hash"], + plan_hash=plan["plan_hash"], + ) + assert result["status"] == "RECONCILED" + calls_after = provider.extend_calls + with pytest.raises(MonotonicAnchorError, match="plan hash differs"): + r15["anchor"].apply_offline_reconciliation( + ledger, preflight_hash=pre["preflight_hash"], + plan_hash=plan["plan_hash"], + ) + assert provider.extend_calls == calls_after + + +def test_r15e_receipt_failure_sidecar_is_advisory_not_outcome_authority(r15): + broker = r15["broker"] + provider = r15["provider"] + tmp_path = r15["tmp_path"] + pre = _preflight(broker, tmp_path, "r15e-sidecar") + provider.fail_before.add(2) + assert broker.execute_preflight("r15e-sidecar")["state"] == "HELD" + provider.fail_before.clear() + + with broker._ledger() as ledger: + plan = r15["anchor"].plan_offline_reconciliation( + ledger, preflight_hash=pre["preflight_hash"] + ) + r15["anchor"].apply_offline_reconciliation( + ledger, preflight_hash=pre["preflight_hash"], + plan_hash=plan["plan_hash"], + ) + + sidecar = Path(broker.ledger_path + ".receipt_failures.jsonl") + sidecar.write_text(json.dumps({ + "status": "EXECUTED_BUT_RECEIPT_FAILED", + "preflight_hash": pre["preflight_hash"], + "process_exit_code": 0, + "instruction": "advisory only", + }) + "\\n", encoding="utf-8") + + with broker._ledger() as ledger: + follow = r15["anchor"].plan_offline_reconciliation( + ledger, preflight_hash=pre["preflight_hash"] + ) + assert follow["action"] == "OUTCOME_EVIDENCE_REQUIRED" + assert provider.extend_calls == 2 + + +def test_r15e_stale_plan_fails_when_pending_boundary_is_no_longer_frontier(r15): + broker = r15["broker"] + provider = r15["provider"] + tmp_path = r15["tmp_path"] + pre = _preflight(broker, tmp_path, "r15e-stale-frontier") + provider.fail_before.add(3) + assert broker.execute_preflight("r15e-stale-frontier")["state"] == "HELD" + + with broker._ledger() as ledger: + plan = r15["anchor"].plan_offline_reconciliation( + ledger, preflight_hash=pre["preflight_hash"] + ) + before_calls = provider.extend_calls + ledger.append("audit_note", {"note": "frontier moved after recovery plan"}) + with pytest.raises(MonotonicAnchorError): + r15["anchor"].apply_offline_reconciliation( + ledger, preflight_hash=pre["preflight_hash"], + plan_hash=plan["plan_hash"], + ) + assert provider.extend_calls == before_calls