"""Gegenproben am iso_exec-Kaefig (Landlock+seccomp+rlimits), ohne systemd-Bus.
Startet NUR python3 / /bin/true im Kaefig, nie codex. Belegt M1/M2/M3(wrapper)/N1-N4."""
import os
import tempfile
import unittest

from tests.iso_helpers import CAGE_PROBE, PYTHON, landlock_abi, run_in_cage, run_probe

PROD_SECRETS = [
    "/home/aria/.env",
    "/home/aria/.claude/.credentials.json",
    "/home/aria/.codex/auth.json",
]
PROD_SECRET_DIRS = ["/home/aria/.ssh", "/home/aria/.claude", "/home/aria/.config/gh"]


class CageBase(unittest.TestCase):
    def setUp(self):
        abi = landlock_abi()
        if abi < 4:
            self.skipTest(f"Landlock ABI {abi} < 4 auf diesem Host (Kaefig nicht einsetzbar)")
        self._tmp = tempfile.TemporaryDirectory()
        self.scratch = self._tmp.name

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

    def assertProbe(self, check, *args, **kw):
        r = run_probe(self.scratch, check, *args, **kw)
        self.assertEqual(r.returncode, 0, f"{check} {args}: rc={r.returncode} err={r.stderr.decode(errors='replace')[:200]}")


class ProdIsolation(CageBase):
    def test_prod_secret_files_read_denied(self):
        for p in PROD_SECRETS:
            if os.path.exists(p):
                with self.subTest(p=p):
                    self.assertProbe("read-denied", p)

    def test_prod_secret_dirs_list_denied(self):
        for p in PROD_SECRET_DIRS:
            if os.path.isdir(p):
                with self.subTest(p=p):
                    self.assertProbe("list-denied", p)

    def test_write_outside_rw_denied(self):
        self.assertProbe("write-denied", "/home/aria/.env")

    def test_write_in_rw_ok(self):
        self.assertProbe("write-ok", os.path.join(self.scratch, "ok.txt"))

    def test_env_allowlist_no_leak(self):
        token = "test-" + os.urandom(8).hex()
        saved = os.environ.get("TELEGRAM_BOT_TOKEN")
        os.environ["TELEGRAM_BOT_TOKEN"] = token
        try:
            self.assertProbe("env-clean")
        finally:
            if saved is None:
                os.environ.pop("TELEGRAM_BOT_TOKEN", None)
            else:
                os.environ["TELEGRAM_BOT_TOKEN"] = saved


class SeccompRules(CageBase):
    def test_memfd_create_denied(self):          # M2
        self.assertProbe("memfd-denied")

    def test_execveat_empty_path_denied(self):   # M2
        self.assertProbe("execveat-empty-denied")

    def test_unshare_denied(self):
        self.assertProbe("unshare-denied")

    def test_keyctl_denied(self):
        self.assertProbe("keyctl-denied")

    def test_io_uring_denied(self):
        self.assertProbe("iouring-denied")

    def test_ptrace_denied(self):
        self.assertProbe("ptrace-denied")

    def test_kill_foreign_pid_denied(self):
        self.assertProbe("kill-denied", "1")     # init

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

    def test_sendto_foreign_unix_denied(self):   # #8a
        self.assertProbe("sendto-unix-denied", "/run/user/1001/bus")

    def test_socketpair_af_unix_still_allowed(self):   # #8a: prozessintern bleibt erlaubt (V8/libuv)
        self.assertProbe("socketpair-afunix-ok")


class SeccompForeignProcess(CageBase):
    """Review #8b/#8c: kein Zugriff auf FREMDE Prozesse (gleiche UID) ueber F_SETOWN/SIGIO,
    prlimit64 oder process_madvise. Wegwerf-Sleeper ausserhalb des Kaefigs als Ziel."""

    def setUp(self):
        super().setUp()
        import subprocess
        self.sleeper = subprocess.Popen(["/usr/bin/sleep", "60"])
        self.addCleanup(self._kill_sleeper)
        self.fpid = str(self.sleeper.pid)

    def _kill_sleeper(self):
        try:
            self.sleeper.terminate()
            self.sleeper.wait(timeout=5)
        except Exception:
            pass

    def test_fcntl_setown_foreign_denied(self):     # #8b
        self.assertProbe("fcntl-setown-denied", self.fpid)

    def test_fcntl_setsig_denied(self):             # #8b
        self.assertProbe("fcntl-setsig-denied", "10")

    def test_prlimit_foreign_denied(self):          # #8c
        self.assertProbe("prlimit-foreign-denied", self.fpid)

    def test_prlimit_self_still_allowed(self):      # #8c: eigene PID (pid==0) bleibt erlaubt
        self.assertProbe("prlimit-self-ok")

    def test_process_madvise_denied(self):          # #8 (reachable auf Basis -> gesperrt)
        self.assertProbe("process-madvise-denied", self.fpid)


class WrapperLimits(CageBase):
    def test_rlimit_as_set_in_wrapper(self):     # M3
        self.assertProbe("rlimit", "AS", str(512 * 1024 * 1024),
                         rlimits={"AS": 512 * 1024 * 1024, "NOFILE": 1024})

    def test_rlimit_cpu_set_in_wrapper(self):    # M3
        self.assertProbe("rlimit", "CPU", "123", rlimits={"CPU": 123, "NOFILE": 1024})

    def test_rlimit_nofile_set(self):
        self.assertProbe("rlimit", "NOFILE", "777", rlimits={"NOFILE": 777})

    def test_umask_077(self):                    # N2
        self.assertProbe("umask-077")

    def test_cwd_set(self):                       # N2
        self.assertProbe("cwd", self.scratch)


