#!/usr/bin/env python3 # SPDX-License-Identifier: GPL-3.0-or-later """Validate Phase-1.0X inactive injected-transport evidence.""" from __future__ import annotations import argparse import ast import hashlib import json from pathlib import Path import subprocess from typing import Any PHASE = "PHASE_1_0X_INACTIVE_INJECTED_TRANSPORT" STATUS = "INACTIVE_INJECTED_TRANSPORT_COMPLETE_LIVE_PROTOCOL_BLOCKED" START_COMMIT = "45228225e3a0e8811b4a5bcaa79e6f125db3a7f8" SHSRV_COMMIT = "6f320637d56d344a0e7797753099e33238bbf146" PARSER_SHA256 = "4701a057a98b4874e49e1bcf11db9a9a3a105e48f2c25e42796bff10f238f7c2" COLLECTOR_SIZE = 7429 COLLECTOR_SHA256 = "f8a306dafee5d135919bec5afda789dd741e57f39803b7683fb8747c186db25c" POLICY_SIZE = 6997 POLICY_SHA256 = "747d23c88f2722e8e8846599c3ac1dae3826eb3fca881caaad36b251f30f3592" TRANSPORT_SIZE = 8040 TRANSPORT_SHA256 = "568d7578482ecf2fcd9e29085b2eb9d8705fc699611508acdf22afd30f2ddd23" TRANSPORT_TEST_SIZE = 13087 TRANSPORT_TEST_SHA256 = "19709ed6ab456be428d262b3f0afb4f6f577b34e4db80e66a73ed63d9bf2cd43" 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", ) NETWORK_MODULES = { "socket", "telnetlib", "urllib", "requests", "http", "ftplib", "asyncio", "selectors", } 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.0X manifest is not an object") return value def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def git(root: Path, *args: str) -> str: result = subprocess.run( ["git", *args], cwd=root, capture_output=True, text=True, check=False) if result.returncode: raise RuntimeError(result.stderr.strip() or "git failed") return result.stdout.strip() def exact_file(path: Path, size: int, digest: str) -> bool: return path.stat().st_size == size and sha256(path) == digest def all_authorizations_false(record: dict[str, Any]) -> bool: return all(record.get(field) is False for field in AUTHORIZATION_FIELDS) def activation_is_inactive(record: dict[str, Any]) -> bool: return record == { "active": False, "policy_sha256": None, "collector_sha256": None, "transport_sha256": None, "run_id": None, "target_address": None, "target_port": None, "window": None, "exact_literal_path": None, "commands": [], "deadline_seconds": None, "expires_at": None, } def architecture_is_injected_only(record: dict[str, Any]) -> bool: return record == { "injected_adapter_only": True, "injected_monotonic_clock_only": True, "policy_cli_present": False, "live_cli_present": False, "network_transport_present": False, "socket_import_present": False, "dns_lookup_present": False, "connect_present": False, "target_address_present": False, "command_byte_formatter_present": False, "telnet_reply_generator_present": False, "live_prompt_detector_present": False, "fake_boundary_labels_only": True, } def evidence_is_bounded(record: dict[str, Any]) -> bool: required_true = { "receipt_before_adapter_open", "exclusive_leaf_create", "file_fsync", "close_and_reopen", "reopened_bytes_verified", "sanitized_output_receipt_hash_binding", "trusted_caller_owned_root_required", } required_false = { "overwrite_supported", "delete_or_cleanup_supported", "raw_transcript_persisted", "target_address_persisted", "partial_file_cleanup", "partial_file_is_valid_evidence", "directory_fsync", } return ( set(record) == required_true | required_false | {"directory_entry_durability"} and all(record.get(field) is True for field in required_true) and all(record.get(field) is False for field in required_false) and record.get("directory_entry_durability") == "UNPROVEN" ) def deadline_is_partial(record: dict[str, Any]) -> bool: return record == { "one_absolute_monotonic_deadline": True, "pre_and_post_boundary_checks": True, "remaining_budget_passed_to_adapter": True, "adapter_close_after_open_attempt": True, "retry_loop_present": False, "second_open_present": False, "blocking_adapter_call_preemption": False, "real_socket_timeout_present": False, "live_cleanup_proven": False, } def decision_is_offline_only(record: dict[str, Any]) -> bool: return record == { "offline_injected_transport_complete": True, "offline_local_evidence_complete_with_limitations": True, "exact_deployed_shsrv_identity": "UNPROVEN", "live_protocol_framing": "BLOCKED_UNPROVEN", "live_network_client_created": False, "live_client_implementation_allowed": False, "live_collection_allowed": False, "device_action_allowed": False, "next_step": "OFFLINE_EXACT_PROMPT_AND_TELNET_FRAMING_AUDIT", } def source_has_no_network_import(path: Path) -> bool: tree = ast.parse(path.read_text(encoding="utf-8")) imports = { alias.name.split(".", 1)[0] for node in ast.walk(tree) if isinstance(node, (ast.Import, ast.ImportFrom)) for alias in node.names } return not imports.intersection(NETWORK_MODULES) def validate(root: Path, shsrv_root: Path) -> list[str]: errors: list[str] = [] try: manifest = load_json( root / "manifests/retroarch/phase-1.0x-inactive-transport.json") if manifest.get("phase") != PHASE or manifest.get("status") != STATUS: errors.append("phase/status mismatch") if manifest.get("start_commit") != START_COMMIT: errors.append("start commit mismatch") if not activation_is_inactive(manifest.get("activation", {})): errors.append("tracked activation is not inert") if not all_authorizations_false(manifest.get("authorizations", {})): errors.append("authorization remains active or missing") if not architecture_is_injected_only(manifest.get("architecture", {})): errors.append("architecture contains live transport capability") if not evidence_is_bounded(manifest.get("local_evidence", {})): errors.append("local evidence contract mismatch") if not deadline_is_partial(manifest.get("deadline_and_cleanup", {})): errors.append("deadline limitations were promoted or changed") if not decision_is_offline_only(manifest.get("decision", {})): errors.append("offline-only decision mismatch") performed = manifest.get("performed_actions", {}) if not performed or not all(value is False for value in performed.values()): errors.append("performed device or network action is present") missing = manifest.get("missing_live_components", {}) if len(missing) != 8 or not all(value is True for value in missing.values()): errors.append("missing live component was promoted") bindings = manifest.get("source_bindings", {}) expected_bindings = { "shsrv_reference_commit": SHSRV_COMMIT, "phase10t_parser_sha256": PARSER_SHA256, "phase10v_collector_size": COLLECTOR_SIZE, "phase10v_collector_sha256": COLLECTOR_SHA256, "phase10w_policy_size": POLICY_SIZE, "phase10w_policy_sha256": POLICY_SHA256, "phase10x_transport_size": TRANSPORT_SIZE, "phase10x_transport_sha256": TRANSPORT_SHA256, "phase10x_transport_tests_size": TRANSPORT_TEST_SIZE, "phase10x_transport_tests_sha256": TRANSPORT_TEST_SHA256, } if bindings != expected_bindings: errors.append("source bindings mismatch") identities = ( (root / "tools/phase10t_shsrv_transcript.py", None, PARSER_SHA256), (root / "tools/phase10v_shsrv_collector_model.py", COLLECTOR_SIZE, COLLECTOR_SHA256), (root / "tools/phase10w_shsrv_client_policy.py", POLICY_SIZE, POLICY_SHA256), (root / "tools/phase10x_inactive_transport.py", TRANSPORT_SIZE, TRANSPORT_SHA256), (root / "tests/test_phase10x_inactive_transport.py", TRANSPORT_TEST_SIZE, TRANSPORT_TEST_SHA256), ) for path, size, digest in identities: if (size is not None and path.stat().st_size != size) or \ sha256(path) != digest: errors.append(f"source identity mismatch: {path.name}") transport_path = root / "tools/phase10x_inactive_transport.py" if not source_has_no_network_import(transport_path): errors.append("inactive transport imports networking") source = transport_path.read_text(encoding="utf-8") for forbidden in ("def main(", "argparse", "target_address", "2323"): if forbidden in source: errors.append(f"inactive transport contains forbidden token: {forbidden}") for required in ( "os.O_EXCL", "os.fsync", "path.read_bytes()", "adapter.close_once()", "receipt = evidence.create_consumed_receipt", ): if required not in source: errors.append(f"transport control missing: {required}") if git(shsrv_root, "rev-parse", "HEAD") != SHSRV_COMMIT: errors.append("shsrv reference commit mismatch") if git(shsrv_root, "status", "--short"): errors.append("shsrv reference tree is dirty") sh_source = (shsrv_root / "sh.c").read_text(encoding="utf-8") for token in ( "sh_prompt(void)", 'getenv("PWD")', 'setenv("PWD", "/", 0)', 'fprintf(stdout, "%s$ ", cwd ? cwd : "(null)")', "sh_prompt();", ): if token not in sh_source: errors.append(f"shsrv prompt evidence missing: {token}") approval = ( root / "docs/approvals/phase-1.0x-inactive-transport.md" ).read_text(encoding="utf-8") for token in ( "active=false", "target_address=null", "target_port=null", "commands=[]", "ps5_connection_authorized=false", "device_request_authorized=false", "automatic_retry=false", ): if token not in approval: errors.append(f"inactive approval token missing: {token}") tests = manifest.get("tests", {}) if not ( tests.get("chimera_gfx_ctest") == "80_OF_80_PASS" and tests.get("phase10x_guardrails") == 18 and tests.get("phase10x_transport_tests") == 18 and tests.get("safety_audit") == "PASS" and tests.get("secret_scan") == "PASS" and tests.get("network_required_by_tests") is False and tests.get("hardware_claim_from_host_test") is False ): errors.append("test evidence mismatch") tracked = git(root, "ls-files").splitlines() if any(path.lower().endswith((".elf", ".self", ".sprx", ".pkg")) for path in tracked): errors.append("target artifact is tracked") except (OSError, RuntimeError, ValueError, json.JSONDecodeError, SyntaxError) as error: errors.append(str(error)) return errors def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--root", type=Path, required=True) parser.add_argument("--shsrv-root", type=Path, required=True) args = parser.parse_args() errors = validate(args.root.resolve(), args.shsrv_root.resolve()) if errors: for error in errors: print(f"ERROR: {error}") return 1 print("Phase-1.0X inactive injected-transport validation passed") return 0 if __name__ == "__main__": raise SystemExit(main())