#!/usr/bin/env python3
"""Pure-function tests for seat_scope_entry (no root, no fixture env)."""

from __future__ import annotations

import argparse
import json
import os
import stat
import sys
import tempfile
import unittest
from unittest.mock import patch

_LIB = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "lib")
if _LIB not in sys.path:
    sys.path.insert(0, _LIB)

from seat_scope_entry import (  # noqa: E402
    INSTALLED_MANIFEST,
    INSTALLED_MODULE,
    INSTALLED_TMUX_MEDIATOR,
    INSTALLED_WRAPPER,
    PR_CAPBSET_DROP,
    PR_SET_NO_NEW_PRIVS,
    SUDOERS_AUTHORIZED_WRAPPER,
    SUDOERS_AUTHORIZED_MEDIATOR,
    SUDOERS_AUTHORIZED_IMPLEMENTER_EXEC,
    SUDOERS_FILE,
    ScopeEntryPaths,
    become_user,
    build_arg_parser,
    build_implementer_invocation,
    die,
    drop_bounding_caps,
    file_sha256,
    load_install_manifest,
    ns_inode,
    parse_proc_status,
    reject_path_aliases,
    reject_symlink_ancestors,
    run_install_check,
    set_no_new_privs,
    validate_args,
    validate_exact_operator_path,
    validate_static,
    verify_tmux_support,
    verify_zero_privileges,
    SEAT_ID_RE,
)


def _fixture_control_base(tmp: str) -> str:
    base = os.path.join(tmp, "seat-control")
    os.makedirs(base, mode=0o711, exist_ok=True)
    os.chmod(base, 0o711)
    return base


def _verify_dir_without_owner(path: str, *, mode: int) -> None:
    st = os.lstat(path)
    if stat.S_ISLNK(st.st_mode):
        die(f"install-check-symlink:{path}", 1)
    if not stat.S_ISDIR(st.st_mode):
        die(f"install-check-not-dir:{path}", 1)
    if stat.S_IMODE(st.st_mode) != mode:
        die(f"install-check-mode:{path}", 1)


