383 lines
16 KiB
Python
383 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Validate the consumed Phase-1.0E inherited result-channel evidence."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
from typing import Any
|
|
|
|
|
|
PHASE = "PHASE_1_0E_INHERITED_RESULT_CHANNEL"
|
|
STATUS = "ONE_SHOT_DEVICE_TEST_COMPLETED_INCOMPLETE_BEFORE_D03"
|
|
RETROARCH_BRANCH = "codex/ps5-inherited-result-channel"
|
|
RETROARCH_COMMIT = "aed1a6e014d56ed25456b8b095955c7d41f7025d"
|
|
OFFLINE_FOLLOWUP_COMMIT = "f1391c3e6717ff4e2b869007fb9627c9edde52e4"
|
|
ARTIFACT_SOURCE_COMMIT = "b9fc037304a14199f35f8229edac26fa5c840509"
|
|
LOADER_COMMIT = "197623058f509eddde18868dafcb92fdcac66464"
|
|
DENYLIST_SHA256 = "e9603b0e3792781ad5b511afb22ef61e3d4fd4c5a16bf928f8609193bcd97783"
|
|
ARTIFACT_SHA256 = "1049c78099a60b472a3fb0e2999e3393b6ad76337a28532a7e53872e7772dedf"
|
|
ARTIFACT_SIZE = 1844880
|
|
MAP_SHA256 = "ae9739f6f578953bc8dc562bb55967ba587912d161b2d6787438450addec3b44"
|
|
TRACE_SHA256 = "4ff27a0eac48283cdc4c7ff964226def2689e808e3adea6594d0e77674a676f0"
|
|
TRACE_SIZE = 1795
|
|
FORBIDDEN_TARGET_IMPORTS = {"socket", "connect", "bind", "listen", "accept", "recv"}
|
|
AUTHORIZATION_FIELDS = (
|
|
"ps5_connection_authorized",
|
|
"device_transfer_authorized",
|
|
"device_execution_authorized",
|
|
"result_receive_authorized",
|
|
"installation_authorized",
|
|
"autoload_authorized",
|
|
"device_write_authorized",
|
|
"automatic_retry",
|
|
)
|
|
PERFORMED_ACTION_FIELDS = (
|
|
"ps5_connected",
|
|
"device_request_performed",
|
|
"files_transferred",
|
|
"target_execution_performed",
|
|
"result_received_from_device",
|
|
)
|
|
FORBIDDEN_ACTION_FIELDS = (
|
|
"device_write_performed",
|
|
"installation_performed",
|
|
"autoload_performed",
|
|
)
|
|
DELIVERABLES = (
|
|
"docs/retroarch/phase-1.0e-device-observations.md",
|
|
"docs/retroarch/phase-1.0e-inherited-result-channel.md",
|
|
"docs/retroarch/phase-1.0e-next-device-test.md",
|
|
"docs/approvals/phase-1.0e-one-shot-result-test.md",
|
|
"manifests/retroarch/phase-1.0e-result-channel.json",
|
|
"tools/validate_retroarch_phase10e.py",
|
|
"tests/test_retroarch_phase10e.py",
|
|
"packaging/retroarch/phase10e/SHA256SUMS.txt",
|
|
)
|
|
|
|
|
|
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 sha256_file(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.strip()
|
|
|
|
|
|
def all_false(record: dict[str, Any], fields: tuple[str, ...]) -> bool:
|
|
return all(record.get(field) is False for field in fields)
|
|
|
|
|
|
def reproducible_artifact(record: dict[str, Any]) -> bool:
|
|
hashes = record.get("clean_build_sha256", [])
|
|
map_hashes = record.get("clean_map_sha256", [])
|
|
return (
|
|
record.get("size") == ARTIFACT_SIZE
|
|
and record.get("sha256") == ARTIFACT_SHA256
|
|
and hashes == [ARTIFACT_SHA256, ARTIFACT_SHA256]
|
|
and record.get("linker_map_sha256") == MAP_SHA256
|
|
and map_hashes == [MAP_SHA256, MAP_SHA256]
|
|
and record.get("execution_eligible") is False
|
|
and record.get("transfer_eligible") is False
|
|
and record.get("installation_eligible") is False
|
|
and record.get("device_action_performed") is True
|
|
)
|
|
|
|
|
|
def wx_closed(headers: list[dict[str, Any]]) -> bool:
|
|
loads = [item for item in headers if item.get("type") == "LOAD"]
|
|
return len(loads) == 3 and all(
|
|
not ({"W", "E"} <= set(str(item.get("flags", "")))) for item in loads
|
|
)
|
|
|
|
|
|
def transport_is_bounded(record: dict[str, Any]) -> bool:
|
|
loader = record.get("loader_transport", {})
|
|
protocol = record.get("result_protocol", {})
|
|
host = record.get("host_contract", {})
|
|
return (
|
|
loader.get("raw_elf_exact_length_read") is True
|
|
and loader.get("payload_stdout_inherits_connection") is True
|
|
and loader.get("controlled_route_supported") is False
|
|
and loader.get("new_loader_change") is False
|
|
and loader.get("new_target_socket") is False
|
|
and loader.get("new_target_connection") is False
|
|
and protocol.get("frame_size") == 64
|
|
and protocol.get("target_write_attempts_per_stage") == 1
|
|
and protocol.get("target_write_retry") is False
|
|
and protocol.get("short_write_retry") is False
|
|
and protocol.get("target_import_added") == "send"
|
|
and set(protocol.get("target_forbidden_imports_absent", []))
|
|
== FORBIDDEN_TARGET_IMPORTS
|
|
and host.get("hash_before_socket_creation") is True
|
|
and host.get("exact_size_before_socket_creation") is True
|
|
and host.get("connection_count") == 1
|
|
and host.get("sendall_count") == 1
|
|
and host.get("shutdown_write_count") == 1
|
|
and host.get("receive_limit_bytes") == 65536
|
|
and host.get("retry") is False
|
|
and host.get("reconnect") is False
|
|
and host.get("resume") is False
|
|
and host.get("trace_overwrite") is False
|
|
and host.get("partial_result_is_success") is False
|
|
)
|
|
|
|
|
|
def observations_are_bounded(record: dict[str, Any]) -> bool:
|
|
prior = record.get("prior_device_observations", {})
|
|
run_a = prior.get("run_a", {})
|
|
run_b = prior.get("run_b", {})
|
|
return (
|
|
prior.get("evidence_class") == "OPERATOR_OBSERVED_ARTIFACT_BOUND"
|
|
and run_a.get("authorization_consumed") is True
|
|
and run_a.get("classification")
|
|
== "CRT_MAIN_AND_NOTIFICATION_PROVEN_ON_FW_9_60"
|
|
and run_b.get("authorization_consumed") is True
|
|
and run_b.get("classification")
|
|
== "PAYLOAD_NOTIFICATION_CODE_EXECUTED_STAGE_UNCLASSIFIED"
|
|
and run_b.get("console_remained_responsive") is True
|
|
)
|
|
|
|
|
|
def device_run_is_exact(record: dict[str, Any]) -> bool:
|
|
run = record.get("device_run", {})
|
|
frames = run.get("validated_frames", [])
|
|
return (
|
|
run.get("run_id") == "RUN_C"
|
|
and run.get("authorization_consumed") is True
|
|
and run.get("firmware") == "9.60"
|
|
and run.get("trace_tracked") is False
|
|
and run.get("trace_size") == TRACE_SIZE
|
|
and run.get("trace_sha256") == TRACE_SHA256
|
|
and run.get("connection_count") == 1
|
|
and run.get("sendall_count") == 1
|
|
and run.get("shutdown_write_count") == 1
|
|
and run.get("retry_count") == 0
|
|
and run.get("reconnect_count") == 0
|
|
and run.get("close_called") is True
|
|
and run.get("remote_eof_observed") is True
|
|
and run.get("timeout_observed") is False
|
|
and run.get("parser_errors") == []
|
|
and [frame.get("stage") for frame in frames] == ["D00", "D01", "D02"]
|
|
and [frame.get("sequence") for frame in frames] == [1, 2, 3]
|
|
and frames[2].get("raw0") == 0
|
|
and all(frame.get("notification_result") == 0 for frame in frames)
|
|
and all(frame.get("terminal") is False for frame in frames)
|
|
and run.get("terminal_stage") is None
|
|
and run.get("last_proven_stage") == "D02"
|
|
and run.get("classification")
|
|
== "REMOTE_PAYLOAD_OUTPUT_PROVEN_INCOMPLETE_BEFORE_D03"
|
|
and "D03_SDL_INIT_BEGIN" in run.get("does_not_prove", [])
|
|
and "SAFE_EXIT_OR_LOADER_CLEANUP" in run.get("does_not_prove", [])
|
|
)
|
|
|
|
|
|
def offline_followup_is_bounded(record: dict[str, Any]) -> bool:
|
|
followup = record.get("offline_followup", {})
|
|
return (
|
|
followup.get("repository") == "chimera-retroarch"
|
|
and followup.get("branch") == RETROARCH_BRANCH
|
|
and followup.get("commit") == OFFLINE_FOLLOWUP_COMMIT
|
|
and followup.get("classification")
|
|
== "D02_TO_D03_SOURCE_INTERVAL_BOUNDED_CAUSE_UNPROVEN"
|
|
and followup.get("bounded_raw_stream_retention") is True
|
|
and followup.get("bounded_ordinary_stdout_retention") is True
|
|
and followup.get("encoding") == "base64"
|
|
and followup.get("sha256_recorded") is True
|
|
and followup.get("maximum_pre_encoding_bytes") == 65536
|
|
and followup.get("exclusive_trace_creation") is True
|
|
and followup.get("consumed_manifest_rejected") is True
|
|
and followup.get("retry") is False
|
|
and followup.get("reconnect") is False
|
|
and followup.get("device_action_performed") is False
|
|
and followup.get("target_code_changed") is False
|
|
and followup.get("target_artifact_created") is False
|
|
and followup.get("authorization_changed") is False
|
|
)
|
|
|
|
|
|
def validate(
|
|
root: Path,
|
|
retroarch_root: Path | None = None,
|
|
loader_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.0e-result-channel.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")
|
|
if record.get("source_commit") != ARTIFACT_SOURCE_COMMIT:
|
|
errors.append("source commit mismatch")
|
|
if record.get("host_client_commit") != RETROARCH_COMMIT:
|
|
errors.append("host client commit mismatch")
|
|
authorizations = record.get("authorizations", {})
|
|
if not all_false(authorizations, AUTHORIZATION_FIELDS):
|
|
errors.append("consumed Phase-1.0E authorization must be fully false")
|
|
actions = record.get("phase_actions", {})
|
|
if not all(actions.get(field) is True for field in PERFORMED_ACTION_FIELDS):
|
|
errors.append("performed one-shot actions are missing")
|
|
if not all_false(actions, FORBIDDEN_ACTION_FIELDS):
|
|
errors.append("a forbidden device action was recorded")
|
|
if not observations_are_bounded(record):
|
|
errors.append("prior device observations are overclaimed or incomplete")
|
|
active = record.get("active_one_shot", {})
|
|
if not (
|
|
active.get("artifact_sha256") == ARTIFACT_SHA256
|
|
and active.get("artifact_size") == ARTIFACT_SIZE
|
|
and active.get("firmware") == "9.60"
|
|
and active.get("connection_count") == 1
|
|
and active.get("transfer_count") == 1
|
|
and active.get("execution_count") == 1
|
|
and active.get("result_receive_count") == 1
|
|
and active.get("timeout_seconds") == 75
|
|
and active.get("automatic_retry") is False
|
|
and active.get("reconnect") is False
|
|
and active.get("installation") is False
|
|
and active.get("autoload") is False
|
|
and active.get("device_write") is False
|
|
and active.get("consumed") is True
|
|
):
|
|
errors.append("one-shot contract is not exact or not consumed")
|
|
if not reproducible_artifact(record.get("artifact", {})):
|
|
errors.append("artifact is not exact, reproducible and post-run ineligible")
|
|
if not device_run_is_exact(record):
|
|
errors.append("RUN C result is missing, malformed or overclaimed")
|
|
if not offline_followup_is_bounded(record):
|
|
errors.append("offline D02-to-D03 follow-up is missing or unsafe")
|
|
|
|
elf = record.get("elf", {})
|
|
imports = set(elf.get("undefined_symbols", []))
|
|
if elf.get("rwx_load_segment_count") != 0 or not wx_closed(
|
|
elf.get("program_headers", [])
|
|
):
|
|
errors.append("ELF load layout is not W^X closed")
|
|
if elf.get("init_array_size") != 0 or elf.get("fini_array_size") != 0:
|
|
errors.append("constructor arrays are not empty")
|
|
if elf.get("tls") is not False:
|
|
errors.append("TLS must be absent")
|
|
if "send" not in imports or imports & FORBIDDEN_TARGET_IMPORTS:
|
|
errors.append("target import closure is not inherited-output-only")
|
|
if elf.get("undefined_symbol_delta_from_phase10d") != ["send"]:
|
|
errors.append("Phase-1.0D import delta must be exactly send")
|
|
if not transport_is_bounded(record):
|
|
errors.append("transport or host contract is not bounded")
|
|
|
|
effects = record.get("startup_effects", {})
|
|
if not (
|
|
effects.get("normal_sdk_crt") is True
|
|
and effects.get("patch_init_reachable_from_start") is True
|
|
and effects.get("kernel_copy_helpers_statically_linked") is True
|
|
and effects.get("side_effect_free") is False
|
|
):
|
|
errors.append("SDK startup effects are hidden or misclassified")
|
|
|
|
denylist = root / "manifests/artifact-denylist.json"
|
|
if not denylist.is_file() or sha256_file(denylist) != DENYLIST_SHA256:
|
|
errors.append("permanent denylist changed")
|
|
tracked = git(root, "ls-files").splitlines()
|
|
for relative in tracked:
|
|
if relative.lower().endswith((".elf", ".self", ".sprx", ".pkg")):
|
|
errors.append(f"tracked target artifact: {relative}")
|
|
path = root / relative
|
|
try:
|
|
text = path.read_text(encoding="utf-8")
|
|
except (OSError, UnicodeDecodeError):
|
|
continue
|
|
if "192.168.10." + "105" in text:
|
|
errors.append(f"tracked device address in {relative}")
|
|
|
|
if retroarch_root is not None:
|
|
ancestor = subprocess.run(
|
|
["git", "merge-base", "--is-ancestor", RETROARCH_COMMIT, "HEAD"],
|
|
cwd=retroarch_root,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
if ancestor.returncode:
|
|
errors.append("tested chimera-retroarch host client commit is not an ancestor")
|
|
followup_ancestor = subprocess.run(
|
|
["git", "merge-base", "--is-ancestor", OFFLINE_FOLLOWUP_COMMIT, "HEAD"],
|
|
cwd=retroarch_root,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
if followup_ancestor.returncode:
|
|
errors.append("offline stdout-capture commit is not an ancestor")
|
|
if git(retroarch_root, "branch", "--show-current") != RETROARCH_BRANCH:
|
|
errors.append("chimera-retroarch branch mismatch")
|
|
artifact = record["artifact"]
|
|
elf_path = retroarch_root / artifact["local_relative_path"]
|
|
map_path = retroarch_root / artifact["linker_map_relative_path"]
|
|
if not elf_path.is_file() or (
|
|
elf_path.stat().st_size != ARTIFACT_SIZE
|
|
or sha256_file(elf_path) != ARTIFACT_SHA256
|
|
):
|
|
errors.append("local result ELF missing or changed")
|
|
if not map_path.is_file() or sha256_file(map_path) != MAP_SHA256:
|
|
errors.append("local result linker map missing or changed")
|
|
run = record["device_run"]
|
|
trace_path = retroarch_root / run["trace_relative_path"]
|
|
if not trace_path.is_file() or (
|
|
trace_path.stat().st_size != TRACE_SIZE
|
|
or sha256_file(trace_path) != TRACE_SHA256
|
|
):
|
|
errors.append("local ignored RUN C trace missing or changed")
|
|
if loader_root is not None:
|
|
if git(loader_root, "rev-parse", "HEAD") != LOADER_COMMIT:
|
|
errors.append("hardened elfldr HEAD mismatch")
|
|
if git(loader_root, "status", "--porcelain"):
|
|
errors.append("hardened elfldr tree is dirty")
|
|
return errors
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--root", type=Path, required=True)
|
|
parser.add_argument("--retroarch-root", type=Path)
|
|
parser.add_argument("--loader-root", type=Path)
|
|
args = parser.parse_args()
|
|
errors = validate(
|
|
args.root.resolve(),
|
|
args.retroarch_root.resolve() if args.retroarch_root else None,
|
|
args.loader_root.resolve() if args.loader_root else None,
|
|
)
|
|
if errors:
|
|
for error in errors:
|
|
print(f"ERROR: {error}", file=sys.stderr)
|
|
return 1
|
|
print("Phase-1.0E inherited result-channel validation passed")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|