#!/usr/bin/env python3 # SPDX-License-Identifier: GPL-3.0-or-later """Validate the bounded Phase-1.0U local shsrv inventory.""" from __future__ import annotations import argparse import hashlib import json from pathlib import Path import subprocess from typing import Any PHASE = "PHASE_1_0U_BOUNDED_LOCAL_SHSRV_INVENTORY" STATUS = "BOUNDED_LOCAL_INVENTORY_COMPLETE_NO_DEPLOYED_CANDIDATE_FOUND" START_COMMIT = "f5b0ff720dabf3ab745ad05aea4ea8edc9666ac7" CURRENT_COMMIT = "6f320637d56d344a0e7797753099e33238bbf146" CURRENT_TREE = "c26ce02b6c3ca4202993e039b3db7c28c353dee4" V07_COMMIT = "74287f5db6b20320efd7892d7b29cf438fe7cb98" V07_TREE = "7184968c702afe038551bf3228cc25f455388bb6" PACBREW_COMMIT = "c2abcfcb60f569128abd0e8e70ad03a67bee5ea7" 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", ) 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.0U manifest is not an object") return value def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() 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 def all_authorizations_false(record: dict[str, Any]) -> bool: return all(record.get(field) is False for field in AUTHORIZATION_FIELDS) def scope_is_bounded(record: dict[str, Any]) -> bool: return record == { "roots": [ "CHIMERA_GFX_REPOSITORY", "KNOWN_CHIMERA_SIBLING_REPOSITORIES", "SUPPLIED_CODEX_ATTACHMENTS", "EVIDENCED_USER_DOWNLOAD_DIRECTORY", ], "full_computer_scan_performed": False, "browser_database_scan_performed": False, "network_share_scan_performed": False, "internet_access_performed": False, "ps5_access_performed": False, } def methods_are_static(record: dict[str, Any]) -> bool: return record == { "case_insensitive_filename_search": True, "zip_entry_name_inventory": True, "zip_extraction_performed": False, "non_zip_unrelated_archives_inspected": False, "downloaded_or_local_code_executed": False, "host_sender_executed": False, "target_artifact_executed": False, "files_modified_by_inventory": False, } def result_is_scoped_absence(record: dict[str, Any]) -> bool: expected_false = ( "local_shsrv_target_candidate_found", "local_shsrv_package_found", "local_shsrv_package_receipt_found", "local_shsrv_transfer_log_found", "zip_entry_name_match_found", "operator_supplied_original_binary_found", "exact_device_path_found", "exact_device_hash_found", "global_absence_proven", ) return all(record.get(field) is False for field in expected_false) and \ record.get("classification") == "NO_CANDIDATE_IN_SCANNED_SCOPE" def references_are_non_deployed(records: Any) -> bool: return isinstance(records, list) and len(records) == 4 and all( item.get("deployed_identity") is False for item in records) def decision_is_blocked(record: dict[str, Any]) -> bool: return record == { "exact_deployed_shsrv_identity": "UNPROVEN", "local_direct_hash_path_available": False, "phase1_launch_context_experiment_allowed": False, "phase10v_inactive_collector_design_allowed": True, "live_collection_allowed": False, "device_action_allowed": False, "next_step": "OFFLINE_INACTIVE_ONE_SHOT_SANITIZING_COLLECTOR_DESIGN", } def exact_file(path: Path, size: int, digest: str) -> bool: return path.stat().st_size == size and sha256(path) == digest def validate( root: Path, shsrv_root: Path, shsrv_v07_root: Path, pacbrew_root: Path, ) -> list[str]: errors: list[str] = [] try: record = load_json( root / "manifests/retroarch/phase-1.0u-local-shsrv-inventory.json") 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 not scope_is_bounded(record.get("scope", {})): errors.append("inventory scope was broadened") if not methods_are_static(record.get("methods", {})): errors.append("inventory method became active or mutating") if not result_is_scoped_absence(record.get("results", {})): errors.append("scoped absence was promoted or changed") references = record.get("reference_objects", []) if not references_are_non_deployed(references): errors.append("reference object was promoted to deployed identity") if not all_authorizations_false(record.get("authorizations", {})): errors.append("authorization remains active or missing") if not decision_is_blocked(record.get("decision", {})): errors.append("device or launch-context decision is not blocked") performed = record.get("performed_actions", {}) if not performed or not all(value is False for value in performed.values()): errors.append("performed action is present") tests = record.get("tests", {}) if not ( tests.get("chimera_gfx_ctest") == "71_OF_71_PASS" and tests.get("phase10u_guardrails") == 16 and tests.get("safety_audit") == "PASS" and tests.get("secret_scan") == "PASS" and tests.get("network_required_by_tests") is False and tests.get("hardware_claim_from_host_test") is False ): errors.append("test evidence mismatch") expected_references = { "OFFICIAL_SHSRV_CURRENT_SOURCE": ( "REFERENCE_SOURCE_ONLY", CURRENT_COMMIT, CURRENT_TREE), "OFFICIAL_SHSRV_V07_SOURCE": ( "HISTORICAL_REFERENCE_SOURCE_ONLY", V07_COMMIT, V07_TREE), } indexed = {item["logical_name"]: item for item in references} for name, expected in expected_references.items(): item = indexed[name] if (item.get("classification"), item.get("commit"), item.get("tree")) != expected: errors.append(f"source reference mismatch: {name}") host = indexed["OFFICIAL_HOST_TELNET_WRAPPER"] if not exact_file( shsrv_root / "host/prospero-shsrv-shell", host["size"], host["sha256"]): errors.append("host wrapper identity mismatch") recipe = indexed["PACBREW_SHSRV_RECIPE"] if not exact_file( pacbrew_root / "shsrv/PKGBUILD", recipe["size"], recipe["sha256"]): errors.append("PacBrew recipe identity mismatch") if host.get("classification") != "HOST_WRAPPER_NOT_TARGET_BINARY" or \ host.get("executed") is not False: errors.append("host wrapper was promoted or executed") if recipe.get("classification") != \ "UNPINNED_RECIPE_NOT_PACKAGE_RECEIPT" or \ recipe.get("executed") is not False: errors.append("package recipe was promoted or executed") for repository, commit, tree_hash, name in ( (shsrv_root, CURRENT_COMMIT, CURRENT_TREE, "current shsrv"), (shsrv_v07_root, V07_COMMIT, V07_TREE, "shsrv v0.7"), ): if git(repository, "rev-parse", "HEAD").strip() != commit: errors.append(f"{name} commit mismatch") if git(repository, "rev-parse", "HEAD^{tree}").strip() != tree_hash: errors.append(f"{name} tree mismatch") if git(repository, "status", "--porcelain"): errors.append(f"{name} worktree is dirty") if git(pacbrew_root, "rev-parse", "HEAD").strip() != PACBREW_COMMIT: errors.append("PacBrew commit mismatch") if git(pacbrew_root, "status", "--porcelain"): errors.append("PacBrew worktree is dirty") host_source = (shsrv_root / "host/prospero-shsrv-shell").read_text( encoding="utf-8") if "telnet $SHSRV_HOST $SHSRV_PORT" not in host_source: errors.append("host wrapper behavior mismatch") recipe_source = (pacbrew_root / "shsrv/PKGBUILD").read_text( encoding="utf-8") for token in ( 'source=("git+https://github.com/ps5-payload-dev/shsrv.git")', "sha256sums=('SKIP')", "cp shsrv-ps5.elf", ): if token not in recipe_source: errors.append(f"package-recipe boundary missing: {token}") tracked = git(root, "ls-files").splitlines() if any(path.lower().endswith((".elf", ".self", ".sprx", ".pkg")) for path in tracked): errors.append("target artifact is tracked") except (OSError, RuntimeError, ValueError, KeyError, json.JSONDecodeError) 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("--shsrv-root", type=Path, required=True) parser.add_argument("--shsrv-v07-root", type=Path, required=True) parser.add_argument("--pacbrew-root", type=Path, required=True) args = parser.parse_args() errors = validate( args.root.resolve(), args.shsrv_root.resolve(), args.shsrv_v07_root.resolve(), args.pacbrew_root.resolve()) if errors: for error in errors: print(f"ERROR: {error}") return 1 print("Phase-1.0U bounded local shsrv inventory validation passed") return 0 if __name__ == "__main__": raise SystemExit(main())