#!/usr/bin/env python3 # SPDX-License-Identifier: GPL-3.0-or-later """Validate Phase-1.0W self-review and inactive client architecture.""" 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_0W_INACTIVE_SHSRV_CLIENT_ARCHITECTURE" STATUS = "INACTIVE_CLIENT_ARCHITECTURE_COMPLETE_NETWORK_IMPLEMENTATION_BLOCKED" START_COMMIT = "665d02995fdb72080a20cd53c92265b1e24afb1c" PARSER_SHA256 = "4701a057a98b4874e49e1bcf11db9a9a3a105e48f2c25e42796bff10f238f7c2" COLLECTOR_SIZE = 7429 COLLECTOR_SHA256 = "f8a306dafee5d135919bec5afda789dd741e57f39803b7683fb8747c186db25c" POLICY_SIZE = 6997 POLICY_SHA256 = "747d23c88f2722e8e8846599c3ac1dae3826eb3fca881caaad36b251f30f3592" FAKE_SIZE = 1870 FAKE_SHA256 = "e2f074e844102177f45fb0f97254108ee641db00931bb037574dee82111f316f" 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", ) 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.0W 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 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, "run_id": None, "target_address": None, "target_port": None, "window": None, "exact_literal_path": None, "commands": [], "deadline_seconds": None, "expires_at": None, } def architecture_has_no_transport(record: dict[str, Any]) -> bool: return record == { "policy_returns_data_only": True, "policy_cli_present": False, "network_transport_present": False, "socket_import_present": False, "dns_lookup_present": False, "connect_present": False, "command_formatter_present": False, "command_sender_present": False, "receive_present": False, "filesystem_output_present": False, "fake_transport_only": True, } def future_contract_is_bounded(record: dict[str, Any]) -> bool: return record == { "source_bound_port": 2323, "maximum_connections": 1, "maximum_deadline_seconds": 10, "maximum_approval_lifetime_minutes": 15, "allowed_windows": { "T2_GREETING_AND_HELP": ["help"], "T3_ONE_EXACT_PATH": ["stat", "sum"], }, "dual_record_exact_match_required": True, "exact_record_shape_required": True, "policy_hash_dual_match_required": True, "operator_attestation_required": True, "listener_attestation_required": True, "spawned_shell_effects_acceptance_required": True, "serial_query_acceptance_required": True, "telemetry_query_acceptance_required": True, "unproven_memory_erasure_acceptance_required": True, "automatic_retry": False, "reconnect": False, "resume": False, "fallback": False, } def live_components_remain_missing(record: dict[str, Any]) -> bool: return set(record) == { "consumed_attempt_receipt", "exclusive_sanitized_output", "monotonic_deadline", "host_clock_acquisition", "network_transport", "telnet_reply_contract", "prompt_completion_contract", "bounded_receive_integration", "deterministic_close", "error_path_redaction_review", } and all(value is True for value in record.values()) def decision_is_offline_only(record: dict[str, Any]) -> bool: return record == { "self_review_complete": True, "review_findings_remediated": True, "offline_client_architecture_complete": True, "exact_deployed_shsrv_identity": "UNPROVEN", "live_network_client_created": False, "live_client_implementation_allowed": False, "phase10x_offline_inactive_transport_implementation_allowed": True, "live_collection_allowed": False, "launch_context_experiment_allowed": False, "device_action_allowed": False, "next_step": "OFFLINE_INACTIVE_FAKE_SOCKET_TRANSPORT_IMPLEMENTATION", } 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({ "socket", "telnetlib", "urllib", "requests", "http", "ftplib"}) def exact_file(path: Path, size: int, digest: str) -> bool: return path.stat().st_size == size and sha256(path) == digest def validate(root: Path) -> list[str]: errors: list[str] = [] try: record = load_json( root / "manifests/retroarch/phase-1.0w-inactive-client-architecture.json") 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 not activation_is_inactive(record.get("activation", {})): errors.append("tracked activation is not inert") if not architecture_has_no_transport(record.get("architecture", {})): errors.append("network or persistence capability was introduced") if not future_contract_is_bounded(record.get("future_contract", {})): errors.append("future contract was relaxed") if not live_components_remain_missing( record.get("missing_live_components", {})): errors.append("missing live component was incorrectly claimed complete") if not all_authorizations_false(record.get("authorizations", {})): errors.append("authorization remains active or missing") if not decision_is_offline_only(record.get("decision", {})): errors.append("live or device decision is enabled") performed = record.get("performed_actions", {}) if not performed or not all(value is False for value in performed.values()): errors.append("performed action is present") bindings = record.get("source_bindings", {}) if bindings != { "phase10t_parser_sha256": PARSER_SHA256, "remediated_phase10v_collector_size": COLLECTOR_SIZE, "remediated_phase10v_collector_sha256": COLLECTOR_SHA256, "phase10w_policy_size": POLICY_SIZE, "phase10w_policy_sha256": POLICY_SHA256, "fake_transport_size": FAKE_SIZE, "fake_transport_sha256": FAKE_SHA256, }: errors.append("source bindings mismatch") tests = record.get("tests", {}) if not ( tests.get("chimera_gfx_ctest") == "77_OF_77_PASS" and tests.get("phase10w_guardrails") == 18 and tests.get("phase10w_policy_tests") == 25 and tests.get("remediated_phase10v_model_tests") == 21 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") 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 / "tests/phase10w_fake_transport.py", FAKE_SIZE, FAKE_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}") policy_path = root / "tools/phase10w_shsrv_client_policy.py" fake_path = root / "tests/phase10w_fake_transport.py" if not source_has_no_network_import(policy_path) or \ not source_has_no_network_import(fake_path): errors.append("policy or fake transport imports networking") policy_source = policy_path.read_text(encoding="utf-8") for token in ( "SOURCE_BOUND_PORT = 2323", "MAX_DEADLINE_SECONDS = 10", '"T2_GREETING_AND_HELP": ("help",)', '"T3_ONE_EXACT_PATH": ("stat", "sum")', "@dataclass(frozen=True)", ): if token not in policy_source: errors.append(f"policy token missing: {token}") if "def main(" in policy_source or "argparse" in policy_source: errors.append("policy unexpectedly exposes a CLI") phase_v = load_json( root / "manifests/retroarch/phase-1.0v-inactive-shsrv-collector.json") remediation = phase_v.get("review_remediation", {}) if not remediation or \ remediation.get("physical_memory_erasure_proven") is not False or \ not all(value is True for key, value in remediation.items() if key != "physical_memory_erasure_proven"): errors.append("Phase-1.0V remediation evidence mismatch") approval = ( root / "docs/approvals/phase-1.0w-shsrv-client.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}") 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) args = parser.parse_args() errors = validate(args.root.resolve()) if errors: for error in errors: print(f"ERROR: {error}") return 1 print("Phase-1.0W inactive client-architecture validation passed") return 0 if __name__ == "__main__": raise SystemExit(main())