From a22cb1e759b9237ee0901c1fd6608a1567e82665 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 21 Oct 2026 21:57:52 +0200 Subject: [PATCH] safety: validate M3 approval records --- docs/CODEX_HANDOFF.md | 3 + .../research/m3-hardware-research-protocol.md | 15 ++ docs/status/phase-0.md | 3 + scripts/prepare_m3_review_packet.py | 7 + scripts/validate_m3_approval.py | 170 ++++++++++++++++++ tests/test_m3_approval.py | 83 +++++++++ tests/test_m3_review_packet.py | 88 ++++----- 7 files changed, 326 insertions(+), 43 deletions(-) create mode 100644 scripts/validate_m3_approval.py create mode 100644 tests/test_m3_approval.py diff --git a/docs/CODEX_HANDOFF.md b/docs/CODEX_HANDOFF.md index e5b91ae..812898e 100644 --- a/docs/CODEX_HANDOFF.md +++ b/docs/CODEX_HANDOFF.md @@ -37,6 +37,9 @@ The repository now provides: - a deterministic M3 review-packet generator that freezes only a `ready_for_review` plan and creates a hash-bound, explicitly unreviewed record without granting hardware access. +- a separate M3 approval-record validator that rejects changed packet files, + reviewer/window drift, late review, incomplete safety statements and any + claim of automatic hardware authorization. Run the ordinary checkpoint with the bundled Codex Python or another supported Python interpreter and a writable `TEMP`/`TMP`: diff --git a/docs/research/m3-hardware-research-protocol.md b/docs/research/m3-hardware-research-protocol.md index fe93162..d27c804 100644 --- a/docs/research/m3-hardware-research-protocol.md +++ b/docs/research/m3-hardware-research-protocol.md @@ -128,6 +128,21 @@ separate approval record rather than edit the packet template. The generated output still states `hardware_access_authorized: false`; generating a packet is not approval. +After an independent reviewer copies and completes the approval template, +validate the separate record against the unchanged packet: + +```powershell +python scripts/validate_m3_approval.py ` + .build/m3-review-packet ` + path/to/approval-record.json +``` + +The validator checks both packet file hashes and sizes, the frozen plan, Git +commit, reviewer identity, exact session window, approval time and every safety +statement. A successful result is `approval_record_valid`, not automatic +permission to access hardware: the result continues to state +`hardware_access_authorized: false` and `live_preflight_required: true`. + ## Session sequence The eventual reviewed session follows this order and stops at the first failed diff --git a/docs/status/phase-0.md b/docs/status/phase-0.md index 87f39f8..c30a658 100644 --- a/docs/status/phase-0.md +++ b/docs/status/phase-0.md @@ -100,6 +100,9 @@ evidence for: `ready_for_review` plan, freezes its exact bytes and session window, records SHA-256 bindings, refuses overwrite and emits an untouched, non-authorizing human approval-record template. +- a fail-closed M3 approval validator that verifies the full packet hash chain, + commit, reviewer, bounded session and explicit safety statements while still + requiring live preflight and returning no automatic hardware authorization. The remaining M1 gates are deliberately explicit: diff --git a/scripts/prepare_m3_review_packet.py b/scripts/prepare_m3_review_packet.py index 38e7ef7..3ea3767 100644 --- a/scripts/prepare_m3_review_packet.py +++ b/scripts/prepare_m3_review_packet.py @@ -68,6 +68,13 @@ def create_packet( "decision": "not_reviewed", "reviewer": None, "reviewed_at": None, + "statements": { + "exact_plan_bytes_reviewed": False, + "operation_evidence_reviewed": False, + "physical_controls_reviewed": False, + "driver_and_rollback_reviewed": False, + "abort_and_recovery_reviewed": False, + }, "notes": "", "hardware_access_authorized": False, "separate_human_approval_required": True, diff --git a/scripts/validate_m3_approval.py b/scripts/validate_m3_approval.py new file mode 100644 index 0000000..300ddc3 --- /dev/null +++ b/scripts/validate_m3_approval.py @@ -0,0 +1,170 @@ +"""Validate an M3 review packet and separate human approval record.""" + +from __future__ import annotations + +import argparse +from datetime import datetime +import hashlib +import json +import re +from pathlib import Path + +if __package__: + from scripts.validate_m3_hardware_protocol import validate as validate_plan +else: + from validate_m3_hardware_protocol import validate as validate_plan + + +SHA256 = re.compile(r"^[0-9a-f]{64}$") +COMMIT = re.compile(r"^[0-9a-f]{40}$") +PACKET_FILES = {"plan.json", "approval-record-template.json"} +STATEMENTS = { + "exact_plan_bytes_reviewed", + "operation_evidence_reviewed", + "physical_controls_reviewed", + "driver_and_rollback_reviewed", + "abort_and_recovery_reviewed", +} + + +def _sha256(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def _timestamp(value: object) -> datetime | None: + if not isinstance(value, str): + return None + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + return parsed if parsed.tzinfo is not None else None + + +def validate(packet: Path, approval_path: Path) -> dict: + packet = packet.resolve() + errors = [] + manifest_path = packet / "packet-manifest.json" + try: + manifest = json.loads(manifest_path.read_text(encoding="utf-8-sig")) + except (OSError, json.JSONDecodeError): + manifest = {} + errors.append("packet manifest is missing or malformed") + + commit = manifest.get("repository_commit") + if ( + manifest.get("schema_version") != 1 + or manifest.get("packet") != "M3 hardware research plan review" + or manifest.get("contains_private_evidence") is not False + or manifest.get("hardware_access_authorized") is not False + or not isinstance(commit, str) + or COMMIT.fullmatch(commit) is None + ): + errors.append("packet manifest contract is invalid") + + records = manifest.get("files", []) + if not isinstance(records, list): + records = [] + record_names = { + record.get("path") for record in records if isinstance(record, dict) + } + if record_names != PACKET_FILES or len(records) != len(PACKET_FILES): + errors.append("packet file set is invalid") + for record in records: + if not isinstance(record, dict) or record.get("path") not in PACKET_FILES: + continue + path = packet / record["path"] + digest = record.get("sha256") + size = record.get("bytes") + if ( + not path.is_file() + or not isinstance(digest, str) + or SHA256.fullmatch(digest) is None + or _sha256(path) != digest + or type(size) is not int + or path.stat().st_size != size + ): + errors.append(f"packet file binding failed: {record['path']}") + + plan_path = packet / "plan.json" + template_path = packet / "approval-record-template.json" + plan = {} + if plan_path.is_file(): + plan_result = validate_plan(plan_path) + if plan_result["status"] != "ready_for_review": + errors.append("frozen plan is not ready for review") + try: + plan = json.loads(plan_path.read_text(encoding="utf-8-sig")) + except json.JSONDecodeError: + errors.append("frozen plan is malformed") + else: + errors.append("frozen plan is missing") + + try: + template = json.loads(template_path.read_text(encoding="utf-8-sig")) + approval = json.loads(approval_path.resolve().read_text(encoding="utf-8-sig")) + except (OSError, json.JSONDecodeError): + template = {} + approval = {} + errors.append("approval template or record is missing or malformed") + + if set(approval) != set(template): + errors.append("approval record fields do not match the packet template") + expected_binding = { + "plan_path": "plan.json", + "plan_sha256": _sha256(plan_path) if plan_path.is_file() else None, + "repository_commit": commit, + } + if approval.get("binding") != expected_binding: + errors.append("approval record is not bound to this plan and commit") + + expected_window = { + "starts_at": plan.get("session", {}).get("planned_at"), + "ends_at": plan.get("session", {}).get("ends_at"), + } + if approval.get("session_window") != expected_window: + errors.append("approval session window does not match the frozen plan") + if approval.get("decision") != "approved": + errors.append("approval decision is not approved") + reviewer = approval.get("reviewer") + if reviewer != plan.get("roles", {}).get("protocol_reviewer"): + errors.append("approval reviewer does not match the plan reviewer") + reviewed_at = _timestamp(approval.get("reviewed_at")) + starts_at = _timestamp(expected_window["starts_at"]) + if reviewed_at is None: + errors.append("reviewed_at is not an ISO-8601 timestamp with timezone") + elif starts_at is not None and reviewed_at >= starts_at: + errors.append("approval must be recorded before the session starts") + + statements = approval.get("statements", {}) + if not isinstance(statements, dict) or set(statements) != STATEMENTS: + errors.append("approval safety statements do not match the contract") + elif any(value is not True for value in statements.values()): + errors.append("every approval safety statement must be confirmed") + if approval.get("hardware_access_authorized") is not False: + errors.append("approval record must not claim automatic hardware authorization") + if approval.get("separate_human_approval_required") is not True: + errors.append("separate human approval requirement cannot be removed") + + valid = not errors + return { + "status": "approval_record_valid" if valid else "invalid", + "errors": errors, + "administrative_approval_valid": valid, + "hardware_access_authorized": False, + "live_preflight_required": True, + } + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("packet", type=Path) + parser.add_argument("approval", type=Path) + args = parser.parse_args() + result = validate(args.packet, args.approval) + print(json.dumps(result, indent=2, sort_keys=True)) + return 0 if result["status"] == "approval_record_valid" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_m3_approval.py b/tests/test_m3_approval.py new file mode 100644 index 0000000..1191ca6 --- /dev/null +++ b/tests/test_m3_approval.py @@ -0,0 +1,83 @@ +import json +import tempfile +import unittest +from pathlib import Path + +from scripts.prepare_m3_review_packet import ROOT, create_packet +from scripts.validate_m3_approval import validate +from tests.test_m3_review_packet import write_ready_plan + + +class M3ApprovalTests(unittest.TestCase): + def _packet_and_approval(self, root: Path) -> tuple[Path, Path]: + packet = root / "packet" + create_packet(write_ready_plan(root), packet, "a" * 40) + template = packet / "approval-record-template.json" + approval = json.loads(template.read_text(encoding="utf-8")) + approval["decision"] = "approved" + approval["reviewer"] = "Synthetic Reviewer" + approval["reviewed_at"] = "2026-10-21T11:30:00+02:00" + approval["statements"] = { + key: True for key in approval["statements"] + } + approval_path = root / "approval-record.json" + approval_path.write_text(json.dumps(approval), encoding="utf-8") + return packet, approval_path + + def test_exact_approval_record_is_valid_but_not_hardware_authorization(self): + with tempfile.TemporaryDirectory(dir=ROOT / ".build") as directory: + packet, approval = self._packet_and_approval(Path(directory)) + result = validate(packet, approval) + self.assertEqual("approval_record_valid", result["status"]) + self.assertTrue(result["administrative_approval_valid"]) + self.assertFalse(result["hardware_access_authorized"]) + self.assertTrue(result["live_preflight_required"]) + + def test_changed_plan_or_manifest_binding_is_rejected(self): + with tempfile.TemporaryDirectory(dir=ROOT / ".build") as directory: + packet, approval = self._packet_and_approval(Path(directory)) + (packet / "plan.json").write_text("{}", encoding="utf-8") + result = validate(packet, approval) + self.assertEqual("invalid", result["status"]) + self.assertTrue(any("binding failed" in item for item in result["errors"])) + self.assertFalse(result["hardware_access_authorized"]) + + def test_reviewer_and_session_window_must_match_plan(self): + with tempfile.TemporaryDirectory(dir=ROOT / ".build") as directory: + root = Path(directory) + packet, approval_path = self._packet_and_approval(root) + approval = json.loads(approval_path.read_text()) + approval["reviewer"] = "Someone Else" + approval["session_window"]["ends_at"] = "2026-10-21T13:00:00+02:00" + approval_path.write_text(json.dumps(approval), encoding="utf-8") + result = validate(packet, approval_path) + self.assertEqual("invalid", result["status"]) + self.assertGreaterEqual(len(result["errors"]), 2) + + def test_late_review_or_unconfirmed_statement_is_rejected(self): + with tempfile.TemporaryDirectory(dir=ROOT / ".build") as directory: + root = Path(directory) + packet, approval_path = self._packet_and_approval(root) + approval = json.loads(approval_path.read_text()) + approval["reviewed_at"] = "2026-10-21T12:00:00+02:00" + approval["statements"]["physical_controls_reviewed"] = False + approval_path.write_text(json.dumps(approval), encoding="utf-8") + result = validate(packet, approval_path) + self.assertEqual("invalid", result["status"]) + self.assertTrue(any("before the session" in item for item in result["errors"])) + self.assertTrue(any("every approval" in item for item in result["errors"])) + + def test_automatic_hardware_authorization_claim_is_rejected(self): + with tempfile.TemporaryDirectory(dir=ROOT / ".build") as directory: + root = Path(directory) + packet, approval_path = self._packet_and_approval(root) + approval = json.loads(approval_path.read_text()) + approval["hardware_access_authorized"] = True + approval_path.write_text(json.dumps(approval), encoding="utf-8") + result = validate(packet, approval_path) + self.assertEqual("invalid", result["status"]) + self.assertFalse(result["hardware_access_authorized"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_m3_review_packet.py b/tests/test_m3_review_packet.py index 837bd0a..602d610 100644 --- a/tests/test_m3_review_packet.py +++ b/tests/test_m3_review_packet.py @@ -11,51 +11,53 @@ from scripts.prepare_m3_review_packet import ROOT, create_packet TEMPLATE = ROOT / "tests" / "fixtures" / "m3-hardware-protocol-template.json" +def write_ready_plan(root: Path) -> Path: + plan = copy.deepcopy(json.loads(TEMPLATE.read_text(encoding="utf-8"))) + plan["roles"] = { + "operator": "Synthetic Operator", + "safety_observer": "Synthetic Observer", + "protocol_reviewer": "Synthetic Reviewer", + } + for key in plan["physical_controls"]: + plan["physical_controls"][key] = True + plan["physical_controls"]["emission_inhibition_method"] = "Synthetic method" + plan["software_controls"]["operation_allowlist_enforced"] = True + plan["software_controls"]["reversible_driver_plan_reviewed"] = True + plan["allowed_operations"] = [{ + "operation_id": "synthetic-read-identity", + "purpose": "Synthetic packet test only", + "evidence_artifact_id": "sha256-0123456789abcdef0123", + "evidence_level": "independently-implemented", + "safety_class": "read-only-no-motion-no-emission", + "direction": "host-to-controller-request-response", + "request_hex": "0102", + "expected_response": "Synthetic fixed response contract", + "timeout_ms": 1000, + "maximum_attempts": 1, + "state_change_expected": False, + }] + plan["recovery"].update({ + "power_down_procedure_reviewed": True, + "driver_rollback_procedure_reviewed": True, + "incident_record_destination_prepared": True, + }) + plan["session"] = { + "machine_location": "Synthetic location", + "planned_at": "2026-10-21T12:00:00+02:00", + "ends_at": "2026-10-21T12:30:00+02:00", + "notes": "Synthetic unit test only", + } + path = root / "source-plan.json" + path.write_text(json.dumps(plan), encoding="utf-8") + return path + + class M3ReviewPacketTests(unittest.TestCase): - def _ready_plan(self, root: Path) -> Path: - plan = copy.deepcopy(json.loads(TEMPLATE.read_text(encoding="utf-8"))) - plan["roles"] = { - "operator": "Synthetic Operator", - "safety_observer": "Synthetic Observer", - "protocol_reviewer": "Synthetic Reviewer", - } - for key in plan["physical_controls"]: - plan["physical_controls"][key] = True - plan["physical_controls"]["emission_inhibition_method"] = "Synthetic method" - plan["software_controls"]["operation_allowlist_enforced"] = True - plan["software_controls"]["reversible_driver_plan_reviewed"] = True - plan["allowed_operations"] = [{ - "operation_id": "synthetic-read-identity", - "purpose": "Synthetic packet test only", - "evidence_artifact_id": "sha256-0123456789abcdef0123", - "evidence_level": "independently-implemented", - "safety_class": "read-only-no-motion-no-emission", - "direction": "host-to-controller-request-response", - "request_hex": "0102", - "expected_response": "Synthetic fixed response contract", - "timeout_ms": 1000, - "maximum_attempts": 1, - "state_change_expected": False, - }] - plan["recovery"].update({ - "power_down_procedure_reviewed": True, - "driver_rollback_procedure_reviewed": True, - "incident_record_destination_prepared": True, - }) - plan["session"] = { - "machine_location": "Synthetic location", - "planned_at": "2026-10-21T12:00:00+02:00", - "ends_at": "2026-10-21T12:30:00+02:00", - "notes": "Synthetic unit test only", - } - path = root / "source-plan.json" - path.write_text(json.dumps(plan), encoding="utf-8") - return path def test_packet_freezes_exact_plan_and_stays_non_authorizing(self): with tempfile.TemporaryDirectory(dir=ROOT / ".build") as directory: root = Path(directory) - source = self._ready_plan(root) + source = write_ready_plan(root) source_bytes = source.read_bytes() output = root / "packet" manifest = create_packet(source, output, repository_commit="a" * 40) @@ -76,7 +78,7 @@ class M3ReviewPacketTests(unittest.TestCase): def test_changed_plan_changes_binding(self): with tempfile.TemporaryDirectory(dir=ROOT / ".build") as directory: root = Path(directory) - source = self._ready_plan(root) + source = write_ready_plan(root) create_packet(source, root / "one", "c" * 40) one = json.loads((root / "one" / "approval-record-template.json").read_text()) plan = json.loads(source.read_text()) @@ -89,7 +91,7 @@ class M3ReviewPacketTests(unittest.TestCase): def test_nonempty_output_is_not_overwritten(self): with tempfile.TemporaryDirectory(dir=ROOT / ".build") as directory: root = Path(directory) - source = self._ready_plan(root) + source = write_ready_plan(root) output = root / "packet" output.mkdir() sentinel = output / "keep.txt" @@ -102,7 +104,7 @@ class M3ReviewPacketTests(unittest.TestCase): with tempfile.TemporaryDirectory(dir=ROOT / ".build") as directory: root = Path(directory) output = root / "packet" - manifest = create_packet(self._ready_plan(root), output, "e" * 40) + manifest = create_packet(write_ready_plan(root), output, "e" * 40) for record in manifest["files"]: path = output / record["path"] self.assertEqual(path.stat().st_size, record["bytes"])