"""Bypass-Flag-Policy (tools nur im Kaefig) und M3-Quercheck (Exit UND JSONL UND last.txt).""" import tempfile import unittest from pathlib import Path from aria_router.adapter_contract import ResultStatus from aria_router.codex_adapter import CodexAdapter, validate_command_template from aria_router.config import ConfigurationError, parse_gateway_config from aria_router.isolation import ISO_EXEC_PATH from aria_router.runner import IsolatedRunner, RunResult, UnsafeCommandError, assert_safe_argv, mock_codex_events from tests.helpers import AdapterHarness, config_dict DANGER = ["/usr/bin/codex", "exec", "-s", "danger-full-access", "-"] CAGE = ["/usr/bin/systemd-run", "--user", "--scope", "--quiet", "--unit", "{unit}", "/usr/bin/env", "-i", "/usr/bin/python3", "-I", ISO_EXEC_PATH, "{cfg}", "--"] NONCAGE = ["/usr/bin/env", "-i"] class BypassPolicy(unittest.TestCase): def test_assert_safe_argv_rejects_bypass_by_default(self): with self.assertRaises(UnsafeCommandError) as c: assert_safe_argv(["/usr/bin/codex", "exec", "-s", "danger-full-access"]) self.assertEqual(c.exception.code, "FORBIDDEN_SANDBOX_FLAG") def test_assert_safe_argv_allows_bypass_when_flagged(self): assert_safe_argv(["/usr/bin/codex", "exec", "-s", "danger-full-access"], allow_bypass_sandbox=True) def test_yolo_always_rejected(self): with self.assertRaises(UnsafeCommandError): assert_safe_argv(["/usr/bin/codex", "--yolo"], allow_bypass_sandbox=True) def test_template_bypass_requires_cage(self): # mit echtem Kaefig erlaubt validate_command_template(DANGER, CAGE, allow_bypass_sandbox=True, require_cage=True) # ohne Kaefig abgelehnt with self.assertRaises(UnsafeCommandError) as c: validate_command_template(DANGER, NONCAGE, allow_bypass_sandbox=True, require_cage=True) self.assertEqual(c.exception.code, "TOOLS_PROFILE_REQUIRES_CAGE") # ohne allow_bypass (text) abgelehnt with self.assertRaises(UnsafeCommandError): validate_command_template(DANGER, CAGE, allow_bypass_sandbox=False) class ConfigProfilePolicy(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.root = Path(self._tmp.name) def tearDown(self): self._tmp.cleanup() def parse(self, codex): return parse_gateway_config(config_dict(self.root, {"codex": codex})) def test_tools_with_cage_accepted(self): cfg = self.parse({"profile": "tools", "command_template": DANGER, "isolation_wrapper": CAGE}) self.assertEqual(cfg.codex.profile, "tools") def test_tools_without_cage_rejected(self): with self.assertRaises(ConfigurationError): self.parse({"profile": "tools", "command_template": DANGER, "isolation_wrapper": NONCAGE}) def test_text_with_bypass_rejected(self): with self.assertRaises(ConfigurationError): self.parse({"profile": "text", "command_template": DANGER}) def test_tools_with_address_space_limit_rejected(self): # RLIMIT_AS bricht den V8-Host der Codex-Werkzeuge (SIGTRAP), siehe test_isolation_tools with self.assertRaises(ConfigurationError) as c: self.parse({"profile": "tools", "command_template": DANGER, "isolation_wrapper": CAGE, "resource_limits": {"address_space_bytes": 4 * 1024 ** 3, "cpu_seconds": 300}}) self.assertIn("address_space_bytes", str(c.exception)) cfg = self.parse({"profile": "tools", "command_template": DANGER, "isolation_wrapper": CAGE, "resource_limits": {"cpu_seconds": 300}}) self.assertIsNone(cfg.codex.resource_limits.address_space_bytes) def test_text_keeps_address_space_limit(self): cfg = self.parse({"resource_limits": {"address_space_bytes": 4 * 1024 ** 3}}) self.assertEqual(cfg.codex.resource_limits.address_space_bytes, 4 * 1024 ** 3) def test_unknown_profile_rejected(self): with self.assertRaises(ConfigurationError): self.parse({"profile": "full"}) def test_tools_protect_startup_defaults_true(self): # Owner-Entscheidung TG 12861: Default ist der Startschutz AN. cfg = self.parse({"profile": "tools", "command_template": DANGER, "isolation_wrapper": CAGE}) self.assertTrue(cfg.codex.tools_protect_startup) def test_tools_protect_startup_can_be_disabled(self): cfg = self.parse({"profile": "tools", "command_template": DANGER, "isolation_wrapper": CAGE, "tools_protect_startup": False}) self.assertFalse(cfg.codex.tools_protect_startup) def test_tools_protect_startup_rejects_non_bool(self): with self.assertRaises(ConfigurationError): self.parse({"profile": "tools", "command_template": DANGER, "isolation_wrapper": CAGE, "tools_protect_startup": "yes"}) class StubRunner(IsolatedRunner): def __init__(self, result): self._result = result def run(self, spec): return self._result class OutputFileCrossCheck(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.harness = AdapterHarness(Path(self._tmp.name)) def tearDown(self): self._tmp.cleanup() def _run(self, *, stdout, artifact): runner = StubRunner(RunResult(True, exit_code=0, stdout=stdout, artifact=artifact)) adapter = CodexAdapter(self.harness.verifier, runner=runner, require_output_file=True) job = self.harness.job("#public hallo") return adapter.execute(self.harness.clear(adapter, job), job) def test_success_when_jsonl_and_last_txt_match(self): # M3 r = self._run(stdout=mock_codex_events("OK"), artifact=b"OK") self.assertEqual((r.status, r.output), (ResultStatus.SUCCESS, "OK")) def test_fail_when_output_file_missing(self): # M3 r = self._run(stdout=mock_codex_events("OK"), artifact=b"") self.assertEqual((r.status, r.error_code), (ResultStatus.FAILED, "OUTPUT_FILE_MISSING")) def test_fail_when_output_file_mismatches_jsonl(self): # M3 r = self._run(stdout=mock_codex_events("OK"), artifact=b"SOMETHING ELSE") self.assertEqual((r.status, r.error_code), (ResultStatus.FAILED, "OUTPUT_MISMATCH")) def _run_tools(self, *, stdout, artifact): runner = StubRunner(RunResult(True, exit_code=0, stdout=stdout, artifact=artifact)) adapter = CodexAdapter(self.harness.verifier, runner=runner, command_template=DANGER, isolation_wrapper=CAGE, allow_sandbox_bypass=True, require_output_file=True) job = self.harness.job("#public hallo") return adapter.execute(self.harness.clear(adapter, job), job) TOOL_RUN = (b'{"type":"turn.started"}\n' b'{"type":"item.completed","item":{"type":"command_execution","command":"id"}}\n' b'{"type":"item.completed","item":{"type":"agent_message","text":"OK"}}\n' b'{"type":"turn.completed"}\n') def test_tools_profile_accepts_tool_events(self): # E2b r = self._run_tools(stdout=self.TOOL_RUN, artifact=b"OK") self.assertEqual((r.status, r.output), (ResultStatus.SUCCESS, "OK")) def test_text_profile_rejects_tool_events(self): r = self._run(stdout=self.TOOL_RUN, artifact=b"OK") self.assertEqual((r.status, r.error_code), (ResultStatus.FAILED, "TOOL_EVENT_SEEN")) def test_fail_when_jsonl_incomplete(self): # M3 (Exit ok, aber JSONL unvollstaendig) r = self._run(stdout=b'{"type":"turn.started"}\n', artifact=b"OK") self.assertEqual(r.status, ResultStatus.FAILED) def test_substring_last_txt_rejected(self): # Review #14 # JSONL letzte agent_message = "NICHT OK", last.txt = "OK" -> KEINE Teilstring-Akzeptanz. r = self._run(stdout=mock_codex_events("NICHT OK"), artifact=b"OK") self.assertEqual((r.status, r.error_code), (ResultStatus.FAILED, "OUTPUT_MISMATCH")) # Umkehrung: last.txt enthaelt die JSONL-Nachricht als Teilstring -> ebenfalls abgelehnt. r2 = self._run(stdout=mock_codex_events("OK"), artifact=b"OK, alles bestens") self.assertEqual((r2.status, r2.error_code), (ResultStatus.FAILED, "OUTPUT_MISMATCH")) # exakte Gleichheit (nach strip) bleibt Erfolg. r3 = self._run(stdout=mock_codex_events("NICHT OK"), artifact=b" NICHT OK\n") self.assertEqual((r3.status, r3.output), (ResultStatus.SUCCESS, "NICHT OK")) if __name__ == "__main__": unittest.main()