"""B+-Gegenproben gegen das ECHTE /home/aria mit dem tools-Profil (Landlock, ohne systemd-Bus).
Laeuft nur mit python3 im Kaefig, nie codex. Belegt: Credentials/ssh/claude/codex gesperrt,
claude nicht ausfuehrbar, normale Tools laufen, Home lesbar aber (E44 Option 2) nur in Allowlist-
Ordnern schreibbar, kill/AF_UNIX gesperrt."""
import os
import shutil
import tempfile
import unittest

from aria_router.isolation import profiles as P
from tests.iso_helpers import (CAGE_PROBE, PYTHON, home_probe_dir, home_project_probe_dir, landlock_abi,
                               run_in_cage)

HOME = "/home/aria"
V8_HOST = ("/home/aria/.local/lib/node_modules/@openai/codex/node_modules/@openai/codex-linux-x64/vendor/"
           "x86_64-unknown-linux-musl/bin/codex-code-mode-host")


@unittest.skipUnless(landlock_abi() >= 4, "Landlock ABI < 4")
class ToolsProfileRealHome(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls._tmp = tempfile.TemporaryDirectory()
        root = os.path.join(cls._tmp.name, "run")
        cls.dirs = P.RunDirs(root, root + "/ws", root + "/home", root + "/codex-home",
                             root + "/authdir", root + "/tmp")
        for d in (cls.dirs.workspace, cls.dirs.home, cls.dirs.codex_home, cls.dirs.authdir, cls.dirs.tmp):
            os.makedirs(d, exist_ok=True)
        with open(cls.dirs.auth_copy, "w", encoding="utf-8") as fh:
            fh.write('{"token":"x"}')
        # Wegwerf-Datei im Home (eigenes Probe-Verzeichnis, kein fremder Worktree) fuer den Schreib-Test;
        # seit E44 Option 2 muss das Schreiben dort SCHEITERN (kein Allowlist-Ordner).
        cls.probe_dir = home_probe_dir()
        fd, cls.home_probe_file = tempfile.mkstemp(prefix=".ariaprobe_w_", dir=cls.probe_dir)
        os.close(fd)
        # cage_probe darf nicht als Secret gewertet und so gesperrt werden
        cls.cfg_secret_files = P.scan_secret_files(HOME, extra_prune_paths=[cls.dirs.root])
        cls.cfg = P.tools_profile(cls.dirs, codex_bin=PYTHON, home_root=HOME,
                                  rlimits={"CPU": 300, "FSIZE": 1024**3, "NOFILE": 4096},
                                  secret_files=cls.cfg_secret_files)

    @classmethod
    def tearDownClass(cls):
        shutil.rmtree(cls.probe_dir, ignore_errors=True)
        cls._tmp.cleanup()

    def probe(self, check, *args):
        r = run_in_cage(self.cfg, [PYTHON, "-I", CAGE_PROBE, check, *args], timeout=30)
        self.assertEqual(r.returncode, 0,
                         f"{check} {args}: rc={r.returncode} err={r.stderr.decode(errors='replace')[:200]}")

    def test_credentials_blocked(self):
        self.probe("read-denied", "/home/aria/.env")
        self.probe("read-denied", "/home/aria/.codex/auth.json")
        if os.path.exists("/home/aria/.claude/settings.json"):
            self.probe("read-denied", "/home/aria/.claude/settings.json")

    def test_ssh_and_gh_blocked(self):
        self.probe("list-denied", "/home/aria/.ssh")
        if os.path.isdir("/home/aria/.config/gh"):
            self.probe("list-denied", "/home/aria/.config/gh")

    def test_claude_not_executable(self):
        self.probe("exec-denied", "/usr/bin/claude")

    def test_normal_tool_executable(self):
        self.probe("exec-ok", "/usr/bin/id")

    def test_home_readable_but_not_writable(self):
        # E44 Option 2 (vorher: test_home_readable_and_writable, Schreiben im Home erlaubt). In EINEM Lauf:
        # Home lesbar, Probe-Datei/-Ordner im Home nicht beschreibbar, Workspace schon (Positivkontrolle).
        ws_ok = os.path.join(self.dirs.workspace, "home-test-ok.txt")
        self.probe("write-matrix", "read-ok:/home/aria/CLAUDE.md", "read-ok:" + self.home_probe_file,
                   "append-denied:" + self.home_probe_file,
                   "create-denied:" + os.path.join(self.probe_dir, "neu.txt"),
                   "write-ok:" + ws_ok)
        self.assertTrue(os.path.isfile(ws_ok))
        self.assertEqual(os.path.getsize(self.home_probe_file), 0)          # unveraendert
        self.assertFalse(os.path.lexists(os.path.join(self.probe_dir, "neu.txt")))

    def test_kill_foreign_denied(self):
        self.probe("kill-denied", "1")

    def _v8_host(self, rlimits):
        # Echter codex-code-mode-host (V8, kein Modell, kein Netz) mit Handshake + 1 JS-Zelle.
        # Codex' Shell-Werkzeug laeuft ueber diesen Host; stirbt er, ist das tools-Profil wertlos.
        import json as _json, struct
        frames = [{"type": "connection/hello", "supportedVersions": [1], "requiredCapabilities": [],
                   "optionalCapabilities": []},
                  {"type": "operation/request", "id": 1, "request": {"method": "session/open", "sessionId": "s1"}},
                  {"type": "operation/request", "id": 2, "request": {"method": "session/execute", "sessionId": "s1",
                   "request": {"tool_call_id": "t1", "source": "1+1", "enabled_tools": []}}}]
        blob = b"".join(struct.pack("<I", len(b)) + b for b in (_json.dumps(f).encode() for f in frames))
        with open(os.path.join(self.dirs.workspace, "frames.bin"), "wb") as fh:
            fh.write(blob)
        cfg = P.tools_profile(self.dirs, codex_bin=V8_HOST, home_root=HOME, rlimits=rlimits,
                              secret_files=self.cfg_secret_files)
        return run_in_cage(cfg, ["/bin/sh", "-c", "(cat %s/frames.bin; sleep 3) | %s"
                                 % (self.dirs.workspace, V8_HOST)], timeout=60)

    @unittest.skipUnless(os.path.isfile(V8_HOST), "codex-code-mode-host nicht installiert")
    def test_v8_host_runs_in_tools_cage(self):                   # E2b: ohne RLIMIT_AS
        r = self._v8_host({"CPU": 300, "FSIZE": 1024**3, "NOFILE": 4096})
        self.assertEqual(r.returncode, 0, r.stderr[-300:])
        self.assertIn(b'"type":"execute/initialResponse"', r.stdout)

    @unittest.skipUnless(os.path.isfile(V8_HOST), "codex-code-mode-host nicht installiert")
    def test_v8_host_dies_under_address_space_limit(self):       # Ursache des SIGTRAP vom 24.10.
        r = self._v8_host({"AS": 4 * 1024**3, "CPU": 300, "FSIZE": 1024**3, "NOFILE": 4096})
        self.assertNotEqual(r.returncode, 0)
        self.assertIn(b"V8 sandbox", r.stderr)

    def test_af_unix_denied(self):
        self.probe("af-unix-denied", "/run/user/1001/bus")

    def test_deferred_exec_ro_targets_denied_rw_ok(self):
        # Re-Review 2: in EINEM Kaefig-Lauf muss das Schreiben auf eine read-only-Startdatei
        # (.cargo/env, von .bashrc/.profile gesourct) UND auf ein Cron-/Startschutz-Ziel
        # (state/run-heartbeat.conf, vom Heartbeat-Cron gesourct) scheitern, das Workspace-Schreiben
        # aber gelingen. O_WRONLY ohne O_TRUNC -> kein Datenverlust an den echten Dateien.
        ro1 = "/home/aria/.cargo/env"
        ro2 = "/home/aria/state/run-heartbeat.conf"
        if not (os.path.isfile(ro1) and os.path.isfile(ro2)):
            self.skipTest("VORAUSSETZUNG: .cargo/env und state/run-heartbeat.conf muessen existieren")
        rw = os.path.join(self.dirs.workspace, "ok.txt")
        self.probe("ro-write-denied-rw-ok", ro1, ro2, rw)
        self.assertTrue(os.path.isfile(rw))         # Workspace-Schreiben hat im selben Lauf funktioniert

    def test_claude_json_read_denied(self):
        # Re-Review 2, 1b: .claude.json (mcpServers) weder lesbar noch schreibbar.
        if os.path.exists("/home/aria/.claude.json"):
            self.probe("read-denied", "/home/aria/.claude.json")

    def test_latent_top_level_dotfile_uncreatable(self):
        # Re-Review 2, 1a: ein noch nicht existentes Top-Level-Dotfile (wie ~/.bash_aliases) darf
        # NICHT neu anlegbar sein, weil home_root keine MAKE_REG-Regel hat. Nichtexistenz vorher gesichert.
        latent = "/home/aria/.aria-latent-probe-%d" % os.getpid()
        self.assertFalse(os.path.lexists(latent))
        self.probe("create-denied", latent)
        self.assertFalse(os.path.lexists(latent), "latente Datei wurde im Home angelegt (LEAK)")


@unittest.skipUnless(landlock_abi() >= 4, "Landlock ABI < 4")
class WriteAllowlistRealHome(unittest.TestCase):
    """E44 Option 2 gegen das ECHTE /home/aria (Landlock, python3 im Kaefig, nie codex).

    Zwei gleich aufgebaute Testprojekte nach dem kostenwache-Muster (melde.sh = Cron-Ziel, per
    cron_targets injiziert, die echte Crontab bleibt unberuehrt; kostenwache.py daneben): `allow` ist
    Allowlist-Ordner, `deny` nicht. Erwartung in EINEM Kaefig-Lauf: Workspace und `allow` schreibbar,
    `deny` und das uebrige Home nicht, Lesen im Home geht, Credentials bleiben unlesbar, im Allowlist-
    Ordner bleiben Cron-Ziel, Git-Hook und Secret geschuetzt."""

    @classmethod
    def setUpClass(cls):
        cls._tmp = tempfile.TemporaryDirectory()
        root = os.path.join(cls._tmp.name, "run")
        cls.dirs = P.RunDirs(root, root + "/ws", root + "/home", root + "/codex-home",
                             root + "/authdir", root + "/tmp")
        for d in (cls.dirs.workspace, cls.dirs.home, cls.dirs.codex_home, cls.dirs.authdir, cls.dirs.tmp):
            os.makedirs(d, exist_ok=True)
        with open(cls.dirs.auth_copy, "w", encoding="utf-8") as fh:
            fh.write('{"token":"x"}')
        cls.allow = home_project_probe_dir()
        cls.deny = home_project_probe_dir()
        cls.body = {}
        for d in (cls.allow, cls.deny):
            for rel, body in (("melde.sh", "#!/bin/sh\npython3 kostenwache.py\n"),
                              ("kostenwache.py", "print('kosten')\n")):
                with open(os.path.join(d, rel), "w", encoding="utf-8") as fh:
                    fh.write(body)
                cls.body[os.path.join(d, rel)] = body
        os.makedirs(os.path.join(cls.allow, ".git", "hooks"))
        os.makedirs(os.path.join(cls.allow, "src"))            # Teilbaum OHNE geschuetzte Eintraege
        with open(os.path.join(cls.allow, ".git", "hooks", "pre-commit"), "w", encoding="utf-8") as fh:
            fh.write("#!/bin/sh\n")
        with open(os.path.join(cls.allow, ".env"), "w", encoding="utf-8") as fh:
            fh.write("CANARY=1\n")
        cls.cron = ["*/10 * * * * %s/melde.sh >/dev/null 2>&1" % d for d in (cls.allow, cls.deny)]
        cls.secret_files = P.scan_secret_files(HOME, extra_prune_paths=[cls.dirs.root])
        cls.rlimits = {"CPU": 300, "FSIZE": 1024**3, "NOFILE": 4096}
        cls.cfg = P.tools_profile(cls.dirs, codex_bin=PYTHON, home_root=HOME, rlimits=cls.rlimits,
                                  secret_files=cls.secret_files, cron_targets=cls.cron,
                                  write_allowlist=[cls.allow])

    @classmethod
    def tearDownClass(cls):
        for d in (cls.allow, cls.deny):
            shutil.rmtree(d, ignore_errors=True)
        cls._tmp.cleanup()

    def probe(self, cfg, *tokens):
        r = run_in_cage(cfg, [PYTHON, "-I", CAGE_PROBE, "write-matrix", *tokens], timeout=30)
        self.assertEqual(r.returncode, 0, f"rc={r.returncode} err={r.stderr.decode(errors='replace')[:300]}")

    def test_write_only_in_workspace_and_allowlist(self):
        ws_ok = os.path.join(self.dirs.workspace, "ok.txt")
        a, d = self.allow, self.deny
        latent = "/home/aria/.aria-e44-latent-%d" % os.getpid()
        self.assertFalse(os.path.lexists(latent))
        self.probe(self.cfg,
                   "write-ok:" + ws_ok,                                     # Positivkontrolle Workspace
                   "write-ok:" + os.path.join(a, "src", "neu.txt"),         # Allowlist: neu anlegen
                   "write-ok:" + os.path.join(a, "kostenwache.py"),         # Allowlist: bestehende Datei
                   # Landlock-Grenze (wie vor E44 fuer jeden Home-Ordner): die Ebene, die selbst geschuetzte
                   # Eintraege enthaelt (.env, melde.sh, .git/hooks), bekommt keine Ordner-Regel - sonst
                   # liessen sich diese per unlink/rename ersetzen. Dort nur Bestandsdateien aenderbar.
                   "create-denied:" + os.path.join(a, "neu-oben.txt"),
                   "append-denied:" + os.path.join(d, "kostenwache.py"),    # E44-Klasse: .py neben Cron-Ziel
                   "append-denied:" + os.path.join(d, "melde.sh"),
                   "create-denied:" + os.path.join(d, "neu.py"),
                   "create-denied:" + latent,                               # uebriges Home
                   "append-denied:" + os.path.join(a, "melde.sh"),          # Cron-Ziel im Allowlist-Ordner ro
                   "append-denied:" + os.path.join(a, ".git", "hooks", "pre-commit"),   # Hook ro
                   "read-denied:" + os.path.join(a, ".env"),                # Secret im Allowlist-Ordner hart
                   "read-ok:" + os.path.join(d, "kostenwache.py"),          # Lesen im Home geht weiter
                   "read-ok:/home/aria/CLAUDE.md",
                   "read-denied:/home/aria/.env",                           # Credentials unlesbar
                   "read-denied:/home/aria/.codex/auth.json")
        # Wirkung ausserhalb des Kaefigs nachpruefen (nicht nur den Exit-Code der Probe)
        self.assertTrue(os.path.isfile(ws_ok))
        with open(os.path.join(a, "src", "neu.txt"), "rb") as fh:
            self.assertEqual(fh.read(), b"x")
        with open(os.path.join(a, "kostenwache.py"), "rb") as fh:
            self.assertEqual(fh.read(), b"x")                               # bewusst freigegeben: geaendert
        self.assertFalse(os.path.lexists(os.path.join(a, "neu-oben.txt")))
        for p in (os.path.join(d, "kostenwache.py"), os.path.join(d, "melde.sh"), os.path.join(a, "melde.sh")):
            with open(p, encoding="utf-8") as fh:
                self.assertEqual(fh.read(), self.body[p], p)
        self.assertFalse(os.path.lexists(os.path.join(d, "neu.py")))
        self.assertFalse(os.path.lexists(latent), "Datei im Home angelegt (LEAK)")

    def test_without_allowlist_the_same_folder_is_read_only(self):
        # Gegenprobe: dieselbe Mappe ohne Allowlist-Eintrag -> nicht beschreibbar; Workspace im selben Lauf ok.
        cfg0 = P.tools_profile(self.dirs, codex_bin=PYTHON, home_root=HOME, rlimits=self.rlimits,
                               secret_files=self.secret_files, cron_targets=self.cron)
        self.probe(cfg0, "write-ok:" + os.path.join(self.dirs.workspace, "ok0.txt"),
                   "append-denied:" + os.path.join(self.allow, "kostenwache.py"),
                   "create-denied:" + os.path.join(self.allow, "neu0.txt"),
                   "read-ok:" + os.path.join(self.allow, "kostenwache.py"))
        self.assertFalse(os.path.lexists(os.path.join(self.allow, "neu0.txt")))


if __name__ == "__main__":
    unittest.main()
