282 lines
11 KiB
Python
282 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Validate Phase-1.0W self-review and inactive client architecture."""
|
|
|
|
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_0W_INACTIVE_SHSRV_CLIENT_ARCHITECTURE"
|
|
STATUS = "INACTIVE_CLIENT_ARCHITECTURE_COMPLETE_NETWORK_IMPLEMENTATION_BLOCKED"
|
|
START_COMMIT = "665d02995fdb72080a20cd53c92265b1e24afb1c"
|
|
PARSER_SHA256 = "4701a057a98b4874e49e1bcf11db9a9a3a105e48f2c25e42796bff10f238f7c2"
|
|
COLLECTOR_SIZE = 7429
|
|
COLLECTOR_SHA256 = "f8a306dafee5d135919bec5afda789dd741e57f39803b7683fb8747c186db25c"
|
|
POLICY_SIZE = 6997
|
|
POLICY_SHA256 = "747d23c88f2722e8e8846599c3ac1dae3826eb3fca881caaad36b251f30f3592"
|
|
FAKE_SIZE = 1870
|
|
FAKE_SHA256 = "e2f074e844102177f45fb0f97254108ee641db00931bb037574dee82111f316f"
|
|
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.0W 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 activation_is_inactive(record: dict[str, Any]) -> bool:
|
|
return record == {
|
|
"active": False,
|
|
"policy_sha256": None,
|
|
"collector_sha256": None,
|
|
"run_id": None,
|
|
"target_address": None,
|
|
"target_port": None,
|
|
"window": None,
|
|
"exact_literal_path": None,
|
|
"commands": [],
|
|
"deadline_seconds": None,
|
|
"expires_at": None,
|
|
}
|
|
|
|
|
|
def architecture_has_no_transport(record: dict[str, Any]) -> bool:
|
|
return record == {
|
|
"policy_returns_data_only": True,
|
|
"policy_cli_present": False,
|
|
"network_transport_present": False,
|
|
"socket_import_present": False,
|
|
"dns_lookup_present": False,
|
|
"connect_present": False,
|
|
"command_formatter_present": False,
|
|
"command_sender_present": False,
|
|
"receive_present": False,
|
|
"filesystem_output_present": False,
|
|
"fake_transport_only": True,
|
|
}
|
|
|
|
|
|
def future_contract_is_bounded(record: dict[str, Any]) -> bool:
|
|
return record == {
|
|
"source_bound_port": 2323,
|
|
"maximum_connections": 1,
|
|
"maximum_deadline_seconds": 10,
|
|
"maximum_approval_lifetime_minutes": 15,
|
|
"allowed_windows": {
|
|
"T2_GREETING_AND_HELP": ["help"],
|
|
"T3_ONE_EXACT_PATH": ["stat", "sum"],
|
|
},
|
|
"dual_record_exact_match_required": True,
|
|
"exact_record_shape_required": True,
|
|
"policy_hash_dual_match_required": True,
|
|
"operator_attestation_required": True,
|
|
"listener_attestation_required": True,
|
|
"spawned_shell_effects_acceptance_required": True,
|
|
"serial_query_acceptance_required": True,
|
|
"telemetry_query_acceptance_required": True,
|
|
"unproven_memory_erasure_acceptance_required": True,
|
|
"automatic_retry": False,
|
|
"reconnect": False,
|
|
"resume": False,
|
|
"fallback": False,
|
|
}
|
|
|
|
|
|
def live_components_remain_missing(record: dict[str, Any]) -> bool:
|
|
return set(record) == {
|
|
"consumed_attempt_receipt", "exclusive_sanitized_output",
|
|
"monotonic_deadline", "host_clock_acquisition", "network_transport",
|
|
"telnet_reply_contract", "prompt_completion_contract",
|
|
"bounded_receive_integration", "deterministic_close",
|
|
"error_path_redaction_review",
|
|
} and all(value is True for value in record.values())
|
|
|
|
|
|
def decision_is_offline_only(record: dict[str, Any]) -> bool:
|
|
return record == {
|
|
"self_review_complete": True,
|
|
"review_findings_remediated": True,
|
|
"offline_client_architecture_complete": True,
|
|
"exact_deployed_shsrv_identity": "UNPROVEN",
|
|
"live_network_client_created": False,
|
|
"live_client_implementation_allowed": False,
|
|
"phase10x_offline_inactive_transport_implementation_allowed": True,
|
|
"live_collection_allowed": False,
|
|
"launch_context_experiment_allowed": False,
|
|
"device_action_allowed": False,
|
|
"next_step": "OFFLINE_INACTIVE_FAKE_SOCKET_TRANSPORT_IMPLEMENTATION",
|
|
}
|
|
|
|
|
|
def source_has_no_network_import(path: Path) -> bool:
|
|
tree = ast.parse(path.read_text(encoding="utf-8"))
|
|
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", "telnetlib", "urllib", "requests", "http", "ftplib"})
|
|
|
|
|
|
def exact_file(path: Path, size: int, digest: str) -> bool:
|
|
return path.stat().st_size == size and sha256(path) == digest
|
|
|
|
|
|
def validate(root: Path) -> list[str]:
|
|
errors: list[str] = []
|
|
try:
|
|
record = load_json(
|
|
root / "manifests/retroarch/phase-1.0w-inactive-client-architecture.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 activation_is_inactive(record.get("activation", {})):
|
|
errors.append("tracked activation is not inert")
|
|
if not architecture_has_no_transport(record.get("architecture", {})):
|
|
errors.append("network or persistence capability was introduced")
|
|
if not future_contract_is_bounded(record.get("future_contract", {})):
|
|
errors.append("future contract was relaxed")
|
|
if not live_components_remain_missing(
|
|
record.get("missing_live_components", {})):
|
|
errors.append("missing live component was incorrectly claimed complete")
|
|
if not all_authorizations_false(record.get("authorizations", {})):
|
|
errors.append("authorization remains active or missing")
|
|
if not decision_is_offline_only(record.get("decision", {})):
|
|
errors.append("live or device decision is enabled")
|
|
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")
|
|
bindings = record.get("source_bindings", {})
|
|
if bindings != {
|
|
"phase10t_parser_sha256": PARSER_SHA256,
|
|
"remediated_phase10v_collector_size": COLLECTOR_SIZE,
|
|
"remediated_phase10v_collector_sha256": COLLECTOR_SHA256,
|
|
"phase10w_policy_size": POLICY_SIZE,
|
|
"phase10w_policy_sha256": POLICY_SHA256,
|
|
"fake_transport_size": FAKE_SIZE,
|
|
"fake_transport_sha256": FAKE_SHA256,
|
|
}:
|
|
errors.append("source bindings mismatch")
|
|
tests = record.get("tests", {})
|
|
if not (
|
|
tests.get("chimera_gfx_ctest") == "77_OF_77_PASS"
|
|
and tests.get("phase10w_guardrails") == 18
|
|
and tests.get("phase10w_policy_tests") == 25
|
|
and tests.get("remediated_phase10v_model_tests") == 21
|
|
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")
|
|
|
|
identities = (
|
|
(root / "tools/phase10t_shsrv_transcript.py", None, PARSER_SHA256),
|
|
(root / "tools/phase10v_shsrv_collector_model.py",
|
|
COLLECTOR_SIZE, COLLECTOR_SHA256),
|
|
(root / "tools/phase10w_shsrv_client_policy.py",
|
|
POLICY_SIZE, POLICY_SHA256),
|
|
(root / "tests/phase10w_fake_transport.py", FAKE_SIZE, FAKE_SHA256),
|
|
)
|
|
for path, size, digest in identities:
|
|
if (size is not None and path.stat().st_size != size) or \
|
|
sha256(path) != digest:
|
|
errors.append(f"source identity mismatch: {path.name}")
|
|
policy_path = root / "tools/phase10w_shsrv_client_policy.py"
|
|
fake_path = root / "tests/phase10w_fake_transport.py"
|
|
if not source_has_no_network_import(policy_path) or \
|
|
not source_has_no_network_import(fake_path):
|
|
errors.append("policy or fake transport imports networking")
|
|
policy_source = policy_path.read_text(encoding="utf-8")
|
|
for token in (
|
|
"SOURCE_BOUND_PORT = 2323", "MAX_DEADLINE_SECONDS = 10",
|
|
'"T2_GREETING_AND_HELP": ("help",)',
|
|
'"T3_ONE_EXACT_PATH": ("stat", "sum")',
|
|
"@dataclass(frozen=True)",
|
|
):
|
|
if token not in policy_source:
|
|
errors.append(f"policy token missing: {token}")
|
|
if "def main(" in policy_source or "argparse" in policy_source:
|
|
errors.append("policy unexpectedly exposes a CLI")
|
|
phase_v = load_json(
|
|
root / "manifests/retroarch/phase-1.0v-inactive-shsrv-collector.json")
|
|
remediation = phase_v.get("review_remediation", {})
|
|
if not remediation or \
|
|
remediation.get("physical_memory_erasure_proven") is not False or \
|
|
not all(value is True for key, value in remediation.items()
|
|
if key != "physical_memory_erasure_proven"):
|
|
errors.append("Phase-1.0V remediation evidence mismatch")
|
|
approval = (
|
|
root / "docs/approvals/phase-1.0w-shsrv-client.md").read_text(
|
|
encoding="utf-8")
|
|
for token in (
|
|
"active=false", "target_address=null", "target_port=null",
|
|
"commands=[]", "ps5_connection_authorized=false",
|
|
"device_request_authorized=false", "automatic_retry=false",
|
|
):
|
|
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, 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)
|
|
args = parser.parse_args()
|
|
errors = validate(args.root.resolve())
|
|
if errors:
|
|
for error in errors:
|
|
print(f"ERROR: {error}")
|
|
return 1
|
|
print("Phase-1.0W inactive client-architecture validation passed")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|