#!/usr/bin/env python3 # SPDX-License-Identifier: GPL-3.0-or-later """Validate Phase-1.0Y offline shsrv framing evidence.""" 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_0Y_OFFLINE_SHSRV_FRAMING_AUDIT" STATUS = "SOURCE_FAMILY_FRAMING_PARTIAL_PROMPT_COMPLETION_UNPROVEN" START_COMMIT = "65c274f5ee723680ca6a576b74aa8b3c280978cd" V07 = "74287f5db6b20320efd7892d7b29cf438fe7cb98" V08 = "8f76139ee69df4b8cb7c3aee401f05bccb2c2a31" V09 = "2f2bc5501d40064c18c06f06f7b1f4cab756389b" V019 = "6f320637d56d344a0e7797753099e33238bbf146" MODEL_SIZE = 6678 MODEL_SHA256 = "5081898ec86be52900670be2f9949a20b9abb7781a6b04d5337178a8340775d4" MODEL_TEST_SIZE = 6485 MODEL_TEST_SHA256 = "802742450d65b237c0865e5820a8523131391988ca9eadd206766fb51693ca95" 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", ) NETWORK_MODULES = { "socket", "telnetlib", "urllib", "requests", "http", "ftplib", "asyncio", "selectors", } LIBTELNET_TAGS = ( "v0.9", "v0.10", "v0.10.1", "v0.11", "v0.12", "v0.13", "v0.13.1", "v0.14", "v0.15", "v0.16", "v0.16.1", "v0.16.2", "v0.17", "v0.18", "v0.18.1", "v0.18.2", "v0.19", ) 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.0Y 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.strip() def exact_file(path: Path, size: int, digest: str) -> bool: return path.stat().st_size == size and sha256(path) == digest 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, "framing_model_sha256": None, "source_family": None, "run_id": None, "target_address": None, "target_port": None, "commands": [], } def model_is_offline(record: dict[str, Any]) -> bool: return record == { "two_source_families": True, "maximum_bytes": 65536, "maximum_chunks": 256, "network_import_present": False, "cli_present": False, "command_formatter_present": False, "target_present": False, "file_output_present": False, "synthetic_negotiation_replies_only": True, "prompt_candidate_never_exact": True, "device_behavior_proven": False, } def prompt_remains_unproven(record: dict[str, Any]) -> bool: return ( record.get("prompt_source_shape") == "PWD_PLUS_DOLLAR_SPACE" and record.get("prompt_has_newline") is False and record.get("prompt_explicitly_flushed") is True and record.get("prompt_precedes_each_readline") is True and record.get("prompt_follows_synchronous_help") is True and record.get("prompt_follows_waited_stat_sum") is True and record.get("pwd_forced_overwrite") is False and record.get("wire_chunk_boundaries_defined") is False and record.get("short_write_completion_loop") is False and record.get("external_client_local_echo_defined") is False and record.get("terminal_shape_candidate_available") is True and record.get("exact_live_completion_proven") is False and len(record) == 12 ) def source_families_are_bound(record: dict[str, Any]) -> bool: return record == { "LEGACY_RAW_V07_V08": { "versions": ["v0.7", "v0.8"], "incoming_telnet_parser": False, "outgoing_nvt_translation": False, "telnet_controls_pass_to_shell": True, "server_side_echo": False, "proactive_negotiation": False, }, "LIBTELNET_NVT_V09_V019": { "versions": list(LIBTELNET_TAGS), "incoming_telnet_parser": True, "outgoing_nvt_translation": True, "empty_option_table": True, "will_reply": "IAC_DONT", "do_reply": "IAC_WONT", "initial_wont_dont_reply": "NONE", "server_side_echo": False, "proactive_negotiation": False, }, } def decision_is_offline_only(record: dict[str, Any]) -> bool: return record == { "official_source_families_identified": True, "offline_framing_model_complete": True, "exact_deployed_shsrv_identity": "UNPROVEN", "exact_live_prompt_completion": "UNPROVEN", "passive_no_negotiation_batch": "OFFLINE_DESIGN_CANDIDATE", "live_network_client_created": False, "live_client_implementation_allowed": False, "phase10z_offline_passive_batch_contract_allowed": True, "live_collection_allowed": False, "device_action_allowed": False, "next_step": "OFFLINE_PASSIVE_SOURCE_FAMILY_TOLERANT_BATCH_CONTRACT", } 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(NETWORK_MODULES) def validate(root: Path, shsrv_root: Path) -> list[str]: errors: list[str] = [] try: manifest = load_json( root / "manifests/retroarch/phase-1.0y-shsrv-framing.json") if manifest.get("phase") != PHASE or manifest.get("status") != STATUS: errors.append("phase/status mismatch") if manifest.get("start_commit") != START_COMMIT: errors.append("start commit mismatch") if not activation_is_inactive(manifest.get("activation", {})): errors.append("tracked activation is not inert") if not all_authorizations_false(manifest.get("authorizations", {})): errors.append("authorization remains active or missing") if not model_is_offline(manifest.get("offline_model", {})): errors.append("framing model gained live capability") if not prompt_remains_unproven(manifest.get("prompt_and_completion", {})): errors.append("prompt/completion evidence was promoted") if not source_families_are_bound(manifest.get("source_families", {})): errors.append("source-family contract mismatch") if not decision_is_offline_only(manifest.get("decision", {})): errors.append("offline-only decision mismatch") performed = manifest.get("performed_actions", {}) if not performed or not all(value is False for value in performed.values()): errors.append("performed network or device action is present") bindings = manifest.get("source_bindings", {}) expected_bindings = { "shsrv_v07_commit": V07, "shsrv_v08_commit": V08, "shsrv_v09_commit": V09, "shsrv_v019_current_commit": V019, "current_sh_c_size": 13278, "current_sh_c_sha256": "3c4b7f76efdd157436ed4b353ee1b550bf3ff9df17c147b4762b767982fc8253", "current_shsrv_c_size": 5293, "current_shsrv_c_sha256": "6ec71b4eb6c2bc1159f21568c1c9834f8aecac8d8881113bf09cf8981be19ee3", "current_libtelnet_c_size": 45904, "current_libtelnet_c_sha256": "64d0b8dc96b128aed30945393d112c2deba000fd00741a9b24b24ae5c596edeb", "current_libtelnet_h_size": 22302, "current_libtelnet_h_sha256": "8d3977ff3993480d18bee8ff91fc14e654b7fb0fdb6581cea91cdfb4120ecca8", "legacy_v07_sh_c_size": 8449, "legacy_v07_sh_c_sha256": "bf97bc6dd3f49345ad8da9a29b28a5d6bcde5e53a6f32c60a538d6e187c7e05a", "phase10x_transport_sha256": "568d7578482ecf2fcd9e29085b2eb9d8705fc699611508acdf22afd30f2ddd23", "phase10y_model_size": MODEL_SIZE, "phase10y_model_sha256": MODEL_SHA256, "phase10y_model_tests_size": MODEL_TEST_SIZE, "phase10y_model_tests_sha256": MODEL_TEST_SHA256, } if bindings != expected_bindings: errors.append("source binding record mismatch") expected_files = { "current_sh_c_size": (shsrv_root / "sh.c", 13278, "3c4b7f76efdd157436ed4b353ee1b550bf3ff9df17c147b4762b767982fc8253"), "current_shsrv_c_size": (shsrv_root / "shsrv.c", 5293, "6ec71b4eb6c2bc1159f21568c1c9834f8aecac8d8881113bf09cf8981be19ee3"), "current_libtelnet_c_size": (shsrv_root / "libtelnet.c", 45904, "64d0b8dc96b128aed30945393d112c2deba000fd00741a9b24b24ae5c596edeb"), "current_libtelnet_h_size": (shsrv_root / "libtelnet.h", 22302, "8d3977ff3993480d18bee8ff91fc14e654b7fb0fdb6581cea91cdfb4120ecca8"), } for size_field, (path, size, digest) in expected_files.items(): digest_field = size_field.replace("_size", "_sha256") if bindings.get(size_field) != size or \ bindings.get(digest_field) != digest or \ not exact_file(path, size, digest): errors.append(f"source identity mismatch: {path.name}") model_path = root / "tools/phase10y_shsrv_framing_model.py" test_path = root / "tests/test_phase10y_shsrv_framing_model.py" if not exact_file(model_path, MODEL_SIZE, MODEL_SHA256): errors.append("framing model identity mismatch") if not exact_file(test_path, MODEL_TEST_SIZE, MODEL_TEST_SHA256): errors.append("framing model test identity mismatch") if bindings.get("phase10y_model_size") != MODEL_SIZE or \ bindings.get("phase10y_model_sha256") != MODEL_SHA256 or \ bindings.get("phase10y_model_tests_size") != MODEL_TEST_SIZE or \ bindings.get("phase10y_model_tests_sha256") != MODEL_TEST_SHA256: errors.append("model manifest binding mismatch") if bindings.get("phase10x_transport_sha256") != \ "568d7578482ecf2fcd9e29085b2eb9d8705fc699611508acdf22afd30f2ddd23": errors.append("Phase-1.0X transport binding mismatch") if git(shsrv_root, "status", "--short"): errors.append("shsrv reference tree is dirty") if git(shsrv_root, "rev-parse", "HEAD") != V019: errors.append("current shsrv commit mismatch") for tag, commit in (("v0.7", V07), ("v0.8", V08), ("v0.9", V09), ("v0.19", V019)): if git(shsrv_root, "rev-parse", tag) != commit: errors.append(f"shsrv {tag} identity mismatch") for tag in ("v0.7", "v0.8"): if git(shsrv_root, "ls-tree", tag, "libtelnet.c"): errors.append(f"legacy {tag} unexpectedly contains libtelnet") if "libtelnet.c" not in git(shsrv_root, "ls-tree", "v0.9", "libtelnet.c"): errors.append("v0.9 libtelnet introduction is missing") for tag in LIBTELNET_TAGS: tagged_sh = git(shsrv_root, "show", f"{tag}:sh.c") tagged_tree = git(shsrv_root, "ls-tree", tag, "libtelnet.c") if not all(token in tagged_sh for token in ( "static const telnet_telopt_t telopts[]", "{-1, 0, 0}", "TELNET_FLAG_NVT_EOL", "telnet_send_text(state.telnet, buf, len)")) or \ "3d09e53bb2d9fe9498634f0998000146ac6bd214" not in tagged_tree: errors.append(f"source-family evidence mismatch: {tag}") current_sh = (shsrv_root / "sh.c").read_text(encoding="utf-8") libtelnet = (shsrv_root / "libtelnet.c").read_text(encoding="utf-8") for token in ( "static const telnet_telopt_t telopts[]", "{-1, 0, 0}", "TELNET_FLAG_NVT_EOL", "telnet_recv(state.telnet, buf, len)", "telnet_send_text(state.telnet, buf, len)", 'fprintf(stdout, "%s$ ", cwd ? cwd : "(null)")', ): if token not in current_sh: errors.append(f"current shsrv evidence missing: {token}") for token in ("_send_negotiate(telnet, TELNET_DONT, telopt)", "_send_negotiate(telnet, TELNET_WONT, telopt)", "_send(telnet, CRNUL, 2)", "_send(telnet, CRLF, 2)"): if token not in libtelnet: errors.append(f"libtelnet evidence missing: {token}") if not source_has_no_network_import(model_path): errors.append("framing model imports networking") model_source = model_path.read_text(encoding="utf-8") for forbidden in ("def main(", "argparse", "target_address", "2323"): if forbidden in model_source: errors.append(f"framing model contains forbidden token: {forbidden}") tests = manifest.get("tests", {}) if not ( tests.get("chimera_gfx_ctest") == "83_OF_83_PASS" and tests.get("phase10y_guardrails") == 18 and tests.get("phase10y_framing_model_tests") == 22 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") 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) parser.add_argument("--shsrv-root", type=Path, required=True) args = parser.parse_args() errors = validate(args.root.resolve(), args.shsrv_root.resolve()) if errors: for error in errors: print(f"ERROR: {error}") return 1 print("Phase-1.0Y offline shsrv-framing validation passed") return 0 if __name__ == "__main__": raise SystemExit(main())