"""Echter Runner (SubprocessRunner): Isolationseigenschaften der Schnittstelle.""" import hashlib import json import os import secrets import signal import tempfile import time import unittest from pathlib import Path from aria_router.codex_adapter import DEFAULT_COMMAND_TEMPLATE, CodexAdapter, validate_command_template from aria_router.config import ConfigurationError, parse_gateway_config from aria_router.hardened_codex import decode_events from aria_router.runner import ( MockRunner, ResourceLimits, RunSpec, UnsafeCommandError, assert_safe_argv, validate_env, ) from tests.helpers import ( AdapterHarness, GuardedSubprocessRunner, RealBinaryGuardError, config_dict, fake_codex_template, ) BASE_ENV = {"PATH": "/usr/bin:/bin", "LANG": "C.UTF-8"} FORBIDDEN_VARIANTS = ( ("--sandbox", "danger-full-access"), ("--sandbox=danger-full-access",), ("-s", "DANGER-FULL-ACCESS"), ("-c", 'sandbox_mode="danger-full-access"'), ("--dangerously-bypass-approvals-and-sandbox",), ("--yolo",), ) def process_gone(pid: int) -> bool: try: with open(f"/proc/{pid}/stat", "r", encoding="utf-8") as handle: state = handle.read().rsplit(")", 1)[1].split()[0] return state in ("Z", "X") except FileNotFoundError: return True class SubprocessRunnerTests(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.root = Path(self._tmp.name) self.work = self.root / "work" self.work.mkdir() self.runner = GuardedSubprocessRunner(self.work) def tearDown(self): self._tmp.cleanup() def spec(self, mode, *extra, timeout=5, max_output=2 * 1024 * 1024, env=None, stdin=b"x", files=()): return RunSpec(tuple(fake_codex_template(mode, *extra)), stdin, dict(env or BASE_ENV), timeout, max_output, files) def test_env_not_inherited(self): token = "test-" + secrets.token_hex(12) # generierter Testwert, kein echtes Secret saved = {k: os.environ.get(k) for k in ("TELEGRAM_BOT_TOKEN", "ARIA_SUPABASE_SERVICE_KEY")} os.environ["TELEGRAM_BOT_TOKEN"] = token os.environ["ARIA_SUPABASE_SERVICE_KEY"] = token try: result = self.runner.run(self.spec("env")) finally: for key, value in saved.items(): if value is None: os.environ.pop(key, None) else: os.environ[key] = value self.assertTrue(result.ok, result) child_env = json.loads(decode_events(result.stdout).output) self.assertNotIn("TELEGRAM_BOT_TOKEN", child_env) self.assertNotIn("ARIA_SUPABASE_SERVICE_KEY", child_env) self.assertNotIn(token.encode(), result.stdout) self.assertLessEqual(set(child_env), set(BASE_ENV) | {"LC_CTYPE"}) self.assertEqual(child_env["PATH"], BASE_ENV["PATH"]) def test_env_allowlist_rejects_secret_names(self): for name in ("TELEGRAM_BOT_TOKEN", "OPENAI_API_KEY", "ARIA_SUPABASE_SERVICE_KEY", "LD_PRELOAD", "PYTHONPATH", "ANTHROPIC_API_KEY", "GITHUB_TOKEN", "DB_PASSWORD"): with self.subTest(name=name): with self.assertRaises(UnsafeCommandError) as ctx: validate_env({name: "x"}) self.assertEqual(ctx.exception.code, "FORBIDDEN_ENV") self.assertEqual(self.runner.run(self.spec("success", env={name: "x"})).error_code, "FORBIDDEN_ENV") with self.assertRaises(ConfigurationError): parse_gateway_config(config_dict(self.root, {"codex": {"env": {"TELEGRAM_BOT_TOKEN": "x"}}})) def test_timeout_kills_process(self): started = time.monotonic() result = self.runner.run(self.spec("sleep", "30", timeout=1)) self.assertEqual(result.error_code, "TIMEOUT") self.assertLess(time.monotonic() - started, 6) def test_crash_reported(self): result = self.runner.run(self.spec("crash")) self.assertEqual((result.ok, result.error_code, result.signal), (False, "CRASHED", signal.SIGKILL)) def test_nonzero_exit(self): result = self.runner.run(self.spec("exit2")) self.assertEqual((result.error_code, result.exit_code), ("NONZERO_EXIT", 2)) def test_output_limit_enforced_during_run(self): started = time.monotonic() result = self.runner.run(self.spec("flood", "100000", timeout=20, max_output=1000)) self.assertEqual(result.error_code, "OUTPUT_LIMIT") self.assertEqual(result.stdout, b"") self.assertLess(time.monotonic() - started, 5) def test_stderr_limit_enforced(self): result = self.runner.run(self.spec("flood_err", "100000", timeout=20, max_output=1000)) self.assertEqual(result.error_code, "OUTPUT_LIMIT") self.assertLessEqual(len(result.stderr_tail), 16 * 1024) def test_process_group_killed(self): pidfile = self.root / "grandchild.pid" result = self.runner.run(self.spec("grandchild", str(pidfile), timeout=2)) self.assertEqual(result.error_code, "TIMEOUT") pid = int(pidfile.read_text()) deadline = time.monotonic() + 5 while not process_gone(pid) and time.monotonic() < deadline: time.sleep(0.05) self.assertTrue(process_gone(pid), f"Enkelprozess {pid} läuft noch") def test_forbidden_flags_rejected(self): for variant in FORBIDDEN_VARIANTS: with self.subTest(variant=variant): argv = ("/usr/bin/codex", "exec") + variant with self.assertRaises(UnsafeCommandError) as ctx: assert_safe_argv(argv) self.assertEqual(ctx.exception.code, "FORBIDDEN_SANDBOX_FLAG") with self.assertRaises(UnsafeCommandError): validate_command_template(list(argv)) with self.assertRaises(UnsafeCommandError): validate_command_template(list(DEFAULT_COMMAND_TEMPLATE), ["/usr/bin/env", *variant]) with self.assertRaises(ConfigurationError): parse_gateway_config(config_dict(self.root, {"codex": {"command_template": list(argv)}})) with self.assertRaises(ConfigurationError): parse_gateway_config(config_dict(self.root, {"codex": {"isolation_wrapper": ["/usr/bin/env", *variant]}})) def test_adapter_refuses_forbidden_template(self): harness = AdapterHarness(self.root / "h") with self.assertRaises(UnsafeCommandError): CodexAdapter(harness.verifier, runner=MockRunner(), command_template=["/usr/bin/codex", "exec", "--sandbox", "danger-full-access", "-"]) def test_adapter_rechecks_template_before_run(self): # Template nach der Konstruktion manipuliert (vor der Freigabe, also im Digest gebunden): # die Prüfung unmittelbar vor dem Lauf blockiert, der (Mock-)Runner wird nie gerufen. from aria_router.adapter_contract import ResultStatus harness = AdapterHarness(self.root / "recheck") runner = MockRunner() adapter = CodexAdapter(harness.verifier, runner=runner, command_template=fake_codex_template()) adapter._argv = ("/usr/bin/codex", "exec", "--sandbox", "danger-full-access", "-") job = harness.job("#public x") result = adapter.execute(harness.clear(adapter, job), job) self.assertEqual((result.status, result.error_code), (ResultStatus.BLOCKED, "FORBIDDEN_SANDBOX_FLAG")) self.assertEqual(runner.calls, []) def test_runners_refuse_forbidden_argv_at_run_time(self): # Test-Interpreter + verbotenes Flag: ohne Runner-Prüfung liefe das Fake-Skript erfolgreich. argv = tuple(fake_codex_template("success")) + ("--dangerously-bypass-approvals-and-sandbox",) spec = RunSpec(argv, b"x", dict(BASE_ENV), 5) self.assertEqual(self.runner.run(spec).error_code, "FORBIDDEN_SANDBOX_FLAG") mock = MockRunner() self.assertEqual(mock.run(spec).error_code, "FORBIDDEN_SANDBOX_FLAG") self.assertEqual(mock.calls, []) def test_relative_command_rejected(self): with self.assertRaises(UnsafeCommandError) as ctx: validate_command_template(["codex", "exec", "-"]) self.assertEqual(ctx.exception.code, "COMMAND_NOT_ABSOLUTE") def test_unknown_placeholder_rejected(self): with self.assertRaises(UnsafeCommandError) as ctx: validate_command_template(["/usr/bin/codex", "exec", "{prompt}"]) self.assertEqual(ctx.exception.code, "UNKNOWN_PLACEHOLDER") def test_no_shell_interpretation(self): payload = "$(id) `whoami` ; echo pwned".encode() result = self.runner.run(self.spec("echo", stdin=payload)) self.assertEqual(decode_events(result.stdout).output, payload.decode()) def test_workspace_missing(self): result = GuardedSubprocessRunner(self.root / "missing").run(self.spec("success")) self.assertEqual(result.error_code, "WORKSPACE_UNAVAILABLE") def test_workdir_private_and_removed(self): result = self.runner.run(self.spec("cwd")) workdir = Path(decode_events(result.stdout).output) self.assertEqual(workdir.parent, self.work) self.assertFalse(workdir.exists()) def test_files_staged(self): result = self.runner.run(self.spec("files", files=(("01-a.txt", b"abc"), ("02-../../x", b"def")))) listing = decode_events(result.stdout).output.splitlines() self.assertIn("01-a.txt:" + hashlib.sha256(b"abc").hexdigest(), listing) self.assertIn("x:" + hashlib.sha256(b"def").hexdigest(), listing) def test_large_stdin_delivered(self): payload = os.urandom(150 * 1024).hex().encode() result = self.runner.run(self.spec("stdin_sha", stdin=payload, timeout=20)) self.assertEqual(decode_events(result.stdout).output, "STDIN_SHA256=" + hashlib.sha256(payload).hexdigest()) def test_stdin_write_failure_is_error(self): # Review #13: Prozess liest stdin nicht und beendet sofort erfolgreich; der Schreibfehler # mit ausstehenden Bytes MUSS als STDIN_FAILED gelten, nicht als Erfolg. payload = os.urandom(400 * 1024).hex().encode() # groesser als der Pipe-Puffer result = self.runner.run(self.spec("noread", stdin=payload, timeout=20)) self.assertFalse(result.ok) self.assertEqual(result.error_code, "STDIN_FAILED") @unittest.skipUnless(os.path.exists("/usr/bin/prlimit"), "prlimit fehlt") def test_cpu_limit_via_prlimit(self): runner = GuardedSubprocessRunner(self.work, resource_limits=ResourceLimits(cpu_seconds=1)) started = time.monotonic() result = runner.run(self.spec("cpu", timeout=30)) self.assertEqual(result.error_code, "CRASHED") self.assertIn(result.signal, (signal.SIGKILL, signal.SIGXCPU)) self.assertLess(time.monotonic() - started, 15) @unittest.skipUnless(os.path.exists("/usr/bin/prlimit"), "prlimit fehlt") def test_memory_limit_via_prlimit(self): runner = GuardedSubprocessRunner(self.work, resource_limits=ResourceLimits(address_space_bytes=512 * 1024 * 1024)) result = runner.run(self.spec("mem", str(1024 * 1024 * 1024), timeout=30)) self.assertEqual(result.error_code, "NONZERO_EXIT") unlimited = self.runner.run(self.spec("mem", str(64 * 1024 * 1024), timeout=30)) self.assertTrue(unlimited.ok) def test_guard_refuses_real_binaries(self): # Test-Infrastruktur: der Guard verweigert, BEVOR irgendetwas gestartet wird. for argv in (("/usr/bin/codex", "exec", "-"), ("/bin/sh", "-c", "true")): with self.subTest(argv=argv), self.assertRaises(RealBinaryGuardError): self.runner.run(RunSpec(argv, b"x", dict(BASE_ENV), 5)) def test_process_wide_model_cli_guard(self): import subprocess from tests import ModelCliGuardError # Nicht existierende Pfade: Ohne Schutz käme FileNotFoundError, nie ein Start. for program in ("/nonexistent/codex", "/nonexistent/claude", "/nonexistent/@openai/codex/bin/x.js"): with self.subTest(program=program), self.assertRaises(ModelCliGuardError): subprocess.Popen([program, "--version"]) def test_resource_limit_prefix(self): limits = ResourceLimits(address_space_bytes=1024, cpu_seconds=2, max_file_bytes=3, max_open_files=4) self.assertEqual(limits.prefix("/usr/bin/prlimit"), ("/usr/bin/prlimit", "--as=1024", "--cpu=2", "--fsize=3", "--nofile=4", "--")) if __name__ == "__main__": unittest.main()