class FdHandling(CageBase):
    def test_inherited_fd_closed_before_restrict(self):   # N1
        fd = os.open("/etc/hostname", os.O_RDONLY)
        try:
            cfg = {
                "min_abi": 4, "cwd": self.scratch, "rx": ["/usr"],
                "file_ro": ["/etc/ld.so.cache", CAGE_PROBE], "rw": [self.scratch],
                "env": {"PATH": "/usr/bin:/bin"}, "rlimits": {"NOFILE": 1024},
            }
            r = run_in_cage(cfg, [PYTHON, "-I", CAGE_PROBE, "fd-closed", str(fd)], pass_fds=(fd,))
            self.assertEqual(r.returncode, 0, r.stderr.decode(errors="replace")[:200])
        finally:
            os.close(fd)


class FailClosed(CageBase):
    def test_abi_requirement_fail_closed(self):   # N3
        cfg = {"min_abi": 99, "cwd": self.scratch, "rw": [self.scratch],
               "env": {"PATH": "/usr/bin:/bin"}, "rlimits": {"NOFILE": 64}}
        r = run_in_cage(cfg, ["/usr/bin/true"])
        self.assertEqual(r.returncode, 70)
        self.assertIn(b"ABI", r.stderr)

    def test_symlink_config_path_rejected(self):  # N4
        target = os.path.join(self.scratch, "real.txt")
        with open(target, "w") as fh:
            fh.write("x")
        link = os.path.join(self.scratch, "link.txt")
        os.symlink(target, link)
        cfg = {"min_abi": 4, "cwd": self.scratch, "rx": ["/usr"],
               "file_ro": ["/etc/ld.so.cache", link], "rw": [self.scratch],
               "env": {"PATH": "/usr/bin:/bin"}, "rlimits": {"NOFILE": 64}}
        r = run_in_cage(cfg, ["/usr/bin/true"])
        self.assertEqual(r.returncode, 70)
        self.assertIn(b"Symlink", r.stderr)

    def test_missing_config_path_fail_closed(self):
        cfg = {"min_abi": 4, "cwd": self.scratch, "file_x": ["/nonexistent/prog"],
               "rw": [self.scratch], "env": {"PATH": "/usr/bin:/bin"}, "rlimits": {"NOFILE": 64}}
        r = run_in_cage(cfg, ["/usr/bin/true"])
        self.assertEqual(r.returncode, 70)

    def test_execute_nonallowlisted_denied(self):
        # cage_probe ist file_ro (lesbar, NICHT ausfuehrbar) -> exec muss EACCES sein
        self.assertProbe("exec-denied", CAGE_PROBE)


class AuthReadonly(CageBase):
    def _stage(self, writable):
        authdir = os.path.join(self.scratch, "authdir")
        codex_home = os.path.join(self.scratch, "codex-home")
        os.mkdir(authdir)
        os.mkdir(codex_home)
        auth = os.path.join(authdir, "auth.json")
        with open(auth, "w") as fh:
            fh.write('{"token":"x"}')
        os.chmod(auth, 0o600 if writable else 0o400)
        link = os.path.join(codex_home, "auth.json")
        os.symlink(auth, link)
        file_rw = [auth] if writable else []
        file_ro = ["/etc/ld.so.cache", CAGE_PROBE] + ([] if writable else [auth])
        cfg = {
            "min_abi": 4, "cwd": self.scratch, "rx": ["/usr"],
            "file_ro": file_ro, "file_rw": file_rw,
            "rw": [self.scratch, codex_home],   # codex_home rw, authdir bewusst NICHT rw
            "env": {"PATH": "/usr/bin:/bin"}, "rlimits": {"NOFILE": 1024},
        }
        return cfg, link

    def test_auth_copy_read_only_by_default(self):   # M1
        cfg, link = self._stage(writable=False)
        ro = run_in_cage(cfg, [PYTHON, "-I", CAGE_PROBE, "read-ok", link])
        self.assertEqual(ro.returncode, 0, ro.stderr.decode(errors="replace")[:200])
        wr = run_in_cage(cfg, [PYTHON, "-I", CAGE_PROBE, "write-denied", link])
        self.assertEqual(wr.returncode, 0, wr.stderr.decode(errors="replace")[:200])

    def test_auth_copy_writable_when_enabled(self):
        cfg, link = self._stage(writable=True)
        wr = run_in_cage(cfg, [PYTHON, "-I", CAGE_PROBE, "write-ok", link])
        self.assertEqual(wr.returncode, 0, wr.stderr.decode(errors="replace")[:200])


class ReadOnlyContentImmutable(CageBase):
    """Re-Review 2: eine file_ro-Datei ist INHALTLICH unveraenderbar (Kernschutz gegen Nutzlast),
    auch wenn utimensat (touch) von Landlock nicht mediiert wird. Synthetische ro-Datei (kein echtes
    Home), deshalb nebenwirkungsfrei."""

    def test_ro_file_content_immutable_touch_may_change_mtime(self):
        import shutil as _sh
        import tempfile as _tf
        d = _tf.mkdtemp()
        self.addCleanup(_sh.rmtree, d, True)
        ro = os.path.join(d, "startup.rc")
        with open(ro, "w", encoding="utf-8") as fh:
            fh.write("ORIGINAL\n")
        # Datei read-only; ihr Verzeichnis read-only (kein MAKE_REG) -> nur Lesen, kein Schreiben/Anlegen.
        r = run_probe(self.scratch, "ro-content-immutable", ro, extra={"file_ro": [ro], "ro": [d]})
        self.assertEqual(r.returncode, 0, r.stderr.decode(errors="replace")[:200])
        with open(ro, encoding="utf-8") as fh:
            self.assertEqual(fh.read(), "ORIGINAL\n")   # Inhalt NICHT veraendert (keine Nutzlast)


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