205 lines
9.3 KiB
Python
205 lines
9.3 KiB
Python
#!/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())
|