class SeatScopeEntryPureTests(unittest.TestCase):
    def test_seat_id_pattern(self) -> None:
        self.assertTrue(SEAT_ID_RE.match("seat-x"))
        self.assertFalse(SEAT_ID_RE.match("bad/id"))

    def test_validate_static_rejects_bad_seat_id(self) -> None:
        parser = build_arg_parser()
        ns = parser.parse_args([
            "--seat-id", "bad/id",
            "--netns-path", "/var/run/netns/overdeck-seat-x",
            "--unit", "agent-seat-x.scope",
            "--socket", "/tmp/s",
            "--launcher", "/tmp/l",
            "--slice", "agent-seat.slice",
            "--seat-host", "debian1",
            "--seat-model", "gpt-5.6-terra",
            "--claude-bin", "/tmp/claude",
            "--guard-bin", "/tmp/guard",
            "--checkout", "/tmp/repo",
            "--account-slug", "roy",
        ])
        with self.assertRaises(SystemExit) as exc:
            validate_static(ns)
        self.assertEqual(exc.exception.code, 2)

    def test_ns_inode_followed_stat_identity(self) -> None:
        import tempfile

        with tempfile.TemporaryDirectory() as tmp:
            ns_file = os.path.join(tmp, "netns-object")
            with open(ns_file, "wb") as fh:
                fh.write(b"\0")
            proc_net = os.path.join(tmp, "proc-net")
            os.symlink(ns_file, proc_net)
            self.assertEqual(ns_inode(ns_file), ns_inode(proc_net))
            other = os.path.join(tmp, "other-netns")
            with open(other, "wb") as fh:
                fh.write(b"\0")
            self.assertNotEqual(ns_inode(ns_file), ns_inode(other))

    def test_validate_args_skip_host_checks(self) -> None:
        parser = build_arg_parser()
        ns = parser.parse_args([
            "--seat-id", "seat-x",
            "--netns-path", "/var/run/netns/overdeck-seat-seat-x",
            "--unit", "agent-seat-seat-x.scope",
            "--socket", "/home/u/.local/state/overdeck/seats/seat-x/tmux.sock",
            "--launcher", "/home/u/.claude/bin/seat-launcher",
            "--slice", "agent-seat.slice",
            "--seat-host", "debian1",
            "--seat-model", "gpt-5.6-terra",
            "--claude-bin", "/home/u/.local/bin/claude",
            "--guard-bin", "/home/u/.local/share/overdeck/seat-guard/current/bin",
            "--checkout", "/home/u/seats/seat-x/repo",
            "--account-slug", "roy",
        ])
        validate_args(ns, "/home/u", skip_host_checks=True)

    def test_die_prefix(self) -> None:
        with self.assertRaises(SystemExit) as exc:
            die("requires root", 1)
        self.assertEqual(exc.exception.code, 1)

    def test_install_check_rejects_extra_args(self) -> None:
        from seat_scope_entry import main

        with self.assertRaises(SystemExit) as exc:
            main(["--install-check", "--seat-id", "x"])
        self.assertEqual(exc.exception.code, 2)

    def test_load_install_manifest_contract(self) -> None:
        import tempfile

        def verify_without_owner(path: str, *, mode: int, must_exec: bool) -> None:
            st = os.lstat(path)
            if stat.S_ISLNK(st.st_mode):
                die(f"install-check-symlink:{path}", 1)
            if not stat.S_ISREG(st.st_mode):
                die(f"install-check-not-file:{path}", 1)
            if stat.S_IMODE(st.st_mode) != mode:
                die(f"install-check-mode:{path}", 1)
            if must_exec and not os.access(path, os.X_OK):
                die(f"install-check-not-executable:{path}", 1)

        with tempfile.TemporaryDirectory() as tmp:
            control_base = _fixture_control_base(tmp)
            module = os.path.join(tmp, "seat_scope_entry.py")
            wrapper = os.path.join(tmp, "overdeck-seat-scope-entry")
            manifest = os.path.join(tmp, "manifest.json")
            sudoers = os.path.join(tmp, "sudoers")
            with open(module, "w", encoding="utf-8") as fh:
                fh.write("# module\n")
            os.chmod(module, 0o644)
            with open(wrapper, "w", encoding="utf-8") as fh:
                fh.write("#!/usr/bin/env python3\n/usr/local/lib/overdeck/seat_scope_entry.py\n")
            os.chmod(wrapper, 0o755)
            payload = {
                "wrapper_sha256": file_sha256(wrapper),
                "module_sha256": file_sha256(module),
                "common_module_sha256": file_sha256(module),
                "implementer_exec_sha256": file_sha256(wrapper),
                "implementer_module_sha256": file_sha256(module),
                "tmux_mediator_sha256": file_sha256(wrapper),
                "tmux_mediator_module_sha256": file_sha256(module),
                "identity_module_sha256": file_sha256(module),
                "ssh_user": "fixture-user",
            }
            with open(manifest, "w", encoding="utf-8") as fh:
                json.dump(payload, fh)
            os.chmod(manifest, 0o600)
            with open(sudoers, "w", encoding="utf-8") as fh:
                fh.write(f"fixture-user ALL=(root) NOPASSWD: {SUDOERS_AUTHORIZED_WRAPPER} *\n")
                fh.write(f"fixture-user ALL=(root) NOPASSWD: {SUDOERS_AUTHORIZED_MEDIATOR} *\n")
                fh.write(f"fixture-user ALL=(root) NOPASSWD: {SUDOERS_AUTHORIZED_IMPLEMENTER_EXEC} *\n")
            os.chmod(sudoers, 0o440)
            with patch("seat_scope_entry.INSTALLED_MANIFEST", manifest), patch(
                "seat_scope_entry.INSTALLED_MODULE", module
            ), patch("seat_scope_entry.INSTALLED_WRAPPER", wrapper), patch(
                "seat_scope_entry.SUDOERS_FILE", sudoers
            ), patch("seat_scope_entry.verify_installed_file", verify_without_owner):
                loaded = load_install_manifest()
                self.assertEqual(loaded, payload)

    def test_install_check_requires_root(self) -> None:
        with patch("seat_scope_entry.os.geteuid", return_value=1000):
            with self.assertRaises(SystemExit) as exc:
                run_install_check()
            self.assertEqual(exc.exception.code, 1)

    def test_install_check_contract(self) -> None:
        import tempfile

        def verify_without_owner(path: str, *, mode: int, must_exec: bool) -> None:
            st = os.lstat(path)
            if stat.S_ISLNK(st.st_mode):
                die(f"install-check-symlink:{path}", 1)
            if not stat.S_ISREG(st.st_mode):
                die(f"install-check-not-file:{path}", 1)
            if stat.S_IMODE(st.st_mode) != mode:
                die(f"install-check-mode:{path}", 1)
            if must_exec and not os.access(path, os.X_OK):
                die(f"install-check-not-executable:{path}", 1)

        def write_fixture_wrapper(path: str, embedded_module: str) -> None:
            with open(path, "w", encoding="utf-8") as fh:
                fh.write(f"#!/usr/bin/env python3\n_MODULE_PATH = \"{embedded_module}\"\n")
            os.chmod(path, 0o755)

        with tempfile.TemporaryDirectory() as tmp:
            control_base = _fixture_control_base(tmp)
            module = os.path.join(tmp, "seat_scope_entry.py")
            wrapper = os.path.join(tmp, "overdeck-seat-scope-entry")
            implementer_wrapper = os.path.join(tmp, "overdeck-seat-implementer-exec")
            mediator_wrapper = os.path.join(tmp, "overdeck-seat-tmux-mediator")
            manifest = os.path.join(tmp, "manifest.json")
            sudoers = os.path.join(tmp, "sudoers")
            with open(module, "w", encoding="utf-8") as fh:
                fh.write("# module\n")
            os.chmod(module, 0o644)
            write_fixture_wrapper(wrapper, "/usr/local/lib/overdeck/seat_scope_entry.py")
            write_fixture_wrapper(implementer_wrapper, "/usr/local/lib/overdeck/seat_implementer_exec.py")
            write_fixture_wrapper(mediator_wrapper, "/usr/local/lib/overdeck/seat_tmux_mediator.py")
            payload = {
                "wrapper_sha256": file_sha256(wrapper),
                "module_sha256": file_sha256(module),
                "common_module_sha256": file_sha256(module),
                "implementer_exec_sha256": file_sha256(implementer_wrapper),
                "implementer_module_sha256": file_sha256(module),
                "tmux_mediator_sha256": file_sha256(mediator_wrapper),
                "tmux_mediator_module_sha256": file_sha256(module),
                "identity_module_sha256": file_sha256(module),
                "ssh_user": "fixture-user",
            }
            with open(manifest, "w", encoding="utf-8") as fh:
                json.dump(payload, fh)
            os.chmod(manifest, 0o600)
            with open(sudoers, "w", encoding="utf-8") as fh:
                fh.write(f"fixture-user ALL=(root) NOPASSWD: {SUDOERS_AUTHORIZED_WRAPPER} *\n")
                fh.write(f"fixture-user ALL=(root) NOPASSWD: {SUDOERS_AUTHORIZED_MEDIATOR} *\n")
                fh.write(f"fixture-user ALL=(root) NOPASSWD: {SUDOERS_AUTHORIZED_IMPLEMENTER_EXEC} *\n")
            os.chmod(sudoers, 0o440)

            with patch("seat_scope_entry.INSTALLED_MANIFEST", manifest), patch(
                "seat_scope_entry.INSTALLED_MODULE", module
            ), patch("seat_scope_entry.INSTALLED_WRAPPER", wrapper), patch(
                "seat_scope_entry.INSTALLED_IMPLEMENTER_EXEC", implementer_wrapper
            ), patch("seat_scope_entry.INSTALLED_IMPLEMENTER_MODULE", module), patch(
                "seat_scope_entry.INSTALLED_TMUX_MEDIATOR", mediator_wrapper
            ), patch("seat_scope_entry.INSTALLED_TMUX_MEDIATOR_MODULE", module), patch(
                "seat_scope_entry.INSTALLED_COMMON_MODULE", module
            ), patch(
                "seat_scope_entry.INSTALLED_IDENTITY_MODULE", module
            ), patch(
                "seat_scope_entry.SUDOERS_FILE", sudoers
            ), patch("seat_scope_entry.verify_installed_file", verify_without_owner), patch(
                "seat_scope_entry.verify_installed_dir", _verify_dir_without_owner
            ), patch("seat_scope_entry.seat_control_base", return_value=control_base
            ), patch("seat_scope_entry.os.geteuid", return_value=0
            ), patch("seat_scope_entry.shutil.which", return_value="/usr/bin/setfacl"):
                run_install_check()

    def test_install_check_rejects_non_traversable_control_base(self) -> None:
        import tempfile

        def verify_without_owner(path: str, *, mode: int, must_exec: bool) -> None:
            st = os.lstat(path)
            if stat.S_ISLNK(st.st_mode):
                die(f"install-check-symlink:{path}", 1)
            if not stat.S_ISREG(st.st_mode):
                die(f"install-check-not-file:{path}", 1)
            if stat.S_IMODE(st.st_mode) != mode:
                die(f"install-check-mode:{path}", 1)
            if must_exec and not os.access(path, os.X_OK):
                die(f"install-check-not-executable:{path}", 1)

        with tempfile.TemporaryDirectory() as tmp:
            control_base = _fixture_control_base(tmp)
            os.chmod(control_base, 0o700)
            module = os.path.join(tmp, "seat_scope_entry.py")
            wrapper = os.path.join(tmp, "overdeck-seat-scope-entry")
            implementer_wrapper = os.path.join(tmp, "overdeck-seat-implementer-exec")
            mediator_wrapper = os.path.join(tmp, "overdeck-seat-tmux-mediator")
            manifest = os.path.join(tmp, "manifest.json")
            sudoers = os.path.join(tmp, "sudoers")
            with open(module, "w", encoding="utf-8") as fh:
                fh.write("# module\n")
            os.chmod(module, 0o644)
            with open(wrapper, "w", encoding="utf-8") as fh:
                fh.write('#!/usr/bin/env python3\n_MODULE_PATH = "/usr/local/lib/overdeck/seat_scope_entry.py"\n')
            os.chmod(wrapper, 0o755)
            for path in (implementer_wrapper, mediator_wrapper):
                with open(path, "w", encoding="utf-8") as fh:
                    fh.write('#!/usr/bin/env python3\n_MODULE_PATH = "/usr/local/lib/overdeck/seat_scope_entry.py"\n')
                os.chmod(path, 0o755)
            payload = {
                "wrapper_sha256": file_sha256(wrapper),
                "module_sha256": file_sha256(module),
                "common_module_sha256": file_sha256(module),
                "implementer_exec_sha256": file_sha256(implementer_wrapper),
                "implementer_module_sha256": file_sha256(module),
                "tmux_mediator_sha256": file_sha256(mediator_wrapper),
                "tmux_mediator_module_sha256": file_sha256(module),
                "identity_module_sha256": file_sha256(module),
                "ssh_user": "fixture-user",
            }
            with open(manifest, "w", encoding="utf-8") as fh:
                json.dump(payload, fh)
            os.chmod(manifest, 0o600)
            with open(sudoers, "w", encoding="utf-8") as fh:
                fh.write(f"fixture-user ALL=(root) NOPASSWD: {SUDOERS_AUTHORIZED_WRAPPER} *\n")
                fh.write(f"fixture-user ALL=(root) NOPASSWD: {SUDOERS_AUTHORIZED_MEDIATOR} *\n")
                fh.write(f"fixture-user ALL=(root) NOPASSWD: {SUDOERS_AUTHORIZED_IMPLEMENTER_EXEC} *\n")
            os.chmod(sudoers, 0o440)

            with patch("seat_scope_entry.INSTALLED_MANIFEST", manifest), patch(
                "seat_scope_entry.INSTALLED_MODULE", module
            ), patch("seat_scope_entry.INSTALLED_WRAPPER", wrapper), patch(
                "seat_scope_entry.INSTALLED_IMPLEMENTER_EXEC", implementer_wrapper
            ), patch("seat_scope_entry.INSTALLED_IMPLEMENTER_MODULE", module), patch(
                "seat_scope_entry.INSTALLED_TMUX_MEDIATOR", mediator_wrapper
            ), patch("seat_scope_entry.INSTALLED_TMUX_MEDIATOR_MODULE", module), patch(
                "seat_scope_entry.INSTALLED_COMMON_MODULE", module
            ), patch(
                "seat_scope_entry.INSTALLED_IDENTITY_MODULE", module
            ), patch(
                "seat_scope_entry.SUDOERS_FILE", sudoers
            ), patch("seat_scope_entry.verify_installed_file", verify_without_owner), patch(
                "seat_scope_entry.verify_installed_dir", _verify_dir_without_owner
            ), patch("seat_scope_entry.seat_control_base", return_value=control_base
            ), patch("seat_scope_entry.os.geteuid", return_value=0
            ), patch("seat_scope_entry.shutil.which", return_value="/usr/bin/setfacl"):
                with self.assertRaises(SystemExit) as exc:
                    run_install_check()
                self.assertEqual(exc.exception.code, 1)

    def test_install_check_rejects_fstring_module_path(self) -> None:
        import tempfile

        def verify_without_owner(path: str, *, mode: int, must_exec: bool) -> None:
            st = os.lstat(path)
            if stat.S_ISLNK(st.st_mode):
                die(f"install-check-symlink:{path}", 1)
            if not stat.S_ISREG(st.st_mode):
                die(f"install-check-not-file:{path}", 1)
            if stat.S_IMODE(st.st_mode) != mode:
                die(f"install-check-mode:{path}", 1)
            if must_exec and not os.access(path, os.X_OK):
                die(f"install-check-not-executable:{path}", 1)

        with tempfile.TemporaryDirectory() as tmp:
            control_base = _fixture_control_base(tmp)
            module = os.path.join(tmp, "seat_scope_entry.py")
            wrapper = os.path.join(tmp, "overdeck-seat-scope-entry")
            implementer_wrapper = os.path.join(tmp, "overdeck-seat-implementer-exec")
            mediator_wrapper = os.path.join(tmp, "overdeck-seat-tmux-mediator")
            manifest = os.path.join(tmp, "manifest.json")
            sudoers = os.path.join(tmp, "sudoers")
            with open(module, "w", encoding="utf-8") as fh:
                fh.write("# module\n")
            os.chmod(module, 0o644)
            with open(wrapper, "w", encoding="utf-8") as fh:
                fh.write('#!/usr/bin/env python3\n_MODULE_PATH = f"{_OVERDECK_LIB}/seat_scope_entry.py"\n')
            os.chmod(wrapper, 0o755)
            for path, embedded in (
                (implementer_wrapper, "/usr/local/lib/overdeck/seat_implementer_exec.py"),
                (mediator_wrapper, "/usr/local/lib/overdeck/seat_tmux_mediator.py"),
            ):
                with open(path, "w", encoding="utf-8") as fh:
                    fh.write(f"#!/usr/bin/env python3\n_MODULE_PATH = \"{embedded}\"\n")
                os.chmod(path, 0o755)
            payload = {
                "wrapper_sha256": file_sha256(wrapper),
                "module_sha256": file_sha256(module),
                "common_module_sha256": file_sha256(module),
                "implementer_exec_sha256": file_sha256(implementer_wrapper),
                "implementer_module_sha256": file_sha256(module),
                "tmux_mediator_sha256": file_sha256(mediator_wrapper),
                "tmux_mediator_module_sha256": file_sha256(module),
                "identity_module_sha256": file_sha256(module),
                "ssh_user": "fixture-user",
            }
            with open(manifest, "w", encoding="utf-8") as fh:
                json.dump(payload, fh)
            os.chmod(manifest, 0o600)
            with open(sudoers, "w", encoding="utf-8") as fh:
                fh.write(f"fixture-user ALL=(root) NOPASSWD: {SUDOERS_AUTHORIZED_WRAPPER} *\n")
                fh.write(f"fixture-user ALL=(root) NOPASSWD: {SUDOERS_AUTHORIZED_MEDIATOR} *\n")
                fh.write(f"fixture-user ALL=(root) NOPASSWD: {SUDOERS_AUTHORIZED_IMPLEMENTER_EXEC} *\n")
            os.chmod(sudoers, 0o440)

            with patch("seat_scope_entry.INSTALLED_MANIFEST", manifest), patch(
                "seat_scope_entry.INSTALLED_MODULE", module
            ), patch("seat_scope_entry.INSTALLED_WRAPPER", wrapper), patch(
                "seat_scope_entry.INSTALLED_IMPLEMENTER_EXEC", implementer_wrapper
            ), patch("seat_scope_entry.INSTALLED_IMPLEMENTER_MODULE", module), patch(
                "seat_scope_entry.INSTALLED_TMUX_MEDIATOR", mediator_wrapper
            ), patch("seat_scope_entry.INSTALLED_TMUX_MEDIATOR_MODULE", module), patch(
                "seat_scope_entry.INSTALLED_COMMON_MODULE", module
            ), patch(
                "seat_scope_entry.INSTALLED_IDENTITY_MODULE", module
            ), patch(
                "seat_scope_entry.SUDOERS_FILE", sudoers
            ), patch("seat_scope_entry.verify_installed_file", verify_without_owner), patch(
                "seat_scope_entry.verify_installed_dir", _verify_dir_without_owner
            ), patch("seat_scope_entry.seat_control_base", return_value=control_base
            ), patch("seat_scope_entry.os.geteuid", return_value=0
            ), patch("seat_scope_entry.shutil.which", return_value="/usr/bin/setfacl"):
                with self.assertRaises(SystemExit) as exc:
                    run_install_check()
                self.assertEqual(exc.exception.code, 1)


