"""CagedCodexRunner: Argv-Komposition, Bypass-/Kaefig-Pflicht, auth-Staging, NO_USER_BUS (ohne Bus),
und - falls der User-Bus erreichbar ist - echter systemd-Scope-Lauf (gegen python3, NIE codex)."""
import os
import shutil
import tempfile
import time
import unittest
from pathlib import Path

from aria_router.isolation import ISO_EXEC_PATH
from aria_router.isolation.caged_runner import CagedCodexRunner, wrapper_is_tools_cage
from aria_router.runner import RunSpec, UnsafeCommandError
from tests.iso_helpers import FAKE_CODEX_CAGE, home_project_probe_dir, landlock_abi, user_bus_reachable

CAGE = ["/usr/bin/systemd-run", "--user", "--scope", "--quiet", "--unit", "{unit}",
        "-p", "RuntimeMaxSec=120", "/usr/bin/env", "-i", "/usr/bin/python3", "-I",
        ISO_EXEC_PATH, "{cfg}", "--"]


def process_gone(pid: int) -> bool:
    try:
        with open(f"/proc/{pid}/stat", "r", encoding="utf-8") as fh:
            return fh.read().rsplit(")", 1)[1].split()[0] in ("Z", "X")
    except FileNotFoundError:
        return True


class CagedUnit(unittest.TestCase):
    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory()
        self.root = Path(self._tmp.name)
        self.work = self.root / "work"
        self.work.mkdir()
        self.auth = self.root / "auth.json"
        self.auth.write_text('{"token":"x"}', encoding="utf-8")

    def tearDown(self):
        self._tmp.cleanup()

    def runner(self, profile="text", **kw):
        return CagedCodexRunner(self.work, profile=profile, auth_file=str(self.auth),
                                home_root=str(self.root), secret_files=set(), **kw)

    def test_no_user_bus_is_fail_closed(self):
        # systemctl auf /bin/false -> Bus gilt als nicht erreichbar -> NO_USER_BUS, kein Start
        r = self.runner(systemctl_path="/bin/false").run(
            RunSpec(tuple(CAGE + ["/usr/bin/true"]), b"", {}, 10))
        self.assertEqual(r.error_code, "NO_USER_BUS")

    def test_tools_profile_rejects_address_space_limit(self):
        from aria_router.runner import ResourceLimits
        with self.assertRaises(ValueError):
            self.runner(profile="tools", resource_limits=ResourceLimits(address_space_bytes=4 * 1024 ** 3))
        self.runner(profile="tools", resource_limits=ResourceLimits(cpu_seconds=300))
        self.runner(profile="text", resource_limits=ResourceLimits(address_space_bytes=4 * 1024 ** 3))

    def test_files_fail_closed(self):
        # Review #15: Anhaenge werden nicht bereitgestellt -> fail-closed (vor dem Bus-Check).
        run = self.runner(profile="tools", systemctl_path="/bin/false")
        spec = RunSpec(tuple(CAGE + ["/usr/bin/true"]), b"prompt", {}, 10,
                       files=(("01-a.txt", b"data"),))
        self.assertEqual(run.run(spec).error_code, "ATTACHMENTS_UNSUPPORTED")
        # ohne Anhaenge kein ATTACHMENTS_UNSUPPORTED (hier: NO_USER_BUS wegen /bin/false)
        spec2 = RunSpec(tuple(CAGE + ["/usr/bin/true"]), b"prompt", {}, 10)
        self.assertEqual(run.run(spec2).error_code, "NO_USER_BUS")

    def test_codex_bin_claude_refused(self):
        run = self.runner(profile="tools")
        with self.assertRaises(UnsafeCommandError) as c:
            run._codex_bin_of(tuple(CAGE + ["/home/aria/.local/bin/claude", "exec"]))
        self.assertEqual(c.exception.code, "CODEX_BIN_IS_CLAUDE")

    def test_compose_argv_substitutes_placeholders(self):
        run = self.runner()
        import aria_router.isolation.profiles as P
        rr = str(self.root / "rr")
        dirs = P.RunDirs(rr, rr + "/ws", rr + "/home", rr + "/codex-home", rr + "/authdir", rr + "/tmp")
        argv = run._compose_argv(tuple(CAGE + ["/usr/bin/true"]), dirs, "/tmp/x.json", "aria-codex-u1")
        self.assertIn("aria-codex-u1", argv)
        self.assertIn("/tmp/x.json", argv)
        self.assertNotIn("{unit}", argv)
        self.assertNotIn("{cfg}", argv)

    def test_tools_profile_requires_cage_in_compose(self):
        run = self.runner(profile="tools")
        import aria_router.isolation.profiles as P
        rr = str(self.root / "rr")
        dirs = P.RunDirs(rr, rr + "/ws", rr + "/home", rr + "/codex-home", rr + "/authdir", rr + "/tmp")
        # Nicht-Kaefig-Wrapper + Bypass -> abgelehnt
        with self.assertRaises(UnsafeCommandError):
            run._compose_argv(("/usr/bin/env", "-i", "--", "/usr/bin/codex", "-s", "danger-full-access"),
                              dirs, "/tmp/x.json", "u1")

    def test_stage_auth_readonly_default(self):
        run = self.runner()
        dirs = run._make_run_dirs()
        try:
            run._stage_auth(dirs)
            mode = os.stat(dirs.auth_copy).st_mode & 0o777
            self.assertEqual(mode, 0o400)
            self.assertTrue(os.path.islink(dirs.codex_home_auth_link))
            self.assertEqual(os.path.realpath(dirs.codex_home_auth_link), os.path.realpath(dirs.auth_copy))
        finally:
            import shutil
            shutil.rmtree(dirs.root, ignore_errors=True)

    def test_stage_auth_writable_when_enabled(self):
        run = self.runner(codex_auth_writable=True)
        dirs = run._make_run_dirs()
        try:
            run._stage_auth(dirs)
            self.assertEqual(os.stat(dirs.auth_copy).st_mode & 0o777, 0o600)
        finally:
            import shutil
            shutil.rmtree(dirs.root, ignore_errors=True)

    def test_wrapper_is_tools_cage_detection(self):
        self.assertTrue(wrapper_is_tools_cage(tuple(CAGE + ["/usr/bin/codex"])))
        self.assertFalse(wrapper_is_tools_cage(("/usr/bin/env", "-i", "--", "/usr/bin/codex")))

    def test_wrapper_structural_rejects_bypass(self):
        # Review #1: systemd-run/--scope/iso_exec als DECOY-Argumente zu python3 -c / sh -c.
        evil_py = ["/usr/bin/systemd-run", "--user", "--scope", "/usr/bin/env", "-i",
                   "/usr/bin/python3", "-c", "PAYLOAD", ISO_EXEC_PATH, "--", "/usr/bin/codex", "exec"]
        evil_sh = ["/usr/bin/systemd-run", "--user", "--scope", "/bin/sh", "-c", "x",
                   ISO_EXEC_PATH, "--", "/usr/bin/codex"]
        self.assertFalse(wrapper_is_tools_cage(evil_py))
        self.assertFalse(wrapper_is_tools_cage(evil_sh))
        # falsche Kopf-Reihenfolge
        self.assertFalse(wrapper_is_tools_cage(
            ["/usr/bin/systemd-run", "--scope", "--user", "/usr/bin/env", "-i", "/usr/bin/python3",
             "-I", ISO_EXEC_PATH, "{cfg}", "--", "/usr/bin/codex"]))
        # fremdes Token in der Mitte
        self.assertFalse(wrapper_is_tools_cage(
            ["/usr/bin/systemd-run", "--user", "--scope", "--evil", "/usr/bin/env", "-i",
             "/usr/bin/python3", "-I", ISO_EXEC_PATH, "{cfg}", "--", "/usr/bin/codex"]))
        # python3 statt /usr/bin/python3 (nicht die feste Kette)
        self.assertFalse(wrapper_is_tools_cage(
            ["/usr/bin/systemd-run", "--user", "--scope", "/usr/bin/env", "-i", "python3",
             "-I", ISO_EXEC_PATH, "{cfg}", "--", "/usr/bin/codex"]))
        # iso-Token ohne iso_exec.py-Basename
        self.assertFalse(wrapper_is_tools_cage(
            ["/usr/bin/systemd-run", "--user", "--scope", "/usr/bin/env", "-i", "/usr/bin/python3",
             "-I", "/tmp/evil.py", "{cfg}", "--", "/usr/bin/codex"]))

    def test_wrapper_structural_accepts_valid(self):
        # -p/-unit/--quiet in der Mitte sind erlaubt; Prod-Pfad fuer iso_exec ebenfalls.
        self.assertTrue(wrapper_is_tools_cage(tuple(CAGE + ["/usr/bin/codex", "exec"])))
        prod = ["/usr/bin/systemd-run", "--user", "--scope", "--quiet", "--unit", "u", "-p",
                "MemoryMax=2G", "-p", "CPUQuota=200%", "/usr/bin/env", "-i", "/usr/bin/python3",
                "-I", "/home/aria/aria-v3/python/aria_router/isolation/iso_exec.py", "/x/cfg.json",
                "--", "/usr/bin/codex"]
        self.assertTrue(wrapper_is_tools_cage(prod))

    def test_build_cfg_threads_state_repo_and_startup(self):
        # Review #6 + H1 + E44: state_dir/repo_root/protect_startup/write_allowlist erreichen das Profil.
        import aria_router.isolation.profiles as P
        H = str(self.root / "home")
        for rel in (".bashrc", "state/budget.json", "v3repo/code.py", "normal.txt",
                    "proj/main.py", "proj/.git/hooks/pre-commit"):
            full = os.path.join(H, rel)
            os.makedirs(os.path.dirname(full), exist_ok=True)
            with open(full, "w") as fh:
                fh.write("x\n")
        state, repo, proj = os.path.join(H, "state"), os.path.join(H, "v3repo"), os.path.join(H, "proj")
        run = CagedCodexRunner(self.work, profile="tools", auth_file=str(self.auth), home_root=H,
                               secret_files=None, state_dir=state, repo_root=repo, protect_startup=True,
                               write_allowlist=[proj])
        rr = str(self.root / "rr")
        dirs = P.RunDirs(rr, rr + "/ws", rr + "/home", rr + "/codex-home", rr + "/authdir", rr + "/tmp")
        cfg = run._build_cfg(dirs, "/usr/bin/true")

        def blocked(c, p):
            if p in set(c["file_rw"]) | set(c["file_ro"]):
                return False
            return not any(p == d or p.startswith(d + os.sep) for d in c["rw"] + c["ro"])

        def writable(c, p):
            return p in set(c["file_rw"]) or any(p == d or p.startswith(d + os.sep) for d in c["rw"])
        self.assertTrue(blocked(cfg, os.path.join(state, "budget.json")))   # state_dir gesperrt
        self.assertTrue(blocked(cfg, os.path.join(repo, "code.py")))        # Repo gesperrt
        self.assertIn(os.path.join(H, ".bashrc"), set(cfg["file_ro"]))      # Startschutz an
        self.assertTrue(writable(cfg, os.path.join(proj, "main.py")))       # Allowlist erreicht das Profil
        self.assertFalse(writable(cfg, os.path.join(proj, ".git", "hooks", "pre-commit")))
        self.assertFalse(writable(cfg, os.path.join(H, "normal.txt")))      # E44: Home sonst nicht rw
        # protect_startup=False: .bashrc bleibt seit E44 trotzdem nur lesbar; die Hooks im Allowlist-
        # Ordner werden schreibbar (einzige verbliebene Wirkung der Abwahl auf Schreibrechte).
        run_off = CagedCodexRunner(self.work, profile="tools", auth_file=str(self.auth), home_root=H,
                                   secret_files=None, state_dir=state, repo_root=repo, protect_startup=False,
                                   write_allowlist=[proj])
        cfg_off = run_off._build_cfg(dirs, "/usr/bin/true")
        self.assertIn(os.path.join(H, ".bashrc"), set(cfg_off["file_ro"]))
        self.assertFalse(writable(cfg_off, os.path.join(H, ".bashrc")))
        self.assertTrue(writable(cfg_off, os.path.join(proj, ".git", "hooks", "pre-commit")))


