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