christian/ThePirateRay
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

safety: validate M3 approval records

Browse source
This commit is contained in:
Codex 2026-10-21 21:57:52 +02:00
parent c382ea9a78
commit a22cb1e759
7 changed files with 326 additions and 43 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
docs/CODEX_HANDOFF.md
View file

@ -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`:

15
docs/research/m3-hardware-research-protocol.md
View file

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

3
docs/status/phase-0.md
View file

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

7
scripts/prepare_m3_review_packet.py
View file

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

170
scripts/validate_m3_approval.py Normal file
View file

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

83
tests/test_m3_approval.py Normal file
View file

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

88
tests/test_m3_review_packet.py
View file

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