#!/usr/bin/env python3 """Integration tests for per-seat identity, ACL, and path validation.""" from __future__ import annotations import os import pwd import shutil import stat import shlex import subprocess import sys import tempfile import unittest from unittest import mock _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 CONTROL_BASE_MODE, CONTROL_DIR_MODE, CONTROL_OWNER_MARKER_MODE, LAUNCH_AUTHORITY_FILES, TMUX_CONF_MODE, ) from seat_implementer_identity import ( # noqa: E402 atomic_root_control_write, ensure_control_base, ensure_control_dir, grant_managed_runtime_access, grant_trusted_config_read_boundary, grant_traverse_acl, implementer_home_for_seat, implementer_username_for_seat, launch_authority_paths, managed_runtime_targets, read_owner_marker, resolve_managed_path, trusted_config_dir, validate_implementer_passwd, verify_implementer_seat_metadata_boundary, write_owner_marker, ) from seat_scope_entry import ( # noqa: E402 ScopeEntryPaths, TMUX_CONF, verify_pid_effective_boundary, write_tmux_conf, ) def _nobody_ids() -> tuple[int, int]: entry = pwd.getpwnam("nobody") return entry.pw_uid, entry.pw_gid def _ensure_wrong_owner_for_production_check(path: str) -> None: if os.geteuid() == 0: uid, gid = _nobody_ids() os.chown(path, uid, gid) class ManagedRuntimeTargetsTests(unittest.TestCase): def test_seat_root_is_traverse_only_not_rw(self) -> None: home = "/home/op" seat_id = "seat-a" traverse, rw, _executables = managed_runtime_targets( home, seat_id, repo=f"{home}/seats/{seat_id}/repo", profile=f"{home}/.claudex-accounts/roy/{seat_id}", state_dir=f"{home}/.local/state/overdeck/seats/{seat_id}", guard_bin=f"{home}/.local/share/overdeck/seat-guard/current/bin", claude_bin=f"{home}/.local/bin/claude", launcher=f"{home}/.claude/bin/seat-launcher", ) self.assertIn(f"{home}/seats/{seat_id}", traverse) self.assertNotIn(f"{home}/seats/{seat_id}", rw) self.assertIn(f"{home}/seats/{seat_id}/repo", rw) class SeatJsonBoundaryTests(unittest.TestCase): def setUp(self) -> None: if shutil.which("setfacl") is None: self.skipTest("setfacl unavailable") self.tmp = tempfile.mkdtemp(prefix="seat-json-acl-") self.operator_uid = os.getuid() self.operator_gid = os.getgid() self.home = os.path.join(self.tmp, "home") self.seat_id = "seat-a" os.makedirs(self.home, mode=0o700) os.chown(self.home, self.operator_uid, self.operator_gid) seat_root = os.path.join(self.home, "seats", self.seat_id) os.makedirs(seat_root, mode=0o700) os.chown(seat_root, self.operator_uid, self.operator_gid) seat_json = os.path.join(seat_root, "seat.json") with open(seat_json, "w", encoding="utf-8") as fh: fh.write('{"schemaVersion":1,"model":"gpt-5.6-terra"}\n') os.chmod(seat_json, 0o600) repo = os.path.join(seat_root, "repo") os.makedirs(repo, mode=0o700) os.chown(repo, self.operator_uid, self.operator_gid) profile = os.path.join(self.home, ".claudex-accounts", "roy", self.seat_id) os.makedirs(profile, mode=0o700) os.chown(profile, self.operator_uid, self.operator_gid) state = os.path.join(self.home, ".local/state/overdeck/seats", self.seat_id) os.makedirs(state, mode=0o700) os.chown(state, self.operator_uid, self.operator_gid) guard = os.path.join(self.home, ".local/share/overdeck/seat-guard/current/bin") claude = os.path.join(self.home, ".local/bin/claude") launcher = os.path.join(self.home, ".claude/bin/seat-launcher") for path in (guard, claude, launcher): os.makedirs(os.path.dirname(path), mode=0o700, exist_ok=True) with open(path, "w", encoding="utf-8") as fh: fh.write("#!/bin/sh\n") os.chmod(path, 0o755) os.chown(os.path.dirname(path), self.operator_uid, self.operator_gid) os.chown(path, self.operator_uid, self.operator_gid) os.chown(os.path.dirname(os.path.dirname(path)), self.operator_uid, self.operator_gid) os.environ["OVERDECK_SEAT_CONTROL_BASE"] = os.path.join(self.tmp, "control") os.environ["OVERDECK_SEAT_RUNTIME_BASE"] = os.path.join(self.tmp, "runtime") os.environ["OVERDECK_SEAT_TEST_MODE"] = "1" self.subject = pwd.getpwuid(self.operator_uid).pw_name grant_managed_runtime_access( self.subject, self.subject, self.home, self.seat_id, operator_uid=self.operator_uid, operator_gid=self.operator_gid, repo=repo, profile=profile, state_dir=state, guard_bin=guard, claude_bin=claude, launcher=launcher, ) def tearDown(self) -> None: shutil.rmtree(self.tmp, ignore_errors=True) def test_seat_json_and_root_protected_from_implementer(self) -> None: seat_root = os.path.join(self.home, "seats", self.seat_id) repo_acl = subprocess.run( ["getfacl", "-p", os.path.join(seat_root, "repo")], capture_output=True, text=True, ).stdout seat_acl = subprocess.run(["getfacl", "-p", seat_root], capture_output=True, text=True).stdout self.assertIn(f"user:{self.subject}:rwx", repo_acl) self.assertNotIn(f"user:{self.subject}:rwx", seat_acl) self.assertNotIn(f"user:{self.subject}:rw-", seat_acl) implementer = implementer_username_for_seat(self.seat_id) try: pwd.getpwnam(implementer) except KeyError: self.skipTest(f"{implementer} not provisioned") verify_implementer_seat_metadata_boundary( implementer, self.home, self.seat_id, sudo="sudo", ) class UsernameNormalizationTests(unittest.TestCase): def test_uppercase_seat_id_hashes(self) -> None: user = implementer_username_for_seat("Seat.UPPER") self.assertTrue(user.startswith("ods-")) self.assertLessEqual(len(user), 32) self.assertRegex(user, r"^ods-[a-f0-9]{16}$") def test_simple_seat_id_keeps_prefix(self) -> None: self.assertEqual(implementer_username_for_seat("seat-a"), "ods-seat-a") class AclBoundaryTests(unittest.TestCase): def setUp(self) -> None: if shutil.which("setfacl") is None: self.skipTest("setfacl unavailable") self.tmp = tempfile.mkdtemp(prefix="seat-acl-") self.operator_uid = os.getuid() self.operator_gid = os.getgid() self.home = os.path.join(self.tmp, "home") os.makedirs(self.home, mode=0o700) os.chown(self.home, self.operator_uid, self.operator_gid) self.seat_a = os.path.join(self.home, "seats", "seat-a") self.seat_b = os.path.join(self.home, "seats", "seat-b") os.makedirs(self.seat_a, mode=0o700) os.makedirs(self.seat_b, mode=0o700) os.chown(self.seat_a, self.operator_uid, self.operator_gid) os.chown(self.seat_b, self.operator_uid, self.operator_gid) with open(os.path.join(self.seat_a, "owned.txt"), "w", encoding="utf-8") as fh: fh.write("a\n") with open(os.path.join(self.seat_b, "owned.txt"), "w", encoding="utf-8") as fh: fh.write("b\n") os.environ["OVERDECK_SEAT_CONTROL_BASE"] = os.path.join(self.tmp, "control") os.environ["OVERDECK_SEAT_RUNTIME_BASE"] = os.path.join(self.tmp, "runtime") os.environ["OVERDECK_SEAT_TEST_MODE"] = "1" def tearDown(self) -> None: shutil.rmtree(self.tmp, ignore_errors=True) def _acl_access(self, username: str, path: str, mode: str) -> bool: proc = subprocess.run( ["sudo", "-n", "-u", username, "test", mode, path], capture_output=True, text=True, ) return proc.returncode == 0 def test_ancestor_traverse_only_not_write(self) -> None: user = pwd.getpwuid(self.operator_uid).pw_name grant_traverse_acl(self.home, user) proc = subprocess.run(["getfacl", "-p", self.home], capture_output=True, text=True) self.assertIn(f"user:{user}:--x", proc.stdout.replace("user:", "user:")) self.assertNotRegex(proc.stdout, rf"user:{user}:rwx") proc = subprocess.run( ["touch", os.path.join(self.home, "rogue")], capture_output=True, text=True, ) if proc.returncode == 0: os.unlink(os.path.join(self.home, "rogue")) def test_sibling_seat_isolation(self) -> None: user_a = pwd.getpwuid(self.operator_uid).pw_name repo_a = os.path.join(self.seat_a, "repo") repo_b = os.path.join(self.seat_b, "repo") os.makedirs(repo_a, mode=0o700) os.makedirs(repo_b, mode=0o700) os.chown(repo_a, self.operator_uid, self.operator_gid) os.chown(repo_b, self.operator_uid, self.operator_gid) profile_a = os.path.join(self.home, ".claudex-accounts", "roy", "seat-a") profile_b = os.path.join(self.home, ".claudex-accounts", "roy", "seat-b") os.makedirs(profile_a, mode=0o700) os.makedirs(profile_b, mode=0o700) os.chown(profile_a, self.operator_uid, self.operator_gid) os.chown(profile_b, self.operator_uid, self.operator_gid) state_a = os.path.join(self.home, ".local/state/overdeck/seats/seat-a") state_b = os.path.join(self.home, ".local/state/overdeck/seats/seat-b") os.makedirs(state_a, mode=0o700) os.makedirs(state_b, mode=0o700) os.chown(state_a, self.operator_uid, self.operator_gid) os.chown(state_b, self.operator_uid, self.operator_gid) guard = os.path.join(self.home, ".local/share/overdeck/seat-guard/current/bin") claude = os.path.join(self.home, ".local/bin/claude") launcher = os.path.join(self.home, ".claude/bin/seat-launcher") os.makedirs(os.path.dirname(guard), mode=0o700) os.makedirs(os.path.dirname(claude), mode=0o700) os.makedirs(os.path.dirname(launcher), mode=0o700) with open(guard, "w", encoding="utf-8") as fh: fh.write("#!/bin/sh\n") with open(claude, "w", encoding="utf-8") as fh: fh.write("#!/bin/sh\n") with open(launcher, "w", encoding="utf-8") as fh: fh.write("#!/bin/sh\n") os.chmod(guard, 0o755) os.chmod(claude, 0o755) os.chmod(launcher, 0o755) grant_managed_runtime_access( user_a, user_a, self.home, "seat-a", operator_uid=self.operator_uid, operator_gid=self.operator_gid, repo=repo_a, profile=profile_a, state_dir=state_a, guard_bin=guard, claude_bin=claude, launcher=launcher, ) repo_a_acl = subprocess.run(["getfacl", "-p", repo_a], capture_output=True, text=True).stdout repo_b_acl = subprocess.run(["getfacl", "-p", repo_b], capture_output=True, text=True).stdout self.assertIn(f"user:{user_a}:rwx", repo_a_acl) self.assertNotIn(f"user:{user_a}:", repo_b_acl) class ManagedSymlinkTests(unittest.TestCase): def setUp(self) -> None: self.tmp = tempfile.mkdtemp(prefix="seat-symlink-") self.uid = os.getuid() self.gid = os.getgid() self.home = os.path.join(self.tmp, "home") os.makedirs(self.home, mode=0o700) os.chown(self.home, self.uid, self.gid) def tearDown(self) -> None: shutil.rmtree(self.tmp, ignore_errors=True) def test_allows_guard_current_symlink(self) -> None: guard_root = os.path.join(self.home, ".local/share/overdeck/seat-guard") release = os.path.join(guard_root, "releases", "r1", "bin") os.makedirs(release, mode=0o700) os.chown(guard_root, self.uid, self.gid) for path in ( os.path.join(self.home, ".local"), os.path.join(self.home, ".local/share"), os.path.join(self.home, ".local/share/overdeck"), guard_root, ): os.makedirs(path, exist_ok=True) os.chown(path, self.uid, self.gid) os.symlink(os.path.join(guard_root, "releases", "r1"), os.path.join(guard_root, "current")) resolved = resolve_managed_path( self.home, os.path.join(guard_root, "current", "bin"), operator_uid=self.uid, operator_gid=self.gid, ) self.assertEqual(resolved, os.path.join(release)) def test_rejects_chained_symlink(self) -> None: bin_dir = os.path.join(self.home, ".local/bin") os.makedirs(bin_dir, mode=0o700) os.chown(bin_dir, self.uid, self.gid) first = os.path.join(bin_dir, "claude-real") with open(first, "w", encoding="utf-8") as fh: fh.write("#!/bin/sh\n") os.chmod(first, 0o755) os.symlink("claude-real", os.path.join(bin_dir, "claude-hop")) os.symlink("claude-hop", os.path.join(bin_dir, "claude")) with self.assertRaises(SystemExit): resolve_managed_path( self.home, os.path.join(bin_dir, "claude"), operator_uid=self.uid, operator_gid=self.gid, ) class OwnerMarkerTests(unittest.TestCase): def setUp(self) -> None: self.tmp = tempfile.mkdtemp(prefix="seat-marker-") os.environ["OVERDECK_SEAT_CONTROL_BASE"] = os.path.join(self.tmp, "control") os.environ["OVERDECK_SEAT_TEST_MODE"] = "1" def tearDown(self) -> None: shutil.rmtree(self.tmp, ignore_errors=True) def test_missing_marker_fails(self) -> None: with self.assertRaises(SystemExit): read_owner_marker("seat-a") def test_marker_roundtrip(self) -> None: write_owner_marker("seat-a", "fixture-user") self.assertEqual(read_owner_marker("seat-a"), "fixture-user") class CgroupTamperTests(unittest.TestCase): def test_each_limit_tamper_fails(self) -> None: tamper_cases = { "memory.max": "max", "memory.swap.max": "0", "cpu.max": "max 100000", "cpu.weight": "100", "pids.max": "max", } for name, bad in tamper_cases.items(): with self.subTest(name=name): 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") with open(os.path.join(cg, name), "w", encoding="utf-8") as fh: fh.write(f"{bad}\n") paths = ScopeEntryPaths( proc_root=os.path.join(tmp, "proc"), cgroup_root=os.path.join(tmp, "sys/fs/cgroup"), ) with self.assertRaises(SystemExit): verify_pid_effective_boundary(paths, 123, "agent-seat-seat-x.scope") class DebianSystemUidPasswdTests(unittest.TestCase): def _passwd_entry(self, seat_id: str, uid: int, operator: str = "fixture-user") -> pwd.struct_passwd: home = implementer_home_for_seat(seat_id) username = implementer_username_for_seat(seat_id) return pwd.struct_passwd( ( username, "x", uid, uid, f"overdeck-seat-owner={operator}", home, "/usr/sbin/nologin", ) ) def test_accepts_debian_system_uid_range(self) -> None: for uid in (100, 999): with self.subTest(uid=uid): validate_implementer_passwd(self._passwd_entry("seat-a", uid), "fixture-user", "seat-a") def test_rejects_root_uid(self) -> None: with self.assertRaises(SystemExit): validate_implementer_passwd(self._passwd_entry("seat-a", 0), "fixture-user", "seat-a") def test_rejects_gecos_marker_mismatch(self) -> None: entry = self._passwd_entry("seat-a", 999) bad = pwd.struct_passwd( ( entry.pw_name, entry.pw_passwd, entry.pw_uid, entry.pw_gid, "wrong-marker", entry.pw_dir, entry.pw_shell, ) ) with self.assertRaises(SystemExit): validate_implementer_passwd(bad, "fixture-user", "seat-a") class ControlDirContractTests(unittest.TestCase): def setUp(self) -> None: self.tmp = tempfile.mkdtemp(prefix="seat-control-") os.environ["OVERDECK_SEAT_CONTROL_BASE"] = os.path.join(self.tmp, "control") os.environ["OVERDECK_SEAT_TEST_MODE"] = "1" def tearDown(self) -> None: shutil.rmtree(self.tmp, ignore_errors=True) def test_ensure_control_base_mode_is_traversable(self) -> None: base = ensure_control_base() st = os.stat(base) self.assertEqual(stat.S_IMODE(st.st_mode), CONTROL_BASE_MODE) def test_ensure_control_dir_mode_is_traversable_not_listable(self) -> None: path = ensure_control_dir("seat-a") st = os.stat(path) self.assertEqual(stat.S_IMODE(st.st_mode), CONTROL_DIR_MODE) base = os.path.join(self.tmp, "control") self.assertEqual(stat.S_IMODE(os.stat(base).st_mode), CONTROL_BASE_MODE) def test_ensure_control_base_rejects_symlink(self) -> None: base = os.path.join(self.tmp, "control") if os.path.exists(base): shutil.rmtree(base) os.symlink(self.tmp, base) with self.assertRaises(SystemExit): ensure_control_base() def test_ensure_control_base_symlink_target_mode_unchanged(self) -> None: target = os.path.join(self.tmp, "target-dir") os.makedirs(target, mode=0o755) before = os.stat(target) base = os.path.join(self.tmp, "control") os.symlink(target, base) with self.assertRaises(SystemExit): ensure_control_base() after = os.stat(target) self.assertEqual(stat.S_IMODE(before.st_mode), stat.S_IMODE(after.st_mode)) def test_ensure_control_base_rejects_non_directory(self) -> None: base = os.path.join(self.tmp, "control") with open(base, "w", encoding="utf-8") as fh: fh.write("x") with self.assertRaises(SystemExit): ensure_control_base() def test_ensure_control_base_production_rejects_wrong_owner_before_mode_repair( self, ) -> None: os.environ.pop("OVERDECK_SEAT_TEST_MODE", None) base = os.path.join(self.tmp, "control") os.makedirs(base, mode=0o700) _ensure_wrong_owner_for_production_check(base) before_mode = stat.S_IMODE(os.stat(base).st_mode) with self.assertRaises(SystemExit): ensure_control_base() self.assertEqual(stat.S_IMODE(os.stat(base).st_mode), before_mode) def test_ensure_control_base_repairs_restrictive_umask(self) -> None: base = os.path.join(self.tmp, "control") old = os.umask(0o077) try: ensure_control_base() finally: os.umask(old) self.assertEqual(stat.S_IMODE(os.stat(base).st_mode), CONTROL_BASE_MODE) def test_ensure_control_dir_rejects_symlink(self) -> None: ensure_control_base() base = os.path.join(self.tmp, "control") seat_link = os.path.join(base, "seat-a") os.symlink(self.tmp, seat_link) with self.assertRaises(SystemExit): ensure_control_dir("seat-a") def test_ensure_control_dir_symlink_target_mode_unchanged(self) -> None: ensure_control_base() base = os.path.join(self.tmp, "control") target = os.path.join(self.tmp, "seat-target") os.makedirs(target, mode=0o755) before = os.stat(target) seat_link = os.path.join(base, "seat-a") os.symlink(target, seat_link) with self.assertRaises(SystemExit): ensure_control_dir("seat-a") after = os.stat(target) self.assertEqual(stat.S_IMODE(before.st_mode), stat.S_IMODE(after.st_mode)) def test_ensure_control_dir_rejects_non_directory(self) -> None: ensure_control_base() base = os.path.join(self.tmp, "control") seat_path = os.path.join(base, "seat-a") with open(seat_path, "w", encoding="utf-8") as fh: fh.write("x") with self.assertRaises(SystemExit): ensure_control_dir("seat-a") def test_ensure_control_dir_production_rejects_wrong_owner_before_mode_repair( self, ) -> None: base = os.path.join(self.tmp, "control") os.makedirs(base, mode=CONTROL_BASE_MODE) seat_path = os.path.join(base, "seat-a") os.makedirs(seat_path, mode=0o700) _ensure_wrong_owner_for_production_check(seat_path) os.environ.pop("OVERDECK_SEAT_TEST_MODE", None) before_mode = stat.S_IMODE(os.stat(seat_path).st_mode) with mock.patch( "seat_implementer_identity.ensure_control_base", return_value=base ): with self.assertRaises(SystemExit): ensure_control_dir("seat-a") self.assertEqual(stat.S_IMODE(os.stat(seat_path).st_mode), before_mode) def test_atomic_root_control_write_owner_marker_mode(self) -> None: write_owner_marker("seat-a", "fixture-user") marker = os.path.join(self.tmp, "control", "seat-a", "owner") st = os.stat(marker) self.assertEqual(stat.S_IMODE(st.st_mode), CONTROL_OWNER_MARKER_MODE) def test_atomic_root_control_write_rejects_symlink_final(self) -> None: ensure_control_dir("seat-a") control = os.path.join(self.tmp, "control", "seat-a") os.symlink("/etc/passwd", os.path.join(control, "owner")) with self.assertRaises(SystemExit): write_owner_marker("seat-a", "fixture-user") def test_atomic_root_control_write_rejects_symlink_temp(self) -> None: ensure_control_dir("seat-a") control = os.path.join(self.tmp, "control", "seat-a") os.symlink("/etc/passwd", os.path.join(control, f".owner.{os.getpid()}.tmp")) with self.assertRaises(SystemExit): atomic_root_control_write("seat-a", "owner", "fixture-user", mode=CONTROL_OWNER_MARKER_MODE) class TmuxConfLinuxAccessTests(unittest.TestCase): def test_write_tmux_conf_mode_is_world_readable(self) -> None: with tempfile.TemporaryDirectory() as tmp: os.environ["OVERDECK_SEAT_CONTROL_BASE"] = os.path.join(tmp, "control") os.environ["OVERDECK_SEAT_TEST_MODE"] = "1" path = write_tmux_conf("seat-x", 1000, 1000) st = os.stat(path) self.assertEqual(stat.S_IMODE(st.st_mode), TMUX_CONF_MODE) with open(path, encoding="utf-8") as fh: self.assertEqual(fh.read(), TMUX_CONF) @unittest.skipUnless(os.geteuid() == 0, "requires root to chown control artifacts") def test_dropped_uid_reads_conf_not_control_dir(self) -> None: with tempfile.TemporaryDirectory() as tmp: os.chmod(tmp, 0o755) os.environ["OVERDECK_SEAT_CONTROL_BASE"] = os.path.join(tmp, "control") os.environ.pop("OVERDECK_SEAT_TEST_MODE", None) seat_id = "seat-x" write_owner_marker(seat_id, "fixture-user") conf = write_tmux_conf(seat_id, 65534, 65534) control_base = os.path.join(tmp, "control") control = os.path.join(control_base, seat_id) owner = os.path.join(control, "owner") subject = "nobody" self.assertEqual(stat.S_IMODE(os.stat(control_base).st_mode), CONTROL_BASE_MODE) self.assertEqual(stat.S_IMODE(os.stat(control).st_mode), CONTROL_DIR_MODE) self.assertEqual(stat.S_IMODE(os.stat(owner).st_mode), CONTROL_OWNER_MARKER_MODE) self.assertEqual(stat.S_IMODE(os.stat(conf).st_mode), TMUX_CONF_MODE) os.chmod(control_base, 0o700) read_blocked = subprocess.run( ["runuser", "-u", subject, "--", "test", "-r", conf], capture_output=True, ) self.assertNotEqual(read_blocked.returncode, 0, "0700 base must block dropped-UID conf read") os.chmod(control_base, CONTROL_BASE_MODE) read_ok = subprocess.run( ["runuser", "-u", subject, "--", "test", "-r", conf], capture_output=True, ) self.assertEqual(read_ok.returncode, 0, read_ok.stderr) read_owner = subprocess.run( ["runuser", "-u", subject, "--", "test", "-r", owner], capture_output=True, ) self.assertNotEqual(read_owner.returncode, 0) list_dir = subprocess.run( ["runuser", "-u", subject, "--", "ls", control], capture_output=True, ) self.assertNotEqual(list_dir.returncode, 0) write_conf = subprocess.run( ["runuser", "-u", subject, "--", "test", "-w", conf], capture_output=True, ) self.assertNotEqual(write_conf.returncode, 0) write_dir = subprocess.run( ["runuser", "-u", subject, "--", "test", "-w", control], capture_output=True, ) self.assertNotEqual(write_dir.returncode, 0) @unittest.skipUnless(os.geteuid() == 0 and shutil.which("setfacl"), "requires root and setfacl for live seat.json boundary") def test_live_root_seat_json_mode_0600_operator_only(self) -> None: with tempfile.TemporaryDirectory() as tmp: os.environ["OVERDECK_SEAT_CONTROL_BASE"] = os.path.join(tmp, "control") os.environ.pop("OVERDECK_SEAT_TEST_MODE", None) seat_id = "seat-live" seat_root = os.path.join(tmp, "home", "seats", seat_id) os.makedirs(seat_root, mode=0o700) seat_json = os.path.join(seat_root, "seat.json") with open(seat_json, "w", encoding="utf-8") as fh: fh.write('{"schemaVersion":1}\n') os.chmod(seat_json, 0o600) st = os.stat(seat_json) self.assertEqual(stat.S_IMODE(st.st_mode), 0o600) subject = "nobody" for mode in ("-r", "-w"): proc = subprocess.run( ["runuser", "-u", subject, "--", "test", mode, seat_json], capture_output=True, ) self.assertNotEqual(proc.returncode, 0, mode) class UmaskControlWriteTests(unittest.TestCase): def test_owner_marker_and_tmux_conf_modes_under_umask_077(self) -> None: with tempfile.TemporaryDirectory() as tmp: os.environ["OVERDECK_SEAT_CONTROL_BASE"] = os.path.join(tmp, "control") os.environ["OVERDECK_SEAT_TEST_MODE"] = "1" old = os.umask(0o077) try: write_owner_marker("seat-a", "fixture-user") conf = write_tmux_conf("seat-a", 1000, 1000) finally: os.umask(old) owner = os.path.join(tmp, "control", "seat-a", "owner") self.assertEqual(stat.S_IMODE(os.stat(owner).st_mode), CONTROL_OWNER_MARKER_MODE) self.assertEqual(stat.S_IMODE(os.stat(conf).st_mode), TMUX_CONF_MODE) class CredentialDenyTests(unittest.TestCase): def setUp(self) -> None: if shutil.which("setfacl") is None: self.skipTest("setfacl unavailable") self.tmp = tempfile.mkdtemp(prefix="seat-cred-deny-") self.operator_uid = os.getuid() self.operator_gid = os.getgid() self.home = os.path.join(self.tmp, "home") os.makedirs(self.home, mode=0o700) os.chown(self.home, self.operator_uid, self.operator_gid) self.gitconfig = os.path.join(self.home, ".gitconfig") with open(self.gitconfig, "w", encoding="utf-8") as fh: fh.write("[user]\n\tname = Operator\n") os.chmod(self.gitconfig, 0o644) os.environ["OVERDECK_SEAT_CONTROL_BASE"] = os.path.join(self.tmp, "control") os.environ["OVERDECK_SEAT_RUNTIME_BASE"] = os.path.join(self.tmp, "runtime") os.environ["OVERDECK_SEAT_TEST_MODE"] = "1" self.implementer = implementer_username_for_seat("seat-a") def tearDown(self) -> None: shutil.rmtree(self.tmp, ignore_errors=True) def test_world_readable_gitconfig_denied_to_implementer(self) -> None: from seat_implementer_identity import deny_operator_credentials operator = pwd.getpwuid(self.operator_uid).pw_name try: pwd.getpwnam(self.implementer) except KeyError: self.skipTest(f"{self.implementer} not provisioned") deny_operator_credentials(self.implementer, self.home) proc = subprocess.run( ["sudo", "-n", "-u", self.implementer, "test", "-r", self.gitconfig], capture_output=True, text=True, ) self.assertNotEqual(proc.returncode, 0) proc_op = subprocess.run( ["sudo", "-n", "-u", operator, "test", "-r", self.gitconfig], capture_output=True, text=True, ) self.assertEqual(proc_op.returncode, 0, proc_op.stderr) class GroupMembershipTests(unittest.TestCase): def test_validate_implementer_group_membership_rejects_supplementary(self) -> None: from seat_implementer_identity import validate_implementer_group_membership entry = pwd.struct_passwd( ( "ods-seat-a", "x", 999, 999, "overdeck-seat-owner=fixture-user", implementer_home_for_seat("seat-a"), "/usr/sbin/nologin", ) ) fake_group = type("G", (), {"gr_name": "ods-seat-a", "gr_mem": ["ods-seat-a"], "gr_gid": 999})() with mock.patch("seat_implementer_identity.pwd.getpwnam", return_value=entry), mock.patch( "seat_implementer_identity.grp.getgrgid", return_value=fake_group ), mock.patch("seat_implementer_identity.grp.getgrall", return_value=[fake_group]): with self.assertRaises(SystemExit): validate_implementer_group_membership("ods-seat-a", 999) class OperatorRetentionTests(unittest.TestCase): def setUp(self) -> None: if shutil.which("setfacl") is None: self.skipTest("setfacl unavailable") self.tmp = tempfile.mkdtemp(prefix="seat-operator-retain-") self.operator_uid = os.getuid() self.operator_gid = os.getgid() self.home = os.path.join(self.tmp, "home") os.makedirs(self.home, mode=0o700) os.chown(self.home, self.operator_uid, self.operator_gid) self.seat_id = "seat-a" self.repo = os.path.join(self.home, "seats", self.seat_id, "repo") os.makedirs(self.repo, mode=0o700) os.chown(self.repo, self.operator_uid, self.operator_gid) self.operator = pwd.getpwuid(self.operator_uid).pw_name self.implementer = implementer_username_for_seat(self.seat_id) os.environ["OVERDECK_SEAT_CONTROL_BASE"] = os.path.join(self.tmp, "control") os.environ["OVERDECK_SEAT_RUNTIME_BASE"] = os.path.join(self.tmp, "runtime") os.environ["OVERDECK_SEAT_TEST_MODE"] = "1" def tearDown(self) -> None: shutil.rmtree(self.tmp, ignore_errors=True) def test_implementer_created_file_stays_operator_writable(self) -> None: try: pwd.getpwnam(self.implementer) except KeyError: self.skipTest(f"{self.implementer} not provisioned") profile = os.path.join(self.home, ".claudex-accounts", "roy", self.seat_id) state = os.path.join(self.home, ".local/state/overdeck/seats", self.seat_id) os.makedirs(profile, mode=0o700) os.makedirs(state, mode=0o700) os.chown(profile, self.operator_uid, self.operator_gid) os.chown(state, self.operator_uid, self.operator_gid) guard = os.path.join(self.home, ".local/share/overdeck/seat-guard/current/bin") claude = os.path.join(self.home, ".local/bin/claude") launcher = os.path.join(self.home, ".claude/bin/seat-launcher") for path in (guard, claude, launcher): os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w", encoding="utf-8") as fh: fh.write("#!/bin/sh\n") os.chmod(path, 0o755) grant_managed_runtime_access( self.implementer, self.operator, self.home, self.seat_id, operator_uid=self.operator_uid, operator_gid=self.operator_gid, repo=self.repo, profile=profile, state_dir=state, guard_bin=guard, claude_bin=claude, launcher=launcher, ) nested = os.path.join(self.repo, "nested", "from-implementer.txt") proc = subprocess.run( ["sudo", "-n", "-u", self.implementer, "sh", "-c", f"mkdir -p {shlex.quote(os.path.dirname(nested))} && printf seat >{shlex.quote(nested)}"], capture_output=True, text=True, ) if proc.returncode != 0: self.skipTest(f"cannot write as implementer: {proc.stderr}") op_write = subprocess.run( ["sudo", "-n", "-u", self.operator, "test", "-w", nested], capture_output=True, text=True, ) self.assertEqual(op_write.returncode, 0, op_write.stderr) class TrustedConfigAclTests(unittest.TestCase): def setUp(self) -> None: if shutil.which("setfacl") is None: self.skipTest("setfacl unavailable") self.tmp = tempfile.mkdtemp(prefix="seat-trusted-acl-") self.operator_uid = os.getuid() self.operator_gid = os.getgid() self.home = os.path.join(self.tmp, "home") self.seat_id = "seat-a" os.makedirs(self.home, mode=0o700) os.chown(self.home, self.operator_uid, self.operator_gid) self.state = os.path.join(self.home, ".local/state/overdeck/seats", self.seat_id) trusted = trusted_config_dir(self.state) os.makedirs(trusted, mode=0o700) os.chown(trusted, self.operator_uid, self.operator_gid) for name in LAUNCH_AUTHORITY_FILES: path = os.path.join(trusted, name) with open(path, "w", encoding="utf-8") as fh: fh.write("{}\n" if name.endswith(".json") else "") os.chmod(path, 0o600) os.chown(path, self.operator_uid, self.operator_gid) self.operator = pwd.getpwuid(self.operator_uid).pw_name self.implementer = implementer_username_for_seat(self.seat_id) try: pwd.getpwnam(self.implementer) except KeyError: self.skipTest(f"{self.implementer} not provisioned") grant_trusted_config_read_boundary(self.state, self.implementer, self.operator) def tearDown(self) -> None: shutil.rmtree(self.tmp, ignore_errors=True) def _sudo_as(self, user: str, argv: list[str]) -> subprocess.CompletedProcess[str]: return subprocess.run(["sudo", "-n", "-u", user, *argv], capture_output=True, text=True) def test_implementer_reads_authority_files_and_traverses_directory(self) -> None: trusted = trusted_config_dir(self.state) self.assertEqual(self._sudo_as(self.implementer, ["test", "-x", trusted]).returncode, 0) for path in launch_authority_paths(self.state): self.assertEqual(self._sudo_as(self.implementer, ["test", "-r", path]).returncode, 0, path) def test_implementer_cannot_mutate_trusted_config(self) -> None: trusted = trusted_config_dir(self.state) authority = launch_authority_paths(self.state)[0] self.assertNotEqual(self._sudo_as(self.implementer, ["test", "-w", trusted]).returncode, 0) self.assertNotEqual(self._sudo_as(self.implementer, ["test", "-w", authority]).returncode, 0) rogue = os.path.join(trusted, "rogue.txt") self.assertNotEqual( self._sudo_as(self.implementer, ["sh", "-c", f"printf x >{shlex.quote(rogue)}"]).returncode, 0, ) self.assertNotEqual(self._sudo_as(self.implementer, ["rm", "-f", authority]).returncode, 0) self.assertNotEqual( self._sudo_as(self.implementer, ["mv", authority, f"{authority}.bak"]).returncode, 0, ) def test_directory_acl_keeps_execute_after_file_acl(self) -> None: trusted = trusted_config_dir(self.state) acl = subprocess.run(["getfacl", "-p", trusted], capture_output=True, text=True).stdout self.assertRegex(acl, rf"user:{self.implementer}:r-x") self.assertNotRegex(acl, rf"user:{self.implementer}:r--\n") if __name__ == "__main__": unittest.main()