#!/usr/bin/env python3 """Pure-function tests for seat_scope_entry (no root, no fixture env).""" from __future__ import annotations import argparse import io 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_common import ( # noqa: E402 SEAT_LAUNCHER_BIN, SEAT_STARTUP_STATUS_FILE, SEAT_STARTUP_STATUS_MARKERS, read_startup_status_marker, remove_startup_status, startup_status_path, validate_startup_status_content, ) 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 _write_fixture_launcher(path: str) -> None: with open(path, "w", encoding="utf-8") as fh: fh.write("#!/bin/sh\n") os.chmod(path, 0o755) 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 StartupStatusTests(unittest.TestCase): def setUp(self) -> None: self.tmp = tempfile.mkdtemp(prefix="seat-startup-status-") self.state_dir = os.path.join(self.tmp, "state") os.makedirs(self.state_dir, mode=0o700) def tearDown(self) -> None: import shutil shutil.rmtree(self.tmp, ignore_errors=True) def _write_status(self, content: str) -> str: path = startup_status_path(self.state_dir) with open(path, "w", encoding="utf-8") as fh: fh.write(content) return path def test_startup_status_path_outside_trusted_config(self) -> None: path = startup_status_path(self.state_dir) self.assertEqual(os.path.basename(path), SEAT_STARTUP_STATUS_FILE) self.assertNotIn("trusted-config", path) def test_validate_accepts_each_marker_class(self) -> None: for marker in sorted(SEAT_STARTUP_STATUS_MARKERS): self.assertEqual(validate_startup_status_content(f"{marker}\n".encode()), marker) self.assertEqual(validate_startup_status_content(marker.encode()), marker) self.assertEqual(validate_startup_status_content(b"exit-42\n"), "exit-42") self.assertEqual(validate_startup_status_content(b"exit-42"), "exit-42") def test_validate_rejects_whitespace_padding(self) -> None: for raw in ( b" starting\n", b"starting \n", b"\tstarting\n", b"starting\t\n", b" starting", b"starting ", ): self.assertIsNone(validate_startup_status_content(raw), repr(raw)) def test_validate_rejects_cr_and_multi_newline(self) -> None: for raw in ( b"starting\r\n", b"starting\r", b"starting\n\n", b"\nstarting\n", b"start\ning\n", ): self.assertIsNone(validate_startup_status_content(raw), repr(raw)) def test_validate_rejects_nul_and_invalid_utf8(self) -> None: self.assertIsNone(validate_startup_status_content(b"starting\x00\n")) self.assertIsNone(validate_startup_status_content(b"\xffstarting\n")) self.assertIsNone(validate_startup_status_content(b"\n")) def test_read_accepts_no_newline_and_single_newline_markers(self) -> None: for marker in ("starting", "exec", "exit-3"): for content in (marker, f"{marker}\n"): with self.subTest(marker=marker, content=content): with open(startup_status_path(self.state_dir), "wb") as fh: fh.write(content.encode()) self.assertEqual(read_startup_status_marker(self.state_dir), marker) def test_read_accepts_each_marker_class(self) -> None: for marker in sorted(SEAT_STARTUP_STATUS_MARKERS): self._write_status(f"{marker}\n") self.assertEqual(read_startup_status_marker(self.state_dir), marker) self._write_status("exit-7\n") self.assertEqual(read_startup_status_marker(self.state_dir), "exit-7") def test_missing_file_reports_unavailable(self) -> None: self.assertEqual(read_startup_status_marker(self.state_dir), "unavailable") def test_symlink_rejected(self) -> None: real = os.path.join(self.tmp, "real-status") with open(real, "w", encoding="utf-8") as fh: fh.write("starting\n") os.symlink(real, startup_status_path(self.state_dir)) self.assertEqual(read_startup_status_marker(self.state_dir), "invalid") def test_toctou_symlink_swap_reads_opened_fd(self) -> None: path = startup_status_path(self.state_dir) evil = os.path.join(self.tmp, "evil-status") with open(evil, "w", encoding="utf-8") as fh: fh.write("not-a-real-marker\n") with open(path, "w", encoding="utf-8") as fh: fh.write("starting\n") original_open = os.open swapped = False def swapping_open(p: str, flags: int, mode: int = 0o777) -> int: nonlocal swapped if p == path and not swapped: fd = original_open(p, flags, mode) os.remove(path) os.symlink(evil, path) swapped = True return fd return original_open(p, flags, mode) with patch("seat_common.os.open", side_effect=swapping_open): self.assertEqual(read_startup_status_marker(self.state_dir), "starting") def test_directory_rejected(self) -> None: path = startup_status_path(self.state_dir) os.rmdir(self.state_dir) os.makedirs(path, mode=0o700) self.assertEqual(read_startup_status_marker(self.state_dir), "invalid") def test_fifo_rejected(self) -> None: path = startup_status_path(self.state_dir) os.mkfifo(path, mode=0o600) self.assertEqual(read_startup_status_marker(self.state_dir), "invalid") def test_oversize_rejected_by_stat_and_read(self) -> None: self._write_status("starting" + ("x" * 80) + "\n") self.assertEqual(read_startup_status_marker(self.state_dir), "invalid") path = startup_status_path(self.state_dir) with open(path, "wb") as fh: fh.write(b"x" * (64 + 1)) self.assertEqual(read_startup_status_marker(self.state_dir), "invalid") def test_whitespace_cr_multi_newline_rejected_on_read(self) -> None: for content in ( " starting\n", "starting \n", "starting\r\n", "starting\n\n", ): with self.subTest(content=content): with open(startup_status_path(self.state_dir), "wb") as fh: fh.write(content.encode()) self.assertEqual(read_startup_status_marker(self.state_dir), "invalid") def test_control_chars_rejected(self) -> None: with open(startup_status_path(self.state_dir), "wb") as fh: fh.write(b"starting\x07\n") self.assertEqual(read_startup_status_marker(self.state_dir), "invalid") def test_arbitrary_content_rejected(self) -> None: self._write_status("not-a-real-marker\n") self.assertEqual(read_startup_status_marker(self.state_dir), "invalid") def test_error_never_reflects_raw_content(self) -> None: secret = "ANTHROPIC_API_KEY=super-secret-path" self._write_status(f"{secret}\n") marker = read_startup_status_marker(self.state_dir) self.assertEqual(marker, "invalid") self.assertNotIn("ANTHROPIC", marker) self.assertNotIn("secret", marker) def test_remove_startup_status_cleans_file(self) -> None: self._write_status("exec\n") remove_startup_status(self.state_dir) self.assertEqual(read_startup_status_marker(self.state_dir), "unavailable") def test_wait_tmux_pane_appends_validated_marker(self) -> None: from seat_scope_entry import ScopeEntryPaths, wait_tmux_pane self._write_status("claude-bin\n") paths = ScopeEntryPaths( tmux_mediator="/bin/false", proc_root=self.tmp, cgroup_root=self.tmp, ) with patch("sys.stderr", new_callable=io.StringIO) as err: with self.assertRaises(SystemExit) as exc: wait_tmux_pane(paths, "seat-x", "/tmp/missing.sock", self.state_dir, timeout=0.01) self.assertEqual(exc.exception.code, 1) self.assertIn("tmux-session-missing:claude-bin", err.getvalue()) self.assertNotIn(self.state_dir, err.getvalue()) 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", SEAT_LAUNCHER_BIN, "--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), "launcher_sha256": file_sha256(wrapper), "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") launcher = os.path.join(tmp, "overdeck-seat-launcher") _write_fixture_launcher(launcher) 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), "launcher_sha256": file_sha256(launcher), "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.INSTALLED_LAUNCHER", launcher ), 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) launcher = os.path.join(tmp, "overdeck-seat-launcher") _write_fixture_launcher(launcher) 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), "launcher_sha256": file_sha256(launcher), "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.INSTALLED_LAUNCHER", launcher ), 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) launcher = os.path.join(tmp, "overdeck-seat-launcher") _write_fixture_launcher(launcher) 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), "launcher_sha256": file_sha256(launcher), "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.INSTALLED_LAUNCHER", launcher ), 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", SEAT_LAUNCHER_BIN, "--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_validate_args_rejects_home_launcher_path(self) -> None: parser = build_arg_parser() ns = self._valid_ns(parser) ns.launcher = "/home/u/.claude/bin/seat-launcher" 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_implementer_writable_launcher_path(self) -> None: parser = build_arg_parser() ns = self._valid_ns(parser) ns.launcher = "/home/u/seats/seat-x/repo/evil-launcher" 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_allows_operator_bin_symlink_when_launcher_is_root_owned(self) -> None: import tempfile parser = build_arg_parser() with tempfile.TemporaryDirectory() as tmp: home = os.path.join(tmp, "home") os.makedirs(home, mode=0o700) real_bin = os.path.join(home, ".claude-real", "bin") os.makedirs(real_bin, mode=0o700) os.symlink(os.path.join(home, ".claude-real"), os.path.join(home, ".claude")) launcher = os.path.join(tmp, "overdeck-seat-launcher") _write_fixture_launcher(launcher) with patch("seat_scope_entry.SEAT_LAUNCHER_BIN", launcher), patch( "seat_scope_entry.os.access", return_value=True ): ns = parser.parse_args([ "--seat-id", "seat-x", "--netns-path", "/var/run/netns/overdeck-seat-seat-x", "--unit", "agent-seat-seat-x.scope", "--socket", f"{home}/.local/state/overdeck/seats/seat-x/tmux.sock", "--launcher", launcher, "--slice", "agent-seat.slice", "--seat-host", "debian1", "--seat-model", "gpt-5.6-terra", "--claude-bin", f"{home}/.local/bin/claude", "--guard-bin", f"{home}/.local/share/overdeck/seat-guard/current/bin", "--checkout", f"{home}/seats/seat-x/repo", "--account-slug", "roy", ]) validate_args(ns, home, 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="/usr/local/bin/overdeck-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", operator_home="/home/op", account_slug="roy", profile_dir="/home/op/.claudex-accounts/roy/seat-x", secure_storage_dir="/home/op/.claudex-accounts/roy/seat-x/secure-storage", netns_path="/var/run/netns/overdeck-seat-seat-x", tmux_cmd=["tmux", "-S", "/tmp/s.sock", "new-session", "--", "/bin/sleep", "1"], ) self.assertEqual(argv[:3], ["sudo", "-n", paths.implementer_exec]) self.assertIn("--operator-home", argv) self.assertIn("/home/op", argv) self.assertIn("--profile-dir", argv) self.assertIn("/home/op/.claudex-accounts/roy/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()