"""Pure validation and process inspection for overdeck-seat-scope-entry.""" from __future__ import annotations import argparse import ctypes import ctypes.util import grp import hashlib import json import os import pwd import re import shutil import stat import subprocess import sys import tempfile import time from seat_common import ( CONTROL_BASE_MODE, CONTROL_RE, SEAT_ID_RE, TMUX_CONF_MODE, die as _common_die, 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 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") 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() INSTALLED_WRAPPER = "/usr/local/bin/overdeck-seat-scope-entry" INSTALLED_MODULE = "/usr/local/lib/overdeck/seat_scope_entry.py" WRAPPER_EMBEDDED_MODULE = "/usr/local/lib/overdeck/seat_scope_entry.py" WRAPPER_EMBEDDED_IMPLEMENTER_MODULE = "/usr/local/lib/overdeck/seat_implementer_exec.py" WRAPPER_EMBEDDED_TMUX_MEDIATOR_MODULE = "/usr/local/lib/overdeck/seat_tmux_mediator.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 = "/usr/local/lib/overdeck/seat-scope-entry.manifest.json" INSTALLED_IMPLEMENTER_EXEC = "/usr/local/bin/overdeck-seat-implementer-exec" INSTALLED_IMPLEMENTER_MODULE = "/usr/local/lib/overdeck/seat_implementer_exec.py" INSTALLED_TMUX_MEDIATOR = "/usr/local/bin/overdeck-seat-tmux-mediator" INSTALLED_TMUX_MEDIATOR_MODULE = "/usr/local/lib/overdeck/seat_tmux_mediator.py" INSTALLED_IDENTITY_MODULE = "/usr/local/lib/overdeck/seat_implementer_identity.py" INSTALLED_COMMON_MODULE = "/usr/local/lib/overdeck/seat_common.py" SUDOERS_FILE = "/etc/sudoers.d/overdeck-seat-scope-entry" 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) if st.st_uid != 0 or st.st_gid != 0: 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: try: with open(wrapper_path, encoding="utf-8") as fh: wrapper_text = fh.read() except OSError: die(f"install-check-missing:{wrapper_path}", 1) if embedded_module not in wrapper_text: die("install-check-wrapper-module-path", 1) def load_install_manifest() -> dict[str, str]: verify_installed_file(INSTALLED_MANIFEST, mode=0o600, must_exec=False) try: with open(INSTALLED_MANIFEST, encoding="utf-8") as fh: data = json.load(fh) except (OSError, json.JSONDecodeError, 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", "identity_module_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: die("install-check-requires-root", 1) manifest = load_install_manifest() verify_installed_file(INSTALLED_MODULE, mode=0o644, must_exec=False) verify_installed_file(INSTALLED_COMMON_MODULE, mode=0o644, must_exec=False) verify_installed_file(INSTALLED_IMPLEMENTER_MODULE, mode=0o644, must_exec=False) verify_installed_file(INSTALLED_TMUX_MEDIATOR_MODULE, mode=0o644, must_exec=False) verify_installed_file(INSTALLED_IDENTITY_MODULE, mode=0o644, must_exec=False) verify_installed_file(SUDOERS_FILE, mode=0o440, must_exec=False) wrapper_checks = ( (INSTALLED_WRAPPER, WRAPPER_EMBEDDED_MODULE, 0o755), (INSTALLED_IMPLEMENTER_EXEC, WRAPPER_EMBEDDED_IMPLEMENTER_MODULE, IMPLEMENTER_EXEC_MODE), (INSTALLED_TMUX_MEDIATOR, WRAPPER_EMBEDDED_TMUX_MEDIATOR_MODULE, 0o755), ) for wrapper_path, embedded_module, wrapper_mode in wrapper_checks: verify_installed_file(wrapper_path, mode=wrapper_mode, must_exec=True) verify_wrapper_embedded_module(wrapper_path, embedded_module) hash_checks = { "wrapper_sha256": INSTALLED_WRAPPER, "module_sha256": INSTALLED_MODULE, "common_module_sha256": INSTALLED_COMMON_MODULE, "implementer_exec_sha256": INSTALLED_IMPLEMENTER_EXEC, "implementer_module_sha256": INSTALLED_IMPLEMENTER_MODULE, "tmux_mediator_sha256": INSTALLED_TMUX_MEDIATOR, "tmux_mediator_module_sha256": INSTALLED_TMUX_MEDIATOR_MODULE, "identity_module_sha256": INSTALLED_IDENTITY_MODULE, } for key, path in hash_checks.items(): if file_sha256(path) != manifest[key]: die(f"install-check-hash:{key}", 1) if shutil.which("setfacl") is None: die("install-check-setfacl-missing", 1) verify_installed_dir(seat_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: with open(SUDOERS_FILE, encoding="utf-8") as fh: installed_rules = [line.strip() for line in fh.read().splitlines() if line.strip()] except OSError: 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") want_launcher = f"{home}/.claude/bin/seat-launcher" validate_exact_operator_path(ns.launcher, want_launcher, 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), ("launcher", ns.launcher), ): reject_symlink_ancestors(path, home, label) for path in (ns.netns_path, ns.checkout, want_state, ns.launcher): 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() 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 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("set -g exit-empty on\n") 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 on": die("tmux-exit-empty-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, implementer_home_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, "--implementer-home", implementer_home_path, "--", *tmux_cmd, ] def verify_implementer_isolation( paths: ScopeEntryPaths, implementer_user: str, implementer_uid: 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, 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 wait_tmux_pane( paths: ScopeEntryPaths, seat_id: str, socket: 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, "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) die("tmux session missing", 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) -> None: subprocess.run( [paths.tmux_mediator, "--seat-id", seat_id, "--socket", socket, "kill-server"], check=False, ) if os.path.lexists(socket): try: os.unlink(socket) except OSError: pass 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, 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, dry_run: bool, ) -> None: verify_implementer_isolation(paths, implementer_user, implementer_uid, 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 = [ paths.tmux, "-f", tmux_conf, "-S", socket, "new-session", "-d", "-s", "main", "--", launcher, ] 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, implementer_home_path=implementer_home_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) 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) 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) if not stat.S_ISSOCK(os.stat(socket).st_mode): die("socket missing", 1) print("scope-entry-ok") finally: if dry_socket: kill_tmux(paths, seat_id, dry_socket) 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("--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") 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.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", ): 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}" 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, 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, dry_run=ns.dry_run, ) if __name__ == "__main__": main()