#!/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 socket import stat import sys import tempfile import unittest from contextlib import redirect_stderr from io import StringIO 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) import seat_scope_entry # noqa: E402 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, expected_local_socket, capture_local_path_identity, file_sha256, load_install_manifest, load_session_manifest, launch_remote_lifecycle_owner, ns_inode, publish_exit_receipt, write_session_manifest, run_local_session, parse_proc_start_time, 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 LocalHostedSessionTests(unittest.TestCase): def test_admitted_executable_metadata_change_fails_revalidation(self) -> None: with tempfile.TemporaryDirectory() as tmp: executable = os.path.join(tmp, "runtime") with open(executable, "w", encoding="utf-8") as output: output.write("#!/bin/sh\n") os.chmod(executable, 0o755) identity = capture_local_path_identity(executable, (tmp,), executable=True, label="runtime") try: os.chmod(executable, 0o700) with self.assertRaises(SystemExit): identity.revalidate() finally: identity.close() def test_admitted_executable_hard_link_is_rejected(self) -> None: with tempfile.TemporaryDirectory() as tmp: executable = os.path.join(tmp, "runtime") alias = os.path.join(tmp, "alias") with open(executable, "w", encoding="utf-8") as output: output.write("#!/bin/sh\n") os.chmod(executable, 0o755) os.link(executable, alias) with self.assertRaises(SystemExit): capture_local_path_identity(executable, (tmp,), executable=True, label="runtime") def test_replaced_executable_fails_before_acl_mutation(self) -> None: with tempfile.TemporaryDirectory() as tmp: executable = os.path.join(tmp, "runtime") with open(executable, "w", encoding="utf-8") as fh: fh.write("#!/bin/sh\n") os.chmod(executable, 0o755) identity = capture_local_path_identity(executable, (tmp,), executable=True, label="runtime") replacement = os.path.join(tmp, "replacement") with open(replacement, "w", encoding="utf-8") as fh: fh.write("#!/bin/sh\nexit 9\n") os.chmod(replacement, 0o755) os.replace(replacement, executable) with patch("seat_scope_entry.subprocess.run") as mutate: with redirect_stderr(StringIO()), self.assertRaises(SystemExit): identity.grant_acl("ods-seat-local", "rx") mutate.assert_not_called() identity.close() def test_acl_targets_retained_descriptor_capability(self) -> None: with tempfile.TemporaryDirectory() as tmp: executable = os.path.join(tmp, "runtime") with open(executable, "w", encoding="utf-8") as fh: fh.write("#!/bin/sh\n") os.chmod(executable, 0o755) identity = capture_local_path_identity(executable, (tmp,), executable=True, label="runtime") completed = __import__("subprocess").CompletedProcess([], 0, "", "") with patch("seat_scope_entry.subprocess.run", return_value=completed) as run: identity.grant_acl("ods-seat-local", "rx") command = run.call_args.args[0] self.assertEqual(command[-1], f"/proc/self/fd/{identity.fds[-1]}") self.assertEqual(run.call_args.kwargs["pass_fds"], (identity.fds[-1],)) identity.close() def test_broker_owner_surfaces_worker_failure_and_cleans_socket(self) -> None: import seat_scope_entry with tempfile.TemporaryDirectory() as tmp: socket_path = os.path.join(tmp, "broker.sock") listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) listener.bind(socket_path) try: owner = seat_scope_entry.LocalExecutionBroker(socket_path, listener, lambda: (_ for _ in ()).throw(RuntimeError("boom"))) owner.start() with self.assertRaisesRegex(RuntimeError, "boom"): owner.wait(1.0) owner.close() self.assertFalse(os.path.exists(socket_path)) self.assertFalse(owner.is_alive()) finally: listener.close() def test_broker_owner_close_unblocks_waiting_accept(self) -> None: import seat_scope_entry with tempfile.TemporaryDirectory() as tmp: socket_path = os.path.join(tmp, "broker.sock") listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) listener.bind(socket_path) listener.listen(1) def accept_once() -> None: connection, _address = listener.accept() connection.close() owner = seat_scope_entry.LocalExecutionBroker(socket_path, listener, accept_once) owner.start() owner.close() self.assertFalse(owner.is_alive()) self.assertFalse(os.path.exists(socket_path)) def test_broker_socket_uses_distinct_implementer_gid(self) -> None: import seat_scope_entry with tempfile.TemporaryDirectory() as tmp, patch("seat_scope_entry.os.chown") as chown, patch( "seat_scope_entry.os.chmod" ), patch("seat_scope_entry.LocalExecutionBroker.start"): manifest = os.path.join(tmp, "manifest.json") identities = {tmp: __import__("unittest").mock.Mock(executable=False, fds=[10]), "/bin/a": __import__("unittest").mock.Mock(executable=True, fds=[11]), "/bin/b": __import__("unittest").mock.Mock(executable=True, fds=[12])} owner = seat_scope_entry.start_local_execution_broker( "seat-x", "11111111-1111-4111-8111-111111111111", 61123, 61234, "ods-seat-x", identities, ["/bin/a", "/bin/b"], manifest) owner.listener.close() os.unlink(owner.socket_path) chown.assert_called_once_with(owner.socket_path, 0, 61234) def test_missing_or_relative_runtime_fails_before_mutation(self) -> None: with tempfile.TemporaryDirectory() as home: admission = os.path.join(home, "admission") with open(admission, "w", encoding="utf-8") as fh: fh.write("#!/bin/sh\n") os.chmod(admission, 0o755) socket_path = expected_local_socket(home, "seat-local") for argv in ([admission], [admission, "relative-runtime"]): with self.subTest(argv=argv), patch("seat_scope_entry.capture_local_path_identity") as capture, patch( "seat_scope_entry.ensure_implementer_user" ) as ensure, patch("seat_scope_entry.grant_local_runtime_access") as grant, patch( "seat_scope_entry.prepare_socket" ) as prepare: with redirect_stderr(StringIO()), self.assertRaises(SystemExit): run_local_session(paths=ScopeEntryPaths(), seat_id="seat-local", socket=socket_path, runtime_argv=argv, cwd=home, attach=False, reap_identity="22222222-2222-4222-8222-222222222222", operator_user="owner", operator_uid=1000, operator_gid=1000, operator_home=home) capture.assert_not_called(); ensure.assert_not_called() grant.assert_not_called(); prepare.assert_not_called() def test_many_local_ledgers_share_one_bounded_implementer_identity(self) -> None: import seat_scope_entry resolved = ("ods-local-hosted", 61123, 61234) with patch("seat_scope_entry.ensure_implementer_user", return_value=resolved) as ensure: identities = [seat_scope_entry.ensure_local_implementer_user(f"ledger-{index}", "owner") for index in range(100)] self.assertEqual(set(identities), {resolved}) self.assertEqual({call.args[0] for call in ensure.call_args_list}, {seat_scope_entry.LOCAL_IMPLEMENTER_ID}) self.assertNotEqual(resolved[1], 1000) def test_local_implementer_home_is_not_ledger_scoped(self) -> None: import seat_scope_entry homes = {seat_scope_entry.local_implementer_home(f"ledger-{index}") for index in range(100)} self.assertEqual(homes, {seat_scope_entry.implementer_home_for_seat(seat_scope_entry.LOCAL_IMPLEMENTER_ID)}) def test_socket_mismatch_fails_before_filesystem_mutation(self) -> None: with patch("seat_scope_entry.ensure_implementer_user") as ensure, patch( "seat_scope_entry.prepare_socket" ) as prepare, patch("seat_scope_entry.os.chown") as chown: with self.assertRaises(SystemExit): run_local_session( paths=ScopeEntryPaths(), seat_id="seat-local", socket="/var/lib/forbidden/tmux.sock", runtime_argv=["/home/owner/.claude/bin/_agent-session-admission", "/usr/bin/true"], cwd="/home/owner/project", attach=False, reap_identity="22222222-2222-4222-8222-222222222222", operator_user="owner", operator_uid=1000, operator_gid=1000, operator_home="/home/owner", ) ensure.assert_not_called() prepare.assert_not_called() chown.assert_not_called() def test_cwd_escape_fails_before_acl_mutation(self) -> None: socket = expected_local_socket("/home/owner", "seat-local") with patch("seat_scope_entry.ensure_implementer_user", return_value=("ods-seat-local", 61123, 61123)), patch( "seat_scope_entry.grant_local_runtime_access" ) as grant, patch("seat_scope_entry.prepare_socket") as prepare: with self.assertRaises(SystemExit): run_local_session( paths=ScopeEntryPaths(), seat_id="seat-local", socket=socket, runtime_argv=["/home/owner/.claude/bin/_agent-session-admission", "/usr/bin/true"], cwd="/etc", attach=False, reap_identity="22222222-2222-4222-8222-222222222222", operator_user="owner", operator_uid=1000, operator_gid=1000, operator_home="/home/owner", ) grant.assert_not_called() prepare.assert_not_called() with tempfile.TemporaryDirectory() as tmp: control = os.path.join(tmp, "control") socket_dir = os.path.join(tmp, "state", "sock") os.makedirs(socket_dir) socket = os.path.join(socket_dir, "seat-local") paths = ScopeEntryPaths(tmux="/usr/bin/tmux", sudo="/usr/bin/sudo") calls = [] events = [] waited = __import__("subprocess").CompletedProcess([], 0, "", "") dead = __import__("subprocess").CompletedProcess([], 0, "1 23\n", "") admission = os.path.join(tmp, "_agent-session-admission") runtime = os.path.join(tmp, "runtime") for executable in (admission, runtime): with open(executable, "w", encoding="utf-8") as fh: fh.write("#!/bin/sh\n") os.chmod(executable, 0o755) broker_owner = __import__("unittest").mock.Mock() broker_owner.socket_path = os.path.join(tmp, "broker.sock") broker_owner.wait.side_effect = lambda *_args: events.append("broker-wait") broker_owner.close.side_effect = lambda: events.append("broker-close") def run_command(*_args, **_kwargs): index = len(calls) calls.append(index) if index == 1: events.append("lifecycle-wait") return waited return dead if index == 2 else __import__("subprocess").CompletedProcess([], 0) original_identity_close = seat_scope_entry.LocalPathIdentity.close def close_identity(identity): events.append("identity-close") original_identity_close(identity) with patch.dict(os.environ, {"OVERDECK_SEAT_CONTROL_BASE": control, "OVERDECK_SEAT_TEST_MODE": "1"}), patch( "seat_scope_entry.expected_local_socket", return_value=socket ), patch( "seat_scope_entry.ensure_implementer_user", return_value=("ods-seat-local", 61123, 61123) ), patch("seat_scope_entry.os.chown"), patch( "seat_scope_entry.write_tmux_conf", return_value=os.path.join(tmp, "tmux.conf") ), patch("seat_implementer_identity.require_setfacl") as require_setfacl, patch( "seat_implementer_identity.grant_link_ancestors_acl" ) as grant_ancestors, patch( "seat_implementer_identity.grant_executable_acl" ) as grant_executable, patch( "seat_implementer_identity.grant_traverse_acl" ) as grant_traverse, patch("seat_scope_entry.LocalPathIdentity.grant_acl"), patch( "seat_scope_entry.LocalPathIdentity.close", autospec=True, side_effect=close_identity ), patch( "seat_scope_entry.start_local_execution_broker", return_value=broker_owner, ), patch( "seat_scope_entry.query_tmux_server_pid", return_value=4321 ), patch( "seat_scope_entry.finalize_session_manifest", side_effect=lambda _p, manifest, _s, _pid, _proc: manifest ), patch( "seat_scope_entry.publish_local_authority_routing" ), patch( "seat_scope_entry.verify_finalized_live_identity" ), patch( "seat_scope_entry.subprocess.run", side_effect=run_command ) as run: status = run_local_session( paths=paths, seat_id="seat-local", socket=socket, runtime_argv=[admission, runtime, "--exit", "23"], cwd=tmp, attach=False, reap_identity="22222222-2222-4222-8222-222222222222", operator_user="owner", operator_uid=1000, operator_gid=1000, operator_home=tmp, ) require_setfacl.assert_called_once_with() grant_traverse.assert_not_called() grant_executable.assert_not_called() grant_ancestors.assert_not_called() self.assertEqual(status, 23) broker_owner.wait.assert_called_once() broker_owner.close.assert_called_once_with() self.assertEqual(events[:4], ["broker-wait", "identity-close", "identity-close", "identity-close"]) self.assertEqual(events[4:], ["lifecycle-wait", "broker-close"]) start = run.call_args_list[0].args[0] self.assertIn(paths.systemd_run, start) self.assertIn("--user", start) self.assertIn("--slice=agent.slice", start) self.assertIn("--service-type=exec", start) hook_index = start.index("set-hook") launch_index = start.index("new-session") self.assertLess(hook_index, launch_index) self.assertEqual(start[hook_index:hook_index + 3], ["set-hook", "-g", "pane-died"]) self.assertRegex(start[hook_index + 3], r"^wait-for -S agent-exit-[0-9a-f-]{36}$") self.assertNotIn("--wait", start) manager_boundary = start.index("--") managed_command = start[manager_boundary + 1:] self.assertEqual(managed_command[:2], [paths.sudo, "-n"]) self.assertEqual(managed_command[2], "/usr/local/bin/overdeck-seat-implementer-exec") self.assertNotIn("-u", managed_command) self.assertIn("--local-session", managed_command) self.assertIn("--seat-id", managed_command) self.assertIn("seat-local", managed_command) self.assertNotIn("ods-seat-local", managed_command) self.assertNotIn(admission, start) self.assertNotIn(runtime, start) self.assertIn("/usr/local/bin/overdeck-seat-execution-client", start) self.assertIn(os.path.join(tmp, "broker.sock"), start) with open(os.path.join(control, "seat-local", "session.json"), encoding="utf-8") as fh: generation = json.load(fh)["generation"] self.assertRegex(generation, r"^[0-9a-f-]{36}$") wait = run.call_args_list[1].args[0] self.assertEqual(wait[-2:], ["wait-for", f"agent-exit-{generation}"]) self.assertEqual(len(run.call_args_list), 3) status_probe = run.call_args_list[2].args[0] self.assertIn("list-panes", status_probe) with open(os.path.join(control, "seat-local", "exit.json"), encoding="utf-8") as fh: receipt = json.load(fh) self.assertEqual(receipt["generation"], generation) self.assertEqual(receipt["exitStatus"], 23) class LocalSessionManifestTests(unittest.TestCase): def test_remote_hook_is_installed_before_fast_pane_can_exit(self) -> None: self.assertTrue(hasattr(seat_scope_entry, "build_remote_tmux_command")) command = seat_scope_entry.build_remote_tmux_command( "tmux", "/control/tmux.conf", "/state/tmux.sock", "/state/launcher", "11111111-1111-4111-8111-111111111111", ) hook = command.index("set-hook") launch = command.index("new-session") self.assertLess(hook, launch) self.assertEqual(command[hook:hook + 4], [ "set-hook", "-g", "pane-died", "wait-for -S agent-exit-11111111-1111-4111-8111-111111111111", ]) def test_remote_lifecycle_owner_is_detached_and_generation_bound(self) -> None: generation = "11111111-1111-4111-8111-111111111111" paths = ScopeEntryPaths(systemd_run="/usr/bin/systemd-run", tmux_mediator="/usr/local/bin/overdeck-seat-tmux-mediator") with patch("seat_scope_entry.subprocess.run", return_value=__import__("subprocess").CompletedProcess([], 0)) as run: launch_remote_lifecycle_owner(paths, "seat-x", "/run/tmux.sock", generation, "owner") command = run.call_args.args[0] self.assertEqual(command[:2], ["/usr/bin/systemd-run", "--quiet"]) self.assertIn("--collect", command) self.assertNotIn("--wait", command) self.assertIn(f"--unit=overdeck-seat-lifecycle-seat-x-{generation}.service", command) self.assertIn("--setenv=OVERDECK_SEAT_SSH_USER=owner", command) self.assertEqual(command[-8:], [ "/usr/local/bin/overdeck-seat-tmux-mediator", "--seat-id", "seat-x", "--socket", "/run/tmux.sock", "--generation", generation, "lifecycle", ]) def test_operator_stop_receipt_is_distinct_from_pane_exit(self) -> None: generation = "11111111-1111-4111-8111-111111111111" with tempfile.TemporaryDirectory() as tmp, patch.dict(os.environ, {"OVERDECK_SEAT_TEST_MODE": "1"}): manifest_path = os.path.join(tmp, "session.json") receipt_path = os.path.join(tmp, "exit.json") write_session_manifest(manifest_path, { "schema": 1, "seatId": "seat-local", "generation": generation, "implementerUid": 61123, "socket": os.path.join(tmp, "tmux.sock"), "session": "main", }) publish_exit_receipt(manifest_path, receipt_path, "seat-local", generation, terminal_reason="operator-stop") with open(receipt_path, encoding="utf-8") as fh: receipt = json.load(fh) self.assertEqual(receipt["terminalReason"], "operator-stop") self.assertNotIn("exitStatus", receipt) def test_manifest_and_receipt_bind_generation_and_exact_exit_status(self) -> None: generation = "11111111-1111-4111-8111-111111111111" with tempfile.TemporaryDirectory() as tmp, patch.dict(os.environ, {"OVERDECK_SEAT_TEST_MODE": "1"}): manifest_path = os.path.join(tmp, "session.json") receipt_path = os.path.join(tmp, "exit.json") manifest = { "schema": 1, "seatId": "seat-local", "generation": generation, "implementerUid": 61123, "socket": os.path.join(tmp, "tmux.sock"), "session": "main", } from seat_scope_entry import write_session_manifest write_session_manifest(manifest_path, manifest) self.assertEqual(load_session_manifest(manifest_path, "seat-local", generation), manifest) publish_exit_receipt(manifest_path, receipt_path, "seat-local", generation, 23) with open(receipt_path, encoding="utf-8") as fh: receipt = json.load(fh) self.assertEqual(receipt, { "schema": 1, "seatId": "seat-local", "generation": generation, "exitStatus": 23, }) def test_manifest_rejects_hard_link(self) -> None: generation = "11111111-1111-4111-8111-111111111111" with tempfile.TemporaryDirectory() as tmp: manifest_path = os.path.join(tmp, "session.json") linked_path = os.path.join(tmp, "linked.json") write_session_manifest(manifest_path, { "schema": 1, "seatId": "seat-local", "generation": generation, "implementerUid": 61123, "socket": os.path.join(tmp, "tmux.sock"), "session": "main", }) os.link(manifest_path, linked_path) with self.assertRaises(SystemExit): load_session_manifest(manifest_path, "seat-local", generation) def test_manifest_rejects_group_readable_mode(self) -> None: generation = "11111111-1111-4111-8111-111111111111" with tempfile.TemporaryDirectory() as tmp: manifest_path = os.path.join(tmp, "session.json") write_session_manifest(manifest_path, { "schema": 1, "seatId": "seat-local", "generation": generation, "implementerUid": 61123, "socket": os.path.join(tmp, "tmux.sock"), "session": "main", }) os.chmod(manifest_path, 0o640) with self.assertRaises(SystemExit): load_session_manifest(manifest_path, "seat-local", generation) def test_manifest_replacement_before_receipt_publish_fails(self) -> None: generation = "11111111-1111-4111-8111-111111111111" replacement = "22222222-2222-4222-8222-222222222222" with tempfile.TemporaryDirectory() as tmp, patch.dict(os.environ, {"OVERDECK_SEAT_TEST_MODE": "1"}): manifest_path = os.path.join(tmp, "session.json") receipt_path = os.path.join(tmp, "exit.json") base = {"schema": 1, "seatId": "seat-local", "generation": generation, "implementerUid": 61123, "socket": os.path.join(tmp, "tmux.sock"), "session": "main"} write_session_manifest(manifest_path, base) original = __import__("seat_scope_entry").load_session_manifest def replace_after_validation(path, seat_id, expected): manifest = original(path, seat_id, expected) newer = dict(base) newer["generation"] = replacement replacement_path = os.path.join(tmp, "replacement.json") __import__("seat_scope_entry")._atomic_json_write(replacement_path, newer) os.replace(replacement_path, manifest_path) return manifest with patch("seat_scope_entry.load_session_manifest", side_effect=replace_after_validation): with self.assertRaises(SystemExit): publish_exit_receipt(manifest_path, receipt_path, "seat-local", generation, 23) self.assertFalse(os.path.exists(receipt_path)) def test_stale_generation_cannot_publish_receipt(self) -> None: generation = "11111111-1111-4111-8111-111111111111" with tempfile.TemporaryDirectory() as tmp: manifest_path = os.path.join(tmp, "session.json") from seat_scope_entry import write_session_manifest write_session_manifest(manifest_path, { "schema": 1, "seatId": "seat-local", "generation": generation, "implementerUid": 61123, "socket": os.path.join(tmp, "tmux.sock"), "session": "main", }) with self.assertRaises(SystemExit): publish_exit_receipt( manifest_path, os.path.join(tmp, "exit.json"), "seat-local", "22222222-2222-4222-8222-222222222222", 0, ) 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, "11111111-1111-4111-8111-111111111111", 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), "execution_broker_module_sha256": file_sha256(module), "execution_client_module_sha256": file_sha256(module), "execution_client_sha256": file_sha256(wrapper), "reap_close_sha256": file_sha256(wrapper), "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.dict(os.environ, {"OVERDECK_SEAT_TEST_MODE": "1"}), 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") client_wrapper = os.path.join(tmp, "overdeck-seat-execution-client") 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") write_fixture_wrapper(client_wrapper, "/usr/local/lib/overdeck/seat_execution_client.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), "execution_broker_module_sha256": file_sha256(module), "execution_client_module_sha256": file_sha256(module), "execution_client_sha256": file_sha256(client_wrapper), "reap_close_sha256": file_sha256(wrapper), "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.dict(os.environ, {"OVERDECK_SEAT_TEST_MODE": "1"}), 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.multiple( "seat_scope_entry", INSTALLED_EXECUTION_BROKER_MODULE=module, INSTALLED_EXECUTION_CLIENT_MODULE=module, INSTALLED_EXECUTION_CLIENT=client_wrapper, ), patch( "seat_scope_entry.INSTALLED_REAP_CLOSE", wrapper ), 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") client_wrapper = os.path.join(tmp, "overdeck-seat-execution-client") 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, client_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), "execution_broker_module_sha256": file_sha256(module), "execution_client_module_sha256": file_sha256(module), "execution_client_sha256": file_sha256(client_wrapper), "reap_close_sha256": file_sha256(wrapper), "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.dict(os.environ, {"OVERDECK_SEAT_TEST_MODE": "1"}), 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.multiple( "seat_scope_entry", INSTALLED_EXECUTION_BROKER_MODULE=module, INSTALLED_EXECUTION_CLIENT_MODULE=module, INSTALLED_EXECUTION_CLIENT=client_wrapper, ), patch( "seat_scope_entry.INSTALLED_REAP_CLOSE", wrapper ), 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") client_wrapper = os.path.join(tmp, "overdeck-seat-execution-client") 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"), (client_wrapper, "/usr/local/lib/overdeck/seat_execution_client.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), "execution_broker_module_sha256": file_sha256(module), "execution_client_module_sha256": file_sha256(module), "execution_client_sha256": file_sha256(client_wrapper), "reap_close_sha256": file_sha256(wrapper), "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.dict(os.environ, {"OVERDECK_SEAT_TEST_MODE": "1"}), 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.multiple( "seat_scope_entry", INSTALLED_EXECUTION_BROKER_MODULE=module, INSTALLED_EXECUTION_CLIENT_MODULE=module, INSTALLED_EXECUTION_CLIENT=client_wrapper, ), patch( "seat_scope_entry.INSTALLED_REAP_CLOSE", wrapper ), 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 ProcessIdentityTests(unittest.TestCase): def test_proc_start_time_allows_spaces_in_comm(self) -> None: suffix = ["S", *(["0"] * 18), "55", "0"] self.assertEqual(parse_proc_start_time(f"1234 (tmux: server) {' '.join(suffix)}\n"), 55) def test_proc_start_time_rejects_nonpositive(self) -> None: suffix = ["S", *(["0"] * 18), "0", "0"] with self.assertRaises(SystemExit): parse_proc_start_time(f"1234 (tmux: server) {' '.join(suffix)}\n") 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", generation="11111111-1111-4111-8111-111111111111", 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 off" exit 0 fi if [ "$#" -eq 2 ] && [ "$1" = "-g" ] && [ "$2" = "remain-on-exit" ]; then echo "remain-on-exit 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()