From c382ea9a7801b3472603a4572a4a78278e73eed9 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 21 Oct 2026 21:55:28 +0200 Subject: [PATCH] safety: add immutable M3 review packets --- docs/CODEX_HANDOFF.md | 3 + .../research/m3-hardware-research-protocol.md | 13 ++ docs/status/phase-0.md | 4 + scripts/prepare_m3_review_packet.py | 130 ++++++++++++++++++ scripts/validate_m3_hardware_protocol.py | 16 ++- .../m3-hardware-protocol-template.json | 1 + tests/test_m3_hardware_protocol.py | 10 ++ tests/test_m3_review_packet.py | 113 +++++++++++++++ 8 files changed, 285 insertions(+), 5 deletions(-) create mode 100644 scripts/prepare_m3_review_packet.py create mode 100644 tests/test_m3_review_packet.py diff --git a/docs/CODEX_HANDOFF.md b/docs/CODEX_HANDOFF.md index 8f43a03..e5b91ae 100644 --- a/docs/CODEX_HANDOFF.md +++ b/docs/CODEX_HANDOFF.md @@ -34,6 +34,9 @@ The repository now provides: - a fail-closed M3 hardware-research protocol, incomplete public plan template and validator; validation can reach only `ready_for_review` and never authorizes hardware access. +- 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. 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 6f1398b..fe93162 100644 --- a/docs/research/m3-hardware-research-protocol.md +++ b/docs/research/m3-hardware-research-protocol.md @@ -115,6 +115,19 @@ plan bytes and SHA-256. A separate approval record must identify that hash, reviewer, approval time with timezone, location and exact session window. Any plan edit invalidates that approval. +Create the frozen, non-authorizing review packet with: + +```powershell +python scripts/prepare_m3_review_packet.py path/to/session-plan.json +``` + +The generator refuses an incomplete or invalid plan, refuses to overwrite a +non-empty output directory, copies the exact plan bytes and hashes both the +plan and untouched `approval-record-template.json`. A reviewer must create a +separate approval record rather than edit the packet template. The generated +output still states `hardware_access_authorized: false`; generating a packet is +not approval. + ## 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 be25d74..87f39f8 100644 --- a/docs/status/phase-0.md +++ b/docs/status/phase-0.md @@ -96,6 +96,10 @@ evidence for: plan template and standard-library validator that requires exact evidence-backed request bytes, distinct safety roles, physical controls and recovery planning while always denying hardware authorization. +- a deterministic M3 review-packet generator that accepts only a + `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. The remaining M1 gates are deliberately explicit: diff --git a/scripts/prepare_m3_review_packet.py b/scripts/prepare_m3_review_packet.py new file mode 100644 index 0000000..38e7ef7 --- /dev/null +++ b/scripts/prepare_m3_review_packet.py @@ -0,0 +1,130 @@ +"""Freeze a review-ready M3 plan into a non-authorizing review packet.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import shutil +import subprocess +from pathlib import Path + +if __package__: + from scripts.validate_m3_hardware_protocol import validate +else: + from validate_m3_hardware_protocol import validate + + +ROOT = Path(__file__).resolve().parents[1] + + +def _sha256(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def _repository_commit() -> str: + result = subprocess.run( + ["git", "rev-parse", "HEAD"], + cwd=ROOT, + check=True, + capture_output=True, + text=True, + ) + return result.stdout.strip() + + +def create_packet( + plan_path: Path, + output: Path, + repository_commit: str | None = None, +) -> dict: + source = plan_path.resolve() + validation = validate(source) + if validation["status"] != "ready_for_review": + raise ValueError("M3 plan is not ready for review") + + output = output.resolve() + if output.exists() and any(output.iterdir()): + raise ValueError(f"review-packet directory is not empty: {output}") + output.mkdir(parents=True, exist_ok=True) + + frozen_plan = output / "plan.json" + shutil.copyfile(source, frozen_plan) + plan = json.loads(frozen_plan.read_text(encoding="utf-8-sig")) + commit = repository_commit or _repository_commit() + + review = { + "schema_version": 1, + "review": "M3 hardware research session plan", + "binding": { + "plan_path": frozen_plan.name, + "plan_sha256": _sha256(frozen_plan), + "repository_commit": commit, + }, + "session_window": { + "starts_at": plan["session"]["planned_at"], + "ends_at": plan["session"]["ends_at"], + }, + "decision": "not_reviewed", + "reviewer": None, + "reviewed_at": None, + "notes": "", + "hardware_access_authorized": False, + "separate_human_approval_required": True, + } + review_path = output / "approval-record-template.json" + review_path.write_text( + json.dumps(review, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + newline="\n", + ) + + manifest = { + "schema_version": 1, + "packet": "M3 hardware research plan review", + "repository_commit": commit, + "contains_private_evidence": False, + "hardware_access_authorized": False, + "files": [ + { + "path": frozen_plan.name, + "bytes": frozen_plan.stat().st_size, + "sha256": _sha256(frozen_plan), + }, + { + "path": review_path.name, + "bytes": review_path.stat().st_size, + "sha256": _sha256(review_path), + }, + ], + } + manifest_path = output / "packet-manifest.json" + manifest_path.write_text( + json.dumps(manifest, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + newline="\n", + ) + return manifest + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("plan", type=Path) + parser.add_argument( + "--output", + type=Path, + default=ROOT / ".build" / "m3-review-packet", + ) + args = parser.parse_args() + try: + result = create_packet(args.plan, args.output) + except ValueError as error: + parser.error(str(error)) + print(f"packet={args.output.resolve()}") + print(f"plan_sha256={result['files'][0]['sha256']}") + print("hardware_access_authorized=false") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/validate_m3_hardware_protocol.py b/scripts/validate_m3_hardware_protocol.py index b39ce10..e218437 100644 --- a/scripts/validate_m3_hardware_protocol.py +++ b/scripts/validate_m3_hardware_protocol.py @@ -61,14 +61,14 @@ OPERATION_KEYS = { } -def _timestamp_with_timezone(value: object) -> bool: +def _timestamp_with_timezone(value: object) -> datetime | None: if not isinstance(value, str): - return False + return None try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: - return False - return parsed.tzinfo is not None + return None + return parsed if parsed.tzinfo is not None else None def _validate_operation(operation: dict, seen: set[str]) -> list[str]: @@ -232,8 +232,14 @@ def validate(path: Path) -> dict: "machine_location", "" ).strip(): incomplete.append("machine location is not recorded") - if not _timestamp_with_timezone(session.get("planned_at")): + starts_at = _timestamp_with_timezone(session.get("planned_at")) + ends_at = _timestamp_with_timezone(session.get("ends_at")) + if starts_at is None: incomplete.append("planned_at is not an ISO-8601 timestamp with timezone") + if ends_at is None: + incomplete.append("ends_at is not an ISO-8601 timestamp with timezone") + if starts_at is not None and ends_at is not None and ends_at <= starts_at: + errors.append("session end must be later than session start") status = "invalid" if errors else "ready_for_review" if not incomplete else "incomplete" return { diff --git a/tests/fixtures/m3-hardware-protocol-template.json b/tests/fixtures/m3-hardware-protocol-template.json index 7d99b52..d044931 100644 --- a/tests/fixtures/m3-hardware-protocol-template.json +++ b/tests/fixtures/m3-hardware-protocol-template.json @@ -70,6 +70,7 @@ "session": { "machine_location": null, "planned_at": null, + "ends_at": null, "notes": "" } } diff --git a/tests/test_m3_hardware_protocol.py b/tests/test_m3_hardware_protocol.py index c945b88..112f3f5 100644 --- a/tests/test_m3_hardware_protocol.py +++ b/tests/test_m3_hardware_protocol.py @@ -57,6 +57,7 @@ class M3HardwareProtocolTests(unittest.TestCase): plan["session"] = { "machine_location": "Synthetic test location", "planned_at": "2026-10-21T12:00:00+02:00", + "ends_at": "2026-10-21T12:30:00+02:00", "notes": "No hardware is used by this unit test.", } return plan @@ -90,6 +91,15 @@ class M3HardwareProtocolTests(unittest.TestCase): ) self.assertFalse(result["hardware_access_authorized"]) + def test_session_window_must_be_bounded_and_ordered(self): + with tempfile.TemporaryDirectory(dir=ROOT / ".build") as directory: + plan = self._complete() + plan["session"]["ends_at"] = plan["session"]["planned_at"] + result = validate(self._write(Path(directory), plan)) + self.assertEqual("invalid", result["status"]) + self.assertIn("session end must be later than session start", result["errors"]) + self.assertFalse(result["hardware_access_authorized"]) + def test_safety_boundary_cannot_be_narrowed(self): with tempfile.TemporaryDirectory(dir=ROOT / ".build") as directory: plan = self._complete() diff --git a/tests/test_m3_review_packet.py b/tests/test_m3_review_packet.py new file mode 100644 index 0000000..837bd0a --- /dev/null +++ b/tests/test_m3_review_packet.py @@ -0,0 +1,113 @@ +import copy +import hashlib +import json +import tempfile +import unittest +from pathlib import Path + +from scripts.prepare_m3_review_packet import ROOT, create_packet + + +TEMPLATE = ROOT / "tests" / "fixtures" / "m3-hardware-protocol-template.json" + + +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_bytes = source.read_bytes() + output = root / "packet" + manifest = create_packet(source, output, repository_commit="a" * 40) + review = json.loads((output / "approval-record-template.json").read_text()) + frozen_bytes = (output / "plan.json").read_bytes() + self.assertEqual(source_bytes, frozen_bytes) + self.assertEqual(hashlib.sha256(source_bytes).hexdigest(), review["binding"]["plan_sha256"]) + self.assertEqual("not_reviewed", review["decision"]) + self.assertFalse(review["hardware_access_authorized"]) + self.assertFalse(manifest["hardware_access_authorized"]) + self.assertEqual("a" * 40, manifest["repository_commit"]) + + def test_incomplete_public_template_is_rejected(self): + with tempfile.TemporaryDirectory(dir=ROOT / ".build") as directory: + with self.assertRaisesRegex(ValueError, "not ready for review"): + create_packet(TEMPLATE, Path(directory) / "packet", "b" * 40) + + def test_changed_plan_changes_binding(self): + with tempfile.TemporaryDirectory(dir=ROOT / ".build") as directory: + root = Path(directory) + source = self._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()) + plan["session"]["notes"] = "Changed synthetic note" + source.write_text(json.dumps(plan), encoding="utf-8") + create_packet(source, root / "two", "c" * 40) + two = json.loads((root / "two" / "approval-record-template.json").read_text()) + self.assertNotEqual(one["binding"]["plan_sha256"], two["binding"]["plan_sha256"]) + + def test_nonempty_output_is_not_overwritten(self): + with tempfile.TemporaryDirectory(dir=ROOT / ".build") as directory: + root = Path(directory) + source = self._ready_plan(root) + output = root / "packet" + output.mkdir() + sentinel = output / "keep.txt" + sentinel.write_text("keep", encoding="utf-8") + with self.assertRaisesRegex(ValueError, "not empty"): + create_packet(source, output, "d" * 40) + self.assertEqual("keep", sentinel.read_text(encoding="utf-8")) + + def test_manifest_hashes_every_packet_file(self): + with tempfile.TemporaryDirectory(dir=ROOT / ".build") as directory: + root = Path(directory) + output = root / "packet" + manifest = create_packet(self._ready_plan(root), output, "e" * 40) + for record in manifest["files"]: + path = output / record["path"] + self.assertEqual(path.stat().st_size, record["bytes"]) + self.assertEqual(hashlib.sha256(path.read_bytes()).hexdigest(), record["sha256"]) + + +if __name__ == "__main__": + unittest.main()