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