#!/usr/bin/env python3 from __future__ import annotations import json import os import subprocess 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) import seat_tmux_mediator as mediator from seat_scope_entry import write_session_manifest class SeatTmuxMediatorTests(unittest.TestCase): @staticmethod def synthetic_binding(): binding = __import__("unittest").mock.Mock() binding.socket = "/run/tmux.sock" binding.parents = [("/run", 41, (1, 2))] binding.socket_identity = (1, 3) return binding def test_lifecycle_waits_event_then_publishes_exact_status_once(self) -> None: generation = "11111111-1111-4111-8111-111111111111" manifest = {"generation": generation, "implementerUid": 61123, "socket": "/run/tmux.sock"} waited = subprocess.CompletedProcess([], 0, "", "") dead = subprocess.CompletedProcess([], 0, "1 23\n", "") with patch("seat_tmux_mediator.os.geteuid", return_value=0), patch( "seat_tmux_mediator.resolve_target", return_value=("op", 1000, 1000, "/home/op") ), patch("seat_tmux_mediator.validate_socket"), patch( "seat_tmux_mediator.bind_socket_ancestry", return_value=self.synthetic_binding() ), patch("seat_tmux_mediator.revalidate_socket_ancestry"), patch( "seat_tmux_mediator.session_manifest_path", return_value="/control/session.json" ), patch("seat_tmux_mediator.load_session_manifest", return_value=manifest), patch( "seat_tmux_mediator.resolve_implementer_for_seat", return_value=("ods-seat-x", 61123, 61123) ), patch("seat_tmux_mediator.verify_live_identity"), patch( "seat_tmux_mediator.run_tmux_bound", side_effect=[waited, dead] ) as run, patch("seat_tmux_mediator.publish_exit_receipt") as publish: mediator.main(["--seat-id", "seat-x", "--socket", "/run/tmux.sock", "--generation", generation, "lifecycle"]) self.assertEqual(run.call_count, 2) self.assertEqual(run.call_args_list[0].args[2], ["wait-for", f"agent-exit-{generation}"]) self.assertEqual(run.call_args_list[1].args[2], ["list-panes", "-t", "main", "-F", "#{pane_dead} #{pane_dead_status}"]) self.assertEqual(publish.call_args.args[-1], 23) def test_operator_stop_publishes_only_after_successful_kill(self) -> None: generation = "11111111-1111-4111-8111-111111111111" manifest = {"generation": generation, "implementerUid": 61123, "socket": "/run/tmux.sock"} events = [] stopped = subprocess.CompletedProcess([], 0, "", "") with patch("seat_tmux_mediator.os.geteuid", return_value=0), patch( "seat_tmux_mediator.resolve_target", return_value=("op", 1000, 1000, "/home/op") ), patch("seat_tmux_mediator.validate_socket"), patch( "seat_tmux_mediator.bind_socket_ancestry", return_value=self.synthetic_binding() ), patch("seat_tmux_mediator.revalidate_socket_ancestry"), patch( "seat_tmux_mediator.session_manifest_path", return_value="/control/session.json" ), patch("seat_tmux_mediator.load_session_manifest", return_value=manifest), patch( "seat_tmux_mediator.resolve_implementer_for_seat", return_value=("ods-seat-x", 61123, 61123) ), patch("seat_tmux_mediator.verify_live_identity"), patch( "seat_tmux_mediator.write_stop_intent", side_effect=lambda *_: events.append("intent") ), patch("seat_tmux_mediator.run_tmux_bound", side_effect=lambda *_: (events.append("kill"), stopped)[1]), patch( "seat_tmux_mediator.verify_stopped_identity", side_effect=lambda *_: events.append("absent") ), patch("seat_tmux_mediator.publish_exit_receipt", side_effect=lambda *_a, **_k: events.append("receipt")), patch( "seat_tmux_mediator.remove_stop_intent", side_effect=lambda *_: events.append("remove") ): mediator.main(["--seat-id", "seat-x", "--socket", "/run/tmux.sock", "--generation", generation, "operator-stop"]) self.assertEqual(events, ["intent", "kill", "absent", "receipt", "remove"]) def test_operator_stop_recovers_after_kill_before_receipt(self) -> None: generation = "11111111-1111-4111-8111-111111111111" manifest = {"seatId": "seat-x", "generation": generation, "implementerUid": 61123, "socket": "/run/tmux.sock", "serverPid": 4321, "serverStartTime": 99, "cgroup": "/agent.slice/session.scope"} with patch("seat_tmux_mediator.os.geteuid", return_value=0), patch( "seat_tmux_mediator.resolve_target", return_value=("op", 1000, 1000, "/home/op") ), patch("seat_tmux_mediator.validate_socket"), patch( "seat_tmux_mediator.bind_socket_ancestry", return_value=self.synthetic_binding() ), patch("seat_tmux_mediator.revalidate_socket_ancestry"), patch( "seat_tmux_mediator.session_manifest_path", return_value="/control/session.json" ), patch("seat_tmux_mediator.load_session_manifest", return_value=manifest), patch( "seat_tmux_mediator.resolve_implementer_for_seat", return_value=("ods-seat-x", 61123, 61123) ), patch("seat_tmux_mediator.load_stop_intent", return_value=manifest), patch( "seat_tmux_mediator.verify_stopped_identity" ), patch("seat_tmux_mediator.publish_exit_receipt") as publish, patch( "seat_tmux_mediator.remove_stop_intent" ) as remove, patch("seat_tmux_mediator.verify_live_identity") as live, patch( "seat_tmux_mediator.run_tmux_bound" ) as run: mediator.main(["--seat-id", "seat-x", "--socket", "/run/tmux.sock", "--generation", generation, "operator-stop"]) publish.assert_called_once() remove.assert_called_once_with("/control/session.json") live.assert_not_called() run.assert_not_called() def test_bound_socket_leaf_replacement_fails_revalidation(self) -> None: with tempfile.TemporaryDirectory() as tmp: parent = os.path.join(tmp, "sock") os.mkdir(parent, 0o700) socket_path = os.path.join(parent, "tmux.sock") first = __import__("socket").socket(__import__("socket").AF_UNIX) second = __import__("socket").socket(__import__("socket").AF_UNIX) first.bind(socket_path) binding = mediator.bind_socket_ancestry(socket_path) try: os.unlink(socket_path) second.bind(socket_path) with self.assertRaises(SystemExit): mediator.revalidate_socket_ancestry(binding) finally: binding.close() first.close() second.close() def test_bound_socket_parent_replacement_fails_revalidation(self) -> None: with tempfile.TemporaryDirectory() as tmp: parent = os.path.join(tmp, "sock") moved = os.path.join(tmp, "moved") os.mkdir(parent, 0o700) socket = os.path.join(parent, "tmux.sock") server = __import__("socket").socket(__import__("socket").AF_UNIX) server.bind(socket) binding = mediator.bind_socket_ancestry(socket) try: os.rename(parent, moved) os.mkdir(parent, 0o700) with self.assertRaises(SystemExit): mediator.revalidate_socket_ancestry(binding) finally: binding.close() server.close() def test_tmux_uses_retained_socket_parent_capability(self) -> None: with tempfile.TemporaryDirectory() as tmp: parent = os.path.join(tmp, "sock") os.mkdir(parent, 0o700) socket = os.path.join(parent, "tmux.sock") server = __import__("socket").socket(__import__("socket").AF_UNIX) server.bind(socket) binding = mediator.bind_socket_ancestry(socket) completed = subprocess.CompletedProcess([], 0, "", "") try: account = __import__("types").SimpleNamespace(pw_uid=61123, pw_gid=61123) with patch("seat_tmux_mediator.pwd.getpwnam", return_value=account), patch( "seat_tmux_mediator.subprocess.run", return_value=completed ) as run: mediator.run_tmux_bound("ods-seat-x", binding, ["has-session", "-t", "main"]) parent_fd = binding.parents[-1][1] self.assertIn(f"/proc/self/fd/{parent_fd}/tmux.sock", run.call_args.args[0]) self.assertEqual(run.call_args.kwargs["pass_fds"], (parent_fd,)) self.assertNotIn(socket, run.call_args.args[0]) finally: binding.close() server.close() def test_attach_execs_with_inherited_terminal_and_socket_capability(self) -> None: binding = __import__("types").SimpleNamespace( parents=[("/run/overdeck", 41)], socket="/run/overdeck/tmux.sock" ) account = __import__("types").SimpleNamespace(pw_uid=61123, pw_gid=61234) with patch("seat_tmux_mediator.pwd.getpwnam", return_value=account), patch( "seat_tmux_mediator.os.set_inheritable" ) as inheritable, patch("seat_tmux_mediator.os.setgroups") as groups, patch( "seat_tmux_mediator.os.setgid" ) as setgid, patch("seat_tmux_mediator.os.setuid") as setuid, patch( "seat_tmux_mediator.os.execvp", side_effect=RuntimeError("exec") ) as execute, patch("seat_tmux_mediator.subprocess.run") as run: with self.assertRaisesRegex(RuntimeError, "exec"): mediator.exec_tmux_attach("ods-local-hosted", binding, ["attach-session", "-t", "main"]) inheritable.assert_called_once_with(41, True) groups.assert_called_once_with([]) setgid.assert_called_once_with(61234) setuid.assert_called_once_with(61123) self.assertEqual(execute.call_args.args[0], "tmux") self.assertEqual(execute.call_args.args[1], [ "tmux", "-S", "/proc/self/fd/41/tmux.sock", "attach-session", "-t", "main", ]) run.assert_not_called() def test_generation_is_required_before_authority_work(self) -> None: with patch("seat_tmux_mediator.os.geteuid") as geteuid: with self.assertRaises(SystemExit): mediator.main([ "--seat-id", "seat-x", "--socket", "/run/overdeck/tmux.sock", "has-session", "-t", "main", ]) geteuid.assert_not_called() def test_stale_generation_fails_before_tmux(self) -> None: current = "11111111-1111-4111-8111-111111111111" stale = "22222222-2222-4222-8222-222222222222" with tempfile.TemporaryDirectory() as tmp, patch.dict(os.environ, {"OVERDECK_SEAT_TEST_MODE": "1"}): socket = os.path.join(tmp, "tmux.sock") write_session_manifest(os.path.join(tmp, "session.json"), { "schema": 1, "seatId": "seat-x", "generation": current, "implementerUid": 61123, "socket": socket, "session": "main", }) with patch("seat_tmux_mediator.os.geteuid", return_value=0), patch( "seat_tmux_mediator.session_manifest_path", return_value=os.path.join(tmp, "session.json") ), patch( "seat_tmux_mediator.resolve_target", return_value=("op", 1000, 1000, "/home/op") ), patch("seat_tmux_mediator.validate_socket"), patch( "seat_tmux_mediator.bind_socket_ancestry", return_value=self.synthetic_binding() ), patch("seat_tmux_mediator.revalidate_socket_ancestry"), patch( "seat_tmux_mediator.run_tmux_bound" ) as run: with self.assertRaises(SystemExit): mediator.main([ "--seat-id", "seat-x", "--socket", socket, "--generation", stale, "has-session", "-t", "main", ]) run.assert_not_called() def test_exact_generation_mediates_has_session(self) -> None: generation = "11111111-1111-4111-8111-111111111111" with tempfile.TemporaryDirectory() as tmp, patch.dict(os.environ, {"OVERDECK_SEAT_TEST_MODE": "1"}): socket = os.path.join(tmp, "tmux.sock") write_session_manifest(os.path.join(tmp, "session.json"), { "schema": 1, "seatId": "seat-x", "generation": generation, "implementerUid": 61123, "socket": socket, "session": "main", }) with patch("seat_tmux_mediator.os.geteuid", return_value=0), patch( "seat_tmux_mediator.session_manifest_path", return_value=os.path.join(tmp, "session.json") ), patch( "seat_tmux_mediator.resolve_target", return_value=("op", 1000, 1000, "/home/op") ), patch("seat_tmux_mediator.validate_socket"), patch( "seat_tmux_mediator.bind_socket_ancestry", return_value=self.synthetic_binding() ), patch("seat_tmux_mediator.revalidate_socket_ancestry"), patch( "seat_tmux_mediator.resolve_implementer_for_seat", return_value=("ods-seat-x", 61123, 61123) ), patch("seat_tmux_mediator.verify_live_identity"), patch("seat_tmux_mediator.run_tmux_bound", return_value=subprocess.CompletedProcess([], 0, "", "")) as run: mediator.main([ "--seat-id", "seat-x", "--socket", socket, "--generation", generation, "has-session", "-t", "main", ]) run.assert_called_once() self.assertEqual(run.call_args.args[0], "ods-seat-x") self.assertEqual(run.call_args.args[2], ["has-session", "-t", "main"]) def test_replaced_socket_fails_before_tmux(self) -> None: generation = "11111111-1111-4111-8111-111111111111" with tempfile.TemporaryDirectory() as tmp, patch.dict(os.environ, {"OVERDECK_SEAT_TEST_MODE": "1"}): socket = os.path.join(tmp, "tmux.sock") with open(socket, "wb") as fh: fh.write(b"replacement") write_session_manifest(os.path.join(tmp, "session.json"), { "schema": 1, "seatId": "seat-x", "generation": generation, "implementerUid": 61123, "socket": socket, "session": "main", "socketDev": 1, "socketIno": 2, "serverPid": 1234, "serverStartTime": 55, "cgroup": "/agent-seat.slice/seat-x.scope", }) with patch("seat_tmux_mediator.os.geteuid", return_value=0), patch( "seat_tmux_mediator.session_manifest_path", return_value=os.path.join(tmp, "session.json") ), patch("seat_tmux_mediator.resolve_target", return_value=("op", 1000, 1000, "/home/op")), patch( "seat_tmux_mediator.validate_socket" ), patch("seat_tmux_mediator.run_tmux_bound") as run: with self.assertRaises(SystemExit): mediator.main(["--seat-id", "seat-x", "--socket", socket, "--generation", generation, "has-session", "-t", "main"]) run.assert_not_called() def test_replaced_server_start_time_fails(self) -> None: manifest = {"socketDev": 10, "socketIno": 20, "serverPid": 1234, "serverStartTime": 55, "cgroup": "/agent-seat.slice/seat-x.scope"} socket_stat = __import__("types").SimpleNamespace(st_mode=__import__("stat").S_IFSOCK, st_dev=10, st_ino=20) stat_line = "1234 (tmux) S " + "0 " * 18 + "56 0\n" cgroup = "0::/agent-seat.slice/seat-x.scope\n" with patch("seat_tmux_mediator.os.stat", return_value=socket_stat), patch( "builtins.open", side_effect=[__import__("unittest").mock.mock_open(read_data=stat_line).return_value, __import__("unittest").mock.mock_open(read_data=cgroup).return_value] ): with self.assertRaises(SystemExit): mediator.verify_live_identity(manifest, "/run/tmux.sock") def test_replaced_server_cgroup_fails(self) -> None: manifest = {"socketDev": 10, "socketIno": 20, "serverPid": 1234, "serverStartTime": 55, "cgroup": "/agent-seat.slice/seat-x.scope"} socket_stat = __import__("types").SimpleNamespace(st_mode=__import__("stat").S_IFSOCK, st_dev=10, st_ino=20) stat_line = "1234 (tmux) S " + "0 " * 18 + "55 0\n" cgroup = "0::/user.slice/replacement.scope\n" with patch("seat_tmux_mediator.os.stat", return_value=socket_stat), patch( "builtins.open", side_effect=[__import__("unittest").mock.mock_open(read_data=stat_line).return_value, __import__("unittest").mock.mock_open(read_data=cgroup).return_value] ): with self.assertRaises(SystemExit): mediator.verify_live_identity(manifest, "/run/tmux.sock") def test_reap_close_delegates_exact_identity_after_authority_validation(self) -> None: generation = "11111111-1111-4111-8111-111111111111" manifest = {"generation": generation, "implementerUid": 61123, "socket": "/run/tmux.sock"} identity = ["123", "$1", "456", generation, "%2", "789", "800", "900", "901"] completed = subprocess.CompletedProcess([], 0, "REAPED\n", "") with patch("seat_tmux_mediator.os.geteuid", return_value=0), patch( "seat_tmux_mediator.resolve_target", return_value=("op", 1000, 1000, "/home/op") ), patch("seat_tmux_mediator.validate_socket"), patch( "seat_tmux_mediator.bind_socket_ancestry", return_value=self.synthetic_binding() ), patch("seat_tmux_mediator.revalidate_socket_ancestry"), patch( "seat_tmux_mediator.session_manifest_path", return_value="/control/session.json" ), patch("seat_tmux_mediator.load_session_manifest", return_value=manifest), patch( "seat_tmux_mediator.resolve_implementer_for_seat", return_value=("ods-seat-x", 61123, 61123) ), patch("seat_tmux_mediator.verify_live_identity"), patch( "seat_tmux_mediator.subprocess.run", return_value=completed ) as run: mediator.main(["--seat-id", "seat-x", "--socket", "/run/tmux.sock", "--generation", generation, "reap-close", *identity]) self.assertEqual(run.call_args.args[0][1:], ["/run/tmux.sock", *identity]) generation = "11111111-1111-4111-8111-111111111111" manifest = {"generation": generation, "implementerUid": 61123, "socket": "/run/tmux.sock"} completed = subprocess.CompletedProcess([], 0, "1 23\n", "") with patch("seat_tmux_mediator.os.geteuid", return_value=0), patch( "seat_tmux_mediator.resolve_target", return_value=("op", 1000, 1000, "/home/op") ), patch("seat_tmux_mediator.validate_socket"), patch( "seat_tmux_mediator.bind_socket_ancestry", return_value=self.synthetic_binding() ), patch("seat_tmux_mediator.revalidate_socket_ancestry"), patch( "seat_tmux_mediator.session_manifest_path", return_value="/control/session.json" ), patch("seat_tmux_mediator.load_session_manifest", return_value=manifest), patch( "seat_tmux_mediator.resolve_implementer_for_seat", return_value=("ods-seat-x", 61123, 61123) ), patch("seat_tmux_mediator.verify_live_identity", side_effect=[None, SystemExit(1)]) as verify, patch( "seat_tmux_mediator.run_tmux_bound", return_value=completed ), patch("seat_tmux_mediator.publish_exit_receipt") as publish: with self.assertRaises(SystemExit): mediator.main(["--seat-id", "seat-x", "--socket", "/run/tmux.sock", "--generation", generation, "receipt"]) self.assertEqual(verify.call_count, 2) publish.assert_not_called() generation = "11111111-1111-4111-8111-111111111111" with tempfile.TemporaryDirectory() as tmp, patch.dict(os.environ, {"OVERDECK_SEAT_TEST_MODE": "1"}): socket = os.path.join(tmp, "tmux.sock") write_session_manifest(os.path.join(tmp, "session.json"), { "schema": 1, "seatId": "seat-x", "generation": generation, "implementerUid": 61123, "socket": socket, "session": "main", }) completed = subprocess.CompletedProcess([], 0, "1 23\n", "") with patch("seat_tmux_mediator.os.geteuid", return_value=0), patch( "seat_tmux_mediator.session_manifest_path", return_value=os.path.join(tmp, "session.json") ), patch( "seat_tmux_mediator.resolve_target", return_value=("op", 1000, 1000, "/home/op") ), patch("seat_tmux_mediator.validate_socket"), patch( "seat_tmux_mediator.bind_socket_ancestry", return_value=self.synthetic_binding() ), patch("seat_tmux_mediator.revalidate_socket_ancestry"), patch( "seat_tmux_mediator.resolve_implementer_for_seat", return_value=("ods-seat-x", 61123, 61123) ), patch("seat_tmux_mediator.verify_live_identity"), patch( "seat_tmux_mediator.exit_receipt_path", return_value=os.path.join(tmp, "exit.json") ), patch("seat_tmux_mediator.run_tmux_bound", return_value=completed): mediator.main([ "--seat-id", "seat-x", "--socket", socket, "--generation", generation, "receipt", ]) with open(os.path.join(tmp, "exit.json"), encoding="utf-8") as fh: self.assertEqual(json.load(fh)["exitStatus"], 23) if __name__ == "__main__": unittest.main()