class PrivilegeDropTests(unittest.TestCase):
    _CLEAN_STATUS = """\
Name:\ttest
CapInh:\t0000000000000000
CapPrm:\t0000000000000000
CapEff:\t0000000000000000
CapBnd:\t0000000000000000
CapAmb:\t0000000000000000
NoNewPrivs:\t1
"""

    def test_parse_proc_status(self) -> None:
        fields = parse_proc_status(self._CLEAN_STATUS)
        self.assertEqual(fields["CapBnd"], "0000000000000000")
        self.assertEqual(fields["NoNewPrivs"], "1")

    def test_verify_zero_privileges_ok(self) -> None:
        import tempfile

        with tempfile.NamedTemporaryFile("w", encoding="utf-8", delete=False) as fh:
            fh.write(self._CLEAN_STATUS)
            path = fh.name
        try:
            verify_zero_privileges(status_path=path)
        finally:
            os.unlink(path)

    def test_verify_zero_privileges_fails_nonzero_bound(self) -> None:
        import tempfile

        bad = self._CLEAN_STATUS.replace("CapBnd:\t0000000000000000", "CapBnd:\t0000000000000001")
        with tempfile.NamedTemporaryFile("w", encoding="utf-8", delete=False) as fh:
            fh.write(bad)
            path = fh.name
        try:
            with self.assertRaises(SystemExit) as exc:
                verify_zero_privileges(status_path=path)
            self.assertEqual(exc.exception.code, 1)
        finally:
            os.unlink(path)

    def test_verify_zero_privileges_fails_missing_no_new_privs(self) -> None:
        import tempfile

        bad = self._CLEAN_STATUS.replace("NoNewPrivs:\t1", "NoNewPrivs:\t0")
        with tempfile.NamedTemporaryFile("w", encoding="utf-8", delete=False) as fh:
            fh.write(bad)
            path = fh.name
        try:
            with self.assertRaises(SystemExit) as exc:
                verify_zero_privileges(status_path=path)
            self.assertEqual(exc.exception.code, 1)
        finally:
            os.unlink(path)

    def test_drop_bounding_caps_calls_prctl_for_each_cap(self) -> None:
        calls: list[tuple[int, int]] = []

        def fake_prctl(option: int, arg2: int, _arg3: int, _arg4: int, _arg5: int) -> int:
            calls.append((option, arg2))
            return 0

        drop_bounding_caps(prctl_fn=fake_prctl, cap_last_cap=2)
        self.assertEqual(calls, [(PR_CAPBSET_DROP, 0), (PR_CAPBSET_DROP, 1), (PR_CAPBSET_DROP, 2)])

    def test_drop_bounding_caps_prctl_failure(self) -> None:
        def fail_prctl(_option: int, _arg2: int, _arg3: int, _arg4: int, _arg5: int) -> int:
            ctypes = __import__("ctypes")
            ctypes.set_errno(1)
            return -1

        with self.assertRaises(SystemExit) as exc:
            drop_bounding_caps(prctl_fn=fail_prctl, cap_last_cap=0)
        self.assertEqual(exc.exception.code, 1)

    def test_set_no_new_privs_calls_prctl(self) -> None:
        calls: list[tuple[int, int]] = []

        def fake_prctl(option: int, arg2: int, _arg3: int, _arg4: int, _arg5: int) -> int:
            calls.append((option, arg2))
            return 0

        set_no_new_privs(prctl_fn=fake_prctl)
        self.assertEqual(calls, [(PR_SET_NO_NEW_PRIVS, 1)])

    def test_become_user_ordering(self) -> None:
        order: list[str] = []

        with patch("seat_scope_entry.grp.getgrall", return_value=[]), patch(
            "seat_scope_entry.os.setgroups", side_effect=lambda _groups: order.append("setgroups")
        ), patch(
            "seat_scope_entry.os.setresgid", side_effect=lambda *_args: order.append("setresgid")
        ), patch(
            "seat_scope_entry.drop_bounding_caps", side_effect=lambda: order.append("drop_bounding_caps")
        ), patch(
            "seat_scope_entry.os.setresuid", side_effect=lambda *_args: order.append("setresuid")
        ), patch(
            "seat_scope_entry.set_no_new_privs", side_effect=lambda: order.append("set_no_new_privs")
        ), patch(
            "seat_scope_entry.verify_zero_privileges", side_effect=lambda: order.append("verify_zero_privileges")
        ):
            become_user(1000, 1000, "fixture-user")

        self.assertEqual(
            order,
            [
                "setgroups",
                "setresgid",
                "drop_bounding_caps",
                "setresuid",
                "set_no_new_privs",
                "verify_zero_privileges",
            ],
        )


