"""Pure validation and process inspection for overdeck-seat-scope-entry.""" from __future__ import annotations import argparse import ctypes import ctypes.util import fcntl import grp import hashlib import json import os import pwd import re import shutil import stat import subprocess import sys import tempfile import threading import time from dataclasses import dataclass from seat_common import ( CONTROL_BASE_MODE, CONTROL_RE, SEAT_ID_RE, SEAT_LAUNCHER_BIN, TMUX_CONF_MODE, die as _common_die, read_startup_status_marker, remove_startup_status, seat_control_base, seat_test_mode, ) SLICE = "agent-seat.slice" MEMORY_MAX = "12884901888" MEMORY_SWAP_MAX = "1073741824" CPU_MAX = "400000 100000" PIDS_MAX = "256" CPU_WEIGHT = "1" TMUX_MIN_VERSION = (3, 7) SCOPE_PROPS = [ ("OOMPolicy", "kill"), ("MemoryMax", "12G"), ("MemorySwapMax", "1G"), ("CPUQuota", "400%"), ("CPUWeight", "1"), ("TasksMax", "256"), ] SCOPE_SHOW = { "OOMPolicy": "kill", "MemoryMax": "12884901888", "MemorySwapMax": "1073741824", "CPUQuotaPerSecUSec": "4s", "CPUWeight": "1", "TasksMax": "256", "Slice": SLICE, } LAUNCHER_WAIT_SEC = 5.0 PANE_WAIT_SEC = 5.0 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}$") SESSION_MANIFEST_BASE_FIELDS = frozenset({"schema", "seatId", "generation", "implementerUid", "socket", "session"}) SESSION_MANIFEST_IDENTITY_FIELDS = frozenset({"socketDev", "socketIno", "serverPid", "serverStartTime", "cgroup"}) SESSION_MANIFEST_MAX_BYTES = 4096 SHELL_NAMES = frozenset({"tmux", "bash", "sh", "dash", "zsh", "sleep"}) CLONE_NEWNET = 0x40000000 PR_CAPBSET_DROP = 24 PR_SET_NO_NEW_PRIVS = 38 PRIV_CAP_FIELDS = ("CapBnd", "CapEff", "CapPrm", "CapInh", "CapAmb") def _atomic_json_write(path: str, payload: dict[str, object]) -> None: parent = os.path.dirname(path) os.makedirs(parent, mode=0o700, exist_ok=True) fd, tmp = tempfile.mkstemp(prefix=".seat-session-", dir=parent) try: with os.fdopen(fd, "w", encoding="utf-8") as fh: json.dump(payload, fh, sort_keys=True, separators=(",", ":")) fh.write("\n") fh.flush() os.fsync(fh.fileno()) os.chmod(tmp, 0o600) os.replace(tmp, path) dir_fd = os.open(parent, os.O_RDONLY | os.O_DIRECTORY) try: os.fsync(dir_fd) finally: os.close(dir_fd) finally: if os.path.exists(tmp): os.unlink(tmp) def _validate_session_manifest(manifest: object) -> dict[str, object]: fields = set(manifest) if isinstance(manifest, dict) else set() if fields not in (SESSION_MANIFEST_BASE_FIELDS, SESSION_MANIFEST_BASE_FIELDS | SESSION_MANIFEST_IDENTITY_FIELDS): die("session-manifest-fields", 1) if manifest.get("schema") != 1: die("session-manifest-schema", 1) seat_id = manifest.get("seatId") generation = manifest.get("generation") uid = manifest.get("implementerUid") socket = manifest.get("socket") if not isinstance(seat_id, str) or not SEAT_ID_RE.match(seat_id): die("session-manifest-seat", 1) if not isinstance(generation, str) or not GENERATION_RE.match(generation): die("session-manifest-generation", 1) if not isinstance(uid, int) or isinstance(uid, bool) or uid <= 0: die("session-manifest-uid", 1) if not isinstance(socket, str) or not socket.startswith("/") or "//" in socket: die("session-manifest-socket", 1) if manifest.get("session") != "main": die("session-manifest-session", 1) if SESSION_MANIFEST_IDENTITY_FIELDS <= fields: for name in ("socketDev", "socketIno", "serverPid", "serverStartTime"): value = manifest.get(name) if not isinstance(value, int) or isinstance(value, bool) or value <= 0: die(f"session-manifest-{name}", 1) cgroup = manifest.get("cgroup") if not isinstance(cgroup, str) or not cgroup.startswith("/") or ".." in cgroup.split("/"): die("session-manifest-cgroup", 1) return manifest def session_manifest_path(seat_id: str) -> str: return os.path.join(seat_control_base(), seat_id, "session.json") def exit_receipt_path(seat_id: str) -> str: return os.path.join(seat_control_base(), seat_id, "exit.json") def _seat_generation_lock(manifest_path: str): parent = os.path.dirname(manifest_path) os.makedirs(parent, mode=0o700, exist_ok=True) fd = os.open(os.path.join(parent, ".generation.lock"), os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600) fcntl.flock(fd, fcntl.LOCK_EX) return fd def write_session_manifest(path: str, manifest: dict[str, object]) -> None: lock_fd = _seat_generation_lock(path) try: current_receipt = os.path.join(os.path.dirname(path), "exit.json") try: os.unlink(current_receipt) except FileNotFoundError: pass _atomic_json_write(path, _validate_session_manifest(manifest)) finally: os.close(lock_fd) def load_current_session_manifest(path: str, seat_id: str) -> dict[str, object]: manifest = load_session_manifest(path, seat_id, "") return manifest def load_session_manifest(path: str, seat_id: str, generation: str) -> dict[str, object]: parent, leaf = os.path.split(path) parent_fd = -1 fd = -1 try: parent_fd = os.open(parent, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) fd = os.open(leaf, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=parent_fd) before = os.fstat(fd) if not stat.S_ISREG(before.st_mode) or before.st_nlink != 1: die("session-manifest-type", 1) if stat.S_IMODE(before.st_mode) != 0o600: die("session-manifest-mode", 1) if not seat_test_mode() and before.st_uid != 0: die("session-manifest-owner", 1) if before.st_size <= 0 or before.st_size > SESSION_MANIFEST_MAX_BYTES: die("session-manifest-size", 1) raw = os.read(fd, SESSION_MANIFEST_MAX_BYTES + 1) after = os.fstat(fd) if (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns, before.st_ctime_ns) != ( after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns, after.st_ctime_ns ): die("session-manifest-changed", 1) manifest = _validate_session_manifest(json.loads(raw.decode("utf-8"))) except (OSError, UnicodeDecodeError, json.JSONDecodeError): die("session-manifest-unreadable", 1) finally: if fd >= 0: os.close(fd) if parent_fd >= 0: os.close(parent_fd) if manifest["seatId"] != seat_id or (generation and manifest["generation"] != generation): die("session-generation-mismatch", 1) return manifest def parse_proc_start_time(text: str) -> int: close = text.rfind(")") if close < 0: die("tmux-server-start-invalid", 1) fields = text[close + 1:].split() if len(fields) < 20 or not fields[19].isdigit() or int(fields[19]) <= 0: die("tmux-server-start-invalid", 1) return int(fields[19]) def read_process_identity(proc_root: str, pid: int) -> tuple[int, str]: try: stat_text = open(os.path.join(proc_root, str(pid), "stat"), encoding="utf-8").read() cgroup_lines = open(os.path.join(proc_root, str(pid), "cgroup"), encoding="utf-8").read().splitlines() except OSError: die("tmux-server-identity-unavailable", 1) start_time = parse_proc_start_time(stat_text) matches = [line.split(":", 2)[2] for line in cgroup_lines if line.startswith("0::")] if len(matches) != 1 or not matches[0].startswith("/"): die("tmux-server-cgroup-invalid", 1) return start_time, matches[0] def finalize_session_manifest( path: str, manifest: dict[str, object], socket: str, server_pid: int, proc_root: str = "/proc" ) -> dict[str, object]: try: socket_stat = os.stat(socket, follow_symlinks=False) except OSError: die("tmux-socket-unavailable", 1) if not stat.S_ISSOCK(socket_stat.st_mode): die("tmux-socket-type", 1) start_time, cgroup = read_process_identity(proc_root, server_pid) finalized = dict(manifest) finalized.update({ "socketDev": socket_stat.st_dev, "socketIno": socket_stat.st_ino, "serverPid": server_pid, "serverStartTime": start_time, "cgroup": cgroup, }) write_session_manifest(path, finalized) return finalized def publish_exit_receipt( manifest_path: str, receipt_path: str, seat_id: str, generation: str, exit_status: int | None = None, *, terminal_reason: str | None = None, ) -> None: if (exit_status is None) == (terminal_reason is None): die("terminal-result-invalid", 1) if exit_status is not None and ( not isinstance(exit_status, int) or isinstance(exit_status, bool) or not 0 <= exit_status <= 255 ): die("exit-status-invalid", 1) if terminal_reason is not None and terminal_reason != "operator-stop": die("terminal-reason-invalid", 1) lock_fd = _seat_generation_lock(manifest_path) try: load_session_manifest(manifest_path, seat_id, generation) payload = { "schema": 1, "seatId": seat_id, "generation": generation, } if exit_status is not None: payload["exitStatus"] = exit_status else: payload["terminalReason"] = terminal_reason generation_path = os.path.join(os.path.dirname(receipt_path), f"exit-{generation}.json") _atomic_json_write(generation_path, payload) load_session_manifest(manifest_path, seat_id, generation) _atomic_json_write(receipt_path, payload) finally: os.close(lock_fd) def query_tmux_server_pid(paths: ScopeEntryPaths, implementer_user: str, socket: str) -> int: proc = subprocess.run( [paths.sudo, "-n", "-u", implementer_user, paths.tmux, "-S", socket, "display-message", "-p", "#{pid}"], text=True, capture_output=True, ) value = proc.stdout.strip() if proc.returncode != 0 or not value.isdigit() or int(value) <= 0: die("tmux-server-pid-unavailable", 1) return int(value) @dataclass class LocalPathIdentity: path: str root: str executable: bool fds: list[int] identities: list[tuple[int, int, int, int, int, int, int, int, int]] def revalidate(self) -> None: for fd, expected in zip(self.fds, self.identities): held = stat_identity(os.fstat(fd)) if held != expected: die("local-path-descriptor-changed", 1) current = capture_local_path_identity(self.path, (self.root,), executable=self.executable, label="revalidate") try: for actual, expected in zip(current.identities, self.identities): if actual != expected: die("local-path-replaced", 1) finally: current.close() def grant_acl(self, username: str, permission: str, *, index: int = -1) -> None: self.revalidate() fd = self.fds[index] proc = subprocess.run( ["setfacl", "-m", f"u:{username}:{permission}", f"/proc/self/fd/{fd}"], capture_output=True, text=True, pass_fds=(fd,), ) if proc.returncode != 0: detail = (proc.stderr or proc.stdout or "").strip() die(f"setfacl-failed:{self.path}:{detail}", 1) self.refresh() def refresh(self) -> None: self.revalidate() self.identities = [stat_identity(os.fstat(fd)) for fd in self.fds] def capability_path(self, *, index: int = -1) -> str: return f"/proc/{os.getpid()}/fd/{self.fds[index]}" def close(self) -> None: for fd in reversed(self.fds): os.close(fd) self.fds.clear() def stat_identity(value: os.stat_result) -> tuple[int, int, int, int, int, int, int, int, int]: return (value.st_dev, value.st_ino, value.st_mode, value.st_uid, value.st_gid, value.st_size, value.st_mtime_ns, value.st_ctime_ns, value.st_nlink) def capture_local_path_identity(path: str, roots: tuple[str, ...], *, executable: bool, label: str) -> LocalPathIdentity: normalized = os.path.normpath(path) root = next((candidate for candidate in roots if normalized == candidate or normalized.startswith(candidate.rstrip("/") + "/")), None) if root is None or not os.path.isabs(normalized) or normalized != path: die(f"local-{label}-outside-root", 1) relative = os.path.relpath(normalized, root) fds: list[int] = [] try: current = os.open(root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) fds.append(current) parts = [] if relative == "." else relative.split(os.sep) for index, part in enumerate(parts): flags = os.O_RDONLY | os.O_NOFOLLOW if index < len(parts) - 1 or not executable: flags |= os.O_DIRECTORY current = os.open(part, flags, dir_fd=current) fds.append(current) identities = [stat_identity(os.fstat(fd)) for fd in fds] leaf = identities[-1] if executable and (not stat.S_ISREG(leaf[2]) or not leaf[2] & 0o111 or leaf[8] != 1): die(f"local-{label}-not-executable", 1) if not executable and not stat.S_ISDIR(leaf[2]): die(f"local-{label}-not-directory", 1) return LocalPathIdentity(normalized, root, executable, fds, identities) except OSError: for fd in reversed(fds): os.close(fd) die(f"local-{label}-invalid", 1) def grant_local_runtime_access( runtime_argv: list[str], cwd: str, implementer_user: str, operator_home: str, identities: dict[str, LocalPathIdentity], ) -> None: from seat_implementer_identity import ( grant_executable_acl, grant_link_ancestors_acl, grant_traverse_acl, require_setfacl, ) executables = [runtime_argv[0]] if len(runtime_argv) > 1 and os.path.isabs(runtime_argv[1]): executables.append(runtime_argv[1]) require_setfacl() identities[cwd].grant_acl(implementer_user, "x") for executable in executables: identity = identities[executable] for index in range(len(identity.fds) - 1): identity.grant_acl(implementer_user, "x", index=index) identity.grant_acl(implementer_user, "rx") def expected_local_socket(operator_home: str, seat_id: str) -> str: if not SEAT_ID_RE.match(seat_id): die("invalid seat id", 2) routing_key = hashlib.sha256(seat_id.encode("utf-8")).hexdigest()[:32] return os.path.join(operator_home, ".local", "state", "agent-sessions", "sock", routing_key, "tmux.sock") def verify_finalized_live_identity(manifest: dict[str, object], socket: str, proc_root: str) -> None: socket_stat = os.stat(socket, follow_symlinks=False) if (not stat.S_ISSOCK(socket_stat.st_mode) or socket_stat.st_dev != manifest.get("socketDev") or socket_stat.st_ino != manifest.get("socketIno")): die("session-socket-identity-mismatch", 1) server_pid = manifest.get("serverPid") if not isinstance(server_pid, int) or server_pid <= 0: die("session-server-identity-invalid", 1) start_time, cgroup = read_process_identity(proc_root, server_pid) if start_time != manifest.get("serverStartTime") or cgroup != manifest.get("cgroup"): die("session-server-identity-mismatch", 1) def publish_local_authority_routing( operator_home: str, seat_id: str, generation: str, operator_uid: int, operator_gid: int, ) -> None: sessions = os.path.join(operator_home, ".local", "state", "agent-sessions", "sessions") path = os.path.join(sessions, f"{seat_id}.json") lock = f"{path}.lock" lock_fd = -1 for _attempt in range(200): try: lock_fd = os.open(lock, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) os.fchown(lock_fd, operator_uid, operator_gid) break except FileExistsError: time.sleep(0.01) if lock_fd < 0: die("local-ledger-locked", 1) tmp = os.path.join(sessions, f".{seat_id}.{os.getpid()}.tmp") try: fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW) try: before = os.fstat(fd) if (not stat.S_ISREG(before.st_mode) or before.st_uid != operator_uid or before.st_gid != operator_gid or stat.S_IMODE(before.st_mode) != 0o600 or before.st_nlink != 1 or before.st_size < 1 or before.st_size > 1 << 20): die("local-ledger-invalid", 1) raw = os.read(fd, (1 << 20) + 1) after = os.fstat(fd) identity = lambda value: (value.st_dev, value.st_ino, value.st_size, value.st_mtime_ns, value.st_ctime_ns) if len(raw) != before.st_size or identity(before) != identity(after): die("local-ledger-changed", 1) finally: os.close(fd) try: entry = json.loads(raw) except (json.JSONDecodeError, UnicodeDecodeError): die("local-ledger-invalid", 1) if entry.get("schemaVersion") != 1 or entry.get("ledgerId") != seat_id: die("local-ledger-invalid", 1) entry["sessionAuthority"] = {"kind": "seat", "seatId": seat_id, "generation": generation} payload = (json.dumps(entry, indent=2) + "\n").encode("utf-8") tmp_fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) try: os.fchown(tmp_fd, operator_uid, operator_gid) os.write(tmp_fd, payload) os.fsync(tmp_fd) finally: os.close(tmp_fd) os.replace(tmp, path) dir_fd = os.open(sessions, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) try: os.fsync(dir_fd) finally: os.close(dir_fd) finally: if os.path.exists(tmp): os.unlink(tmp) os.close(lock_fd) os.unlink(lock) class LocalExecutionBroker: def __init__(self, socket_path: str, listener, target) -> None: self.socket_path = socket_path self.listener = listener self.target = target self.error: BaseException | None = None self.thread = threading.Thread(target=self._run, name="seat-execution-broker", daemon=False) def _run(self) -> None: try: self.target() except BaseException as exc: self.error = exc finally: self.listener.close() try: os.unlink(self.socket_path) except FileNotFoundError: pass def start(self) -> None: self.thread.start() def wait(self, timeout: float) -> None: self.thread.join(timeout) if self.thread.is_alive(): raise TimeoutError("broker-join-timeout") if self.error is not None: raise self.error def close(self) -> None: if self.thread.is_alive(): import socket as socket_module wake = socket_module.socket(socket_module.AF_UNIX, socket_module.SOCK_STREAM) try: wake.connect(self.socket_path) except OSError: pass finally: wake.close() self.listener.close() try: os.unlink(self.socket_path) except FileNotFoundError: pass self.thread.join(1.0) if self.thread.is_alive(): raise TimeoutError("broker-close-timeout") def is_alive(self) -> bool: return self.thread.is_alive() def close_local_path_identities(identities: dict[str, LocalPathIdentity]) -> None: closed: set[int] = set() for identity in identities.values(): marker = id(identity) if marker not in closed and identity.fds: identity.close() closed.add(marker) def local_execution_environment(seat_id: str, generation: str, implementer_user: str) -> dict[str, str]: from seat_execution_broker import EXECUTION_ENV_KEYS, validate_execution_environment environment = { "HOME": local_implementer_home(seat_id), "USER": implementer_user, "LOGNAME": implementer_user, "PATH": "/usr/local/bin:/usr/bin:/bin", "OVERDECK_SEAT_ID": seat_id, "OVERDECK_SEAT_GENERATION": generation, } inherited = EXECUTION_ENV_KEYS - set(environment) for key in inherited: value = os.environ.get(key) if value: environment[key] = value return validate_execution_environment(environment) def start_local_execution_broker(seat_id: str, generation: str, implementer_uid: int, implementer_gid: int, implementer_user: str, identities: dict[str, LocalPathIdentity], runtime_argv: list[str], manifest_path: str) -> LocalExecutionBroker: import socket as socket_module from seat_execution_broker import serve_two_phase broker_socket = os.path.join(os.path.dirname(manifest_path), f"broker-{generation}.sock") listener = socket_module.socket(socket_module.AF_UNIX, socket_module.SOCK_STREAM) listener.bind(broker_socket) os.chown(broker_socket, 0, implementer_gid) os.chmod(broker_socket, 0o660) listener.listen(1) cwd_fd = identities[next(path for path, identity in identities.items() if not identity.executable)].fds[-1] admission_fd = identities[runtime_argv[0]].fds[-1] runtime_fd = identities[runtime_argv[1]].fds[-1] def load_manifest() -> dict[str, object]: return load_session_manifest(manifest_path, seat_id, generation) environment = local_execution_environment(seat_id, generation, implementer_user) def owner_target() -> None: serve_two_phase(listener, seat_id, generation, implementer_uid, cwd_fd, admission_fd, runtime_fd, load_manifest, environment) owner = LocalExecutionBroker(broker_socket, listener, owner_target) owner.start() return owner def run_local_session( *, paths: ScopeEntryPaths, seat_id: str, socket: str, runtime_argv: list[str], cwd: str, attach: bool, reap_identity: str, operator_user: str, operator_uid: int, operator_gid: int, operator_home: str, ) -> int: if len(runtime_argv) < 2 or not all(os.path.isabs(path) for path in runtime_argv[:2]): die("local-runtime-invalid", 2) if socket != expected_local_socket(operator_home, seat_id): die("local-socket-mismatch", 1) if not GENERATION_RE.match(reap_identity): die("local-reap-identity-invalid", 2) identities = { cwd: capture_local_path_identity(cwd, (operator_home,), executable=False, label="cwd"), runtime_argv[0]: capture_local_path_identity( runtime_argv[0], (operator_home, "/usr/bin", "/bin"), executable=True, label="admission" ), } if len(runtime_argv) > 1 and os.path.isabs(runtime_argv[1]): identities[runtime_argv[1]] = capture_local_path_identity( runtime_argv[1], (operator_home, "/usr/bin", "/bin"), executable=True, label="runtime" ) implementer_user, implementer_uid, implementer_gid = ensure_local_implementer_user( seat_id, operator_user ) if implementer_uid == operator_uid: die("implementer-operator-same-uid", 1) grant_local_runtime_access(runtime_argv, cwd, implementer_user, operator_home, identities) generation = str(__import__("uuid").uuid4()) prepare_socket(socket) socket_parent = os.path.dirname(socket) os.chown(socket_parent, implementer_uid, implementer_gid) os.chmod(socket_parent, 0o700) tmux_conf = write_tmux_conf(seat_id, implementer_uid, implementer_gid) manifest_path = session_manifest_path(seat_id) manifest = { "schema": 1, "seatId": seat_id, "generation": generation, "implementerUid": implementer_uid, "socket": socket, "session": "main", } write_session_manifest(manifest_path, manifest) _atomic_json_write(os.path.join(socket_parent, "session.json"), manifest) broker = start_local_execution_broker( seat_id, generation, implementer_uid, implementer_gid, implementer_user, identities, runtime_argv, manifest_path ) try: client = "/usr/local/bin/overdeck-seat-execution-client" runtime_args = runtime_argv[2:] pane_argv = [ client, "--socket", broker.socket_path, "--seat-id", seat_id, "--generation", generation, "--phase", "admission", "--", client, "--socket", broker.socket_path, "--seat-id", seat_id, "--generation", generation, "--phase", "runtime", "--", *runtime_args, ] command = [ paths.sudo, "-n", "-u", operator_user, "env", f"XDG_RUNTIME_DIR=/run/user/{operator_uid}", f"HOME={operator_home}", paths.systemd_run, "--user", "--quiet", "--collect", "--slice=agent.slice", "--service-type=exec", "--", paths.sudo, "-n", "/usr/local/bin/overdeck-seat-implementer-exec", "--local-session", "--seat-id", seat_id, "--socket", socket, "--generation", generation, "--cwd", "/", "--", "/usr/bin/tmux", "-f", tmux_conf, "-S", socket, "set-hook", "-g", "pane-died", f"wait-for -S agent-exit-{generation}", ";", "new-session", "-d", "-s", "main", "--", *pane_argv, ";", "set-option", "-t", "main", "@agent_reap_identity", reap_identity, ] started = subprocess.run(command, cwd=cwd, check=False) if started.returncode != 0: die("local-tmux-start-failed", 1) broker.wait(35.0) close_local_path_identities(identities) server_pid = query_tmux_server_pid(paths, implementer_user, socket) manifest = finalize_session_manifest(manifest_path, manifest, socket, server_pid, paths.proc_root) publish_local_authority_routing(operator_home, seat_id, generation, operator_uid, operator_gid) _atomic_json_write(os.path.join(socket_parent, "session.json"), manifest) if attach: subprocess.run([ paths.tmux_mediator, "--seat-id", seat_id, "--socket", socket, "--generation", generation, "attach-session", "-t", "main", ], check=False) waited = subprocess.run([ paths.sudo, "-n", "-u", implementer_user, paths.tmux, "-S", socket, "wait-for", f"agent-exit-{generation}", ], check=False) if waited.returncode != 0: die("local-lifecycle-wait-failed", 1) verify_finalized_live_identity(manifest, socket, paths.proc_root) pane = subprocess.run([ paths.sudo, "-n", "-u", implementer_user, paths.tmux, "-S", socket, "list-panes", "-t", "main", "-F", "#{pane_dead} #{pane_dead_status}", ], text=True, capture_output=True) fields = pane.stdout.strip().split() if pane.returncode != 0 or len(fields) != 2 or fields[0] != "1" or not fields[1].isdigit(): die("local-exit-status-unavailable", 1) status = int(fields[1]) publish_exit_receipt(manifest_path, exit_receipt_path(seat_id), seat_id, generation, status) return status finally: broker.close() close_local_path_identities(identities) class ScopeEntryPaths: """Injectable executable and proc paths for test harnesses only.""" def __init__( self, proc_root: str = "/proc", cgroup_root: str = "/sys/fs/cgroup", tmux: str = "tmux", systemd_run: str = "systemd-run", systemctl: str = "systemctl", implementer_exec: str = "/usr/local/bin/overdeck-seat-implementer-exec", tmux_mediator: str = "/usr/local/bin/overdeck-seat-tmux-mediator", sudo: str = "sudo", dry_sleep_sec: float = 30.0, ) -> None: self.proc_root = proc_root self.cgroup_root = cgroup_root self.tmux = tmux self.systemd_run = systemd_run self.systemctl = systemctl self.implementer_exec = implementer_exec self.tmux_mediator = tmux_mediator self.sudo = sudo self.dry_sleep_sec = dry_sleep_sec DEFAULT_PATHS = ScopeEntryPaths() _INSTALL_ROOT = os.environ.get("OVERDECK_SEAT_INSTALL_CHECK_ROOT", "") if _INSTALL_ROOT: _INSTALL_ROOT = os.path.realpath(_INSTALL_ROOT) if (os.geteuid() != 0 and not seat_test_mode()) or not os.path.isabs(_INSTALL_ROOT): raise SystemExit(1) def _installed(name: str) -> str: return os.path.join(_INSTALL_ROOT, name) if _INSTALL_ROOT else os.path.join("/usr/local/lib/overdeck", name) INSTALLED_WRAPPER = _installed("overdeck-seat-scope-entry") INSTALLED_MODULE = _installed("seat_scope_entry.py") WRAPPER_EMBEDDED_MODULE = "seat_scope_entry.py" WRAPPER_EMBEDDED_IMPLEMENTER_MODULE = "seat_implementer_exec.py" WRAPPER_EMBEDDED_TMUX_MEDIATOR_MODULE = "seat_tmux_mediator.py" WRAPPER_EMBEDDED_EXECUTION_CLIENT_MODULE = "seat_execution_client.py" SUDOERS_AUTHORIZED_WRAPPER = "/usr/local/bin/overdeck-seat-scope-entry" SUDOERS_AUTHORIZED_MEDIATOR = "/usr/local/bin/overdeck-seat-tmux-mediator" SUDOERS_AUTHORIZED_IMPLEMENTER_EXEC = "/usr/local/bin/overdeck-seat-implementer-exec" INSTALLED_MANIFEST = _installed("seat-scope-entry.manifest.json") INSTALLED_IMPLEMENTER_EXEC = _installed("overdeck-seat-implementer-exec") INSTALLED_IMPLEMENTER_MODULE = _installed("seat_implementer_exec.py") INSTALLED_TMUX_MEDIATOR = _installed("overdeck-seat-tmux-mediator") INSTALLED_TMUX_MEDIATOR_MODULE = _installed("seat_tmux_mediator.py") INSTALLED_EXECUTION_BROKER_MODULE = _installed("seat_execution_broker.py") INSTALLED_EXECUTION_CLIENT = _installed("overdeck-seat-execution-client") INSTALLED_EXECUTION_CLIENT_MODULE = _installed("seat_execution_client.py") INSTALLED_REAP_CLOSE = _installed("agent-session-reap-close.py") INSTALLED_IDENTITY_MODULE = _installed("seat_implementer_identity.py") INSTALLED_COMMON_MODULE = _installed("seat_common.py") INSTALLED_LAUNCHER = _installed("seat-launcher") SUDOERS_FILE = _installed("sudoers") IMPLEMENTER_EXEC_MODE = 0o755 def die(msg: str, code: int = 2) -> None: _common_die(f"overdeck-seat-scope-entry: {msg}", code) def file_sha256(path: str) -> str: digest = hashlib.sha256() with open(path, "rb") as fh: for chunk in iter(lambda: fh.read(65536), b""): digest.update(chunk) return digest.hexdigest() def verify_installed_dir(path: str, *, mode: int) -> None: try: st = os.lstat(path) except OSError: die(f"install-check-missing:{path}", 1) 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) expected_uid = os.getuid() if seat_test_mode() else 0 expected_gid = os.getgid() if seat_test_mode() else 0 if st.st_uid != expected_uid or st.st_gid != expected_gid: die(f"install-check-owner:{path}", 1) def verify_installed_file(path: str, *, mode: int, must_exec: bool) -> None: try: st = os.lstat(path) except OSError: die(f"install-check-missing:{path}", 1) 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) actual_mode = stat.S_IMODE(st.st_mode) if actual_mode != mode: die(f"install-check-mode:{path}", 1) if st.st_uid != 0 or st.st_gid != 0: die(f"install-check-owner:{path}", 1) if st.st_mode & (stat.S_ISUID | stat.S_ISGID): die(f"install-check-setid:{path}", 1) if must_exec and not os.access(path, os.X_OK): die(f"install-check-not-executable:{path}", 1) def verify_wrapper_embedded_module(wrapper_path: str, embedded_module: str) -> None: wrapper_text = read_installed_file(wrapper_path, mode=0o755).decode("utf-8") if embedded_module not in wrapper_text: die("install-check-wrapper-module-path", 1) def read_installed_file(path: str, *, mode: int, limit: int = 1 << 20, must_exec: bool = False) -> bytes: fd = -1 try: fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW) before = os.fstat(fd) test_mode = os.environ.get("OVERDECK_SEAT_TEST_MODE") == "1" expected_uid = os.getuid() if test_mode else 0 expected_gid = os.getgid() if test_mode else 0 if (not stat.S_ISREG(before.st_mode) or before.st_uid != expected_uid or before.st_gid != expected_gid or stat.S_IMODE(before.st_mode) != mode or before.st_nlink != 1 or (must_exec and not before.st_mode & 0o111) or before.st_size < 1 or before.st_size > limit): die(f"install-check-file:{path}", 1) data = os.read(fd, limit + 1) after = os.fstat(fd) identity = lambda st: (st.st_dev, st.st_ino, st.st_size, st.st_mtime_ns, st.st_ctime_ns) if len(data) != before.st_size or identity(before) != identity(after): die(f"install-check-file-changed:{path}", 1) return data except OSError: die(f"install-check-missing:{path}", 1) finally: if fd >= 0: os.close(fd) def load_install_manifest() -> dict[str, str]: try: data = json.loads(read_installed_file(INSTALLED_MANIFEST, mode=0o600, limit=4096)) except (json.JSONDecodeError, UnicodeDecodeError, TypeError, ValueError): die("install-check-manifest-invalid", 1) if not isinstance(data, dict): die("install-check-manifest-invalid", 1) required = ( "wrapper_sha256", "module_sha256", "common_module_sha256", "implementer_exec_sha256", "implementer_module_sha256", "tmux_mediator_sha256", "tmux_mediator_module_sha256", "execution_broker_module_sha256", "execution_client_module_sha256", "execution_client_sha256", "reap_close_sha256", "identity_module_sha256", "launcher_sha256", "ssh_user", ) out: dict[str, str] = {} for key in required: val = str(data.get(key, "")).strip() if not val: die("install-check-manifest-invalid", 1) out[key] = val if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*", out["ssh_user"]): die("install-check-manifest-invalid", 1) for key in required: if key == "ssh_user": continue if not re.fullmatch(r"[0-9a-f]{64}", out[key]): die("install-check-manifest-invalid", 1) return out def run_install_check() -> None: if os.geteuid() != 0 and not seat_test_mode(): die("install-check-requires-root", 1) manifest = load_install_manifest() artifact_contract = { INSTALLED_WRAPPER: (0o755, True, "wrapper_sha256"), INSTALLED_MODULE: (0o644, False, "module_sha256"), INSTALLED_COMMON_MODULE: (0o644, False, "common_module_sha256"), INSTALLED_IMPLEMENTER_EXEC: (IMPLEMENTER_EXEC_MODE, True, "implementer_exec_sha256"), INSTALLED_IMPLEMENTER_MODULE: (0o644, False, "implementer_module_sha256"), INSTALLED_TMUX_MEDIATOR: (0o755, True, "tmux_mediator_sha256"), INSTALLED_TMUX_MEDIATOR_MODULE: (0o644, False, "tmux_mediator_module_sha256"), INSTALLED_EXECUTION_BROKER_MODULE: (0o644, False, "execution_broker_module_sha256"), INSTALLED_EXECUTION_CLIENT: (0o755, True, "execution_client_sha256"), INSTALLED_EXECUTION_CLIENT_MODULE: (0o644, False, "execution_client_module_sha256"), INSTALLED_REAP_CLOSE: (0o755, True, "reap_close_sha256"), INSTALLED_IDENTITY_MODULE: (0o644, False, "identity_module_sha256"), INSTALLED_LAUNCHER: (0o755, True, "launcher_sha256"), SUDOERS_FILE: (0o440, False, None), } artifacts = { path: read_installed_file(path, mode=mode, must_exec=must_exec, limit=4096 if path == SUDOERS_FILE else 1 << 20) for path, (mode, must_exec, _key) in artifact_contract.items() } wrapper_checks = ( (INSTALLED_WRAPPER, WRAPPER_EMBEDDED_MODULE), (INSTALLED_IMPLEMENTER_EXEC, WRAPPER_EMBEDDED_IMPLEMENTER_MODULE), (INSTALLED_TMUX_MEDIATOR, WRAPPER_EMBEDDED_TMUX_MEDIATOR_MODULE), (INSTALLED_EXECUTION_CLIENT, WRAPPER_EMBEDDED_EXECUTION_CLIENT_MODULE), ) for wrapper_path, embedded_module in wrapper_checks: try: wrapper_text = artifacts[wrapper_path].decode("utf-8") except UnicodeDecodeError: die("install-check-wrapper-module-path", 1) if embedded_module not in wrapper_text or 'f"{_OVERDECK_LIB}' in wrapper_text: die("install-check-wrapper-module-path", 1) for path, (_mode, _must_exec, key) in artifact_contract.items(): if key is not None and hashlib.sha256(artifacts[path]).hexdigest() != manifest[key]: die(f"install-check-hash:{key}", 1) if shutil.which("setfacl") is None: die("install-check-setfacl-missing", 1) control_base = seat_control_base() if seat_test_mode(): control_base = os.environ.get("OVERDECK_SEAT_TEST_CONTROL_BASE", control_base) if not os.path.isabs(control_base): die("install-check-control-base", 1) verify_installed_dir(control_base, mode=CONTROL_BASE_MODE) ssh_user = manifest["ssh_user"] want_rules = [ f"{ssh_user} ALL=(root) NOPASSWD: {SUDOERS_AUTHORIZED_WRAPPER} *", f"{ssh_user} ALL=(root) NOPASSWD: {SUDOERS_AUTHORIZED_MEDIATOR} *", f"{ssh_user} ALL=(root) NOPASSWD: {SUDOERS_AUTHORIZED_IMPLEMENTER_EXEC} *", ] try: installed_rules = [line.strip() for line in artifacts[SUDOERS_FILE].decode("utf-8").splitlines() if line.strip()] except UnicodeDecodeError: die("install-check-sudoers-missing", 1) if installed_rules != want_rules: die("install-check-sudoers-drift", 1) print("scope-entry-install-check-ok") def libc() -> ctypes.CDLL: name = ctypes.util.find_library("c") if not name: die("libc-missing", 1) return ctypes.CDLL(name, use_errno=True) def setns(fd: int, nstype: int) -> None: lib = libc() if lib.setns(fd, nstype) != 0: err = ctypes.get_errno() die(f"setns-failed:{os.strerror(err)}", 1) def prctl( option: int, arg2: int = 0, arg3: int = 0, arg4: int = 0, arg5: int = 0, *, prctl_fn=None, ) -> int: lib = libc() call = prctl_fn or lib.prctl rc = call(option, arg2, arg3, arg4, arg5) if rc < 0: err = ctypes.get_errno() die(f"prctl-failed:{option}:{os.strerror(err)}", 1) return rc def read_cap_last_cap(*, cap_last_cap_path: str = "/proc/sys/kernel/cap_last_cap") -> int: try: with open(cap_last_cap_path, encoding="utf-8") as fh: return int(fh.read().strip()) except (OSError, ValueError): die("cap-last-cap-read-failed", 1) def drop_bounding_caps( *, prctl_fn=None, cap_last_cap: int | None = None, cap_last_cap_path: str = "/proc/sys/kernel/cap_last_cap", ) -> None: last = cap_last_cap if cap_last_cap is not None else read_cap_last_cap(cap_last_cap_path=cap_last_cap_path) for cap in range(last + 1): prctl(PR_CAPBSET_DROP, cap, 0, 0, 0, prctl_fn=prctl_fn) def set_no_new_privs(*, prctl_fn=None) -> None: prctl(PR_SET_NO_NEW_PRIVS, 1, 0, 0, 0, prctl_fn=prctl_fn) def parse_proc_status(text: str) -> dict[str, str]: fields: dict[str, str] = {} for line in text.splitlines(): if ":" not in line: continue key, val = line.split(":", 1) fields[key.strip()] = val.strip() return fields def verify_zero_privileges(*, status_path: str = "/proc/self/status") -> None: try: with open(status_path, encoding="utf-8") as fh: fields = parse_proc_status(fh.read()) except OSError as exc: die(f"proc-status-read-failed:{exc.strerror}", 1) for key in PRIV_CAP_FIELDS: raw = fields.get(key) if raw is None: die(f"proc-status-missing:{key}", 1) try: if int(raw, 16) != 0: die(f"proc-status-cap-nonzero:{key}:{raw}", 1) except ValueError: die(f"proc-status-cap-invalid:{key}:{raw}", 1) no_new = fields.get("NoNewPrivs") if no_new != "1": die(f"proc-status-no-new-privs:{no_new or 'missing'}", 1) def resolve_target() -> tuple[str, int, int, str]: uid_s = os.environ.get("SUDO_UID", "").strip() user = os.environ.get("SUDO_USER", "").strip() if uid_s and user: try: uid = int(uid_s) except ValueError: die("invalid sudo uid") try: pw = pwd.getpwuid(uid) except KeyError: die("target user missing") if pw.pw_name != user: die("sudo user mismatch") return user, uid, pw.pw_gid, pw.pw_dir fallback = os.environ.get("OVERDECK_SEAT_SSH_USER", "").strip() if not fallback: die("target user unknown", 1) try: pw = pwd.getpwnam(fallback) except KeyError: die("target home missing", 1) return pw.pw_name, pw.pw_uid, pw.pw_gid, pw.pw_dir def validate_text(name: str, value: str) -> None: if not value or CONTROL_RE.search(value): die(f"invalid {name}") def reject_path_aliases(path: str, label: str) -> None: if not path.startswith("/"): die(f"invalid {label}") if path.endswith("/") or "//" in path: die(f"invalid {label}") for part in path.split("/"): if part in (".", ".."): die(f"invalid {label}") def reject_symlink_ancestors(path: str, home: str, label: str) -> None: if path == home: return if not path.startswith(home + "/"): die(f"invalid {label}") current = home for part in path[len(home) + 1 :].split("/"): if not part: continue current = os.path.join(current, part) if os.path.islink(current): die(f"{label}-symlink-ancestor:{current}", 1) def validate_exact_operator_path(path: str, want: str, *, label: str) -> None: reject_path_aliases(path, label) if path != want: die(f"{label} mismatch") def validate_static(ns: argparse.Namespace) -> None: if not SEAT_ID_RE.match(ns.seat_id): die("invalid seat id") if ns.slice != SLICE: die("slice mismatch") validate_text("seat-host", ns.seat_host) validate_text("seat-model", ns.seat_model) for name in ("claude_bin", "guard_bin", "checkout"): path = getattr(ns, name) reject_path_aliases(path, name.replace("_", "-")) if CONTROL_RE.search(path): die(f"invalid {name}") def validate_args(ns: argparse.Namespace, home: str, *, skip_host_checks: bool = False) -> None: if not SEAT_ID_RE.match(ns.seat_id): die("invalid seat id") if not SEAT_ID_RE.match(ns.account_slug): die("invalid account slug") want_netns = f"/var/run/netns/overdeck-seat-{ns.seat_id}" if ns.netns_path != want_netns: die("netns path mismatch") want_unit = f"agent-seat-{ns.seat_id}.scope" if ns.unit != want_unit: die("unit mismatch") want_state = f"{home}/.local/state/overdeck/seats/{ns.seat_id}" want_socket = f"{want_state}/tmux.sock" if ns.socket != want_socket: die("socket mismatch") validate_exact_operator_path(ns.launcher, SEAT_LAUNCHER_BIN, label="launcher") want_checkout = f"{home}/seats/{ns.seat_id}/repo" validate_exact_operator_path(ns.checkout, want_checkout, label="checkout") want_claude = f"{home}/.local/bin/claude" validate_exact_operator_path(ns.claude_bin, want_claude, label="claude-bin") want_guard = f"{home}/.local/share/overdeck/seat-guard/current/bin" validate_exact_operator_path(ns.guard_bin, want_guard, label="guard-bin") if ns.slice != SLICE: die("slice mismatch") validate_text("seat-host", ns.seat_host) validate_text("seat-model", ns.seat_model) validate_text("account-slug", ns.account_slug) if skip_host_checks: return for label, path in ( ("checkout", ns.checkout), ("socket-parent", want_state), ): reject_symlink_ancestors(path, home, label) for path in (ns.netns_path, ns.checkout, want_state): if os.path.islink(path): die(f"{path}:symlink-forbidden", 1) if not os.path.exists(ns.netns_path): die("netns missing") st = os.stat(ns.netns_path, follow_symlinks=False) if st.st_uid != 0: die("netns owner mismatch") if not os.access(ns.launcher, os.X_OK): die("launcher missing") if not os.path.isdir(ns.checkout): die("checkout missing") def proc_path(paths: ScopeEntryPaths, pid: int, *parts: str) -> str: return os.path.join(paths.proc_root, str(pid), *parts) def ns_inode(path: str) -> str: st = os.stat(path) return f"{st.st_dev}:{st.st_ino}" def proc_field(paths: ScopeEntryPaths, pid: int, *parts: str) -> str: with open(proc_path(paths, pid, *parts), "rb") as fh: return fh.read().decode("utf-8", "replace") def proc_environ(paths: ScopeEntryPaths, pid: int) -> dict[str, str]: raw = proc_field(paths, pid, "environ") out: dict[str, str] = {} for item in raw.split("\0"): if not item or "=" not in item: continue key, val = item.split("=", 1) out[key] = val return out def proc_cmdline(paths: ScopeEntryPaths, pid: int) -> list[str]: return [part for part in proc_field(paths, pid, "cmdline").split("\0") if part] def proc_exe(paths: ScopeEntryPaths, pid: int) -> str: try: return os.readlink(proc_path(paths, pid, "exe")) except OSError: return "" def child_pids(paths: ScopeEntryPaths, pid: int) -> list[int]: try: raw = proc_field(paths, pid, "task", str(pid), "children") except OSError: return [] return [int(item) for item in raw.split() if item.isdigit()] def join_netns(netns_path: str) -> None: fd = os.open(netns_path, os.O_RDONLY) try: setns(fd, CLONE_NEWNET) finally: os.close(fd) def become_implementer(uid: int, gid: int, username: str) -> None: become_user(uid, gid, username, primary_group_only=True) def operator_systemd_env(uid: int, home: str, username: str) -> dict[str, str]: runtime = f"/run/user/{uid}" env = { "HOME": home, "USER": username, "LOGNAME": username, "XDG_RUNTIME_DIR": runtime, "DBUS_SESSION_BUS_ADDRESS": f"unix:path={runtime}/bus", } for key in ("PATH", "LANG", "LC_ALL", "SHELL"): if key in os.environ: env[key] = os.environ[key] if "PATH" not in env: env["PATH"] = "/usr/local/bin:/usr/bin:/bin" return env def become_user(uid: int, gid: int, username: str, *, primary_group_only: bool = False) -> None: if primary_group_only: from seat_implementer_identity import validate_implementer_group_membership validate_implementer_group_membership(username, gid) groups = [gid] else: groups = [g.gr_gid for g in grp.getgrall() if username in g.gr_mem] groups.append(gid) os.setgroups(sorted(set(groups))) os.setresgid(gid, gid, gid) drop_bounding_caps() os.setresuid(uid, uid, uid) set_no_new_privs() verify_zero_privileges() LOCAL_IMPLEMENTER_ID = "local-hosted" def local_implementer_home(_ledger_id: str) -> str: return implementer_home_for_seat(LOCAL_IMPLEMENTER_ID) def ensure_local_implementer_user(_ledger_id: str, operator_user: str) -> tuple[str, int, int]: return ensure_implementer_user(LOCAL_IMPLEMENTER_ID, operator_user) def implementer_home_for_seat(seat_id: str) -> str: from seat_implementer_identity import implementer_home_for_seat as _home return _home(seat_id) def resolve_implementer_for_seat(seat_id: str, operator_user: str) -> tuple[str, int, int]: from seat_implementer_identity import resolve_implementer_for_seat as _resolve return _resolve(seat_id, operator_user) def ensure_implementer_user(seat_id: str, operator_user: str) -> tuple[str, int, int]: from seat_implementer_identity import ensure_implementer_user as _ensure return _ensure(seat_id, operator_user) TMUX_CONF = """set -g exit-empty off set -g remain-on-exit on """ def parse_tmux_version(text: str) -> tuple[int, int, str]: match = re.search(r"tmux\s+(\d+)\.(\d+)([a-z]*)", text.strip(), re.IGNORECASE) if not match: die("tmux-version-unparseable", 1) return int(match.group(1)), int(match.group(2)), match.group(3).lower() def verify_tmux_support(paths: ScopeEntryPaths) -> None: proc = subprocess.run([paths.tmux, "-V"], capture_output=True, text=True) if proc.returncode != 0: die("tmux-version-failed", 1) major, minor, suffix = parse_tmux_version(proc.stdout or proc.stderr) if (major, minor) < TMUX_MIN_VERSION: die(f"tmux-version-too-old:{major}.{minor}{suffix}", 1) with tempfile.TemporaryDirectory(prefix="overdeck-tmux-probe-") as tmp: socket = os.path.join(tmp, "probe.sock") conf = os.path.join(tmp, "probe.conf") with open(conf, "w", encoding="utf-8") as fh: fh.write(TMUX_CONF) start = subprocess.run( [ paths.tmux, "-f", conf, "-S", socket, "new-session", "-d", "-s", "overdeck-probe", "sleep", "30", ], capture_output=True, text=True, ) if start.returncode != 0: detail = (start.stderr or start.stdout or "").strip() die(f"tmux-probe-start-failed:{detail}", 1) try: opt = subprocess.run( [paths.tmux, "-f", conf, "-S", socket, "show-options", "-s", "exit-empty"], capture_output=True, text=True, ) if opt.returncode != 0: detail = (opt.stderr or opt.stdout or "").strip() die(f"tmux-exit-empty-probe-failed:{detail}", 1) if (opt.stdout or "").strip() != "exit-empty off": die("tmux-exit-empty-option-missing", 1) remain = subprocess.run( [paths.tmux, "-f", conf, "-S", socket, "show-options", "-g", "remain-on-exit"], capture_output=True, text=True, ) if remain.returncode != 0 or (remain.stdout or "").strip() != "remain-on-exit on": die("tmux-remain-on-exit-option-missing", 1) finally: subprocess.run( [paths.tmux, "-f", conf, "-S", socket, "kill-server"], capture_output=True, check=False, ) if os.path.exists(socket): try: os.unlink(socket) except OSError: pass def write_tmux_conf(seat_id: str, implementer_uid: int, implementer_gid: int) -> str: from seat_implementer_identity import atomic_root_control_write, tmux_conf_path path = tmux_conf_path(seat_id) if os.path.isfile(path): st = os.lstat(path) if stat.S_ISLNK(st.st_mode): die("tmux-conf-symlink", 1) if not seat_test_mode() and (st.st_uid != 0 or st.st_gid != 0): die("tmux-conf-owner", 1) if not seat_test_mode() and stat.S_IMODE(st.st_mode) != TMUX_CONF_MODE: die("tmux-conf-mode", 1) with open(path, encoding="utf-8") as fh: if fh.read() == TMUX_CONF: os.chmod(path, TMUX_CONF_MODE) return path atomic_root_control_write(seat_id, "tmux-seat.conf", TMUX_CONF, mode=TMUX_CONF_MODE) st = os.lstat(path) if stat.S_ISLNK(st.st_mode) or (not seat_test_mode() and st.st_uid != 0): die("tmux-conf-finalize", 1) return path def scope_argv(paths: ScopeEntryPaths, unit: str, slice_name: str, command: list[str]) -> list[str]: argv = [ paths.systemd_run, "--user", "--scope", "--collect", f"--unit={unit}", f"--slice={slice_name}", ] for key, val in SCOPE_PROPS: argv.extend(["-p", f"{key}={val}"]) argv.append("--") argv.extend(command) return argv def run_checked( argv: list[str], env: dict[str, str], *, quiet: bool = False, ) -> subprocess.CompletedProcess[str]: proc = subprocess.run(argv, env=env, text=True, capture_output=True) if proc.returncode != 0: detail = (proc.stderr or proc.stdout or "").strip() die(f"command-failed:{argv[0]}:{detail}", 1) if not quiet and proc.stdout: sys.stdout.write(proc.stdout) return proc def run_as_operator( paths: ScopeEntryPaths, operator_user: str, operator_uid: int, operator_home: str, argv: list[str], ) -> subprocess.CompletedProcess[str]: env = operator_systemd_env(operator_uid, operator_home, operator_user) sudo_argv = [paths.sudo, "-n", "-u", operator_user, "-g", operator_user, "env"] flat_env = [f"{key}={value}" for key, value in sorted(env.items())] return run_checked(sudo_argv + flat_env + argv, os.environ.copy()) def build_implementer_invocation( paths: ScopeEntryPaths, *, seat_id: str, seat_host: str, seat_model: str, socket: str, launcher: str, claude_bin: str, guard_bin: str, checkout: str, state_dir: str, generation: str, implementer_home_path: str, operator_home: str, account_slug: str, profile_dir: str, secure_storage_dir: str, netns_path: str, tmux_cmd: list[str], ) -> list[str]: return [ paths.sudo, "-n", paths.implementer_exec, "--seat-id", seat_id, "--seat-host", seat_host, "--seat-model", seat_model, "--socket", socket, "--launcher", launcher, "--claude-bin", claude_bin, "--guard-bin", guard_bin, "--checkout", checkout, "--state-dir", state_dir, "--generation", generation, "--implementer-home", implementer_home_path, "--operator-home", operator_home, "--account-slug", account_slug, "--profile-dir", profile_dir, "--secure-storage-dir", secure_storage_dir, "--netns-path", netns_path, "--", *tmux_cmd, ] def verify_implementer_isolation( paths: ScopeEntryPaths, implementer_user: str, implementer_uid: int, implementer_gid: int, operator_home: str, operator_uid: int, ) -> None: from seat_implementer_identity import ( verify_implementer_credential_boundary, verify_implementer_runtime_bus_absent, verify_implementer_systemd_absent, ) verify_implementer_credential_boundary(implementer_user, operator_home, sudo=paths.sudo) verify_implementer_systemd_absent(implementer_user, sudo=paths.sudo) verify_implementer_runtime_bus_absent( implementer_user, implementer_uid, implementer_gid, operator_uid, sudo=paths.sudo, ) def scope_properties_ok( paths: ScopeEntryPaths, unit: str, operator_user: str, operator_uid: int, operator_home: str, ) -> None: props = ",".join(SCOPE_SHOW.keys()) proc = run_as_operator( paths, operator_user, operator_uid, operator_home, [paths.systemctl, "--user", "show", unit, f"--property={props}"], ) got: dict[str, str] = {} for line in proc.stdout.splitlines(): if "=" in line: key, val = line.split("=", 1) got[key] = val for key, want in SCOPE_SHOW.items(): actual = got.get(key, "") if actual != want: die(f"scope-prop-mismatch:{key}:{actual}", 1) def verify_pid_netns(paths: ScopeEntryPaths, pid: int, want_inode: str) -> None: inode = ns_inode(proc_path(paths, pid, "ns", "net")) if inode != want_inode: die("netns not retained", 1) def cgroup_v2_path(paths: ScopeEntryPaths, pid: int) -> tuple[str, str]: cg = proc_field(paths, pid, "cgroup") for line in cg.splitlines(): if line.startswith("0::"): rel = line[3:].strip() return rel, os.path.join(paths.cgroup_root, rel.lstrip("/")) die("cgroup-v2-missing", 1) def read_cgroup_file(cg_path: str, name: str) -> str: try: with open(os.path.join(cg_path, name), encoding="utf-8") as fh: return fh.read().strip() except OSError as exc: die(f"cgroup-prop-missing:{name}:{exc.strerror}", 1) def verify_pid_effective_boundary(paths: ScopeEntryPaths, pid: int, unit: str) -> None: rel, cg_path = cgroup_v2_path(paths, pid) scope_token = unit.replace(".scope", "") if "tmux-spawn" in rel and scope_token not in rel: die("tmux-spawn-outside-seat-boundary", 1) if "agent-seat.slice" not in rel: die("cgroup-slice-ancestry", 1) if scope_token not in rel: die("cgroup-seat-scope-missing", 1) mem = read_cgroup_file(cg_path, "memory.max") if mem != MEMORY_MAX: die(f"effective-memory-max:{mem}", 1) swap = read_cgroup_file(cg_path, "memory.swap.max") if swap != MEMORY_SWAP_MAX: die(f"effective-memory-swap-max:{swap}", 1) cpu = read_cgroup_file(cg_path, "cpu.max") if cpu != CPU_MAX: die(f"effective-cpu-max:{cpu}", 1) weight = read_cgroup_file(cg_path, "cpu.weight") if weight != CPU_WEIGHT: die(f"effective-cpu-weight:{weight}", 1) pids = read_cgroup_file(cg_path, "pids.max") if pids != PIDS_MAX: die(f"effective-pids-max:{pids}", 1) def verify_pid_uid(paths: ScopeEntryPaths, pid: int, want_uid: int) -> None: fields = parse_proc_status(proc_field(paths, pid, "status")) uid_raw = fields.get("Uid", "").split() if not uid_raw: die("proc-status-uid-missing", 1) try: uid = int(uid_raw[0]) except ValueError: die("proc-status-uid-invalid", 1) if uid != want_uid: die(f"implementer-uid-mismatch:{uid}", 1) def verify_pid_implementer_env(paths: ScopeEntryPaths, pid: int) -> None: env = proc_environ(paths, pid) for key in ("DBUS_SESSION_BUS_ADDRESS", "XDG_RUNTIME_DIR"): if env.get(key): die(f"implementer-env-forbidden:{key}", 1) def verify_all_descendants_bounded(paths: ScopeEntryPaths, root_pid: int, unit: str, implementer_uid: int) -> None: for pid in collect_descendants(paths, root_pid): verify_pid_effective_boundary(paths, pid, unit) verify_pid_uid(paths, pid, implementer_uid) verify_pid_implementer_env(paths, pid) def verify_launcher_env(pid_paths: ScopeEntryPaths, pid: int, seat_id: str, seat_host: str, seat_model: str) -> None: env = proc_environ(pid_paths, pid) checks = { "OVERDECK_SEAT_ID": seat_id, "OVERDECK_SEAT_HOST": seat_host, "OVERDECK_SEAT_MODEL": seat_model, } for key, want in checks.items(): if env.get(key) != want: die(f"launcher-env-mismatch:{key}", 1) def basename(path: str) -> str: return os.path.basename(path.rstrip("/")) def is_shell_or_tmux(paths: ScopeEntryPaths, pid: int) -> bool: exe = proc_exe(paths, pid) if exe and basename(exe) in SHELL_NAMES: return True cmd = proc_cmdline(paths, pid) if cmd and basename(cmd[0]) in SHELL_NAMES: return True return False def is_claude_process(paths: ScopeEntryPaths, pid: int, launcher: str) -> bool: exe = proc_exe(paths, pid) if exe and basename(exe) == "claude": return True cmd = proc_cmdline(paths, pid) if not cmd: return False if basename(cmd[0]) == "claude": return True if cmd[0] == launcher and "--model" in cmd: return True return False def collect_descendants(paths: ScopeEntryPaths, root_pid: int) -> list[int]: seen = {root_pid} queue = [root_pid] order = [root_pid] while queue: pid = queue.pop(0) for child in child_pids(paths, pid): if child in seen: continue seen.add(child) order.append(child) queue.append(child) return order def build_remote_tmux_command(tmux: str, tmux_conf: str, socket: str, launcher: str, generation: str) -> list[str]: return [ tmux, "-f", tmux_conf, "-S", socket, "set-hook", "-g", "pane-died", f"wait-for -S agent-exit-{generation}", ";", "new-session", "-d", "-s", "main", "--", launcher, ] def wait_tmux_pane( paths: ScopeEntryPaths, seat_id: str, socket: str, state_dir: str, generation: str, timeout: float = PANE_WAIT_SEC, ) -> int: deadline = time.time() + timeout while time.time() < deadline: pane = subprocess.run( [ paths.tmux_mediator, "--seat-id", seat_id, "--socket", socket, "--generation", generation, "list-panes", "-F", "#{pane_pid}", "-t", "main", ], text=True, capture_output=True, ) if pane.returncode == 0 and pane.stdout.strip().isdigit(): return int(pane.stdout.strip()) time.sleep(0.05) marker = read_startup_status_marker(state_dir) die(f"tmux-session-missing:{marker}", 1) def wait_launcher_target( paths: ScopeEntryPaths, pane_pid: int, launcher: str, timeout: float = LAUNCHER_WAIT_SEC, ) -> int: deadline = time.time() + timeout while time.time() < deadline: claude_hits: list[int] = [] for pid in collect_descendants(paths, pane_pid): if is_shell_or_tmux(paths, pid): continue if is_claude_process(paths, pid, launcher): claude_hits.append(pid) if claude_hits: return claude_hits[0] time.sleep(0.05) die("launcher-process-missing", 1) def verify_claude_process(paths: ScopeEntryPaths, pid: int, launcher: str) -> None: if is_shell_or_tmux(paths, pid): die("launcher-shell-rejected", 1) if not is_claude_process(paths, pid, launcher): die("launcher-process-mismatch", 1) def prepare_socket(socket: str) -> None: parent = os.path.dirname(socket) os.makedirs(parent, mode=0o700, exist_ok=True) if os.path.lexists(socket): os.unlink(socket) def stop_scope( paths: ScopeEntryPaths, unit: str, operator_user: str, operator_uid: int, operator_home: str, ) -> None: subprocess.run( [paths.sudo, "-n", "-u", operator_user, paths.systemctl, "--user", "stop", unit], env=operator_systemd_env(operator_uid, operator_home, operator_user), check=False, ) def kill_tmux(paths: ScopeEntryPaths, seat_id: str, socket: str, generation: str) -> None: subprocess.run( [paths.tmux_mediator, "--seat-id", seat_id, "--socket", socket, "--generation", generation, "kill-server"], check=False, ) if os.path.lexists(socket): try: os.unlink(socket) except OSError: pass def launch_remote_lifecycle_owner(paths: ScopeEntryPaths, seat_id: str, socket: str, generation: str, operator_user: str) -> None: unit = f"overdeck-seat-lifecycle-{seat_id}-{generation}.service" proc = subprocess.run([ paths.systemd_run, "--quiet", "--collect", f"--unit={unit}", f"--setenv=OVERDECK_SEAT_SSH_USER={operator_user}", "--service-type=exec", "--", paths.tmux_mediator, "--seat-id", seat_id, "--socket", socket, "--generation", generation, "lifecycle", ], check=False) if proc.returncode != 0: die("remote-lifecycle-owner-start-failed", 1) def run_scope( *, paths: ScopeEntryPaths, unit: str, slice_name: str, socket: str, launcher: str, claude_bin: str, guard_bin: str, checkout: str, state_dir: str, seat_id: str, seat_host: str, seat_model: str, generation: str, netns_inode: str, operator_user: str, operator_uid: int, operator_home: str, implementer_user: str, implementer_uid: int, implementer_gid: int, implementer_home_path: str, account_slug: str, profile_dir: str, secure_storage_dir: str, netns_path: str, dry_run: bool, ) -> None: verify_implementer_isolation( paths, implementer_user, implementer_uid, implementer_gid, operator_home, operator_uid, ) from seat_implementer_identity import verify_implementer_launch_authority_boundary verify_implementer_launch_authority_boundary( implementer_user, state_dir, sudo=paths.sudo ) verify_tmux_support(paths) tmux_conf = write_tmux_conf(seat_id, implementer_uid, implementer_gid) dry_socket = "" work_socket = socket if dry_run: dry_socket = os.path.join(os.path.dirname(socket), f".{seat_id}-scope-dry.sock") prepare_socket(dry_socket) work_socket = dry_socket tmux_cmd = [ paths.tmux, "-f", tmux_conf, "-S", work_socket, "new-session", "-d", "-s", "main", "--", "sleep", str(paths.dry_sleep_sec), ] else: prepare_socket(socket) tmux_cmd = build_remote_tmux_command( paths.tmux, tmux_conf, socket, launcher, generation ) manifest_path = session_manifest_path(seat_id) manifest = { "schema": 1, "seatId": seat_id, "generation": generation, "implementerUid": implementer_uid, "socket": work_socket, "session": "main", } write_session_manifest(manifest_path, manifest) _atomic_json_write(os.path.join(state_dir, "session.json"), manifest) inner = build_implementer_invocation( paths, seat_id=seat_id, seat_host=seat_host, seat_model=seat_model, socket=work_socket, launcher=launcher, claude_bin=claude_bin, guard_bin=guard_bin, checkout=checkout, state_dir=state_dir, generation=generation, implementer_home_path=implementer_home_path, operator_home=operator_home, account_slug=account_slug, profile_dir=profile_dir, secure_storage_dir=secure_storage_dir, netns_path=netns_path, tmux_cmd=tmux_cmd, ) scope_cmd = scope_argv(paths, unit, slice_name, inner) try: run_as_operator(paths, operator_user, operator_uid, operator_home, scope_cmd) run_as_operator( paths, operator_user, operator_uid, operator_home, [paths.systemctl, "--user", "is-active", "--quiet", unit], ) pane_pid = wait_tmux_pane(paths, seat_id, work_socket, state_dir, generation) verify_pid_netns(paths, pane_pid, netns_inode) verify_pid_effective_boundary(paths, pane_pid, unit) verify_pid_uid(paths, pane_pid, implementer_uid) verify_pid_implementer_env(paths, pane_pid) verify_all_descendants_bounded(paths, pane_pid, unit, implementer_uid) server_pid = query_tmux_server_pid(paths, implementer_user, work_socket) manifest = finalize_session_manifest(manifest_path, manifest, work_socket, server_pid, paths.proc_root) _atomic_json_write(os.path.join(state_dir, "session.json"), manifest) scope_properties_ok(paths, unit, operator_user, operator_uid, operator_home) if dry_run: verify_launcher_env(paths, pane_pid, seat_id, seat_host, seat_model) print("scope-entry-dry-run-ok") return target_pid = wait_launcher_target(paths, pane_pid, launcher) verify_pid_netns(paths, target_pid, netns_inode) verify_pid_effective_boundary(paths, target_pid, unit) verify_pid_uid(paths, target_pid, implementer_uid) verify_pid_implementer_env(paths, target_pid) verify_all_descendants_bounded(paths, pane_pid, unit, implementer_uid) verify_launcher_env(paths, target_pid, seat_id, seat_host, seat_model) verify_claude_process(paths, target_pid, launcher) remove_startup_status(state_dir) if not stat.S_ISSOCK(os.stat(socket).st_mode): die("socket missing", 1) launch_remote_lifecycle_owner(paths, seat_id, socket, generation, operator_user) print(f"scope-entry-ok generation={generation}") finally: if dry_socket: kill_tmux(paths, seat_id, dry_socket, generation) stop_scope(paths, unit, operator_user, operator_uid, operator_home) def build_arg_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="overdeck-seat-scope-entry", add_help=True) parser.add_argument("--ensure-implementer-only", action="store_true") parser.add_argument("--local-session", action="store_true") parser.add_argument("--attach", action="store_true") parser.add_argument("--reap-identity") parser.add_argument("--cwd") parser.add_argument("--seat-id", required=False) parser.add_argument("--netns-path") parser.add_argument("--unit") parser.add_argument("--socket") parser.add_argument("--launcher") parser.add_argument("--slice") parser.add_argument("--seat-host") parser.add_argument("--seat-model") parser.add_argument("--claude-bin") parser.add_argument("--guard-bin") parser.add_argument("--checkout") parser.add_argument("--account-slug") parser.add_argument("--dry-run", action="store_true") parser.add_argument("command", nargs=argparse.REMAINDER) return parser def run_ensure_implementer_only(seat_id: str) -> None: if not SEAT_ID_RE.match(seat_id): die("invalid seat id") if os.geteuid() != 0: die("requires root", 1) operator_user, _operator_uid, _operator_gid, _operator_home = resolve_target() username, uid, gid = ensure_implementer_user(seat_id, operator_user) print(f"implementer-user={username}") print(f"implementer-uid={uid}") print(f"implementer-gid={gid}") def main(argv: list[str] | None = None, paths: ScopeEntryPaths = DEFAULT_PATHS) -> None: args = list(argv) if argv is not None else sys.argv[1:] if args and args[0] == "--install-check": if len(args) != 1: die("install-check-no-extra-args", 2) run_install_check() return parser = build_arg_parser() try: ns = parser.parse_args(argv) except SystemExit: die("unknown arg", 2) if ns.local_session: if not ns.seat_id or not ns.socket or not ns.cwd or not ns.reap_identity: die("local-session-required-args", 2) if ns.command[:1] != ["--"] or len(ns.command) < 2: die("local-session-command-required", 2) if os.geteuid() != 0: die("requires root", 1) operator_user, operator_uid, operator_gid, operator_home = resolve_target() status = run_local_session( paths=paths, seat_id=ns.seat_id, socket=ns.socket, runtime_argv=ns.command[1:], cwd=ns.cwd, attach=ns.attach, reap_identity=ns.reap_identity, operator_user=operator_user, operator_uid=operator_uid, operator_gid=operator_gid, operator_home=operator_home, ) raise SystemExit(status) if ns.ensure_implementer_only: if not ns.seat_id: die("seat-id-required", 2) for name in ( "netns_path", "unit", "socket", "launcher", "slice", "seat_host", "seat_model", "claude_bin", "guard_bin", "checkout", "account_slug", "generation", ): if getattr(ns, name, None): die("ensure-implementer-extra-args", 2) run_ensure_implementer_only(ns.seat_id) return if not ns.seat_id: die("seat-id-required", 2) required_scope = ( "netns_path", "unit", "socket", "launcher", "slice", "seat_host", "seat_model", "claude_bin", "guard_bin", "checkout", "account_slug", ) for name in required_scope: if not getattr(ns, name, None): die(f"missing-{name.replace('_', '-')}", 2) validate_static(ns) if os.geteuid() != 0: die("requires root", 1) operator_user, operator_uid, operator_gid, operator_home = resolve_target() implementer_user, implementer_uid, implementer_gid = ensure_implementer_user(ns.seat_id, operator_user) if operator_uid == implementer_uid: die("implementer-operator-same-uid", 1) validate_args(ns, operator_home) netns_inode = ns_inode(ns.netns_path) state_dir = os.path.dirname(ns.socket) implementer_home_path = implementer_home_for_seat(ns.seat_id) from seat_implementer_identity import ( configure_seat_git_author, grant_managed_runtime_access, verify_implementer_launch_authority_boundary, verify_seat_git_author, ) profile = f"{operator_home}/.claudex-accounts/{ns.account_slug}/{ns.seat_id}" secure_storage = f"{profile}/secure-storage" grant_managed_runtime_access( implementer_user, operator_user, operator_home, ns.seat_id, operator_uid=operator_uid, operator_gid=operator_gid, repo=ns.checkout, profile=profile, state_dir=state_dir, guard_bin=ns.guard_bin, claude_bin=ns.claude_bin, launcher=ns.launcher, ) configure_seat_git_author(ns.checkout, ns.seat_id) verify_seat_git_author(ns.checkout, ns.seat_id) os.makedirs(implementer_home_path, mode=0o700, exist_ok=True) os.chown(implementer_home_path, implementer_uid, implementer_gid) join_netns(ns.netns_path) run_scope( paths=paths, unit=ns.unit, slice_name=ns.slice, socket=ns.socket, launcher=ns.launcher, claude_bin=ns.claude_bin, guard_bin=ns.guard_bin, checkout=ns.checkout, state_dir=state_dir, seat_id=ns.seat_id, seat_host=ns.seat_host, seat_model=ns.seat_model, generation=str(__import__("uuid").uuid4()), netns_inode=netns_inode, operator_user=operator_user, operator_uid=operator_uid, operator_home=operator_home, implementer_user=implementer_user, implementer_uid=implementer_uid, implementer_gid=implementer_gid, implementer_home_path=implementer_home_path, account_slug=ns.account_slug, profile_dir=profile, secure_storage_dir=secure_storage, netns_path=ns.netns_path, dry_run=ns.dry_run, ) if __name__ == "__main__": main()