#!/usr/bin/env python3 from __future__ import annotations import os import socket import subprocess import sys import tempfile import threading import time 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_execution_broker as broker import seat_execution_client as client class SeatExecutionClientTests(unittest.TestCase): def test_runtime_environment_preserves_allowlist_and_drops_dangerous_values(self) -> None: request = {"seatId": "seat-x", "generation": broker.TEST_GENERATION, "phase": "runtime"} response = {**request, "fdCount": 2, "environment": { "HOME": "/var/lib/overdeck/implementers/seat-x", "USER": "ods-seat-x", "LOGNAME": "ods-seat-x", "PATH": "/usr/local/bin:/usr/bin:/bin", "OVERDECK_SEAT_ID": "seat-x", "OVERDECK_SEAT_GENERATION": broker.TEST_GENERATION, "CLAUDE_CONFIG_DIR": "/var/lib/overdeck/profiles/seat-x", }} fds = [os.open("/dev/null", os.O_RDONLY) for _ in range(2)] with patch.dict(os.environ, {"LD_PRELOAD": "/tmp/evil", "ANTHROPIC_API_KEY": "secret"}, clear=False), patch( "seat_execution_client.connect_and_receive", return_value=(response, fds) ), patch("seat_execution_client.fd_exec") as execute: client.main(["--socket", "/run/broker.sock", "--seat-id", "seat-x", "--generation", broker.TEST_GENERATION, "--phase", "runtime"]) env = execute.call_args.args[3] self.assertEqual(env["HOME"], "/var/lib/overdeck/implementers/seat-x") self.assertEqual(env["CLAUDE_CONFIG_DIR"], "/var/lib/overdeck/profiles/seat-x") self.assertNotIn("LD_PRELOAD", env) self.assertNotIn("ANTHROPIC_API_KEY", env) for fd in fds: with self.assertRaises(OSError): os.fstat(fd) def test_response_environment_rejects_unknown_key(self) -> None: request = {"seatId": "seat-x", "generation": broker.TEST_GENERATION, "phase": "runtime"} response = {**request, "fdCount": 2, "environment": {"HOME": "/safe", "LD_PRELOAD": "/tmp/evil"}} with patch("seat_execution_client.connect_and_receive", return_value=(response, [10, 11])), patch( "seat_execution_client.fd_exec" ) as execute, patch("os.close"): with self.assertRaises(SystemExit): client.main(["--socket", "/run/broker.sock", "--seat-id", "seat-x", "--generation", broker.TEST_GENERATION, "--phase", "runtime"]) execute.assert_not_called() def test_runtime_phase_fd_executes_received_runtime(self) -> None: request = {"seatId": "seat-x", "generation": broker.TEST_GENERATION, "phase": "runtime"} response = {**request, "fdCount": 2, "environment": self.valid_environment()} fds = [os.open("/dev/null", os.O_RDONLY) for _ in range(2)] with patch("seat_execution_client.connect_and_receive", return_value=(response, fds)), patch( "seat_execution_client.fd_exec" ) as execute: client.main(["--socket", "/run/broker.sock", "--seat-id", "seat-x", "--generation", broker.TEST_GENERATION, "--phase", "runtime", "--", "--exit", "23"]) execute.assert_called_once_with(fds[0], fds[1], ["seat-runtime", "--exit", "23"], self.valid_environment()) for fd in fds: with self.assertRaises(OSError): os.fstat(fd) @staticmethod def valid_environment() -> dict[str, str]: return { "HOME": "/var/lib/overdeck/implementers/seat-x", "USER": "ods-seat-x", "LOGNAME": "ods-seat-x", "PATH": "/usr/local/bin:/usr/bin:/bin", "OVERDECK_SEAT_ID": "seat-x", "OVERDECK_SEAT_GENERATION": broker.TEST_GENERATION, } class SeatExecutionBrokerTests(unittest.TestCase): ENVIRONMENT = { "HOME": "/var/lib/overdeck/implementers/seat-x", "USER": "ods-seat-x", "LOGNAME": "ods-seat-x", "PATH": "/usr/local/bin:/usr/bin:/bin", "OVERDECK_SEAT_ID": "seat-x", "OVERDECK_SEAT_GENERATION": broker.TEST_GENERATION, } def test_replacement_after_admission_does_not_change_transferred_object(self) -> None: with tempfile.TemporaryDirectory() as tmp: target = os.path.join(tmp, "runtime") with open(target, "w", encoding="utf-8") as fh: fh.write("#!/bin/sh\nexit 23\n") os.chmod(target, 0o755) fd = os.open(target, os.O_RDONLY | os.O_NOFOLLOW) 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, target) left, right = socket.socketpair() try: broker.send_capabilities(left, "seat-x", broker.TEST_GENERATION, "runtime", [fd], self.ENVIRONMENT) request, received = broker.receive_capabilities(right, expected_count=1) self.assertEqual(request["generation"], broker.TEST_GENERATION) self.assertEqual(os.read(received[0], 128), b"#!/bin/sh\nexit 23\n") finally: os.close(fd) left.close() right.close() for received_fd in locals().get("received", []): os.close(received_fd) def test_wrong_generation_is_rejected_before_descriptor_send(self) -> None: left, right = socket.socketpair() try: with patch("seat_execution_broker.sendmsg_with_fds") as send: with self.assertRaises(SystemExit): broker.serve_request(left, "seat-x", broker.TEST_GENERATION, 61123, [], {"seatId": "seat-x", "generation": "22222222-2222-4222-8222-222222222222"}, peer_uid=61123, expected_phase="runtime", environment=self.ENVIRONMENT) send.assert_not_called() finally: left.close() right.close() def test_wrong_peer_uid_is_rejected_before_descriptor_send(self) -> None: left, right = socket.socketpair() try: with patch("seat_execution_broker.sendmsg_with_fds") as send: with self.assertRaises(SystemExit): broker.serve_request(left, "seat-x", broker.TEST_GENERATION, 61123, [], {"seatId": "seat-x", "generation": broker.TEST_GENERATION}, peer_uid=1000, expected_phase="runtime", environment=self.ENVIRONMENT) send.assert_not_called() finally: left.close() right.close() def test_extra_request_key_is_rejected_before_descriptor_send(self) -> None: left, right = socket.socketpair() try: request = { "seatId": "seat-x", "generation": broker.TEST_GENERATION, "phase": "runtime", "extra": "forbidden", } with patch("seat_execution_broker.sendmsg_with_fds") as send: with self.assertRaises(SystemExit): broker.serve_request(left, "seat-x", broker.TEST_GENERATION, 61123, [], request, peer_uid=61123, expected_phase="runtime", environment=self.ENVIRONMENT) send.assert_not_called() finally: left.close() right.close() def test_request_reader_accepts_fragmented_single_frame(self) -> None: left, right = socket.socketpair() try: payload = (b'{"seatId":"seat-x","generation":"' + broker.TEST_GENERATION.encode() + b'","phase":"runtime"}\n') left.sendall(payload[:17]) left.sendall(payload[17:]) self.assertEqual(broker.read_request(right), { "seatId": "seat-x", "generation": broker.TEST_GENERATION, "phase": "runtime", }) finally: left.close() right.close() def test_two_phase_server_times_out_without_client(self) -> None: with tempfile.TemporaryDirectory() as tmp: path = os.path.join(tmp, "broker.sock") listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) listener.bind(path) listener.listen(1) started = time.monotonic() try: with self.assertRaises(TimeoutError): broker.serve_two_phase(listener, "seat-x", broker.TEST_GENERATION, os.getuid(), -1, -1, -1, lambda: {}, self.ENVIRONMENT, deadline_seconds=0.02) finally: listener.close() self.assertLess(time.monotonic() - started, 0.5) def test_runtime_phase_receives_a_fresh_handshake_deadline(self) -> None: class Connection: def __init__(self) -> None: self.timeouts = [] def settimeout(self, value) -> None: self.timeouts.append(value) def close(self) -> None: pass class Listener: def __init__(self) -> None: self.timeouts = [] self.connections = [Connection(), Connection()] def settimeout(self, value) -> None: self.timeouts.append(value) def accept(self): return self.connections.pop(0), None listener = Listener() clock = iter([0.0, 0.0, 29.0, 40.0, 40.0, 40.0]) requests = iter([ {"seatId": "seat-x", "generation": broker.TEST_GENERATION, "phase": "admission"}, {"seatId": "seat-x", "generation": broker.TEST_GENERATION, "phase": "runtime"}, ]) manifest = {"seatId": "seat-x", "generation": broker.TEST_GENERATION} with patch("seat_execution_broker.time.monotonic", side_effect=lambda: next(clock)), patch( "seat_execution_broker.read_request", side_effect=lambda _connection: next(requests) ), patch("seat_execution_broker.serve_request"): broker.serve_two_phase(listener, "seat-x", broker.TEST_GENERATION, os.getuid(), -1, -1, -1, lambda: manifest, self.ENVIRONMENT, deadline_seconds=30.0) self.assertEqual(listener.timeouts, [30.0, 30.0]) def test_disconnected_partial_request_fails_without_descriptor_send(self) -> None: left, right = socket.socketpair() try: left.sendall(b'{"seatId":"seat-x"') left.close() with patch("seat_execution_broker.sendmsg_with_fds") as send, self.assertRaises(SystemExit): broker.read_request(right) send.assert_not_called() finally: right.close() def test_real_tmux_two_phase_descriptor_execution_exits_23_without_leaks(self) -> None: with tempfile.TemporaryDirectory() as tmp: broker_path = os.path.join(tmp, "broker.sock") tmux_path = os.path.join(tmp, "tmux.sock") client_path = os.path.join(tmp, "overdeck-seat-execution-client") admission_path = os.path.join(tmp, "admission") runtime_path = os.path.join(tmp, "runtime") with open(client_path, "w", encoding="utf-8") as fh: fh.write(f"#!/bin/sh\nexec {sys.executable} {_LIB}/seat_execution_client.py \"$@\"\n") with open(admission_path, "w", encoding="utf-8") as fh: fh.write("#!/bin/sh\nexec \"$@\"\n") with open(runtime_path, "w", encoding="utf-8") as fh: fh.write("#!/bin/sh\nexit 23\n") for path in (client_path, admission_path, runtime_path): os.chmod(path, 0o755) cwd_fd = os.open(tmp, os.O_RDONLY | os.O_DIRECTORY) admission_fd = os.open(admission_path, os.O_RDONLY | os.O_NOFOLLOW) runtime_fd = os.open(runtime_path, os.O_RDONLY | os.O_NOFOLLOW) listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) listener.bind(broker_path) listener.listen(1) error = [] worker = threading.Thread(target=lambda: self._run_broker( error, listener, cwd_fd, admission_fd, runtime_fd), daemon=False) worker.start() command = [ "tmux", "-S", tmux_path, "new-session", "-d", "-s", "main", "--", client_path, "--socket", broker_path, "--seat-id", "seat-x", "--generation", broker.TEST_GENERATION, "--phase", "admission", "--", client_path, "--socket", broker_path, "--seat-id", "seat-x", "--generation", broker.TEST_GENERATION, "--phase", "runtime", "--", ";", "set-option", "-t", "main", "remain-on-exit", "on", ";", "set-hook", "-t", "main", "pane-died", "wait-for -S broker-proof", ] try: subprocess.run(command, check=True, pass_fds=(cwd_fd, admission_fd, runtime_fd)) subprocess.run(["tmux", "-S", tmux_path, "wait-for", "broker-proof"], check=True, timeout=5) pane = subprocess.run(["tmux", "-S", tmux_path, "list-panes", "-t", "main", "-F", "#{pane_dead} #{pane_dead_status}"], check=True, text=True, capture_output=True) self.assertEqual(pane.stdout.strip(), "1 23") worker.join(5) self.assertFalse(worker.is_alive()) self.assertEqual(error, []) finally: subprocess.run(["tmux", "-S", tmux_path, "kill-server"], check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) listener.close() for fd in (cwd_fd, admission_fd, runtime_fd): os.close(fd) try: os.unlink(broker_path) except FileNotFoundError: pass try: os.unlink(tmux_path) except FileNotFoundError: pass self.assertFalse(os.path.exists(broker_path)) self.assertFalse(os.path.exists(tmux_path)) @staticmethod def _run_broker(error, listener, cwd_fd, admission_fd, runtime_fd) -> None: try: broker.serve_two_phase(listener, "seat-x", broker.TEST_GENERATION, os.getuid(), cwd_fd, admission_fd, runtime_fd, lambda: {"seatId": "seat-x", "generation": broker.TEST_GENERATION}, SeatExecutionBrokerTests.ENVIRONMENT, deadline_seconds=5) except BaseException as exc: error.append(exc) finally: listener.close() def test_fd_count_mismatch_is_rejected(self) -> None: left, right = socket.socketpair() fd = os.open("/dev/null", os.O_RDONLY) try: broker.send_capabilities(left, "seat-x", broker.TEST_GENERATION, "runtime", [fd], self.ENVIRONMENT) with self.assertRaises(SystemExit): broker.receive_capabilities(right, expected_count=2) finally: os.close(fd) left.close() right.close() if __name__ == "__main__": unittest.main()