"""E44 Option 2 (Kais TG 12923, 25.10.2026): codex.tools_write_allowlist. Validierung (Funktion, Konfiguration, Runner, Profilbau) und Regelaufbau des tools-Profils: das Home ist nur noch LESBAR, geschrieben wird nur in den ephemeren Lauf-Verzeichnissen und in ausdruecklich freigegebenen Projektordnern. Synthetischer Home-Baum, kein Kaefig-Start (echte Landlock-Laeufe stehen in test_isolation_tools.WriteAllowlistRealHome).""" import os import tempfile import unittest from pathlib import Path from aria_router.config import ConfigurationError, parse_gateway_config from aria_router.gateway import Gateway from aria_router.isolation import ISO_EXEC_PATH from aria_router.isolation import profiles as P from aria_router.isolation.caged_runner import CagedCodexRunner from aria_router.runner import RunSpec from tests.helpers import config_dict, write_key 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}", "--"] def touch(path, body="x"): os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w", encoding="utf-8") as fh: fh.write(body) class SyntheticHomeBase(unittest.TestCase): """Home-Baum mit Projektordnern, einem kostenwache-Muster (melde.sh = Cron-Ziel, kostenwache.py daneben), Startschutz-Bereichen, Caches, Credentials, einer aria-v3-Kopie und Symlinks.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.base = self._tmp.name self.H = os.path.join(self.base, "home") for rel, body in { "projekte/app/main.py": "print('app')\n", "projekte/app/src/util.py": "x = 1\n", "projekte/app/.env": "SECRET=1\n", "projekte/app/.git/hooks/pre-commit": "#!/bin/sh\n", "projekte/app/.git/config": "[core]\n", "projekte/aws-kostenwache/melde.sh": "#!/bin/sh\npython3 kostenwache.py\n", "projekte/aws-kostenwache/kostenwache.py": "print('kosten')\n", "projekte/cronproj/run.py": "print('cron')\n", "projekte/runs/keep.txt": "x", "brain/note.md": "n", "scripts/wrapper.sh": "#!/bin/sh\n", "scripts/sub/x.sh": "#!/bin/sh\n", "state/run-heartbeat.conf": "CONF=1\n", "gwstate/sub/budget.json": "{}", ".cache/pip/wheel.whl": "w", ".cargo/registry/src/lib.rs": "fn x(){}\n", ".ssh/id_x": "KEY", "keystore/key": "KEY", "aria-v3-copy/x.py": "copy", "notes.txt": "n", }.items(): touch(os.path.join(self.H, rel), body) os.symlink(os.path.join(self.H, "projekte"), os.path.join(self.H, "link")) os.symlink(os.path.join(self.H, "scripts"), os.path.join(self.H, "alias")) self.app = os.path.join(self.H, "projekte", "app") self.kosten = os.path.join(self.H, "projekte", "aws-kostenwache") self.cronproj = os.path.join(self.H, "projekte", "cronproj") self.cron = ["*/10 * * * * %s/melde.sh >/dev/null 2>&1" % self.kosten, "*/5 * * * * cd %s && python3 run.py" % self.cronproj] def tearDown(self): self._tmp.cleanup() def p(self, rel): return os.path.join(self.H, rel) class ValidateWriteAllowlist(SyntheticHomeBase): def v(self, entries, **kw): kw.setdefault("home_root", self.H) return P.validate_write_allowlist(entries, **kw) def rejected(self, entry, needle, **kw): with self.assertRaises(P.WriteAllowlistError) as c: self.v([entry], **kw) self.assertIn(needle, str(c.exception), entry) def test_default_empty_is_valid(self): self.assertEqual(self.v([]), ()) self.assertEqual(self.v(()), ()) def test_valid_project_dirs_accepted_and_deduplicated(self): self.assertEqual(self.v([self.app, self.kosten, self.app]), (self.app, self.kosten)) def test_ancestor_of_blocked_paths_is_allowed(self): # Ein Eintrag OBERHALB gesperrter Pfade ist zulaessig; tools_profile schneidet sie heraus. secret = self.p("projekte/app/.env") hooks = self.p("projekte/app/.git/hooks") self.assertEqual(self.v([self.p("projekte")], blocked={secret}, read_only={hooks}), (self.p("projekte"),)) def test_rejects_wrong_types(self): for bad in (self.app, None, 3, {"x": 1}): with self.assertRaises(P.WriteAllowlistError): self.v(bad) for bad in ([""], [3], ["a\x00b"]): with self.assertRaises(P.WriteAllowlistError): self.v(bad) def test_rejects_relative_and_non_canonical(self): self.rejected("projekte/app", "absolute") self.rejected(self.app + "/", "canonical") self.rejected(self.p("projekte/../projekte/app"), "canonical") self.rejected(self.H + "//projekte/app", "canonical") self.rejected(self.p("projekte/./app"), "canonical") def test_rejects_home_root_and_outside(self): self.rejected(self.H, "home_root itself") self.rejected(self.base, "below home_root") self.rejected("/tmp", "below home_root") self.rejected("/etc", "below home_root") def test_rejects_hidden_components(self): # Caches/Konfig/Zugangsdaten/Hooks sind keine Projektordner (Cache-Vergiftung = spaeter # ungekaefigte Ausfuehrung, z. B. pip-Wheel oder Crate-Quelltext). for rel in (".cache", ".cache/pip", ".cargo/registry", "projekte/app/.git", ".ssh"): self.rejected(self.p(rel), "hidden") def test_rejects_missing_file_and_symlinks(self): self.rejected(self.p("projekte/fehlt"), "does not exist") self.rejected(self.p("notes.txt"), "directory expected") self.rejected(self.p("link"), "symlink rejected") self.rejected(self.p("link/app"), "path contains a symlink") def test_symlinked_ancestor_cannot_smuggle_protected_area(self): # alias -> scripts: der rohe Pfad alias/sub liegt NICHT unter scripts; erst die realpath-Pruefung # verhindert eine rw-Regel auf ~/scripts/sub (Startschutz-Bereich). self.assertTrue(os.path.isdir(self.p("alias/sub"))) self.rejected(self.p("alias/sub"), "path contains a symlink") def test_rejects_startup_protected_dirs(self): for rel in ("brain", "scripts", "scripts/sub", "state"): self.rejected(self.p(rel), "inside blocked area") def test_rejects_state_dir_repo_and_family(self): self.rejected(self.p("gwstate"), "inside blocked area", state_dir=self.p("gwstate")) self.rejected(self.p("gwstate/sub"), "inside blocked area", state_dir=self.p("gwstate")) self.rejected(self.app, "inside blocked area", repo_root=self.app) # Geschwister-/Familien-Sperre (1g): Stamm aus dem laufenden Repo (aria-v3-dev -> aria-v3). from aria_router.isolation import PKG_REPO_ROOT self.rejected(self.p("aria-v3-copy"), "inside blocked area", repo_root=PKG_REPO_ROOT) def test_rejects_credential_symlink_target(self): # .ssh -> keystore: das realpath-Ziel ist hart gesperrt (#5) und darf nicht per Allowlist rw werden. import shutil shutil.rmtree(self.p(".ssh")) os.symlink(self.p("keystore"), self.p(".ssh")) self.rejected(self.p("keystore"), "inside blocked area") def test_rejects_dynamic_blocked_and_read_only(self): # erst beim Profilbau bekannt: Secrets/Lauf-Root (blocked) und Cron-/Hook-Ziele (read_only) self.rejected(self.app, "inside blocked area", blocked={self.app}) self.rejected(self.cronproj, "inside read-only protected area", read_only={self.cronproj}) self.rejected(self.p("projekte/app"), "inside read-only protected area", read_only={self.p("projekte")}) def test_rejects_overlap_with_run_container(self): runs = self.p("projekte/runs") for entry in (self.p("projekte"), runs): self.rejected(entry, "run container", runs_root=runs) os.makedirs(self.p("projekte/runs/codexrun-x")) self.rejected(self.p("projekte/runs/codexrun-x"), "run container", runs_root=runs) self.assertEqual(self.v([self.app], runs_root=runs), (self.app,)) class ConfigWriteAllowlist(SyntheticHomeBase): def parse(self, codex, **top): data = config_dict(Path(self.base), {"codex": codex}) data.update(top) return parse_gateway_config(data) def tools(self, **extra): return {"profile": "tools", "command_template": DANGER, "isolation_wrapper": CAGE, "home_root": self.H, **extra} def test_default_is_empty(self): self.assertEqual(self.parse(self.tools()).codex.tools_write_allowlist, ()) self.assertEqual(self.parse({}).codex.tools_write_allowlist, ()) def test_valid_allowlist_parsed(self): cfg = self.parse(self.tools(tools_write_allowlist=[self.app, self.kosten])) self.assertEqual(cfg.codex.tools_write_allowlist, (self.app, self.kosten)) def test_invalid_entry_fails_closed_with_clear_error(self): for entry, needle in ((self.p(".cache"), "hidden"), (self.H, "home_root itself"), (self.p("brain"), "inside blocked area"), (self.app + "/", "canonical"), (self.p("projekte/fehlt"), "does not exist"), ("/etc", "below home_root")): with self.subTest(entry=entry): with self.assertRaises(ConfigurationError) as c: self.parse(self.tools(tools_write_allowlist=[entry])) self.assertIn("codex.tools_write_allowlist", str(c.exception)) self.assertIn(needle, str(c.exception)) def test_inside_state_dir_rejected(self): with self.assertRaises(ConfigurationError) as c: self.parse(self.tools(tools_write_allowlist=[self.p("gwstate/sub")]), state_dir=self.p("gwstate")) self.assertIn("inside blocked area", str(c.exception)) def test_overlap_with_workspace_root_rejected(self): with self.assertRaises(ConfigurationError) as c: self.parse(self.tools(tools_write_allowlist=[self.p("projekte")], workspace_root=self.p("projekte/runs"))) self.assertIn("run container", str(c.exception)) def test_text_profile_rejects_allowlist(self): with self.assertRaises(ConfigurationError) as c: self.parse({"home_root": self.H, "tools_write_allowlist": [self.app]}) self.assertIn("only valid for codex.profile tools", str(c.exception)) self.assertEqual(self.parse({"home_root": self.H, "tools_write_allowlist": []}).codex.tools_write_allowlist, ()) def test_type_errors_rejected(self): for bad in (self.app, [1], [""], {"a": 1}): with self.subTest(bad=bad): with self.assertRaises(ConfigurationError): self.parse(self.tools(tools_write_allowlist=bad)) def test_gateway_threads_allowlist_to_caged_runner(self): write_key(Path(self.base) / "keys" / "clearance.key") os.makedirs(os.path.join(self.base, "work"), exist_ok=True) cfg = self.parse(self.tools(real_execution=True, tools_write_allowlist=[self.app], resource_limits={"cpu_seconds": 60}, auth_file=os.path.join(self.base, "auth.json"))) runner = Gateway(cfg).adapters["codex"].runner self.assertIsInstance(runner, CagedCodexRunner) self.assertEqual(runner.write_allowlist, (self.app,)) class ProfileWriteRules(SyntheticHomeBase): def dirs(self): rr = os.path.join(self.base, "run") d = P.RunDirs(rr, rr + "/ws", rr + "/home", rr + "/codex-home", rr + "/authdir", rr + "/tmp") for x in (d.workspace, d.home, d.codex_home, d.authdir, d.tmp): os.makedirs(x, exist_ok=True) touch(d.auth_copy, "{}") return d def cfg(self, **kw): kw.setdefault("cron_targets", self.cron) kw.setdefault("protect_startup", True) return P.tools_profile(self.dirs(), codex_bin="/usr/bin/true", rlimits={}, home_root=self.H, exec_roots=(), **kw) def writable(self, cfg, rel): p = self.p(rel) return p in set(cfg["file_rw"]) or any(p == d or p.startswith(d + os.sep) for d in cfg["rw"]) def readable(self, cfg, rel): p = self.p(rel) return (self.writable(cfg, rel) or p in set(cfg["file_ro"]) or any(p == d or p.startswith(d + os.sep) for d in cfg["ro"])) def test_without_allowlist_home_is_read_only(self): cfg = self.cfg() for rel in ("projekte/app/main.py", "projekte/aws-kostenwache/kostenwache.py", "projekte/aws-kostenwache/melde.sh", "projekte/cronproj/run.py", "notes.txt", "brain/note.md", ".cache/pip/wheel.whl", ".cargo/registry/src/lib.rs"): with self.subTest(rel=rel): self.assertTrue(self.readable(cfg, rel), f"{rel} nicht mehr lesbar") self.assertFalse(self.writable(cfg, rel), f"{rel} schreibbar (E44-Klasse offen)") self.assertNotIn(self.H, set(cfg["rw"])) def test_only_run_dirs_and_dev_null_writable_by_default(self): d = self.dirs() cfg = P.tools_profile(d, codex_bin="/usr/bin/true", rlimits={}, home_root=self.H, exec_roots=(), cron_targets=self.cron) self.assertEqual(set(cfg["rw"]), {d.workspace, d.home, d.codex_home, d.tmp}) self.assertEqual(cfg["file_rw"], ["/dev/null"]) self.assertNotIn(d.authdir, cfg["rw"]) # auth-Kopie bleibt read-only (M1) rw_auth = P.tools_profile(d, codex_bin="/usr/bin/true", rlimits={}, home_root=self.H, exec_roots=(), cron_targets=self.cron, writable_auth=True) self.assertEqual(rw_auth["file_rw"], ["/dev/null", d.auth_copy]) # nur mit E36-Schalter def test_kostenwache_class_closed(self): # Das E44-Beispiel: melde.sh steht in der Crontab (ro seit 1e), kostenwache.py daneben war bis # E44 schreibbar und wird von melde.sh ungekaefigt ausgefuehrt. Jetzt: beide nur lesbar. cfg = self.cfg() for rel in ("projekte/aws-kostenwache/melde.sh", "projekte/aws-kostenwache/kostenwache.py"): self.assertTrue(self.readable(cfg, rel)) self.assertFalse(self.writable(cfg, rel), rel) def test_allowlist_dir_writable_neighbours_not(self): cfg = self.cfg(write_allowlist=[self.app]) self.assertTrue(self.writable(cfg, "projekte/app/main.py")) self.assertIn(self.p("projekte/app/src"), set(cfg["rw"])) # Teilbaum ohne Sperren: ganz rw for rel in ("projekte/aws-kostenwache/kostenwache.py", "projekte/cronproj/run.py", "notes.txt"): self.assertFalse(self.writable(cfg, rel), rel) self.assertTrue(self.readable(cfg, rel), rel) self.assertNotIn(self.p("projekte"), set(cfg["rw"])) def test_dir_level_with_protected_entries_gets_no_dir_rule(self): # Landlock-Grenze (galt vor E44 fuer jeden Home-Ordner): app enthaelt .env und .git/hooks, darf # also selbst KEINE Ordner-Regel bekommen (MAKE_REG/REMOVE/REFER erben sich auf alles darunter -> # .env/Hooks waeren per unlink/rename ersetzbar). Folge: dort nur Bestandsdateien aenderbar. cfg = self.cfg(write_allowlist=[self.app]) self.assertNotIn(self.app, set(cfg["rw"])) self.assertNotIn(self.p("projekte/app/.git"), set(cfg["rw"])) self.assertIn(self.p("projekte/app/main.py"), set(cfg["file_rw"])) def test_allowlist_carves_secret_hooks_and_cron_targets(self): cfg = self.cfg(write_allowlist=[self.app, self.kosten]) self.assertTrue(self.writable(cfg, "projekte/app/main.py")) self.assertTrue(self.writable(cfg, "projekte/aws-kostenwache/kostenwache.py")) # bewusst freigegeben self.assertFalse(self.readable(cfg, "projekte/app/.env"), "Secret im Allowlist-Ordner erreichbar") for rel in ("projekte/app/.git/hooks/pre-commit", "projekte/app/.git/config", "projekte/aws-kostenwache/melde.sh"): self.assertTrue(self.readable(cfg, rel), rel) self.assertFalse(self.writable(cfg, rel), f"{rel} im Allowlist-Ordner schreibbar") def test_protect_startup_off_opens_hooks_only_inside_allowlist(self): cfg = self.cfg(write_allowlist=[self.app], protect_startup=False) self.assertTrue(self.writable(cfg, "projekte/app/.git/hooks/pre-commit")) self.assertFalse(self.readable(cfg, "projekte/app/.env")) # Secrets bleiben in jedem Modus hart self.assertFalse(self.writable(cfg, "brain/note.md")) # ausserhalb der Allowlist nie rw def test_build_time_rejects_entry_inside_cron_dir(self): with self.assertRaises(P.WriteAllowlistError) as c: self.cfg(write_allowlist=[self.cronproj]) self.assertIn("read-only protected", str(c.exception)) def test_build_time_rejects_overlap_with_runs_root(self): with self.assertRaises(P.WriteAllowlistError): self.cfg(write_allowlist=[self.p("projekte")], runs_root=self.p("projekte/runs")) def test_text_profile_has_no_home_rules(self): d = self.dirs() cfg = P.text_profile(d, codex_bin="/usr/bin/true", rlimits={}) self.assertEqual(set(cfg["rw"]), {d.workspace, d.home, d.codex_home, d.tmp}) for key in ("ro", "rw", "file_ro", "file_rw", "file_x"): self.assertFalse([x for x in cfg.get(key, []) if x.startswith(self.H)], key) class RunnerWriteAllowlist(SyntheticHomeBase): def runner(self, **kw): os.makedirs(os.path.join(self.base, "work"), exist_ok=True) auth = os.path.join(self.base, "auth.json") touch(auth, "{}") kw.setdefault("profile", "tools") return CagedCodexRunner(os.path.join(self.base, "work"), auth_file=auth, home_root=self.H, secret_files=None, protect_startup=True, **kw) def test_runner_validates_at_construction(self): for entry in (self.p(".cache"), self.H, self.p("brain"), self.p("link/app")): with self.subTest(entry=entry): with self.assertRaises(P.WriteAllowlistError): self.runner(write_allowlist=[entry]) self.assertEqual(self.runner().write_allowlist, ()) def test_runner_text_profile_rejects_allowlist(self): with self.assertRaises(ValueError): self.runner(profile="text", write_allowlist=[self.app]) def test_runner_threads_allowlist_into_profile(self): run = self.runner(write_allowlist=[self.app]) self.assertEqual(run.write_allowlist, (self.app,)) rr = os.path.join(self.base, "rr") dirs = P.RunDirs(rr, rr + "/ws", rr + "/home", rr + "/codex-home", rr + "/authdir", rr + "/tmp") cfg = run._build_cfg(dirs, "/usr/bin/true") main = self.p("projekte/app/main.py") self.assertTrue(main in set(cfg["file_rw"]) or any(main.startswith(d + os.sep) for d in cfg["rw"])) kosten = self.p("projekte/aws-kostenwache/kostenwache.py") self.assertFalse(kosten in set(cfg["file_rw"]) or any(kosten.startswith(d + os.sep) for d in cfg["rw"])) def test_entry_swapped_for_symlink_after_construction_fails_closed(self): # Konstruktion mit gueltigem Ordner, danach wird er gegen einen Symlink auf ~/scripts getauscht: # der Profilbau je Lauf prueft erneut -> WRITE_ALLOWLIST_INVALID, es startet nichts. swap = self.p("projekte/swap") os.makedirs(swap) class NoBus(CagedCodexRunner): def bus_reachable(self): return True auth = os.path.join(self.base, "auth.json") touch(auth, "{}") os.makedirs(os.path.join(self.base, "work"), exist_ok=True) run = NoBus(os.path.join(self.base, "work"), profile="tools", auth_file=auth, home_root=self.H, secret_files=set(), protect_startup=True, write_allowlist=[swap], systemctl_path="/bin/false") os.rmdir(swap) os.symlink(self.p("scripts"), swap) # Nicht-Kaefig-argv: selbst wenn die Pruefung (z. B. in einer Mutationsprobe) fehlt, endet der Lauf # vor jedem Prozessstart (TOOLS_PROFILE_REQUIRES_CAGE) - der Test startet nie systemd-run. argv = ("/usr/bin/env", "-i", "--", "/usr/bin/true") r = run.run(RunSpec(argv, b"", {}, 10)) self.assertEqual(r.error_code, "WRITE_ALLOWLIST_INVALID") self.assertEqual(os.listdir(os.path.join(self.base, "work")), []) # Lauf-Root aufgeraeumt if __name__ == "__main__": unittest.main()