#!/usr/bin/env python3 # SPDX-License-Identifier: GPL-3.0-or-later """Validate Phase-1.0Z offline passive-batch evidence.""" from __future__ import annotations import argparse import ast import hashlib import json from pathlib import Path from typing import Any PHASE = "PHASE_1_0Z_OFFLINE_PASSIVE_BATCH_CONTRACT" STATUS = "PASSIVE_LF_BATCH_CONTRACT_COMPLETE_LIVE_TRANSPORT_BLOCKED" START_COMMIT = "0433b73958f466b50c3b3f301f29e6af4ad9d319" AUTHORIZATION_FIELDS = { "target_build_authorized", "ps5_connection_authorized", "device_request_authorized", "result_receive_authorized", "device_transfer_authorized", "device_execution_authorized", "installation_authorized", "autoload_authorized", "device_write_authorized", "automatic_retry", "reconnect_authorized", "resume_authorized", } SOURCE_BINDINGS = { "phase10v_collector_size": 7429, "phase10v_collector_sha256": "f8a306dafee5d135919bec5afda789dd741e57f39803b7683fb8747c186db25c", "phase10w_policy_size": 6997, "phase10w_policy_sha256": "747d23c88f2722e8e8846599c3ac1dae3826eb3fca881caaad36b251f30f3592", "phase10x_transport_size": 8040, "phase10x_transport_sha256": "568d7578482ecf2fcd9e29085b2eb9d8705fc699611508acdf22afd30f2ddd23", "phase10y_framing_size": 6678, "phase10y_framing_sha256": "5081898ec86be52900670be2f9949a20b9abb7781a6b04d5337178a8340775d4", "phase10z_contract_size": 10487, "phase10z_contract_sha256": "0728c2be7f368e0a7f4b68efe86f6e0c5c2f50704a41d0e1992b0bfec19dde06", "phase10z_contract_tests_size": 9792, "phase10z_contract_tests_sha256": "7706cc212a0fc683eb32acaef26ecaa64cbe5784ea6aeeed7e29ade11612490c", } SOURCE_FILES = { "phase10v_collector": "tools/phase10v_shsrv_collector_model.py", "phase10w_policy": "tools/phase10w_shsrv_client_policy.py", "phase10x_transport": "tools/phase10x_inactive_transport.py", "phase10y_framing": "tools/phase10y_shsrv_framing_model.py", "phase10z_contract": "tools/phase10z_passive_batch_contract.py", "phase10z_contract_tests": "tests/test_phase10z_passive_batch_contract.py", } NETWORK_MODULES = { "socket", "asyncio", "selectors", "urllib", "http", "ftplib", "requests", "telnetlib", "paramiko", } def load_json(path: Path) -> dict[str, Any]: value = json.loads(path.read_text(encoding="utf-8")) if not isinstance(value, dict): raise ValueError("Phase-1.0Z manifest is not an object") return value def exact_file(path: Path, size: int, digest: str) -> bool: try: payload = path.read_bytes() except OSError: return False return len(payload) == size and hashlib.sha256(payload).hexdigest() == digest def _imports(tree: ast.AST) -> set[str]: names: set[str] = set() for node in ast.walk(tree): if isinstance(node, ast.Import): names.update(alias.name.split(".")[0] for alias in node.names) elif isinstance(node, ast.ImportFrom) and node.module: names.add(node.module.split(".")[0]) return names def validate_record(record: dict[str, Any], root: Path | None = None) -> list[str]: errors: list[str] = [] if record.get("phase") != PHASE or record.get("status") != STATUS: errors.append("phase/status mismatch") if record.get("start_commit") != START_COMMIT: errors.append("start commit mismatch") if record.get("activation") != { "active": False, "batch_contract_sha256": None, "run_id": None, "target_address": None, "target_port": None, "window": None, "exact_literal_path": None, "deadline_seconds": None, "commands": [], }: errors.append("activation is not inert") if record.get("source_bindings") != SOURCE_BINDINGS: errors.append("source bindings mismatch") contract = record.get("batch_contract", {}) expected_contract = { "outbound_batches": 1, "ascii_only": True, "lf_only": True, "nul_allowed": False, "cr_allowed": False, "iac_allowed": False, "telnet_negotiation_emitted": False, "server_echo_required": False, "retry_allowed": False, "reconnect_allowed": False, "resume_allowed": False, "target_retained": False, "maximum_literal_path_bytes": 512, "maximum_batch_bytes": 1035, "windows": { "T2_GREETING_AND_HELP": { "commands": ["help"], "exact_payload_hex": "68656c700a", "exact_payload_bytes": 5, "completion_requirement": "KNOWN_COMPLETE_HELP_FINGERPRINT", }, "T3_ONE_EXACT_PATH": { "commands": ["stat", "sum"], "payload_shape": "stat PATH LF sum PATH LF", "completion_requirement": "STAT_SIZE_AND_WEAK_SUM_FOR_EXACT_PATH", }, }, } if contract != expected_contract: errors.append("batch contract mismatch") receive = record.get("receive_contract", {}) expected_receive = { "input_source": "ALREADY_SUPPLIED_SYNTHETIC_BYTES_ONLY", "incoming_iac": "FAIL_CLOSED", "prompt_completion_used": False, "remote_eof_completion_used": False, "completion_event": "SYNTHETIC_HARD_DEADLINE_ONLY", "partial_result": "INVALID", "unknown_help_fingerprint": "INVALID", "source_family_selected": False, "network_transport_present": False, "clock_present": False, "cli_present": False, "file_output_present": False, "device_behavior_proven": False, "exact_identity_proven": False, } if receive != expected_receive: errors.append("receive contract mismatch") authority = record.get("authorizations", {}) if set(authority) != AUTHORIZATION_FIELDS or any( authority.get(field) is not False for field in AUTHORIZATION_FIELDS): errors.append("authorization fields are not exactly false") decision = record.get("decision", {}) if decision != { "offline_passive_batch_contract_complete": True, "exact_deployed_shsrv_identity": "UNPROVEN", "live_hard_deadline_preemption": "UNPROVEN", "live_transport_created": False, "live_collection_allowed": False, "device_action_allowed": False, "phase10aa_offline_fake_adapter_integration_allowed": True, "next_step": "OFFLINE_FAKE_ADAPTER_BATCH_AND_DEADLINE_INTEGRATION", }: errors.append("decision mismatch") performed = record.get("performed_actions", {}) if not performed or any(value is not False for value in performed.values()): errors.append("a performed action is true or missing") tests = record.get("tests", {}) if tests != { "chimera_gfx_ctest": "86_OF_86_PASS", "phase10z_guardrails": 18, "phase10z_contract_tests": 25, "safety_audit": "PASS", "secret_scan": "PASS", "network_required_by_tests": False, "hardware_claim_from_host_test": False, }: errors.append("test evidence mismatch") side_effects = record.get("side_effects", {}) if side_effects.get("offline_model") != "NONE" or \ side_effects.get("effects_claimed_absent_on_device") is not False or \ "EXPLICIT_ACCEPTANCE_REQUIRED" not in side_effects.values(): errors.append("side effects are incomplete or promoted") if root is not None: for prefix, relative in SOURCE_FILES.items(): size = SOURCE_BINDINGS[f"{prefix}_size"] digest = SOURCE_BINDINGS[f"{prefix}_sha256"] if not exact_file(root / relative, size, digest): errors.append(f"source identity mismatch: {relative}") model_path = root / SOURCE_FILES["phase10z_contract"] try: source = model_path.read_text(encoding="utf-8") tree = ast.parse(source) except (OSError, SyntaxError, UnicodeError): errors.append("contract source cannot be parsed") else: if _imports(tree) & NETWORK_MODULES: errors.append("contract imports a network module") function_names = { node.name for node in ast.walk(tree) if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) } if {"main", "connect", "send", "recv", "seal_at_prompt", "seal_at_eof"} & function_names: errors.append("contract exposes a forbidden live or completion API") if "target_address" in source or "target_port" in source: errors.append("contract retains target data") if "argparse" in source or "__main__" in source: errors.append("contract exposes a CLI") approval = (root / "docs/approvals/phase-1.0z-passive-batch.md").read_text( encoding="utf-8") if "active=false" not in approval or "attested=false" not in approval or \ "target_address=null" not in approval: errors.append("approval template is not inert") phase_output_roots = ( root / "tools", root / "tests", root / "docs", root / "manifests", ) if any( path.suffix.lower() in {".elf", ".self", ".sprx", ".pkg"} for base in phase_output_roots for path in base.rglob("*") if path.is_file() ): errors.append("target artifact exists in a Phase-1.0Z output area") return errors def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--root", type=Path, required=True) args = parser.parse_args() root = args.root.resolve() try: record = load_json( root / "manifests/retroarch/phase-1.0z-passive-batch.json") errors = validate_record(record, root) except (OSError, ValueError, json.JSONDecodeError) as error: errors = [f"validation input failed: {error}"] if errors: for error in errors: print(f"ERROR: {error}") return 1 print("Phase-1.0Z offline passive-batch validation passed") return 0 if __name__ == "__main__": raise SystemExit(main())