#!/usr/bin/env python3 # SPDX-License-Identifier: GPL-3.0-or-later """Validate the inactive Phase-1.0T shsrv identity-collection gate.""" from __future__ import annotations import argparse import ast import hashlib import json from pathlib import Path import subprocess from typing import Any PHASE = "PHASE_1_0T_INACTIVE_SHSRV_IDENTITY_GATE" STATUS = "INACTIVE_METADATA_GATE_DESIGNED_EXACT_IDENTITY_UNAVAILABLE" START_COMMIT = "4c1944ef1412ed8a0bb18ae534244cde92db00ba" CURRENT_COMMIT = "6f320637d56d344a0e7797753099e33238bbf146" CURRENT_TREE = "c26ce02b6c3ca4202993e039b3db7c28c353dee4" V07_COMMIT = "74287f5db6b20320efd7892d7b29cf438fe7cb98" V07_TREE = "7184968c702afe038551bf3228cc25f455388bb6" CURRENT_COMMAND_HASH = \ "f41168292e205590bda1d243cdf727044e0af280a89fb0c070f4c5d6c92f2fd7" V07_COMMAND_HASH = \ "40313637116b532f3c7f9bebe2c23c0018fe7d4093840cf463a22ba0314ca021" 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", "app_termination_authorized", "system_remount_authorized", "automatic_retry", "reconnect_authorized", ) SOURCE_IDENTITIES = { "bundles/core/sum.c": ( 1965, "cd03227b0fbad40c342946b3ba93cc3119e94a1dd99a9c8dff04a2a9854bd93b"), "bundles/core/stat.c": ( 2183, "caf25c94f5edaacd0d824187edc47e9e804d9ed51177d89c125fbe959020e638"), "bundles/core/ps.c": ( 2761, "34d21ffef89341d29fbeec04d7b1361a15d3ee1c005681b16bd06c4f7ba6e5dc"), "builtin.c": ( 4182, "4b15b395562f62d547a8c0c2d27da570f9b7776263a0ad5dc23c0b238ba52643"), "sh.c": ( 13278, "3c4b7f76efdd157436ed4b353ee1b550bf3ff9df17c147b4762b767982fc8253"), } 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.0T 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_false(record: dict[str, Any]) -> bool: return all(record.get(field) is False for field in AUTHORIZATION_FIELDS) def activation_is_inactive(record: dict[str, Any]) -> bool: return record == { "active": False, "target_address": None, "listener_already_running_attested": False, "window": None, "exact_literal_path": None, "commands": [], "expires_at": None, } def connection_effects_are_complete(record: dict[str, Any]) -> bool: return record == { "accepted_socket_spawns_shell": True, "new_session_created": True, "pipes_created": True, "thread_created": True, "heap_allocations": True, "process_environment_changed": True, "model_queried": True, "serial_queried_and_transmitted": True, "firmware_queried": True, "temperature_queried_and_transmitted": True, "cpu_frequency_queried_and_transmitted": True, "filesystem_write_found": False, "autoload_change_found": False, "target_payload_launched": False, } def identity_is_non_exact(record: dict[str, Any]) -> bool: return record == { "greeting_compile_date_time": "METADATA_ONLY", "help_command_fingerprint": "SOURCE_FAMILY_FINGERPRINT_ONLY", "stat": "FILE_METADATA_ONLY", "sum": "BSD_ROTATE_16_WEAK_CHECKSUM_ONLY", "sha256_command_available": False, "binary_safe_file_read_available": False, "exact_deployed_identity_possible": False, "strongest_possible_classification": "WEAK_FILE_CORRELATION_ONLY", } def command_policy_is_fail_closed(record: dict[str, Any]) -> bool: forbidden = record.get("forbidden_mutating_or_launch_commands", []) return ( record.get("initial_candidate") == ["help"] and record.get("exact_path_candidates") == ["stat", "sum"] and record.get("fakeapp_metadata_candidate") == ["stat"] and all(command in forbidden for command in ( "hbldr", "launch", "exec", "rm", "mv", "mount", "touch")) and all(record.get(field) is False for field in ( "wildcards_allowed", "relative_paths_allowed", "pipes_allowed", "redirection_allowed", "multiple_commands_per_line_allowed", "path_guessing_allowed")) ) def sanitization_is_strict(record: dict[str, Any]) -> bool: return record == { "raw_transcript_persistence_allowed": False, "serial_value_retained": False, "model_value_retained": False, "temperature_values_retained": False, "cpu_frequency_value_retained": False, "unknown_paths_retained": False, "compile_metadata_retained": True, "firmware_retained": True, "approved_path_metadata_retained": True, } def future_windows_are_inactive(record: dict[str, Any]) -> bool: return set(record) == { "T1_MANUAL_HOST_FACTS", "T2_GREETING_AND_HELP", "T3_ONE_EXACT_PATH", "T4_FAKEAPP_METADATA", } and all(value == "DESIGNED_NOT_ACTIVE" for value in record.values()) def decision_is_inactive(record: dict[str, Any]) -> bool: return ( record.get("exact_deployed_identity_obtainable_from_existing_shsrv") is False and record.get("metadata_collection_design_complete") is True and record.get("network_client_created") is False and record.get("live_collection_allowed") is False and record.get("target_change_allowed") is False and record.get("device_action_allowed") is False and record.get("safe_next_steps") == [ "OFFLINE_INACTIVE_ONE_SHOT_SANITIZING_CLIENT_DESIGN", "MANUAL_HOST_ARTIFACT_INVENTORY", "HOST_OR_SOFTWARE_ONLY_INTEGRATION", ] ) def parser_is_offline(path: Path) -> bool: source = path.read_text(encoding="utf-8") tree = ast.parse(source) imports = { alias.name.split(".", 1)[0] for node in ast.walk(tree) if isinstance(node, (ast.Import, ast.ImportFrom)) for alias in node.names } return not imports.intersection({ "socket", "urllib", "requests", "http", "ftplib", "telnetlib"}) \ and "sys.stdin.read()" in source \ and "open(" not in source \ and "write_text(" not in source \ and "write_bytes(" not in source def file_identity_is_exact(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) -> list[str]: errors: list[str] = [] try: record = load_json( root / "manifests/retroarch/phase-1.0t-shsrv-identity-gate.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 all_false(record.get("authorizations", {})): errors.append("authorization remains active or is missing") if not activation_is_inactive(record.get("activation", {})): errors.append("activation record is not inert") if record.get("source_identities") != { "official_current_tag": "v0.19", "official_current_commit": CURRENT_COMMIT, "historical_tag": "v0.7", "historical_commit": V07_COMMIT, }: errors.append("source identity mismatch") if not connection_effects_are_complete( record.get("mandatory_connection_effects", {})): errors.append("connection effects were hidden or promoted") if not identity_is_non_exact(record.get("identity_capabilities", {})): errors.append("identity capability was promoted") fingerprints = record.get("command_fingerprints", {}) if fingerprints != { "v0_7": {"command_count": 44, "sha256": V07_COMMAND_HASH, "proves_exact_binary": False}, "v0_19": {"command_count": 50, "sha256": CURRENT_COMMAND_HASH, "proves_exact_binary": False}, }: errors.append("command fingerprint mismatch") if not command_policy_is_fail_closed(record.get("command_policy", {})): errors.append("command policy is not fail closed") if not sanitization_is_strict(record.get("sanitization", {})): errors.append("sanitization boundary is relaxed") if not future_windows_are_inactive(record.get("future_windows", {})): errors.append("a future collection window is active") if not decision_is_inactive(record.get("decision", {})): errors.append("decision is not inactive") 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") == "69_OF_69_PASS" and tests.get("phase10t_guardrails") == 20 and tests.get("phase10t_transcript_tests") == 12 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") 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(shsrv_root, "remote", "get-url", "origin").strip() != \ "https://github.com/ps5-payload-dev/shsrv.git": errors.append("shsrv origin is not official") for relative, identity in SOURCE_IDENTITIES.items(): if not file_identity_is_exact( shsrv_root / relative, identity[0], identity[1]): errors.append(f"source identity mismatch: {relative}") shell = (shsrv_root / "sh.c").read_text(encoding="utf-8") for token in ( 'printf("S/N:', 'printf("SoC temp:', 'printf("CPU temp:', 'printf("CPU freq:', "sh_greet();", "setsid();", "pthread_create(&trd", "telnet_init(", ): if token not in shell: errors.append(f"mandatory shell effect token missing: {token}") server = (shsrv_root / "shsrv.c").read_text(encoding="utf-8") if "accept(srvfd" not in server or \ "elfldr_spawn(connfd, connfd, -1, sh_elf, argv)" not in server: errors.append("connection-to-shell contract mismatch") builtin = (shsrv_root / "builtin.c").read_text(encoding="utf-8") if "rfork_thread(" not in builtin or "sceKernelSetBudget(0)" not in builtin: errors.append("forked builtin effects missing") sum_source = (shsrv_root / "bundles/core/sum.c").read_text( encoding="utf-8") if "open(name, O_RDONLY)" not in sum_source or \ "& 0xffff" not in sum_source or "sha256" in sum_source.lower(): errors.append("sum read-only weak-checksum contract mismatch") stat_source = (shsrv_root / "bundles/core/stat.c").read_text( encoding="utf-8") if "stat(path, &statbuf)" not in stat_source: errors.append("stat metadata contract mismatch") parser_path = root / "tools/phase10t_shsrv_transcript.py" if not parser_is_offline(parser_path): errors.append("transcript parser has network or persistence capability") approval = (root / "docs/approvals/phase-1.0t-shsrv-metadata-collection.md").read_text( encoding="utf-8") for token in ( "active=false", "ps5_connection_authorized=false", "device_request_authorized=false", "automatic_retry=false", "target_address=null", ): if token not in approval: errors.append(f"inactive approval token 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, SyntaxError) 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) args = parser.parse_args() errors = validate( args.root.resolve(), args.shsrv_root.resolve(), args.shsrv_v07_root.resolve()) if errors: for error in errors: print(f"ERROR: {error}") return 1 print("Phase-1.0T inactive shsrv identity gate validation passed") return 0 if __name__ == "__main__": raise SystemExit(main())