"""Validated drop-in exec for seat implementer processes (per-seat UID).""" from __future__ import annotations import argparse import os import sys from seat_common import CONTROL_RE, SEAT_ID_RE, SEAT_LAUNCHER_BIN, die as _common_die, seat_test_mode from seat_implementer_identity import ( implementer_home_for_seat, resolve_implementer_for_seat, tmux_conf_path, ) from seat_implementer_identity import record_netns_inode from seat_scope_entry import ( TMUX_CONF, TMUX_CONF_MODE, become_implementer, join_netns, resolve_target, verify_zero_privileges, ) INSTALLED_MODULE = "/usr/local/lib/overdeck/seat_implementer_exec.py" INSTALLED_BIN = "/usr/local/bin/overdeck-seat-implementer-exec" FORBIDDEN_ENV = frozenset({ "DBUS_SESSION_BUS_ADDRESS", "XDG_RUNTIME_DIR", "SYSTEMD_EXEC_PID", "GIT_CONFIG_GLOBAL", "GIT_CONFIG_SYSTEM", }) def die(msg: str, code: int = 2) -> None: _common_die(f"overdeck-seat-implementer-exec: {msg}", code) def validate_text(name: str, value: str) -> None: if not value or CONTROL_RE.search(value): die(f"invalid {name}") def validate_absolute_path(name: str, path: str) -> None: if not path.startswith("/") or "//" in path: die(f"invalid {name}") if "$" in path or "`" in path: die(f"shell-home-forbidden:{name}", 1) validate_text(name, path) def validate_exact_path(path: str, want: str, *, label: str) -> None: validate_absolute_path(label, path) if path != want: die(f"{label} mismatch", 1) def validate_operator_paths(ns: argparse.Namespace) -> str: if not SEAT_ID_RE.match(ns.account_slug): die("invalid account slug") validate_absolute_path("operator-home", ns.operator_home) _operator_user, _operator_uid, _operator_gid, authenticated_home = resolve_target() if ns.operator_home != authenticated_home: die("operator-home-mismatch", 1) want_checkout = f"{authenticated_home}/seats/{ns.seat_id}/repo" want_profile = f"{authenticated_home}/.claudex-accounts/{ns.account_slug}/{ns.seat_id}" want_secure = f"{want_profile}/secure-storage" validate_exact_path(ns.checkout, want_checkout, label="checkout") validate_exact_path(ns.profile_dir, want_profile, label="profile-dir") validate_exact_path(ns.secure_storage_dir, want_secure, label="secure-storage-dir") return authenticated_home def read_tmux_conf(seat_id: str) -> str: path = tmux_conf_path(seat_id) if not os.path.isfile(path): die("tmux-conf-missing", 1) st = os.lstat(path) if os.path.islink(path): 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 (st.st_mode & 0o777) != TMUX_CONF_MODE: die("tmux-conf-mode", 1) with open(path, encoding="utf-8") as fh: text = fh.read() if text != TMUX_CONF: die("tmux-conf-content", 1) return path def build_implementer_env(ns: argparse.Namespace, uid: int, username: str) -> dict[str, str]: env: dict[str, str] = { "HOME": ns.implementer_home, "USER": username, "LOGNAME": username, "PATH": ns.path or "/usr/local/bin:/usr/bin:/bin", "OVERDECK_SEAT_ID": ns.seat_id, "OVERDECK_SEAT_HOST": ns.seat_host, "OVERDECK_SEAT_MODEL": ns.seat_model, "OVERDECK_CLAUDE_BIN": ns.claude_bin, "OVERDECK_SEAT_GUARD_BIN": ns.guard_bin, "OVERDECK_SEAT_CHECKOUT": ns.checkout, "OVERDECK_SEAT_STATE_DIR": ns.state_dir, "OVERDECK_SEAT_GENERATION": ns.generation, "CLAUDE_CONFIG_DIR": ns.profile_dir, "CLAUDE_SECURESTORAGE_CONFIG_DIR": ns.secure_storage_dir, "SYSTRAY_CLAUDE_ACCOUNT_HOME": ns.profile_dir, "SYSTRAY_CLAUDE_ACCOUNT_SLUG": ns.account_slug, } for key in ("LANG", "LC_ALL", "TERM", "SHELL"): if key in os.environ: env[key] = os.environ[key] return env def validate_command(cmd: list[str], launcher: str, socket: str, tmux_conf: str) -> None: if len(cmd) < 2: die("command-too-short", 1) if os.path.basename(cmd[0]) != "tmux": die("command-not-tmux", 1) if "-S" not in cmd or socket not in cmd: die("tmux-socket-mismatch", 1) if "-f" not in cmd or tmux_conf not in cmd: die("tmux-conf-missing", 1) if "new-session" not in cmd: die("tmux-new-session-missing", 1) try: dash_dash = cmd.index("--") except ValueError: die("tmux-launcher-missing", 1) if dash_dash + 1 >= len(cmd) or cmd[dash_dash + 1] != launcher: die("launcher-mismatch", 1) def build_arg_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="overdeck-seat-implementer-exec", add_help=True) parser.add_argument("--seat-id", required=True) parser.add_argument("--seat-host", required=True) parser.add_argument("--seat-model", required=True) parser.add_argument("--socket", required=True) parser.add_argument("--launcher", required=True) parser.add_argument("--claude-bin", required=True) parser.add_argument("--guard-bin", required=True) parser.add_argument("--checkout", required=True) parser.add_argument("--state-dir", required=True) parser.add_argument("--generation", required=True) parser.add_argument("--implementer-home", required=True) parser.add_argument("--operator-home", required=True) parser.add_argument("--account-slug", required=True) parser.add_argument("--profile-dir", required=True) parser.add_argument("--secure-storage-dir", required=True) parser.add_argument("--netns-path", required=True) parser.add_argument("--path", default="") parser.add_argument("command", nargs=argparse.REMAINDER) return parser def validate_args(ns: argparse.Namespace) -> list[str]: if not SEAT_ID_RE.match(ns.seat_id): die("invalid seat id") validate_text("seat-host", ns.seat_host) validate_text("seat-model", ns.seat_model) for name in ( "socket", "launcher", "claude_bin", "guard_bin", "checkout", "state_dir", "implementer_home", "profile_dir", "secure_storage_dir", ): validate_absolute_path(name, getattr(ns, name)) validate_text("account-slug", ns.account_slug) from seat_scope_entry import GENERATION_RE if not GENERATION_RE.match(ns.generation): die("invalid generation", 1) validate_absolute_path("netns-path", ns.netns_path) validate_operator_paths(ns) if ns.launcher != SEAT_LAUNCHER_BIN: die("launcher-mismatch", 1) want_home = implementer_home_for_seat(ns.seat_id) if os.path.normpath(ns.implementer_home) != os.path.normpath(want_home): die("implementer-home-mismatch", 1) if not ns.command or ns.command[0] != "--": die("missing-command-separator") cmd = ns.command[1:] if not cmd: die("empty-command") return cmd def require_root_transition(operator_uid: int) -> None: if os.geteuid() != 0: die("requires-root", 1) sudo_uid = os.environ.get("SUDO_UID", "").strip() if not sudo_uid: die("operator-invocation-required", 1) try: invoker_uid = int(sudo_uid) except ValueError: die("invalid sudo uid", 1) if invoker_uid != operator_uid: die("operator-invocation-required", 1) def import_capability(path: str, *, directory: bool) -> tuple[int, str]: if not __import__("re").fullmatch(r"/proc/[1-9][0-9]*/fd/[0-9]+", path): die("local-capability-invalid", 1) flags = os.O_RDONLY | os.O_NOFOLLOW if directory: flags |= os.O_DIRECTORY try: fd = os.open(path, flags) except OSError: die("local-capability-unavailable", 1) os.set_inheritable(fd, True) return fd, f"/proc/self/fd/{fd}" def run_local_session(argv: list[str]) -> None: parser = argparse.ArgumentParser(prog="overdeck-seat-implementer-exec --local-session") parser.add_argument("--seat-id", required=True) parser.add_argument("--socket", required=True) parser.add_argument("--generation", required=True) parser.add_argument("--cwd", required=True) parser.add_argument("command", nargs=argparse.REMAINDER) ns = parser.parse_args(argv) if not SEAT_ID_RE.match(ns.seat_id) or not ns.command or ns.command[0] != "--": die("local-session-invalid", 2) operator_user, operator_uid, _operator_gid, _operator_home = resolve_target() require_root_transition(operator_uid) user, uid, gid = resolve_implementer_for_seat(ns.seat_id, operator_user) from seat_scope_entry import load_session_manifest, session_manifest_path manifest = load_session_manifest(session_manifest_path(ns.seat_id), ns.seat_id, ns.generation) if manifest["socket"] != ns.socket or manifest["implementerUid"] != uid: die("manifest-authority-mismatch", 1) tmux_conf = read_tmux_conf(ns.seat_id) cmd = ns.command[1:] if not cmd or cmd[0] != "/usr/bin/tmux": die("local-tmux-path", 1) if ("-S" not in cmd or ns.socket not in cmd or "-f" not in cmd or tmux_conf not in cmd or "new-session" not in cmd or "--" not in cmd): die("local-tmux-command", 1) launcher_index = cmd.index("--") + 1 if launcher_index >= len(cmd): die("local-launcher", 1) if ns.cwd != "/": die("local-cwd-invalid", 1) env = { "HOME": implementer_home_for_seat(ns.seat_id), "USER": user, "LOGNAME": user, "PATH": "/usr/local/bin:/usr/bin:/bin", "OVERDECK_SEAT_ID": ns.seat_id, "OVERDECK_SEAT_GENERATION": ns.generation, } for key in ("LANG", "LC_ALL", "TERM"): if key in os.environ: env[key] = os.environ[key] become_implementer(uid, gid, user) os.chdir("/") os.execve(cmd[0], cmd, env) def main(argv: list[str] | None = None) -> None: argv = list(sys.argv[1:] if argv is None else argv) if argv[:1] == ["--local-session"]: run_local_session(argv[1:]) return parser = build_arg_parser() try: ns = parser.parse_args(argv) except SystemExit: die("unknown arg", 2) cmd = validate_args(ns) operator_user, operator_uid, _operator_gid, _operator_home = resolve_target() require_root_transition(operator_uid) user, uid, gid = resolve_implementer_for_seat(ns.seat_id, operator_user) from seat_scope_entry import load_session_manifest, session_manifest_path manifest = load_session_manifest( session_manifest_path(ns.seat_id), ns.seat_id, ns.generation ) if manifest["socket"] != ns.socket or manifest["implementerUid"] != uid: die("manifest-authority-mismatch", 1) tmux_conf = read_tmux_conf(ns.seat_id) validate_command(cmd, ns.launcher, ns.socket, tmux_conf) env = build_implementer_env(ns, uid, user) for key in FORBIDDEN_ENV: env.pop(key, None) join_netns(ns.netns_path) record_netns_inode(ns.seat_id) become_implementer(uid, gid, user) os.chdir(ns.checkout) os.execvpe(cmd[0], cmd, env) if __name__ == "__main__": main()