safety: add immutable M3 review packets

This commit is contained in:
Codex 2026-10-21 21:55:28 +02:00
parent ed59b928c9
commit c382ea9a78
8 changed files with 285 additions and 5 deletions

View file

@ -34,6 +34,9 @@ The repository now provides:
- a fail-closed M3 hardware-research protocol, incomplete public plan template - a fail-closed M3 hardware-research protocol, incomplete public plan template
and validator; validation can reach only `ready_for_review` and never and validator; validation can reach only `ready_for_review` and never
authorizes hardware access. 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 Run the ordinary checkpoint with the bundled Codex Python or another supported
Python interpreter and a writable `TEMP`/`TMP`: Python interpreter and a writable `TEMP`/`TMP`:

View file

@ -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 reviewer, approval time with timezone, location and exact session window. Any
plan edit invalidates that approval. 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 ## Session sequence
The eventual reviewed session follows this order and stops at the first failed The eventual reviewed session follows this order and stops at the first failed

View file

@ -96,6 +96,10 @@ evidence for:
plan template and standard-library validator that requires exact plan template and standard-library validator that requires exact
evidence-backed request bytes, distinct safety roles, physical controls and evidence-backed request bytes, distinct safety roles, physical controls and
recovery planning while always denying hardware authorization. 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: The remaining M1 gates are deliberately explicit:

View file

@ -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())

View file

@ -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): if not isinstance(value, str):
return False return None
try: try:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError: except ValueError:
return False return None
return parsed.tzinfo is not None return parsed if parsed.tzinfo is not None else None
def _validate_operation(operation: dict, seen: set[str]) -> list[str]: def _validate_operation(operation: dict, seen: set[str]) -> list[str]:
@ -232,8 +232,14 @@ def validate(path: Path) -> dict:
"machine_location", "" "machine_location", ""
).strip(): ).strip():
incomplete.append("machine location is not recorded") 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") 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" status = "invalid" if errors else "ready_for_review" if not incomplete else "incomplete"
return { return {

View file

@ -70,6 +70,7 @@
"session": { "session": {
"machine_location": null, "machine_location": null,
"planned_at": null, "planned_at": null,
"ends_at": null,
"notes": "" "notes": ""
} }
} }

View file

@ -57,6 +57,7 @@ class M3HardwareProtocolTests(unittest.TestCase):
plan["session"] = { plan["session"] = {
"machine_location": "Synthetic test location", "machine_location": "Synthetic test location",
"planned_at": "2026-10-21T12:00:00+02:00", "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.", "notes": "No hardware is used by this unit test.",
} }
return plan return plan
@ -90,6 +91,15 @@ class M3HardwareProtocolTests(unittest.TestCase):
) )
self.assertFalse(result["hardware_access_authorized"]) 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): def test_safety_boundary_cannot_be_narrowed(self):
with tempfile.TemporaryDirectory(dir=ROOT / ".build") as directory: with tempfile.TemporaryDirectory(dir=ROOT / ".build") as directory:
plan = self._complete() plan = self._complete()

View file

@ -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()