"""Root-owned descriptor broker for local hosted session execution.""" from __future__ import annotations import array import json import os import re import socket import struct import time from seat_common import SEAT_ID_RE, die as _common_die TEST_GENERATION = "11111111-1111-4111-8111-111111111111" GENERATION_RE = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$") MAX_MESSAGE = 4096 EXECUTION_ENV_KEYS = frozenset({ "HOME", "USER", "LOGNAME", "PATH", "LANG", "LC_ALL", "TERM", "ANTHROPIC_BASE_URL", "CLAUDE_CONFIG_DIR", "CLAUDE_SECURESTORAGE_CONFIG_DIR", "SYSTRAY_CLAUDE_ACCOUNT_HOME", "SYSTRAY_CLAUDE_ACCOUNT_SLUG", "OVERDECK_SEAT_HOST", "OVERDECK_SEAT_ID", "OVERDECK_SEAT_MODEL", "OVERDECK_CLAUDE_BIN", "OVERDECK_SEAT_GUARD_BIN", "OVERDECK_SEAT_CHECKOUT", "OVERDECK_SEAT_STATE_DIR", "OVERDECK_SEAT_GENERATION", "CCP_CONFIG_DIR", "CCP_BIND_ADDRESS", "CCP_CODEX_MODEL", "CCP_CODEX_SERVICE_TIER", }) def die(message: str, code: int = 1) -> None: _common_die(f"overdeck-seat-execution-broker: {message}", code) def sendmsg_with_fds(sock: socket.socket, payload: bytes, fds: list[int]) -> None: rights = array.array("i", fds) sent = sock.sendmsg([payload], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, rights)]) if sent != len(payload): die("descriptor-send-short") def validate_execution_environment(environment: dict[str, str]) -> dict[str, str]: if not isinstance(environment, dict) or not set(environment).issubset(EXECUTION_ENV_KEYS): die("environment-invalid") out = {} for key, value in environment.items(): if not isinstance(value, str) or not value or "\x00" in value or len(value) > 4096: die("environment-invalid") out[key] = value return out def send_capabilities(sock: socket.socket, seat_id: str, generation: str, phase: str, fds: list[int], environment: dict[str, str]) -> None: if (not SEAT_ID_RE.match(seat_id) or not GENERATION_RE.match(generation) or phase not in ("admission", "runtime")): die("request-identity-invalid") payload = (json.dumps({"seatId": seat_id, "generation": generation, "phase": phase, "fdCount": len(fds), "environment": validate_execution_environment(environment)}, separators=(",", ":")) + "\n").encode() if len(payload) > MAX_MESSAGE: die("request-too-large") sendmsg_with_fds(sock, payload, fds) def receive_capabilities(sock: socket.socket, *, expected_count: int) -> tuple[dict[str, object], list[int]]: data, ancillary, flags, _address = sock.recvmsg(MAX_MESSAGE + 1, socket.CMSG_SPACE(expected_count * struct.calcsize("i"))) received: list[int] = [] try: if flags & (socket.MSG_TRUNC | socket.MSG_CTRUNC) or len(data) > MAX_MESSAGE: die("response-truncated") for level, kind, value in ancillary: if level == socket.SOL_SOCKET and kind == socket.SCM_RIGHTS: rights = array.array("i") rights.frombytes(value[:len(value) - (len(value) % rights.itemsize)]) received.extend(rights.tolist()) request = json.loads(data) if (not isinstance(request, dict) or request.get("fdCount") != expected_count or len(received) != expected_count): die("descriptor-count-mismatch") return request, received except (json.JSONDecodeError, UnicodeDecodeError): die("response-invalid") except BaseException: for fd in received: os.close(fd) raise def peer_uid(sock: socket.socket) -> int: raw = sock.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, struct.calcsize("3i")) _pid, uid, _gid = struct.unpack("3i", raw) return uid def read_request(sock: socket.socket) -> dict[str, object]: chunks = bytearray() while b"\n" not in chunks: data = sock.recv(MAX_MESSAGE + 1 - len(chunks)) if not data: die("request-invalid") chunks.extend(data) if len(chunks) > MAX_MESSAGE: die("request-invalid") if chunks[-1:] != b"\n" or b"\n" in chunks[:-1]: die("request-invalid") try: request = json.loads(chunks) except (json.JSONDecodeError, UnicodeDecodeError): die("request-invalid") if not isinstance(request, dict): die("request-invalid") return request def serve_two_phase(listener: socket.socket, seat_id: str, generation: str, implementer_uid: int, cwd_fd: int, admission_fd: int, runtime_fd: int, manifest_loader, environment: dict[str, str], *, deadline_seconds: float = 30.0) -> None: for phase, executable_fd in (("admission", admission_fd), ("runtime", runtime_fd)): deadline = time.monotonic() + deadline_seconds remaining = deadline - time.monotonic() if remaining <= 0: raise TimeoutError("broker-deadline") listener.settimeout(remaining) try: connection, _address = listener.accept() except socket.timeout: raise TimeoutError("broker-deadline") from None try: remaining = deadline - time.monotonic() if remaining <= 0: raise TimeoutError("broker-deadline") connection.settimeout(remaining) try: request = read_request(connection) except socket.timeout: raise TimeoutError("broker-deadline") from None manifest = manifest_loader() if (manifest.get("seatId") != seat_id or manifest.get("generation") != generation): die("manifest-identity-mismatch") if request.get("phase") != phase: die("request-phase-mismatch") serve_request(connection, seat_id, generation, implementer_uid, [cwd_fd, executable_fd], request, peer_uid=None, expected_phase=phase, environment=environment) finally: connection.close() def serve_request(sock: socket.socket, seat_id: str, generation: str, implementer_uid: int, fds: list[int], request: dict[str, object], *, peer_uid: int | None = None, expected_phase: str, environment: dict[str, str]) -> None: actual_uid = globals()["peer_uid"](sock) if peer_uid is None else peer_uid if actual_uid != implementer_uid: die("peer-uid-mismatch") expected = {"seatId": seat_id, "generation": generation, "phase": expected_phase} if request != expected: die("request-identity-mismatch") send_capabilities(sock, seat_id, generation, expected_phase, fds, environment) def fd_exec(cwd_fd: int, executable_fd: int, argv: list[str], env: dict[str, str]) -> None: os.fchdir(cwd_fd) os.execve(executable_fd, argv, env)