147 lines
8.6 KiB
Python
147 lines
8.6 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Validate Phase-1.0AB offline live-adapter feasibility evidence."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import ast
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
PHASE = "PHASE_1_0AB_OFFLINE_LIVE_ADAPTER_FEASIBILITY"
|
|
STATUS = "PARTIAL_FEASIBILITY_LIVE_IMPLEMENTATION_BLOCKED"
|
|
START_COMMIT = "06d2fe831959b71562401722ae36821faa197636"
|
|
MODEL_SIZE = 8738
|
|
MODEL_SHA256 = "7d1aa32d49b91b1e5cf3a085dda033767bdf17ab34389ff044f7403f86287959"
|
|
MODEL_TEST_SIZE = 7558
|
|
MODEL_TEST_SHA256 = "39597e991b15bcfa9aa28cb2f68c87ed048c38482c6dbf81a28c56ccfceb0a48"
|
|
NETWORK_MODULES = {"socket", "selectors", "select", "asyncio", "urllib", "http", "requests", "telnetlib"}
|
|
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",
|
|
}
|
|
|
|
|
|
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.0AB 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()
|
|
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,
|
|
python_root: Path | None = None) -> list[str]:
|
|
errors=[]
|
|
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, "trace_model_sha256": None, "target_address": None, "target_port": None, "run_id": None}: errors.append("activation is not inert")
|
|
runtime=record.get("local_runtime", {})
|
|
if runtime != {
|
|
"python_version": "3.13.2", "compiler": "MSC_V_1942_64_BIT_AMD64",
|
|
"platform": "WINDOWS", "default_selector": "SelectSelector",
|
|
"monotonic_implementation": "QueryPerformanceCounter()", "monotonic": True,
|
|
"monotonic_adjustable": False, "reported_resolution_seconds": 1e-7,
|
|
"socket_py_size": 38741, "socket_py_sha256": "523695ac3383799547b421b4fe18004de1e80181e97181b6d7a10533b47f4c49",
|
|
"selectors_py_size": 20060, "selectors_py_sha256": "b3d6cebd4a3a03b4a614f12f171622ce4e4ba3295b9e8b89e2bde051003106eb",
|
|
"socket_pyd_size": 84984, "socket_pyd_sha256": "8daefaff53e6956f5aea5279a7c71f17d8c63e2b0d54031c3b9e82fcb0fb84db",
|
|
"select_pyd_size": 32248, "select_pyd_sha256": "baee284995b22d495fd12fa8378077e470978db1522c61bfb9af37fb827f33d1",
|
|
}: errors.append("local runtime record mismatch")
|
|
if record.get("source_bindings") != {
|
|
"phase10aa_integration_sha256": "8e1cac255f85d2cd14baf8fbc27d631c9b607fc7d19fc57c65462089f0574055",
|
|
"phase10ab_trace_model_size": MODEL_SIZE, "phase10ab_trace_model_sha256": MODEL_SHA256,
|
|
"phase10ab_trace_tests_size": MODEL_TEST_SIZE, "phase10ab_trace_tests_sha256": MODEL_TEST_SHA256,
|
|
}: errors.append("source bindings mismatch")
|
|
if record.get("feasibility") != {
|
|
"receipt_before_socket": "FEASIBLE_FROM_EXISTING_HOST_MODEL",
|
|
"numeric_address_only": "DESIGN_REQUIRED",
|
|
"nonblocking_before_connect": "FEASIBLE_FROM_LOCAL_RUNTIME",
|
|
"pending_connect": "PARTIAL", "complete_send_loop": "FEASIBLE_FROM_LOCAL_RUNTIME",
|
|
"bounded_receive_memory": "FEASIBLE_FROM_EXISTING_MODEL",
|
|
"hard_wall_clock_deadline": "PARTIAL",
|
|
"prompt_independent_completion": "FEASIBLE_FROM_Z",
|
|
"remote_eof": "FEASIBLE_FAIL_CLOSED",
|
|
"local_descriptor_cleanup": "FEASIBLE_BY_DESIGN",
|
|
"remote_shell_cleanup": "UNPROVEN", "retry_reconnect_resume": "EXCLUDED",
|
|
}: errors.append("feasibility matrix mismatch")
|
|
if record.get("trace_model") != {
|
|
"synthetic_input_only": True, "maximum_events": 512,
|
|
"maximum_batch_bytes": 1035, "maximum_receive_bytes": 65536,
|
|
"maximum_deadline_seconds": 10, "network_import_present": False,
|
|
"selector_import_present": False, "real_clock_present": False,
|
|
"address_present": False, "cli_present": False, "file_output_present": False,
|
|
"exact_identity_proven": False, "device_behavior_proven": False,
|
|
}: errors.append("trace model record mismatch")
|
|
stops=record.get("hard_stops", {})
|
|
if set(stops) != {"live_adapter_implementation","socket_creation","dns","target_retention","connection","request","retry","device_action"} or any(value is not True for value in stops.values()): errors.append("hard stops mismatch")
|
|
auth=record.get("authorizations", {})
|
|
if set(auth) != AUTHORIZATION_FIELDS or any(auth.get(field) is not False for field in AUTHORIZATION_FIELDS): errors.append("authorization fields are not exactly false")
|
|
if record.get("decision") != {
|
|
"overall": STATUS, "live_adapter_created": False, "live_adapter_allowed": False,
|
|
"device_action_allowed": False, "phase10ac_offline_dormant_syscall_facade_allowed": True,
|
|
"next_step": "OFFLINE_DORMANT_TARGET_FREE_ADAPTER_WITH_FAKE_SYSCALLS",
|
|
}: 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": "92_OF_92_PASS", "phase10ab_guardrails": 20,
|
|
"phase10ab_trace_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:
|
|
model=root / "tools/phase10ab_nonblocking_trace_model.py"; tests=root / "tests/test_phase10ab_nonblocking_trace_model.py"
|
|
if not exact_file(model, MODEL_SIZE, MODEL_SHA256): errors.append("trace model identity mismatch")
|
|
if not exact_file(tests, MODEL_TEST_SIZE, MODEL_TEST_SHA256): errors.append("trace tests identity mismatch")
|
|
try: source=model.read_text(encoding="utf-8"); tree=ast.parse(source)
|
|
except (OSError,SyntaxError,UnicodeError): errors.append("trace model cannot be parsed")
|
|
else:
|
|
if _imports(tree) & NETWORK_MODULES or "time" in _imports(tree): errors.append("trace model imports live capability")
|
|
names={node.name for node in ast.walk(tree) if isinstance(node,(ast.FunctionDef,ast.AsyncFunctionDef))}
|
|
if {"main","connect","send","recv"} & names: errors.append("trace model exposes live API")
|
|
if "target_address" in source or "target_port" in source: errors.append("trace model contains target fields")
|
|
approval=(root / "docs/approvals/phase-1.0ab-live-adapter-feasibility.md").read_text(encoding="utf-8")
|
|
if "active=false" not in approval or "ps5_connection_authorized=false" not in approval: errors.append("approval is not inert")
|
|
if python_root is not None:
|
|
paths={
|
|
"socket_py": python_root / "Lib/socket.py", "selectors_py": python_root / "Lib/selectors.py",
|
|
"socket_pyd": python_root / "DLLs/_socket.pyd", "select_pyd": python_root / "DLLs/select.pyd",
|
|
}
|
|
for prefix,path in paths.items():
|
|
if not exact_file(path, runtime[f"{prefix}_size"], runtime[f"{prefix}_sha256"]): errors.append(f"local runtime identity mismatch: {prefix}")
|
|
return errors
|
|
|
|
|
|
def main() -> int:
|
|
parser=argparse.ArgumentParser();parser.add_argument("--root",type=Path,required=True);parser.add_argument("--python-root",type=Path)
|
|
args=parser.parse_args();root=args.root.resolve()
|
|
try: record=load_json(root / "manifests/retroarch/phase-1.0ab-live-adapter-feasibility.json");errors=validate_record(record,root,args.python_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.0AB offline live-adapter feasibility validation passed");return 0
|
|
|
|
|
|
if __name__ == "__main__": raise SystemExit(main())
|