class EffectiveBoundaryTests(unittest.TestCase):
    def test_verify_pid_effective_boundary_rejects_tmux_spawn(self) -> None:
        import tempfile
        from seat_scope_entry import ScopeEntryPaths, verify_pid_effective_boundary, die

        with tempfile.TemporaryDirectory() as tmp:
            proc = os.path.join(tmp, "proc", "123")
            cg = os.path.join(tmp, "sys/fs/cgroup/user.slice/tmux-spawn-1.scope")
            os.makedirs(proc, exist_ok=True)
            os.makedirs(cg, exist_ok=True)
            with open(os.path.join(proc, "cgroup"), "w", encoding="utf-8") as fh:
                fh.write("0::/user.slice/tmux-spawn-1.scope\n")
            with open(os.path.join(cg, "memory.max"), "w", encoding="utf-8") as fh:
                fh.write("max\n")
            with open(os.path.join(cg, "cpu.max"), "w", encoding="utf-8") as fh:
                fh.write("max 100000\n")
            paths = ScopeEntryPaths(proc_root=os.path.join(tmp, "proc"), cgroup_root=os.path.join(tmp, "sys/fs/cgroup"))
            with self.assertRaises(SystemExit) as exc:
                verify_pid_effective_boundary(paths, 123, "agent-seat-seat-x.scope")
            self.assertEqual(exc.exception.code, 1)

    def test_verify_pid_effective_boundary_accepts_seat_scope(self) -> None:
        import tempfile
        from seat_scope_entry import ScopeEntryPaths, verify_pid_effective_boundary

        with tempfile.TemporaryDirectory() as tmp:
            proc = os.path.join(tmp, "proc", "123")
            cg = os.path.join(tmp, "sys/fs/cgroup/user.slice/agent-seat.slice/agent-seat-seat-x.scope")
            os.makedirs(proc, exist_ok=True)
            os.makedirs(cg, exist_ok=True)
            with open(os.path.join(proc, "cgroup"), "w", encoding="utf-8") as fh:
                fh.write("0::/user.slice/agent-seat.slice/agent-seat-seat-x.scope\n")
            with open(os.path.join(cg, "memory.max"), "w", encoding="utf-8") as fh:
                fh.write("12884901888\n")
            with open(os.path.join(cg, "memory.swap.max"), "w", encoding="utf-8") as fh:
                fh.write("1073741824\n")
            with open(os.path.join(cg, "cpu.max"), "w", encoding="utf-8") as fh:
                fh.write("400000 100000\n")
            with open(os.path.join(cg, "cpu.weight"), "w", encoding="utf-8") as fh:
                fh.write("1\n")
            with open(os.path.join(cg, "pids.max"), "w", encoding="utf-8") as fh:
                fh.write("256\n")
            paths = ScopeEntryPaths(proc_root=os.path.join(tmp, "proc"), cgroup_root=os.path.join(tmp, "sys/fs/cgroup"))
            verify_pid_effective_boundary(paths, 123, "agent-seat-seat-x.scope")


