#!/usr/bin/env python3 # SPDX-License-Identifier: GPL-3.0-or-later """Host-only policy tests for Phase-1.0E.""" from __future__ import annotations import argparse import importlib.util from pathlib import Path import sys from typing import Callable def load_validator(path: Path): spec = importlib.util.spec_from_file_location("phase10e_validator", path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module def require(condition: bool, message: str) -> None: if not condition: raise RuntimeError(message) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--root", type=Path, required=True) args = parser.parse_args() root = args.root.resolve() validator = load_validator(root / "tools/validate_retroarch_phase10e.py") record = validator.load_json( root / "manifests/retroarch/phase-1.0e-result-channel.json" ) cases: list[tuple[str, Callable[[], None]]] = [] def case(name: str): def register(function: Callable[[], None]) -> Callable[[], None]: cases.append((name, function)) return function return register @case("01 consumed authorization is fully false") def _() -> None: require( validator.all_false( record["authorizations"], validator.AUTHORIZATION_FIELDS ), "consumed authorization remains active", ) @case("02 any later authorization true fails") def _() -> None: values = dict(record["authorizations"]) values["ps5_connection_authorized"] = True require( not validator.all_false(values, validator.AUTHORIZATION_FIELDS), "later connection authorization accepted", ) @case("03 reproducible exact artifact passes") def _() -> None: require(validator.reproducible_artifact(record["artifact"]), "artifact rejected") @case("04 differing clean build fails") def _() -> None: value = dict(record["artifact"]) value["clean_build_sha256"] = [validator.ARTIFACT_SHA256, "0" * 64] require(not validator.reproducible_artifact(value), "mismatch accepted") @case("05 renewed execution eligibility fails") def _() -> None: value = dict(record["artifact"]) value["execution_eligible"] = True require(not validator.reproducible_artifact(value), "renewed eligibility accepted") @case("06 bounded inherited transport passes") def _() -> None: require(validator.transport_is_bounded(record), "bounded transport rejected") @case("07 controlled route claim fails") def _() -> None: value = dict(record) value["loader_transport"] = dict(record["loader_transport"]) value["loader_transport"]["controlled_route_supported"] = True require(not validator.transport_is_bounded(value), "controlled route accepted") @case("08 target socket fails") def _() -> None: value = dict(record) value["loader_transport"] = dict(record["loader_transport"]) value["loader_transport"]["new_target_socket"] = True require(not validator.transport_is_bounded(value), "target socket accepted") @case("09 send retry fails") def _() -> None: value = dict(record) value["result_protocol"] = dict(record["result_protocol"]) value["result_protocol"]["target_write_retry"] = True require(not validator.transport_is_bounded(value), "send retry accepted") @case("10 short write retry fails") def _() -> None: value = dict(record) value["result_protocol"] = dict(record["result_protocol"]) value["result_protocol"]["short_write_retry"] = True require(not validator.transport_is_bounded(value), "short retry accepted") @case("11 second connection fails") def _() -> None: value = dict(record) value["host_contract"] = dict(record["host_contract"]) value["host_contract"]["connection_count"] = 2 require(not validator.transport_is_bounded(value), "second connect accepted") @case("12 reconnect fails") def _() -> None: value = dict(record) value["host_contract"] = dict(record["host_contract"]) value["host_contract"]["reconnect"] = True require(not validator.transport_is_bounded(value), "reconnect accepted") @case("13 trace overwrite fails") def _() -> None: value = dict(record) value["host_contract"] = dict(record["host_contract"]) value["host_contract"]["trace_overwrite"] = True require(not validator.transport_is_bounded(value), "trace overwrite accepted") @case("14 partial result success fails") def _() -> None: value = dict(record) value["host_contract"] = dict(record["host_contract"]) value["host_contract"]["partial_result_is_success"] = True require(not validator.transport_is_bounded(value), "partial success accepted") @case("15 prior observations stay bounded") def _() -> None: require(validator.observations_are_bounded(record), "observations rejected") @case("16 unread RUN B stage stays unclassified") def _() -> None: value = dict(record) value["prior_device_observations"] = dict(record["prior_device_observations"]) value["prior_device_observations"]["run_b"] = dict( record["prior_device_observations"]["run_b"] ) value["prior_device_observations"]["run_b"]["classification"] = "D12_PROVEN" require(not validator.observations_are_bounded(value), "invented D12 accepted") @case("17 W X separation passes") def _() -> None: require(validator.wx_closed(record["elf"]["program_headers"]), "W^X rejected") @case("18 RWX fails") def _() -> None: headers = [dict(item) for item in record["elf"]["program_headers"]] headers[0]["flags"] = "RWE" require(not validator.wx_closed(headers), "RWX accepted") @case("19 import delta is exactly send") def _() -> None: require(record["elf"]["undefined_symbol_delta_from_phase10d"] == ["send"], "delta changed") @case("20 SDK patch effects remain explicit") def _() -> None: effects = record["startup_effects"] require( effects["patch_init_reachable_from_start"] is True and effects["side_effect_free"] is False, "startup effect hidden", ) @case("21 approval is consumed and non-authorizing") def _() -> None: text = (root / "docs/retroarch/phase-1.0e-next-device-test.md").read_text( encoding="utf-8" ) require( "result_receive_authorized=false" in text and "consumed" in text.lower(), "consumed authorization boundary absent", ) @case("22 result document disclaims controlled route") def _() -> None: text = ( root / "docs/retroarch/phase-1.0e-inherited-result-channel.md" ).read_text(encoding="utf-8") require("stdio=-1" in text and "does **not** apply" in text, "route overclaim") @case("23 no target artifact is tracked") def _() -> None: tracked = validator.git(root, "ls-files").splitlines() require( not any(path.lower().endswith((".elf", ".self", ".sprx", ".pkg")) for path in tracked), "target artifact tracked", ) @case("24 Phase-1.0E hardware evidence is exact and bounded") def _() -> None: require( record["tests"]["hardware_evidence_from_phase10e"] is True and validator.device_run_is_exact(record), "RUN C evidence rejected", ) @case("25 RUN C has no retry or reconnect") def _() -> None: run = record["device_run"] require( run["retry_count"] == 0 and run["reconnect_count"] == 0, "repeat action", ) @case("26 missing D03 cannot be promoted to SDL") def _() -> None: value = dict(record) value["device_run"] = dict(record["device_run"]) value["device_run"]["last_proven_stage"] = "D03" require(not validator.device_run_is_exact(value), "invented D03 accepted") @case("27 missing terminal cannot be promoted") def _() -> None: value = dict(record) value["device_run"] = dict(record["device_run"]) value["device_run"]["terminal_stage"] = "D12" require(not validator.device_run_is_exact(value), "invented terminal accepted") @case("28 performed and forbidden actions remain separate") def _() -> None: actions = record["phase_actions"] require( all(actions[field] is True for field in validator.PERFORMED_ACTION_FIELDS) and validator.all_false(actions, validator.FORBIDDEN_ACTION_FIELDS), "action record is inconsistent", ) @case("29 offline stdout retention is bounded") def _() -> None: require( validator.offline_followup_is_bounded(record), "bounded offline follow-up rejected", ) @case("30 offline follow-up cannot add target or authority") def _() -> None: value = dict(record) value["offline_followup"] = dict(record["offline_followup"]) value["offline_followup"]["target_code_changed"] = True value["offline_followup"]["authorization_changed"] = True require( not validator.offline_followup_is_bounded(value), "target or authorization expansion accepted", ) failures: list[str] = [] for name, function in cases: try: function() except Exception as error: # noqa: BLE001 - isolated test report failures.append(f"{name}: {error}") if len(cases) != 30: failures.append(f"expected 30 cases, got {len(cases)}") for failure in failures: print(f"FAIL: {failure}", file=sys.stderr) if failures: return 1 print("Phase-1.0E guardrails passed (30 cases)") return 0 if __name__ == "__main__": raise SystemExit(main())