#!/usr/bin/env python3 # SPDX-License-Identifier: GPL-3.0-or-later """Validate Phase-1.0AA offline fake-adapter integration evidence.""" from __future__ import annotations import argparse import ast import hashlib import json from pathlib import Path from typing import Any PHASE = "PHASE_1_0AA_OFFLINE_FAKE_ADAPTER_INTEGRATION" STATUS = "OFFLINE_FAKE_BATCH_INTEGRATION_COMPLETE_LIVE_ADAPTER_BLOCKED" START_COMMIT = "57ff9a1c5937575b00df05f2cd9897118eab1f2a" SOURCE_BINDINGS = { "phase10x_transport_size": 8040, "phase10x_transport_sha256": "568d7578482ecf2fcd9e29085b2eb9d8705fc699611508acdf22afd30f2ddd23", "phase10z_contract_size": 10487, "phase10z_contract_sha256": "0728c2be7f368e0a7f4b68efe86f6e0c5c2f50704a41d0e1992b0bfec19dde06", "phase10aa_integration_size": 11200, "phase10aa_integration_sha256": "8e1cac255f85d2cd14baf8fbc27d631c9b607fc7d19fc57c65462089f0574055", "phase10aa_integration_tests_size": 10667, "phase10aa_integration_tests_sha256": "18a5470a651fcbe4b23f2a68cba499dd19d3da15a229623c054be0b598025699", } SOURCE_FILES = { "phase10x_transport": "tools/phase10x_inactive_transport.py", "phase10z_contract": "tools/phase10z_passive_batch_contract.py", "phase10aa_integration": "tools/phase10aa_offline_fake_batch.py", "phase10aa_integration_tests": "tests/test_phase10aa_offline_fake_batch.py", } 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", } 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.0AA 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]: values: set[str] = set() for node in ast.walk(tree): if isinstance(node, ast.Import): values.update(alias.name.split(".")[0] for alias in node.names) elif isinstance(node, ast.ImportFrom) and node.module: values.add(node.module.split(".")[0]) return values 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, "integration_sha256": None, "run_id": None, "target_address": None, "target_port": None, "window": None, "deadline_seconds": None, }: errors.append("activation is not inert") if record.get("source_bindings") != SOURCE_BINDINGS: errors.append("source bindings mismatch") if record.get("fake_boundary") != { "exact_builtin_adapter_required": True, "adapter_subclasses_allowed": False, "exact_builtin_clock_required": True, "clock_subclasses_allowed": False, "exact_fake_evidence_store_required": True, "live_adapter_protocol_present": False, "network_import_present": False, "real_clock_present": False, "target_present": False, "cli_present": False, "maximum_fake_events": 257, "event_kinds": ["DATA", "HARD_DEADLINE", "REMOTE_EOF", "BLOCKED"], }: errors.append("fake boundary mismatch") if record.get("ordering_contract") != { "receipt_before_fake_open": True, "fake_open_count": 1, "complete_batch_send_count": 1, "fake_close_count": 1, "completion_event": "SYNTHETIC_HARD_DEADLINE_ONLY", "early_deadline": "INVALID", "remote_eof": "INVALID", "blocked_receive": "INVALID", "missing_deadline": "INVALID", "data_at_or_after_deadline": "INVALID", "partial_result": "INVALID", "incoming_iac": "INVALID", "retry_allowed": False, "reconnect_allowed": False, "resume_allowed": False, }: errors.append("ordering contract mismatch") evidence = record.get("evidence_contract", {}) if evidence != { "exclusive_create": True, "consumed_receipt_retained_on_failure": True, "failure_output_created": False, "sanitized_output_receipt_bound": True, "batch_sha256_recorded": True, "batch_size_recorded": True, "target_retained": False, "raw_transcript_persisted": False, "logical_event_buffer_discarded": True, "physical_memory_erasure_proven": False, "directory_entry_durability_proven": False, "device_behavior_proven": False, "exact_identity_proven": False, }: errors.append("evidence 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") if record.get("decision") != { "offline_fake_batch_integration_complete": True, "live_adapter_created": False, "live_adapter_allowed": False, "live_collection_allowed": False, "device_action_allowed": False, "phase10ab_offline_live_adapter_feasibility_review_allowed": True, "next_step": "OFFLINE_LIVE_ADAPTER_TIMEOUT_AND_CLEANUP_FEASIBILITY_REVIEW", }: 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("performed actions are missing or true") if record.get("tests") != { "chimera_gfx_ctest": "89_OF_89_PASS", "phase10aa_guardrails": 20, "phase10aa_integration_tests": 25, "safety_audit": "PASS", "secret_scan": "PASS", "network_required_by_tests": False, "hardware_claim_from_host_test": False, }: errors.append("test evidence mismatch") if root is not None: for prefix, relative in SOURCE_FILES.items(): if not exact_file(root / relative, SOURCE_BINDINGS[f"{prefix}_size"], SOURCE_BINDINGS[f"{prefix}_sha256"]): errors.append(f"source identity mismatch: {relative}") source_path = root / SOURCE_FILES["phase10aa_integration"] try: source = source_path.read_text(encoding="utf-8") tree = ast.parse(source) except (OSError, SyntaxError, UnicodeError): errors.append("integration source cannot be parsed") else: if _imports(tree) & NETWORK_MODULES or "time" in _imports(tree): errors.append("integration imports network or real-clock support") names = {node.name for node in ast.walk(tree) if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))} if {"main", "connect", "recv", "open_socket"} & names: errors.append("integration exposes a live API or CLI") if "target_address" in source or "target_port" in source: errors.append("integration contains target fields") required_shapes = ( "type(adapter) is not OfflineFakeBatchAdapter", "type(clock) is not OfflineFakeClock", "type(evidence) is not OfflineFakeEvidenceStore", "receipt = evidence.create_fake_consumed_receipt", "adapter.send_one_batch(batch)", "accumulator.seal_at_hard_deadline(True)", ) if any(shape not in source for shape in required_shapes): errors.append("exact fake ordering source shape is missing") approval = (root / "docs/approvals/phase-1.0aa-offline-fake-adapter.md").read_text(encoding="utf-8") if "active=false" not in approval or "attested=false" not in approval or \ "ps5_connection_authorized=false" not in approval: errors.append("approval template is not inert") for base in (root / "tools", root / "tests", root / "docs", root / "manifests"): if any(path.is_file() and path.suffix.lower() in {".elf", ".self", ".sprx", ".pkg"} for path in base.rglob("*")): errors.append("target artifact exists in a Phase-1.0AA output area") break 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.0aa-offline-fake-adapter.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.0AA offline fake-adapter validation passed") return 0 if __name__ == "__main__": raise SystemExit(main())