class ArgumentValidationTests(unittest.TestCase):
    def _valid_ns(self, parser: argparse.ArgumentParser, *, checkout: str = "/home/u/seats/seat-x/repo") -> argparse.Namespace:
        return parser.parse_args([
            "--seat-id", "seat-x",
            "--netns-path", "/var/run/netns/overdeck-seat-seat-x",
            "--unit", "agent-seat-seat-x.scope",
            "--socket", "/home/u/.local/state/overdeck/seats/seat-x/tmux.sock",
            "--launcher", "/home/u/.claude/bin/seat-launcher",
            "--slice", "agent-seat.slice",
            "--seat-host", "debian1",
            "--seat-model", "gpt-5.6-terra",
            "--claude-bin", "/home/u/.local/bin/claude",
            "--guard-bin", "/home/u/.local/share/overdeck/seat-guard/current/bin",
            "--checkout", checkout,
            "--account-slug", "roy",
        ])

    def test_validate_args_rejects_checkout_drift(self) -> None:
        parser = build_arg_parser()
        ns = self._valid_ns(parser, checkout="/tmp/evil/repo")
        with self.assertRaises(SystemExit) as exc:
            validate_args(ns, "/home/u", skip_host_checks=True)
        self.assertEqual(exc.exception.code, 2)

    def test_validate_args_rejects_checkout_traversal_alias(self) -> None:
        parser = build_arg_parser()
        ns = self._valid_ns(parser, checkout="/home/u/seats/seat-x/../seat-x/repo")
        with self.assertRaises(SystemExit):
            validate_args(ns, "/home/u", skip_host_checks=True)

    def test_validate_args_rejects_checkout_trailing_slash(self) -> None:
        parser = build_arg_parser()
        ns = self._valid_ns(parser, checkout="/home/u/seats/seat-x/repo/")
        with self.assertRaises(SystemExit):
            validate_args(ns, "/home/u", skip_host_checks=True)

    def test_validate_args_rejects_alternate_claude_binary(self) -> None:
        parser = build_arg_parser()
        ns = self._valid_ns(parser)
        ns.claude_bin = "/home/u/.local/bin/claude-real"
        with self.assertRaises(SystemExit):
            validate_args(ns, "/home/u", skip_host_checks=True)

    def test_reject_path_aliases_flags_dot_segments(self) -> None:
        with self.assertRaises(SystemExit):
            reject_path_aliases("/home/u/./seats/seat-x/repo", "checkout")

    def test_reject_symlink_ancestors_flags_parent_symlink(self) -> None:
        import tempfile

        with tempfile.TemporaryDirectory() as tmp:
            home = os.path.join(tmp, "home")
            os.makedirs(home, mode=0o700)
            seats = os.path.join(home, "seats")
            os.makedirs(seats, mode=0o700)
            os.symlink(seats, os.path.join(home, "seats-link"))
            checkout = os.path.join(home, "seats-link", "seat-x", "repo")
            os.makedirs(os.path.dirname(checkout), exist_ok=True)
            with self.assertRaises(SystemExit):
                reject_symlink_ancestors(checkout, home, "checkout")

    def test_validate_exact_operator_path_requires_byte_match(self) -> None:
        with self.assertRaises(SystemExit):
            validate_exact_operator_path(
                "/home/u/seats/seat-x/../seat-x/repo",
                "/home/u/seats/seat-x/repo",
                label="checkout",
            )

    def test_become_implementer_uses_primary_group_only(self) -> None:
        order: list[str] = []

        with patch("seat_scope_entry.grp.getgrall", return_value=[]), patch(
            "seat_scope_entry.os.setgroups", side_effect=lambda groups: order.append(f"setgroups:{groups}")
        ), patch(
            "seat_scope_entry.os.setresgid", side_effect=lambda *_args: order.append("setresgid")
        ), patch(
            "seat_scope_entry.drop_bounding_caps", side_effect=lambda: order.append("drop_bounding_caps")
        ), patch(
            "seat_scope_entry.os.setresuid", side_effect=lambda *_args: order.append("setresuid")
        ), patch(
            "seat_scope_entry.set_no_new_privs", side_effect=lambda: order.append("set_no_new_privs")
        ), patch(
            "seat_scope_entry.verify_zero_privileges", side_effect=lambda: order.append("verify_zero_privileges")
        ), patch(
            "seat_implementer_identity.validate_implementer_group_membership", side_effect=lambda *_a: order.append("validate-groups")
        ):
            from seat_scope_entry import become_implementer

            become_implementer(1001, 1001, "ods-seat-x")

        self.assertIn("setgroups:[1001]", order)
        self.assertIn("validate-groups", order)