class ReadLastHardening(unittest.TestCase):
    """Review #7: Scope-Stop vor dem Lesen + rennfreies _read_last (O_NOFOLLOW/fstat/Pfadcheck)."""

    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory()
        self.root = Path(self._tmp.name)
        self.auth = self.root / "auth.json"
        self.auth.write_text("{}", encoding="utf-8")
        (self.root / "w").mkdir()

    def tearDown(self):
        self._tmp.cleanup()

    def _runner(self, order):
        import aria_router.isolation.profiles as P
        from aria_router.runner import RunResult

        class Rec(CagedCodexRunner):
            def stop_scope(self, u):
                order.append("stop")

            def reset_scope(self, u):
                pass

            def _read_last(self, dirs, limit):
                order.append("read")
                return b""

            def bus_reachable(self):
                return True

            def _supervise(self, proc, spec, started):
                try:
                    proc.kill()
                    proc.wait(timeout=5)
                except Exception:
                    pass
                for stream in (proc.stdin, proc.stdout, proc.stderr):
                    try:
                        stream.close()
                    except Exception:
                        pass
                order.append("supervise")
                return RunResult(True, exit_code=0, stdout=b"")

        return Rec(self.root / "w", profile="tools", auth_file=str(self.auth),
                   home_root=str(self.root), secret_files=set(), repo_root=str(self.root),
                   protect_startup=False)

    def test_scope_stopped_before_read(self):
        order = []
        r = self._runner(order)
        r.run(RunSpec(tuple(CAGE + ["/bin/true"]), b"", {}, 10))
        self.assertIn("read", order)
        self.assertIn("stop", order)
        self.assertLess(order.index("stop"), order.index("read"),
                        f"Scope muss VOR dem Lesen gestoppt sein: {order}")

    def _read(self, make):
        import aria_router.isolation.profiles as P
        run = CagedCodexRunner(self.root / "w", profile="tools", auth_file=str(self.auth),
                               home_root=str(self.root), secret_files=set(), protect_startup=False)
        ws = self.root / "ws"
        ws.mkdir()
        dirs = P.RunDirs(str(self.root / "rr"), str(ws), "", "", "", "")
        make(ws)
        return run._read_last(dirs, 1024)

    def test_reads_regular_last_txt(self):
        self.assertEqual(self._read(lambda ws: (ws / "last.txt").write_text("HELLO")), b"HELLO")

    def test_symlink_last_txt_rejected(self):
        secret = self.root / "secret.txt"
        secret.write_text("TOPSECRET")
        self.assertEqual(self._read(lambda ws: os.symlink(secret, ws / "last.txt")), b"")

    def test_fifo_last_txt_rejected(self):
        self.assertEqual(self._read(lambda ws: os.mkfifo(ws / "last.txt")), b"")


