#!/usr/bin/env python3 # SPDX-License-Identifier: GPL-3.0-or-later """Validate the inactive Phase-1.0N one-shot runner contract.""" from __future__ import annotations import argparse import hashlib import json from pathlib import Path import subprocess from typing import Any PHASE = "PHASE_1_0N_WRITE_FREE_DEFAULTS_ONE_SHOT_RUNNER" STATUS = "OFFLINE_RUNNER_PREPARED_NO_DEVICE_AUTHORIZATION" ARTIFACT_SHA256 = "c99a0856309a357ad2667d89b4924e4063ad214cae09c8a419457b0732f583cd" RUNNER_SHA256 = "1d46510369349c1e75ef3b5f983a2ef6fa5896fdd4398788bf58708821d0385d" RUNNER_COMMIT = "606909706f91d7213751c245081333f56c2cce89" WIRE_STAGES = ( tuple(f"D{index:02d}" for index in range(13)) + tuple(f"I{index:02d}" for index in range(15)) + ("C1", "D13") ) AUTHORIZATION_FIELDS = ( "ps5_connection_authorized", "device_transfer_authorized", "device_execution_authorized", "result_receive_authorized", "installation_authorized", "autoload_authorized", "device_write_authorized", "automatic_retry", ) ACTION_FIELDS = ( "ps5_connected", "device_request_performed", "files_transferred", "target_execution_performed", "result_received_from_device", "target_build_performed", "target_artifact_created", "device_write_performed", "installation_performed", "autoload_performed", "retry_performed", "reconnect_performed", ) DELIVERABLES = ( "docs/retroarch/phase-1.0n-inactive-one-shot-runner.md", "docs/approvals/phase-1.0n-write-free-one-shot-template.md", "manifests/retroarch/phase-1.0n-write-free-one-shot-runner.json", "manifests/retroarch/phase-1.0n-one-shot-approval-template.json", "packaging/retroarch/phase10n/SHA256SUMS.txt", "tools/validate_retroarch_phase10n.py", "tests/test_retroarch_phase10n.py", ) def load_json(path: Path) -> dict[str, Any]: value = json.loads(path.read_text(encoding="utf-8")) if not isinstance(value, dict): raise ValueError(f"{path} is not a JSON object") return value def all_false(record: dict[str, Any], fields: tuple[str, ...]) -> bool: return all(record.get(field) is False for field in fields) def artifact_is_ineligible(record: dict[str, Any]) -> bool: return ( record.get("name") == "retroarch_ps5_write_diag.elf" and record.get("profile") == "write-diag" and record.get("size") == 1845208 and record.get("sha256") == ARTIFACT_SHA256 and record.get("source_unchanged_from_phase10m") is True and record.get("execution_eligible") is False and record.get("transfer_eligible") is False and record.get("installation_eligible") is False and record.get("tracked") is False ) def protocol_is_exact_and_inactive(record: dict[str, Any]) -> bool: return ( record.get("name") == "PHASE_1_0N_WRITE_FREE_DEFAULTS" and record.get("magic") == "CHD10J01" and record.get("version") == 1 and record.get("frame_size") == 64 and record.get("byte_order") == "BIG_ENDIAN" and record.get("wire_stages") == list(WIRE_STAGES) and record.get("c1_wire_index") == 28 and record.get("d13_wire_index") == 29 and record.get("terminal_stage") == "D12" and record.get("protocol_activation_authorized") is False and record.get("tracked_target") is None and record.get("tracked_port") is None and record.get("tracked_run_id") is None ) def runner_is_fail_closed(record: dict[str, Any]) -> bool: return ( record.get("implementation_available") is True and record.get("source_size") == 21179 and record.get("source_sha256") == RUNNER_SHA256 and record.get("protocol_selection") == "MANIFEST_ONLY" and record.get("free_protocol_selector") is False and record.get("active_manifest_required") is True and record.get("separate_untracked_approval_required") is True and record.get("actual_artifact_rehashed_before_socket") is True and record.get("attempt_receipt_required") is True and record.get("attempt_receipt_exclusive_create") is True and record.get("attempt_receipt_durable_fsync") is True and record.get("attempt_receipt_written_before_connect") is True and all(record.get(field) == 1 for field in ( "maximum_connections", "maximum_transfers", "maximum_executions", "maximum_result_receives", )) and record.get("receive_limit_bytes") == 65536 and all(record.get(field) is False for field in ( "retry", "reconnect", "resume", "trace_overwrite", )) and record.get("trace_exclusive_create") is True ) def approval_is_inactive(record: dict[str, Any]) -> bool: return ( record.get("phase") == PHASE and record.get("authorized") is False and record.get("consumed") is False and record.get("authorization_scope") == "EXACT_ONE_SHOT_PHASE_1_0N" and record.get("authorized_by") is None and record.get("approval_reference") is None and record.get("protocol_magic") == "CHD10J01" and record.get("run_id") is None and record.get("target") is None and record.get("port") is None and record.get("artifact_sha256") == ARTIFACT_SHA256 and all(record.get(field) == 0 for field in ( "connection_count", "transfer_count", "execution_count", "result_receive_count", )) and all(record.get(field) is False for field in ( "installation", "autoload", "device_write", "retry", "reconnect", "resume", "reboot", )) ) def git(root: Path, *args: str) -> str: result = subprocess.run( ["git", *args], cwd=root, capture_output=True, text=True, check=False) if result.returncode: raise RuntimeError(result.stderr.strip() or "git failed") return result.stdout.strip() def git_bytes(root: Path, *args: str) -> bytes: result = subprocess.run(["git", *args], cwd=root, capture_output=True, check=False) if result.returncode: raise RuntimeError(result.stderr.decode(errors="replace").strip() or "git failed") return result.stdout def validate(root: Path, retroarch_root: Path | None = None) -> list[str]: errors: list[str] = [] for relative in DELIVERABLES: if not (root / relative).is_file(): errors.append(f"missing deliverable: {relative}") try: record = load_json(root / "manifests/retroarch/phase-1.0n-write-free-one-shot-runner.json") approval = load_json(root / "manifests/retroarch/phase-1.0n-one-shot-approval-template.json") except (OSError, ValueError, json.JSONDecodeError) as error: return errors + [str(error)] if record.get("phase") != PHASE or record.get("status") != STATUS: errors.append("phase/status mismatch") commits = record.get("source_commits", {}) if commits.get("retroarch_runner") != RUNNER_COMMIT: errors.append("runner commit mismatch") if not all_false(record.get("authorizations", {}), AUTHORIZATION_FIELDS): errors.append("an authorization is active") if not all_false(record.get("phase_actions", {}), ACTION_FIELDS): errors.append("Phase-1.0N records a target or device action") if not artifact_is_ineligible(record.get("artifact", {})): errors.append("artifact is not exact and ineligible") if not protocol_is_exact_and_inactive(record.get("result_protocol", {})): errors.append("protocol is misindexed, mislabeled or active") if not runner_is_fail_closed(record.get("runner", {})): errors.append("runner is widened or incomplete") if not approval_is_inactive(approval): errors.append("tracked approval template is active or incomplete") activation = record.get("activation_requirements", {}) if (activation.get("current_requirements_satisfied") is not False or activation.get("phase10k_authority_reusable") is not False): errors.append("activation or consumed-authority boundary is widened") tests = record.get("tests", {}) if not ( tests.get("retroarch_duplex_python_cases") == 28 and tests.get("retroarch_phase10n_guardrails") == 5 and tests.get("chimera_gfx_ctest") == "56_OF_56_PASS" and tests.get("chimera_gfx_phase10n_guardrails") == 20 and tests.get("fake_socket_only") is True and tests.get("hardware_evidence_from_phase10n") is False ): errors.append("test evidence is incomplete or promoted") sums = (root / "packaging/retroarch/phase10n/SHA256SUMS.txt").read_text(encoding="utf-8") if ARTIFACT_SHA256 not in sums or RUNNER_SHA256 not in sums: errors.append("checksum record is incomplete") tracked = git(root, "ls-files").splitlines() if any(path.lower().endswith((".elf", ".self", ".sprx", ".pkg", ".map")) for path in tracked): errors.append("target artifact or map is tracked") if retroarch_root is not None: try: git(retroarch_root, "cat-file", "-e", f"{RUNNER_COMMIT}^{{commit}}") runner_bytes = git_bytes( retroarch_root, "show", f"{RUNNER_COMMIT}:tools/ps5_diag_duplex.py") runner = runner_bytes.decode("utf-8") if len(runner_bytes) != 21179 or hashlib.sha256(runner_bytes).hexdigest() != RUNNER_SHA256: errors.append("runner source identity mismatch") for token in ( 'WRITE_FREE_ACTIVE_PHASE = "PHASE_1_0N_WRITE_FREE_DEFAULTS_ONE_SHOT_RUNNER"', '"PHASE_1_0N_WRITE_FREE_DEFAULTS", WRITE_DIAG_MAGIC, WRITE_DIAG_STAGES', 'expected_scope = "EXACT_ONE_SHOT_PHASE_1_0N"', ): if token not in runner: errors.append(f"runner source omits {token}") if 'parser.add_argument("--protocol"' in runner: errors.append("runner source has a free protocol selector") except RuntimeError as error: errors.append(str(error)) return errors def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--root", type=Path, required=True) parser.add_argument("--retroarch-root", type=Path) args = parser.parse_args() errors = validate(args.root.resolve(), args.retroarch_root) if errors: for error in errors: print(f"ERROR: {error}") return 1 print("Phase-1.0N inactive one-shot runner validation passed") return 0 if __name__ == "__main__": raise SystemExit(main())