class ImplementerInvocationTests(unittest.TestCase):
    def test_build_implementer_invocation_uses_sudo_not_setuid(self) -> None:
        paths = ScopeEntryPaths(implementer_exec="/usr/local/bin/overdeck-seat-implementer-exec")
        argv = build_implementer_invocation(
            paths,
            seat_id="seat-x",
            seat_host="debian1",
            seat_model="gpt-5.6-terra",
            socket="/home/op/.local/state/overdeck/seats/seat-x/tmux.sock",
            launcher="/home/op/.claude/bin/seat-launcher",
            claude_bin="/home/op/.local/bin/claude",
            guard_bin="/home/op/.local/share/overdeck/seat-guard/current/bin",
            checkout="/home/op/seats/seat-x/repo",
            state_dir="/home/op/.local/state/overdeck/seats/seat-x",
            implementer_home_path="/var/lib/overdeck/seat-runtime/seat-x/home",
            tmux_cmd=["tmux", "-S", "/tmp/s.sock", "new-session", "--", "/bin/sleep", "1"],
        )
        self.assertEqual(argv[:3], ["sudo", "-n", paths.implementer_exec])
        self.assertIn("--seat-id", argv)
        self.assertIn("seat-x", argv)


class TmuxSupportProbeTests(unittest.TestCase):
    def test_verify_tmux_support_without_default_server(self) -> None:
        import shutil

        if shutil.which("tmux") is None:
            self.skipTest("tmux unavailable")
        verify_tmux_support(ScopeEntryPaths())

    def test_verify_tmux_support_accepts_exit_empty_probe(self) -> None:
        import tempfile

        with tempfile.TemporaryDirectory() as tmp:
            fake = os.path.join(tmp, "tmux")
            with open(fake, "w", encoding="utf-8") as fh:
                fh.write("""#!/bin/bash
set -euo pipefail
while [ $# -gt 0 ]; do
  case "$1" in
    -V) echo "tmux 3.7"; exit 0 ;;
    -f) shift 2 ;;
    -S) shift 2 ;;
    new-session)
      shift
      while [ $# -gt 0 ]; do shift; done
      exit 0 ;;
    show-options)
      shift
      if [ "$#" -eq 2 ] && [ "$1" = "-s" ] && [ "$2" = "exit-empty" ]; then
        echo "exit-empty on"
        exit 0
      fi
      echo "invalid show-options argv: $*" >&2
      exit 1 ;;
    kill-server) exit 0 ;;
    *) echo "unknown command: $1" >&2; exit 1 ;;
  esac
done
exit 1
""")
            os.chmod(fake, 0o755)
            verify_tmux_support(ScopeEntryPaths(tmux=fake))

    def test_verify_tmux_support_rejects_missing_exit_empty_option(self) -> None:
        import tempfile

        with tempfile.TemporaryDirectory() as tmp:
            fake = os.path.join(tmp, "tmux")
            with open(fake, "w", encoding="utf-8") as fh:
                fh.write("""#!/bin/bash
set -euo pipefail
socket=""
conf=""
while [ $# -gt 0 ]; do
  case "$1" in
    -V) echo "tmux 3.7"; exit 0 ;;
    -f) conf="$2"; shift 2 ;;
    -S) socket="$2"; shift 2 ;;
    new-session)
      shift
      while [ $# -gt 0 ]; do
        case "$1" in
          -d|-s) shift 2 ;;
          *) shift ;;
        esac
      done
      exit 0 ;;
    show-options)
      shift
      if [ "$#" -eq 2 ] && [ "$1" = "-s" ] && [ "$2" = "exit-empty" ]; then
        echo "exit-empty off"
        exit 0
      fi
      echo "invalid show-options argv: $*" >&2
      exit 1 ;;
    kill-server) exit 0 ;;
    *) echo "unknown command: $1" >&2; exit 1 ;;
  esac
done
echo "unknown command" >&2
exit 1
""")
            os.chmod(fake, 0o755)
            with self.assertRaises(SystemExit) as exc:
                verify_tmux_support(ScopeEntryPaths(tmux=fake))
            self.assertEqual(exc.exception.code, 1)

    def test_verify_tmux_support_rejects_mis_scoped_show_options(self) -> None:
        import tempfile

        with tempfile.TemporaryDirectory() as tmp:
            fake = os.path.join(tmp, "tmux")
            with open(fake, "w", encoding="utf-8") as fh:
                fh.write("""#!/bin/bash
set -euo pipefail
while [ $# -gt 0 ]; do
  case "$1" in
    -V) echo "tmux 3.7"; exit 0 ;;
    -f) shift 2 ;;
    -S) shift 2 ;;
    new-session)
      shift
      while [ $# -gt 0 ]; do shift; done
      exit 0 ;;
    show-options)
      shift
      if [ "$#" -eq 2 ] && [ "$1" = "-g" ] && [ "$2" = "exit-empty" ]; then
        echo "exit-empty on"
        exit 0
      fi
      echo "invalid show-options argv: $*" >&2
      exit 1 ;;
    kill-server) exit 0 ;;
    *) echo "unknown command: $1" >&2; exit 1 ;;
  esac
done
exit 1
""")
            os.chmod(fake, 0o755)
            with self.assertRaises(SystemExit) as exc:
                verify_tmux_support(ScopeEntryPaths(tmux=fake))
            self.assertEqual(exc.exception.code, 1)


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