199 lines
8.3 KiB
Python
199 lines
8.3 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Validate the inactive Phase-1.0AC dormant-adapter evidence."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import ast
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
from typing import Any
|
|
|
|
|
|
PHASE = "PHASE_1_0AC_OFFLINE_DORMANT_ADAPTER"
|
|
STATUS = "OFFLINE_DORMANT_FAKE_SYSCALL_ADAPTER_COMPLETE_LIVE_ADAPTER_BLOCKED"
|
|
START_COMMIT = "3bc8ac09615dda3c4ee3ad02f19440ca9f4f8f96"
|
|
ADAPTER_SIZE = 13282
|
|
ADAPTER_SHA256 = "6f28926b59fd9afa6de1ff36d4fa9b013d7e027c9adc0bffd9445d7c89acf939"
|
|
ADAPTER_TEST_SIZE = 11681
|
|
ADAPTER_TEST_SHA256 = "9c8c611dbab5e43df71d523169d9a1bf7579ed1918943df7531bf75ef4b9abfc"
|
|
NETWORK_MODULES = {
|
|
"socket", "selectors", "select", "asyncio", "urllib", "http",
|
|
"requests", "ftplib", "telnetlib", "subprocess",
|
|
}
|
|
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.0AC 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, "adapter_sha256": None, "target_address": None,
|
|
"target_port": None, "run_id": None,
|
|
}:
|
|
errors.append("activation is not inert")
|
|
if record.get("source_bindings") != {
|
|
"phase10ab_trace_model_sha256": "7d1aa32d49b91b1e5cf3a085dda033767bdf17ab34389ff044f7403f86287959",
|
|
"phase10ac_adapter_size": ADAPTER_SIZE,
|
|
"phase10ac_adapter_sha256": ADAPTER_SHA256,
|
|
"phase10ac_tests_size": ADAPTER_TEST_SIZE,
|
|
"phase10ac_tests_sha256": ADAPTER_TEST_SHA256,
|
|
}:
|
|
errors.append("source bindings mismatch")
|
|
if record.get("adapter") != {
|
|
"exact_builtin_fake_facade_required": True,
|
|
"exact_builtin_fake_clock_required": True,
|
|
"precommitted_receipt_required": True,
|
|
"maximum_fake_steps": 1024, "maximum_receive_bytes": 65536,
|
|
"maximum_batch_bytes": 1035, "maximum_deadline_seconds": 10,
|
|
"network_import_present": False, "selector_import_present": False,
|
|
"dns_present": False, "real_clock_present": False,
|
|
"address_present": False, "cli_present": False,
|
|
"file_output_present": False,
|
|
"live_adapter_protocol_present": False, "target_retained": False,
|
|
"device_behavior_proven": False,
|
|
}:
|
|
errors.append("adapter boundary mismatch")
|
|
if record.get("lifecycle") != {
|
|
"create_count": 1, "nonblocking_before_connect": True,
|
|
"pending_connect_requires_write_ready": True,
|
|
"pending_connect_requires_zero_so_error": True,
|
|
"partial_write_loop": True, "zero_write_rejected": True,
|
|
"remote_eof_rejected": True,
|
|
"deadline_wins_readiness_race": True,
|
|
"deadline_only_completion": True,
|
|
"local_close_on_success": True, "local_close_on_failure": True,
|
|
"remote_cleanup_proven": False, "automatic_retry": False,
|
|
"reconnect": False, "resume": False,
|
|
}:
|
|
errors.append("lifecycle contract mismatch")
|
|
stops = record.get("hard_stops", {})
|
|
if set(stops) != {
|
|
"live_adapter_implementation", "socket_creation", "dns",
|
|
"target_retention", "connection", "device_request", "retry",
|
|
"device_action",
|
|
} or any(value is not True for value in stops.values()):
|
|
errors.append("hard stops mismatch")
|
|
authorizations = record.get("authorizations", {})
|
|
if set(authorizations) != AUTHORIZATION_FIELDS or any(
|
|
authorizations.get(field) is not False
|
|
for field in AUTHORIZATION_FIELDS):
|
|
errors.append("authorization fields are not exactly false")
|
|
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("decision") != {
|
|
"overall": STATUS, "dormant_fake_adapter_created": True,
|
|
"live_adapter_created": False, "live_adapter_allowed": False,
|
|
"device_action_allowed": False,
|
|
"phase10ad_offline_inactive_activation_design_allowed": True,
|
|
"next_step": "OFFLINE_NUMERIC_TARGET_AND_INACTIVE_ACTIVATION_CONTRACT",
|
|
}:
|
|
errors.append("decision mismatch")
|
|
if record.get("tests") != {
|
|
"chimera_gfx_ctest": "95_OF_95_PASS", "phase10ac_guardrails": 20,
|
|
"phase10ac_adapter_tests": 32, "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:
|
|
adapter = root / "tools/phase10ac_dormant_adapter.py"
|
|
tests = root / "tests/test_phase10ac_dormant_adapter.py"
|
|
if not exact_file(adapter, ADAPTER_SIZE, ADAPTER_SHA256):
|
|
errors.append("dormant adapter identity mismatch")
|
|
if not exact_file(tests, ADAPTER_TEST_SIZE, ADAPTER_TEST_SHA256):
|
|
errors.append("dormant adapter tests identity mismatch")
|
|
try:
|
|
source = adapter.read_text(encoding="utf-8")
|
|
tree = ast.parse(source)
|
|
except (OSError, SyntaxError, UnicodeError):
|
|
errors.append("dormant adapter cannot be parsed")
|
|
else:
|
|
imports = _imports(tree)
|
|
if imports & NETWORK_MODULES or "time" in imports:
|
|
errors.append("dormant adapter imports a live capability")
|
|
names = {
|
|
node.name for node in ast.walk(tree)
|
|
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
|
|
}
|
|
if {"connect", "send", "recv", "main"} & names:
|
|
errors.append("dormant adapter exposes a live API")
|
|
if re.search(r"(?<!\d)(?:\d{1,3}\.){3}\d{1,3}(?!\d)", source):
|
|
errors.append("dormant adapter embeds an address")
|
|
if "target_address" in source or "target_port" in source:
|
|
errors.append("dormant adapter contains target fields")
|
|
approval = (root / "docs/approvals/phase-1.0ac-dormant-adapter.md").read_text(
|
|
encoding="utf-8")
|
|
if "active=false" not in approval or \
|
|
"ps5_connection_authorized=false" not in approval or \
|
|
"automatic_retry=false" not in approval:
|
|
errors.append("approval record is not inert")
|
|
forbidden = list(root.glob("**/*phase10ac*.elf")) + \
|
|
list(root.glob("**/*phase-1.0ac*.elf"))
|
|
if forbidden:
|
|
errors.append("Phase-1.0AC target artifact exists")
|
|
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.0ac-dormant-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.0AC dormant-adapter validation passed")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|