from __future__ import annotations import hashlib import importlib.util import json import os import subprocess import sys from pathlib import Path import pytest from types import ModuleType from typing import Protocol, cast, runtime_checkable @runtime_checkable class _SupportProtocol(Protocol): def load_launcher(self, path: Path) -> object: ... def run_launcher(self, launcher: Path, evidence: Path, environment: dict[str, str], *, env: dict[str, str] | None = None, flags: tuple[str, ...] = ()) -> subprocess.CompletedProcess[str]: ... def verify_success_output(self, content: str, leaves: int) -> None: ... def prepare_mutated_artifacts(self, evidence: Path, destination: Path, names: tuple[str, ...], mutation: str) -> None: ... def bootstrap(self, launcher: object) -> bytes: ... def sealed_memfd(self, launcher: object, name: str, content: bytes) -> int: ... def source_snapshots(self, launcher: object) -> tuple[dict[str, list[object]], list[int]]: ... def source_closure(self, launcher: object) -> dict[str, str]: ... def validate_source_allowlist(self, launcher: object, pins: dict[str, str]) -> None: ... def snapshot_entry(self, entry: list[object], label: str) -> tuple[str, str, bool]: ... def strict_environment(self, launcher: object) -> dict[str, str]: ... def run_snapshot_child(self, launcher: object, entries: dict[str, list[object]], cli_descriptor: int, cli_digest: str, after_fork: object | None = None) -> int: ... def json_object(self, content: str | bytes, label: str) -> dict[str, object]: ... def required_object(self, value: object, label: str) -> dict[str, object]: ... def _load_support(path: Path) -> _SupportProtocol: spec = importlib.util.spec_from_file_location("task22_launcher_test_support", path) if spec is None or spec.loader is None: raise AssertionError("test support has no import loader") module: ModuleType = importlib.util.module_from_spec(spec) _ = spec.loader.exec_module(module) if not isinstance(module, _SupportProtocol): raise AssertionError("test support does not expose the required surface") return module EVIDENCE = Path(__file__).resolve().parent LAUNCHER = EVIDENCE / "task22-trainer-authority-removal-canonical.py" SUPPORT = _load_support(EVIDENCE / "task22_launcher_test_support.py") STRICT_ENV = {"HOME": "/home/cube", "LANG": "C.UTF-8", "LC_ALL": "C.UTF-8", "PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"} APPLICATION_MODULES = { "gateway.platforms.task22_callback_storage_adoption", "gateway.platforms.task22_protected_session_inventory", "gateway.platforms.task22_trainer_authority_recovery", "gateway.platforms.task22_trainer_authority_removal", "gateway.platforms.telegram_nutrition_onboarding_publication_outbox", } SYNTHETIC_PACKAGES = {"gateway", "gateway.platforms"} def test_exact_verify_only_isolated_command_succeeds() -> None: result = SUPPORT.run_launcher(LAUNCHER, EVIDENCE, STRICT_ENV, flags=("-I", "-B")) assert result.returncode == 0, result.stderr SUPPORT.verify_success_output(result.stdout, 177) @pytest.mark.parametrize("flags", (("-O",), ("-OO",))) def test_optimized_outer_interpreter_fails_closed(flags: tuple[str, ...]) -> None: result = SUPPORT.run_launcher(LAUNCHER, EVIDENCE, STRICT_ENV, flags=flags) assert result.returncode == 1 assert '"ready": false' in result.stderr assert "requires unoptimized isolated Python" in result.stderr @pytest.mark.parametrize( "name", ("PYTHONOPTIMIZE", "PYTHONPATH", "PYTHONHOME", "PYTHONINSPECT", "PYTHONWARNINGS", "PYTHONSTARTUP"), ) def test_inherited_python_environment_fails_before_verification(name: str, tmp_path: Path) -> None: value = str(tmp_path) if name in {"PYTHONPATH", "PYTHONHOME", "PYTHONSTARTUP"} else "1" result = SUPPORT.run_launcher(LAUNCHER, EVIDENCE, STRICT_ENV, env={**STRICT_ENV, name: value}, flags=("-I", "-B")) assert result.returncode == 1 assert '"ready": false' in result.stderr assert name in result.stderr @pytest.mark.parametrize("mutation", ("leaf", "status", "obsolete-symlink")) def test_exact_launcher_rejects_stale_successor_artifacts( tmp_path: Path, mutation: str ) -> None: SUPPORT.prepare_mutated_artifacts(EVIDENCE, tmp_path, ( LAUNCHER.name, "task22_launcher_runtime.py", "task22_dependency_closure.py", "task22_child_protocol.py", "task22_resource_ownership.py", "task22_lifecycle.py", "verify-dualcoach-trainer-removal-candidate.py", "test_task22_trainer_removal_launcher_hardening.py", "test_task22_launcher_venv_fork.py", "test_task22_launcher_complete_closure.py", "test_task22_pidfd_timeout.py", "test_task22_pidfd_contract.py", "test_task22_sealed_module_identity.py", "test_task22_parent_death_lifecycle.py", "task22_launcher_test_support.py", "dualcoach-trainer-removal-successor-manifest.json", "dualcoach-trainer-removal-successor-freeze-receipt.json", "dualcoach-task-22-pre-callback-adoption-evidence.json", ), mutation) result = subprocess.run( [sys.executable, "-I", "-B", str(tmp_path / LAUNCHER.name)], cwd=tmp_path, env=STRICT_ENV, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, ) assert result.returncode == 1 assert '"ready": false' in result.stderr assert "successor candidate verification failed" in result.stderr @pytest.mark.parametrize("mutation", ("leaf", "status", "obsolete-symlink")) @pytest.mark.parametrize("optimize", ("-O", "-OO")) def test_explicit_verifier_checks_survive_optimization( tmp_path: Path, mutation: str, optimize: str ) -> None: SUPPORT.prepare_mutated_artifacts(EVIDENCE, tmp_path, ( "verify-dualcoach-trainer-removal-candidate.py", "dualcoach-trainer-removal-successor-manifest.json", "dualcoach-trainer-removal-successor-freeze-receipt.json", "dualcoach-task-22-pre-callback-adoption-evidence.json", ), mutation) result = subprocess.run( [sys.executable, optimize, str(tmp_path / "verify-dualcoach-trainer-removal-candidate.py"), "--evidence-root", str(tmp_path)], cwd=tmp_path, env=STRICT_ENV, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, ) assert result.returncode != 0 assert "RuntimeError" in result.stderr def test_verify_only_never_opens_the_live_profile_root() -> None: profile_root = "/home/cube/.hermes/profiles/dualcoachtest" code = ( "import runpy,sys\n" f"forbidden={profile_root!r}\n" "def audit(event,args):\n" " if event == 'open' and args and isinstance(args[0],str) and args[0].startswith(forbidden): raise RuntimeError('live profile access denied')\n" "sys.addaudithook(audit)\n" f"sys.argv={[str(EVIDENCE / 'verify-dualcoach-trainer-removal-candidate.py'), '--evidence-root', str(EVIDENCE)]!r}\n" f"runpy.run_path({str(EVIDENCE / 'verify-dualcoach-trainer-removal-candidate.py')!r},run_name='__main__')\n" ) result = subprocess.run( [sys.executable, "-I", "-B", "-c", code], cwd=EVIDENCE, env=STRICT_ENV, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, ) assert result.returncode == 0, result.stderr assert '"leaves": 177' in result.stdout def test_verified_memfd_cannot_be_rebound_or_written() -> None: launcher = SUPPORT.load_launcher(LAUNCHER) original = b"print('verified')\n" descriptor = SUPPORT.sealed_memfd(launcher, "test-task22-snapshot", original) try: path = Path(f"/proc/self/fd/{descriptor}") assert path.read_bytes() == original with pytest.raises(OSError): _ = os.write(descriptor, b"replacement") assert hashlib.sha256(path.read_bytes()).hexdigest() == hashlib.sha256(original).hexdigest() finally: os.close(descriptor) def test_manifest_owned_execution_closure_is_sealed() -> None: launcher = SUPPORT.load_launcher(LAUNCHER) entries, descriptors = SUPPORT.source_snapshots(launcher) try: assert set(entries) == APPLICATION_MODULES | SYNTHETIC_PACKAGES for module, entry in entries.items(): path, expected, package = SUPPORT.snapshot_entry(entry, module) content = Path(path).read_bytes() assert hashlib.sha256(content).hexdigest() == expected if module in {"gateway", "gateway.platforms", "checkin_cli"}: assert package is True assert content == b"" finally: for descriptor in descriptors: os.close(descriptor) def test_static_closure_equals_explicit_hash_pinned_allowlist() -> None: launcher = SUPPORT.load_launcher(LAUNCHER) closure = SUPPORT.source_closure(launcher) assert set(closure) == APPLICATION_MODULES assert all(len(digest) == 64 for digest in closure.values()) def test_missing_required_and_extra_unused_allowlist_modules_fail() -> None: launcher = SUPPORT.load_launcher(LAUNCHER) closure = SUPPORT.source_closure(launcher) missing = dict(closure) del missing["gateway.platforms.task22_trainer_authority_recovery"] with pytest.raises(RuntimeError, match="missing required"): SUPPORT.validate_source_allowlist(launcher, missing) extra = {**closure, "gateway.platforms.nutrition_coaching": "0" * 64} with pytest.raises(RuntimeError, match="extra unused"): SUPPORT.validate_source_allowlist(launcher, extra) def test_every_included_module_loads_sealed_and_every_excluded_module_is_denied( tmp_path: Path, capfd: pytest.CaptureFixture[str] ) -> None: launcher = SUPPORT.load_launcher(LAUNCHER) entries, module_fds = SUPPORT.source_snapshots(launcher) manifest = SUPPORT.json_object( (EVIDENCE / "dualcoach-trainer-removal-successor-manifest.json").read_text(encoding="utf-8"), "successor manifest", ) leaves = manifest.get("candidate_files") if not isinstance(leaves, list): raise AssertionError("successor manifest leaves are invalid") inventory: set[str] = set() for index, item in enumerate(cast(list[object], leaves)): leaf = SUPPORT.required_object(item, f"successor manifest leaf {index}") relative = leaf.get("path") if isinstance(relative, str) and relative.endswith(".py"): inventory.add(relative[:-3].replace("/", ".").removesuffix(".__init__")) excluded = sorted( name for name in inventory - APPLICATION_MODULES - SYNTHETIC_PACKAGES if name == "gateway" or name.startswith("gateway.") or name == "checkin_cli" or name.startswith("checkin_cli.") ) hostile = tmp_path / "hostile" hostile.mkdir() for module in APPLICATION_MODULES: replacement = hostile.joinpath(*module.split(".")).with_suffix(".py") replacement.parent.mkdir(parents=True, exist_ok=True) written = replacement.write_text( "raise RuntimeError('canonical disk replacement executed')\n", encoding="utf-8" ) assert written > 0 cli = ( "import importlib,json,sys\n" f"sys.path.insert(0,{str(hostile)!r})\n" f"included={sorted(APPLICATION_MODULES)!r}\n" f"excluded={excluded!r}\n" "for name in included: importlib.import_module(name)\n" "denied=[]\n" "for name in excluded:\n" " try: importlib.import_module(name)\n" " except ImportError: denied.append(name)\n" " else: raise RuntimeError('excluded application module loaded: '+name)\n" "loaded={name for name in sys.modules if name in included}\n" "identity={name:[sys.modules[name].__file__,sys.modules[name].__spec__.origin,sys.modules[name].__loader__.path] for name in loaded}\n" "print(json.dumps({'loaded':sorted(loaded),'denied':denied,'identity':identity},sort_keys=True))\n" ).encode() cli_fd = SUPPORT.sealed_memfd(launcher, "closure-audit-cli", cli) try: status = SUPPORT.run_snapshot_child( launcher, entries, cli_fd, hashlib.sha256(cli).hexdigest() ) finally: for descriptor in (*module_fds, cli_fd): os.close(descriptor) assert status == 0 trace = SUPPORT.json_object(capfd.readouterr().out, "closure trace") assert trace["loaded"] == sorted(APPLICATION_MODULES) assert trace["denied"] == excluded identity = SUPPORT.required_object(trace.get("identity"), "closure identities") for raw_values in identity.values(): if not isinstance(raw_values, list): raise AssertionError("closure module identity is invalid") values: list[str] = [] for item in cast(list[object], raw_values): if not isinstance(item, str): raise AssertionError("closure module identity is invalid") values.append(item) assert len(set(values)) == 1 assert values[0].startswith("/proc/self/fd/") def test_exact_sealed_removal_and_canonical_loader_import_closure_executes(capfd: pytest.CaptureFixture[str]) -> None: launcher = SUPPORT.load_launcher(LAUNCHER) entries, module_fds = SUPPORT.source_snapshots(launcher) cli = b"""import sys\nimport gateway.platforms.task22_callback_storage_adoption as callback\nimport gateway.platforms.task22_protected_session_inventory as inventory\nimport gateway.platforms.task22_trainer_authority_recovery as recovery\nimport gateway.platforms.task22_trainer_authority_removal as removal\nimport gateway.platforms.telegram_nutrition_onboarding_publication_outbox as outbox\nassert sys.argv[1:] == [\"--execute\"]\nfor module in (callback, inventory, recovery, removal, outbox): print(module.__name__, module.__spec__.origin)\n""" cli_fd = SUPPORT.sealed_memfd(launcher, "closure-cli", cli) descriptors = (*module_fds, cli_fd) try: status = SUPPORT.run_snapshot_child(launcher, entries, cli_fd, hashlib.sha256(cli).hexdigest()) finally: for descriptor in descriptors: os.close(descriptor) assert status == 0 lines = capfd.readouterr().out.splitlines() assert [line.split(" ", 1)[0] for line in lines] == sorted(APPLICATION_MODULES) for line in lines: assert line.split(" ", 1)[1].startswith("/proc/self/fd/") def test_strict_child_environment_constructs_trusted_user_bus( monkeypatch: pytest.MonkeyPatch, ) -> None: launcher = SUPPORT.load_launcher(LAUNCHER) monkeypatch.setenv("XDG_RUNTIME_DIR", "/tmp/attacker-runtime") monkeypatch.setenv("DBUS_SESSION_BUS_ADDRESS", "unix:path=/tmp/attacker-bus") child = SUPPORT.strict_environment(launcher) runtime = f"/run/user/{os.geteuid()}" assert child["XDG_RUNTIME_DIR"] == runtime assert child["DBUS_SESSION_BUS_ADDRESS"] == f"unix:path={runtime}/bus" assert "/tmp/attacker" not in json.dumps(child) def test_disposable_execute_uses_only_exact_sealed_cli_and_module_snapshot(tmp_path: Path) -> None: launcher = SUPPORT.load_launcher(LAUNCHER) selected = tmp_path / "selected" escaped = tmp_path / "escaped" cli = b"import gateway.platforms.fixture as fixture\nfixture.run()\n" module = f"from pathlib import Path\ndef run(): Path({str(selected)!r}).write_text('snapshot')\n".encode() replacement = f"from pathlib import Path\ndef run(): Path({str(escaped)!r}).write_text('rebound')\n".encode() cli_fd = SUPPORT.sealed_memfd(launcher, "fixture-cli", cli) module_fd = SUPPORT.sealed_memfd(launcher, "fixture-module", module) gateway_fd = SUPPORT.sealed_memfd(launcher, "fixture-gateway", b"") platforms_fd = SUPPORT.sealed_memfd(launcher, "fixture-platforms", b"") entries: dict[str, list[object]] = { "gateway": [f"/proc/self/fd/{gateway_fd}", hashlib.sha256(b"").hexdigest(), True], "gateway.platforms": [f"/proc/self/fd/{platforms_fd}", hashlib.sha256(b"").hexdigest(), True], "gateway.platforms.fixture": [f"/proc/self/fd/{module_fd}", hashlib.sha256(module).hexdigest(), False], } live = tmp_path / "gateway" / "platforms" _ = live.mkdir(parents=True) descriptors = (cli_fd, module_fd, gateway_fd, platforms_fd) try: status = SUPPORT.run_snapshot_child( launcher, entries, cli_fd, hashlib.sha256(cli).hexdigest(), lambda: (live / "fixture.py").write_bytes(replacement), ) finally: for descriptor in descriptors: os.close(descriptor) assert status == 0 assert selected.read_text() == "snapshot" assert not escaped.exists()