@unittest.skipUnless(landlock_abi() >= 4, "Landlock ABI < 4")
class CagedIntegration(unittest.TestCase):
    """Echter systemd --user --scope. Braucht den User-Bus; ohne Bus wird NICHT still
    uebersprungen, sondern mit klarer Meldung auf die Voraussetzung geprueft (gezaehlt)."""

    def setUp(self):
        if not user_bus_reachable():
            self.skipTest("VORAUSSETZUNG FEHLT: systemd --user Bus nicht erreichbar "
                          "(XDG_RUNTIME_DIR/DBUS). Caged-Integration kann nicht laufen.")
        self._tmp = tempfile.TemporaryDirectory()
        self.root = Path(self._tmp.name)
        self.work = self.root / "work"
        self.work.mkdir()
        self.auth = self.root / "auth.json"
        self.auth.write_text('{"token":"x"}', encoding="utf-8")

    def tearDown(self):
        self._tmp.cleanup()

    def runner(self, write_allowlist=()):
        # home_root=/home/aria, damit das Fixture-Skript (im Worktree) im Home lesbar ist;
        # secret_files=set() ueberspringt den langsamen Realscan (Mechanik-Test, kein Carve-Test);
        # startup_git ueber protect_startup=False aus, und repo_root=tmp: der Mechanik-Test fuehrt ein
        # Fake-Codex-Skript AUS DEM Repo aus (in Prod liegt Codex ausserhalb des gesperrten Repos).
        # write_allowlist (E44): das Home ist sonst nur lesbar; die PID-Datei der Kill-Tests liegt darum
        # in einem eigenen, ausdruecklich freigegebenen Probe-Ordner.
        return CagedCodexRunner(self.work, profile="tools", auth_file=str(self.auth),
                                home_root="/home/aria", secret_files=set(),
                                repo_root=str(self.root), protect_startup=False,
                                write_allowlist=write_allowlist)

    def test_real_scope_runs_fake_codex(self):
        argv = tuple(CAGE + ["/usr/bin/python3", "-I", FAKE_CODEX_CAGE, "ok", "HELLO"])
        r = self.runner().run(RunSpec(argv, b"", {}, 60))
        self.assertTrue(r.ok, f"err={r.error_code} stderr={r.stderr_tail.decode(errors='replace')[:300]}")
        self.assertEqual(r.artifact.strip(), b"HELLO")        # last.txt gelesen (M3)
        self.assertIn(b"turn.completed", r.stdout)

    def test_sigterm_stops_scope_and_grandchild(self):
        # Review #9: SIGTERM an die Bridge (hier: an den Testprozess) -> run() räumt den Scope auf.
        import signal
        import threading
        probe_dir = home_project_probe_dir()
        self.addCleanup(shutil.rmtree, probe_dir, True)
        fd, pidfile = tempfile.mkstemp(prefix=".ariaprobe_pid_", dir=probe_dir)
        os.close(fd)
        argv = tuple(CAGE + ["/usr/bin/python3", "-I", FAKE_CODEX_CAGE, "sleep", "30", pidfile])

        def send_term():
            deadline = time.monotonic() + 8
            while time.monotonic() < deadline:
                try:
                    if int(Path(pidfile).read_text()) > 0:
                        break
                except (OSError, ValueError):
                    time.sleep(0.05)
            os.kill(os.getpid(), signal.SIGTERM)

        t = threading.Thread(target=send_term)
        t.start()
        r = self.runner(write_allowlist=[probe_dir]).run(RunSpec(argv, b"", {}, 30))
        t.join(10)
        self.assertEqual(r.error_code, "BRIDGE_TERMINATED")
        deadline = time.monotonic() + 8
        pid = int(Path(pidfile).read_text())
        while not process_gone(pid) and time.monotonic() < deadline:
            time.sleep(0.1)
        self.assertTrue(process_gone(pid), f"Enkelprozess {pid} laeuft noch (Scope nach SIGTERM nicht gestoppt)")

    def test_timeout_kills_scope_and_grandchild(self):
        # Enkel-PID in eine Datei im freigegebenen Probe-Ordner schreiben (ueberlebt das Aufraeumen des
        # Run-Roots); eigenes Wegwerf-Verzeichnis unter /home/aria, kein fremder Worktree.
        probe_dir = home_project_probe_dir()
        self.addCleanup(shutil.rmtree, probe_dir, True)
        fd, pidfile = tempfile.mkstemp(prefix=".ariaprobe_pid_", dir=probe_dir)
        os.close(fd)
        argv = tuple(CAGE + ["/usr/bin/python3", "-I", FAKE_CODEX_CAGE, "sleep", "30", pidfile])
        started = time.monotonic()
        r = self.runner(write_allowlist=[probe_dir]).run(RunSpec(argv, b"", {}, 2))
        self.assertEqual(r.error_code, "TIMEOUT")
        self.assertLess(time.monotonic() - started, 20)
        # Enkel muss durch den cgroup-Stop beendet sein
        deadline = time.monotonic() + 8
        pid = None
        while time.monotonic() < deadline:
            try:
                pid = int(Path(pidfile).read_text())
                break
            except (OSError, ValueError):
                time.sleep(0.1)
        self.assertIsNotNone(pid, "Enkel-PID wurde nicht geschrieben")
        while not process_gone(pid) and time.monotonic() < deadline:
            time.sleep(0.1)
        self.assertTrue(process_gone(pid), f"Enkelprozess {pid} laeuft noch (Scope nicht gekillt)")


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