"""Per-seat implementer system identity, ACL grants, and path validation.""" from __future__ import annotations import hashlib import grp import os import pwd import shutil import shlex import stat import subprocess import errno import re from seat_common import ( CONTROL_BASE_MODE, CONTROL_DIR_MODE, CONTROL_OWNER_MARKER_MODE, CONTROL_NETNS_INODE_MODE, CONTROL_RE, IMPLEMENTER_USER_PREFIX, LAUNCH_AUTHORITY_FILES, LINUX_USERNAME_RE, MAX_LINUX_USERNAME, NOLOGIN_SHELL, OWNER_MARKER_PREFIX, SEAT_GIT_EMAIL_DOMAIN, SEAT_ID_RE, SEAT_LAUNCHER_BIN, SEAT_TRUSTED_CONFIG_DIR, die, implementer_runtime_base, seat_control_base, seat_test_mode, ) _O_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0) MSG_PREFIX = "overdeck-seat-identity:" def _die(msg: str, code: int = 2) -> None: die(f"{MSG_PREFIX} {msg}", code) def _lstat_strict(path: str) -> os.stat_result: try: return os.lstat(path) except OSError as exc: _die(f"path-missing:{path}:{exc.strerror}", 1) def implementer_username_for_seat(seat_id: str) -> str: if not SEAT_ID_RE.match(seat_id): _die("invalid seat id") digest = hashlib.sha256(seat_id.encode("utf-8")).hexdigest()[:16] if ( seat_id != seat_id.lower() or "." in seat_id or len(f"{IMPLEMENTER_USER_PREFIX}{seat_id.lower()}") > MAX_LINUX_USERNAME ): return f"{IMPLEMENTER_USER_PREFIX}{digest}" candidate = f"{IMPLEMENTER_USER_PREFIX}{seat_id.lower()}" if LINUX_USERNAME_RE.match(candidate): return candidate return f"{IMPLEMENTER_USER_PREFIX}{digest}" def implementer_home_for_seat(seat_id: str) -> str: return f"{implementer_runtime_base()}/{seat_id}/home" def implementer_runtime_dir(seat_id: str) -> str: return f"{implementer_runtime_base()}/{seat_id}" def control_dir_for_seat(seat_id: str) -> str: return f"{seat_control_base()}/{seat_id}" def owner_marker_path(seat_id: str) -> str: return f"{control_dir_for_seat(seat_id)}/owner" def tmux_conf_path(seat_id: str) -> str: return f"{control_dir_for_seat(seat_id)}/tmux-seat.conf" def netns_inode_marker_path(seat_id: str) -> str: return f"{control_dir_for_seat(seat_id)}/netns-inode" def read_netns_inode(path: str) -> str: st = os.stat(path) return f"{st.st_dev}:{st.st_ino}" def record_netns_inode(seat_id: str) -> None: inode = read_netns_inode("/proc/self/ns/net") atomic_root_control_write(seat_id, "netns-inode", f"{inode}\n", mode=CONTROL_NETNS_INODE_MODE) def read_recorded_netns_inode(seat_id: str) -> str: path = netns_inode_marker_path(seat_id) if not os.path.isfile(path): die("netns-inode-missing", 1) st = os.lstat(path) if stat.S_ISLNK(st.st_mode) or not stat.S_ISREG(st.st_mode): die("netns-inode-unreadable", 1) if not seat_test_mode() and (st.st_uid != 0 or st.st_gid != 0): die("netns-inode-owner", 1) if not seat_test_mode() and stat.S_IMODE(st.st_mode) != CONTROL_NETNS_INODE_MODE: die("netns-inode-mode", 1) with open(path, encoding="utf-8") as fh: raw = fh.read().strip() if not raw or not re.fullmatch(r"\d+:\d+", raw): die("netns-inode-nonnumeric", 1) return raw def expected_socket_for_seat(operator_home: str, seat_id: str) -> str: return f"{operator_home}/.local/state/overdeck/seats/{seat_id}/tmux.sock" def assert_safe_path(path: str, *, label: str) -> None: if not path.startswith("/") or "//" in path or CONTROL_RE.search(path): _die(f"invalid-{label}") st = _lstat_strict(path) if stat.S_ISLNK(st.st_mode): _die(f"symlink-forbidden:{path}", 1) def assert_owned_regular(path: str, *, uid: int, gid: int, mode: int) -> None: st = _lstat_strict(path) if stat.S_ISLNK(st.st_mode) or not stat.S_ISREG(st.st_mode): _die(f"not-regular-file:{path}", 1) if st.st_uid != uid or st.st_gid != gid: _die(f"ownership-drift:{path}", 1) if stat.S_IMODE(st.st_mode) != mode: _die(f"mode-drift:{path}", 1) def ancestor_paths(path: str) -> list[str]: parts: list[str] = [] current = os.path.normpath(path) while True: parts.append(current) parent = os.path.dirname(current) if parent == current: break current = parent return list(reversed(parts)) def _relpath_from_home(path: str, operator_home: str) -> str: rel = os.path.relpath(path, operator_home) if rel.startswith("..") or rel == ".": _die(f"path-outside-home:{path}", 1) return rel def _allowed_symlink(parent: str, name: str, operator_home: str) -> bool: rel_parent = _relpath_from_home(parent, operator_home) if rel_parent == ".local/share/overdeck/seat-guard" and name == "current": return True if rel_parent == ".local/bin" and name == "claude": return True return False def _read_symlink_target(link_path: str, parent: str) -> str: target = os.readlink(link_path) if target.startswith("/"): resolved = os.path.normpath(target) else: resolved = os.path.normpath(os.path.join(parent, target)) if not resolved.startswith("/") or "//" in resolved or CONTROL_RE.search(resolved): _die(f"symlink-target-invalid:{link_path}", 1) return resolved def resolve_managed_path( operator_home: str, logical_path: str, *, operator_uid: int, operator_gid: int, ) -> str: """Resolve one-hop managed symlinks; validate ownership on every step.""" if not logical_path.startswith("/"): logical_path = os.path.join(operator_home, logical_path) logical_path = os.path.normpath(logical_path) if not logical_path.startswith(operator_home + "/") and logical_path != operator_home: _die(f"managed-path-outside-home:{logical_path}", 1) current = operator_home parts = logical_path[len(operator_home) :].strip("/").split("/") idx = 0 while idx < len(parts): part = parts[idx] candidate = os.path.join(current, part) st = _lstat_strict(candidate) if stat.S_ISLNK(st.st_mode): if not _allowed_symlink(os.path.dirname(candidate), part, operator_home): _die(f"symlink-unmanaged:{candidate}", 1) parent = os.path.dirname(candidate) resolved = _read_symlink_target(candidate, parent) if not resolved.startswith(operator_home + "/"): _die(f"symlink-escape:{candidate}", 1) target_st = _lstat_strict(resolved) if stat.S_ISLNK(target_st.st_mode): _die(f"symlink-chained:{candidate}", 1) if target_st.st_uid != operator_uid or target_st.st_gid != operator_gid: _die(f"symlink-target-owner:{resolved}", 1) current = resolved idx += 1 continue if st.st_uid != operator_uid or st.st_gid != operator_gid: _die(f"path-owner-mismatch:{candidate}", 1) current = candidate idx += 1 return current def managed_runtime_targets( operator_home: str, seat_id: str, *, repo: str, profile: str, state_dir: str, guard_bin: str, claude_bin: str, launcher: str, ) -> tuple[list[str], list[str], list[str]]: seat_root = f"{operator_home}/seats/{seat_id}" traverse_only = [ operator_home, f"{operator_home}/seats", seat_root, f"{operator_home}/.claudex-accounts", os.path.dirname(profile), f"{operator_home}/.local", f"{operator_home}/.local/state", f"{operator_home}/.local/state/overdeck", f"{operator_home}/.local/state/overdeck/seats", f"{operator_home}/.local/share", f"{operator_home}/.local/share/overdeck", f"{operator_home}/.local/share/overdeck/seat-guard", f"{operator_home}/.local/bin", ] rw_roots = [ os.path.normpath(state_dir), os.path.normpath(repo), os.path.normpath(profile), ] executables = [guard_bin, claude_bin] return traverse_only, rw_roots, executables def require_setfacl() -> None: if shutil.which("setfacl") is None: _die("setfacl-missing", 1) def _run_setfacl(argv: list[str], *, label: str) -> None: proc = subprocess.run(argv, capture_output=True, text=True) if proc.returncode != 0: detail = (proc.stderr or proc.stdout or "").strip() _die(f"{label}:{detail}", 1) def grant_traverse_acl(path: str, username: str) -> None: if not os.path.lexists(path): return assert_safe_path(path, label="acl-target") st = _lstat_strict(path) if stat.S_ISLNK(st.st_mode): _die(f"symlink-forbidden:{path}", 1) _run_setfacl(["setfacl", "-m", f"u:{username}:x", path], label=f"setfacl-failed:{path}") def trusted_config_dir(state_dir: str) -> str: return os.path.join(os.path.normpath(state_dir), SEAT_TRUSTED_CONFIG_DIR) def launch_authority_paths(state_dir: str) -> list[str]: base = trusted_config_dir(state_dir) return [os.path.join(base, name) for name in LAUNCH_AUTHORITY_FILES] def seat_git_author_name(seat_id: str) -> str: return f"Overdeck Seat {seat_id}" def seat_git_author_email(seat_id: str) -> str: return f"{seat_id}@{SEAT_GIT_EMAIL_DOMAIN}" def configure_seat_git_author(checkout: str, seat_id: str) -> None: if not os.path.isdir(checkout): _die("checkout-missing", 1) name = seat_git_author_name(seat_id) email = seat_git_author_email(seat_id) for key, value in (("user.name", name), ("user.email", email)): proc = subprocess.run( ["git", "-C", checkout, "config", "--local", key, value], capture_output=True, text=True, ) if proc.returncode != 0: detail = (proc.stderr or proc.stdout or "").strip() _die(f"seat-git-config-failed:{key}:{detail}", 1) def verify_seat_git_author(checkout: str, seat_id: str) -> None: if not os.path.isdir(checkout): _die("checkout-missing", 1) want_name = seat_git_author_name(seat_id) want_email = seat_git_author_email(seat_id) for key, want in (("user.name", want_name), ("user.email", want_email)): proc = subprocess.run( ["git", "-C", checkout, "config", "--local", "--get", key], capture_output=True, text=True, ) if proc.returncode != 0 or proc.stdout.strip() != want: _die(f"seat-git-config-drift:{key}", 1) def validate_implementer_group_membership(username: str, primary_gid: int) -> None: try: pw = pwd.getpwnam(username) except KeyError: _die("implementer-user-missing", 1) if pw.pw_gid != primary_gid: _die("implementer-primary-gid-mismatch", 1) try: primary = grp.getgrgid(primary_gid) except KeyError: _die("implementer-primary-group-missing", 1) if primary.gr_name != username: _die("implementer-primary-group-name-mismatch", 1) for group in grp.getgrall(): if username in group.gr_mem: _die(f"implementer-supplementary-group:{group.gr_name}", 1) def deny_operator_credentials(implementer_user: str, operator_home: str) -> None: for path in operator_credential_paths(operator_home): if not os.path.lexists(path): continue assert_safe_path(path, label="credential-deny-target") st = _lstat_strict(path) if stat.S_ISLNK(st.st_mode): _die(f"symlink-forbidden:{path}", 1) _run_setfacl( ["setfacl", "-m", f"u:{implementer_user}:---", path], label=f"credential-deny-failed:{path}", ) def grant_rw_root_acl(path: str, implementer_user: str, operator_user: str) -> None: if not os.path.exists(path): return assert_safe_path(path, label="acl-rw-target") st = _lstat_strict(path) if stat.S_ISLNK(st.st_mode): _die(f"symlink-forbidden:{path}", 1) for username in (implementer_user, operator_user): perm = "rwX" if stat.S_ISDIR(st.st_mode) else "rw-" _run_setfacl(["setfacl", "-m", f"u:{username}:{perm}", path], label=f"setfacl-failed:{path}") if stat.S_ISDIR(st.st_mode): _run_setfacl( ["setfacl", "-R", "-m", f"u:{username}:rwX", path], label=f"setfacl-recursive-failed:{path}", ) _run_setfacl( ["setfacl", "-R", "-d", "-m", f"u:{username}:rwX", path], label=f"setfacl-default-recursive-failed:{path}", ) def grant_trusted_config_read_boundary( state_dir: str, implementer_user: str, operator_user: str, ) -> None: trusted = trusted_config_dir(state_dir) if not os.path.lexists(trusted): os.makedirs(trusted, mode=0o700, exist_ok=True) assert_safe_path(trusted, label="trusted-config") st = _lstat_strict(trusted) if stat.S_ISLNK(st.st_mode): _die(f"symlink-forbidden:{trusted}", 1) _run_setfacl( ["setfacl", "-m", f"u:{operator_user}:rwX", trusted], label=f"trusted-config-operator:{trusted}", ) _run_setfacl( ["setfacl", "-d", "-m", f"u:{operator_user}:rwX", trusted], label=f"trusted-config-operator-default:{trusted}", ) _run_setfacl( ["setfacl", "-m", f"u:{implementer_user}:r-x", trusted], label=f"trusted-config-implementer:{trusted}", ) _run_setfacl( ["setfacl", "-d", "-m", f"u:{implementer_user}:r--", trusted], label=f"trusted-config-implementer-default:{trusted}", ) if os.path.isdir(trusted): for root, dirnames, filenames in os.walk(trusted): for name in dirnames: dir_path = os.path.join(root, name) _run_setfacl( ["setfacl", "-m", f"u:{implementer_user}:r-x", dir_path], label=f"trusted-config-implementer-dir:{dir_path}", ) for name in filenames: file_path = os.path.join(root, name) _run_setfacl( ["setfacl", "-m", f"u:{implementer_user}:r--", file_path], label=f"trusted-config-implementer-file:{file_path}", ) def _verify_implementer_path_boundary( implementer_user: str, paths: list[str], *, sudo: str, require_readable: bool, label: str, ) -> None: for path in paths: if not os.path.lexists(path): continue if require_readable: proc = subprocess.run( [sudo, "-n", "-u", implementer_user, "test", "-r", path], capture_output=True, text=True, ) if proc.returncode != 0: _die(f"{label}-unreadable:{path}", 1) for mode in ("-r", "-w"): if require_readable and mode == "-r": continue proc = subprocess.run( [sudo, "-n", "-u", implementer_user, "test", mode, path], capture_output=True, text=True, ) if proc.returncode == 0: _die(f"{label}-readable:{path}" if mode == "-r" else f"{label}-writable:{path}", 1) proc = subprocess.run( [sudo, "-n", "-u", implementer_user, "rm", "-f", path], capture_output=True, text=True, ) if proc.returncode == 0: _die(f"{label}-deletable:{path}", 1) proc = subprocess.run( [sudo, "-n", "-u", implementer_user, "sh", "-c", f"printf x >{shlex.quote(path)}"], capture_output=True, text=True, ) if proc.returncode == 0: _die(f"{label}-replaceable:{path}", 1) def verify_implementer_launch_authority_boundary( implementer_user: str, state_dir: str, *, sudo: str = "sudo", ) -> None: trusted = trusted_config_dir(state_dir) _verify_implementer_path_boundary( implementer_user, launch_authority_paths(state_dir), sudo=sudo, require_readable=True, label="launch-authority", ) for mode in ("-w",): proc = subprocess.run( [sudo, "-n", "-u", implementer_user, "test", mode, trusted], capture_output=True, text=True, ) if proc.returncode == 0: _die(f"launch-authority-dir-writable:{trusted}", 1) for path in launch_authority_paths(state_dir): if not os.path.lexists(path): continue for mode in ("-w",): proc = subprocess.run( [sudo, "-n", "-u", implementer_user, "test", mode, path], capture_output=True, text=True, ) if proc.returncode == 0: _die(f"launch-authority-writable:{path}", 1) def grant_executable_acl(path: str, username: str) -> None: if not os.path.lexists(path): _die(f"executable-missing:{path}", 1) assert_safe_path(path, label="acl-exec-target") st = _lstat_strict(path) if stat.S_ISLNK(st.st_mode): _die(f"symlink-forbidden:{path}", 1) perm = "rx" if stat.S_ISREG(st.st_mode) else "x" _run_setfacl(["setfacl", "-m", f"u:{username}:{perm}", path], label=f"setfacl-failed:{path}") def grant_link_ancestors_acl(path: str, username: str, operator_home: str) -> None: for ancestor in ancestor_paths(path): if ancestor == path: continue if ancestor != operator_home and not ancestor.startswith(operator_home + "/"): continue grant_traverse_acl(ancestor, username) def verify_root_launcher(path: str) -> None: if path != SEAT_LAUNCHER_BIN: _die("launcher-mismatch", 1) if not os.path.lexists(path): _die(f"launcher-missing:{path}", 1) st = _lstat_strict(path) if stat.S_ISLNK(st.st_mode): _die(f"launcher-symlink:{path}", 1) if not stat.S_ISREG(st.st_mode): _die(f"launcher-not-file:{path}", 1) if stat.S_IMODE(st.st_mode) != 0o755: _die(f"launcher-mode:{path}", 1) if not seat_test_mode() and (st.st_uid != 0 or st.st_gid != 0): _die(f"launcher-owner:{path}", 1) if not os.access(path, os.X_OK): _die(f"launcher-not-executable:{path}", 1) def grant_managed_runtime_access( implementer_user: str, operator_user: str, operator_home: str, seat_id: str, *, operator_uid: int, operator_gid: int, repo: str, profile: str, state_dir: str, guard_bin: str, claude_bin: str, launcher: str, ) -> None: require_setfacl() deny_operator_credentials(implementer_user, operator_home) resolved_guard = resolve_managed_path( operator_home, guard_bin, operator_uid=operator_uid, operator_gid=operator_gid ) resolved_claude = resolve_managed_path( operator_home, claude_bin, operator_uid=operator_uid, operator_gid=operator_gid ) verify_root_launcher(launcher) traverse_only, rw_roots, _executables = managed_runtime_targets( operator_home, seat_id, repo=repo, profile=profile, state_dir=state_dir, guard_bin=resolved_guard, claude_bin=resolved_claude, launcher=launcher, ) seen: set[str] = set() for path in traverse_only: norm = os.path.normpath(path) if norm in seen: continue seen.add(norm) grant_traverse_acl(norm, implementer_user) for path in rw_roots: grant_rw_root_acl(path, implementer_user, operator_user) grant_trusted_config_read_boundary(state_dir, implementer_user, operator_user) grant_link_ancestors_acl(resolved_guard, implementer_user, operator_home) grant_executable_acl(resolved_guard, implementer_user) grant_link_ancestors_acl(resolved_claude, implementer_user, operator_home) grant_executable_acl(resolved_claude, implementer_user) def _test_mode() -> bool: from seat_common import seat_test_mode return seat_test_mode() def _ensure_root_control_dir( path: str, mode: int, *, symlink_msg: str, invariant_msg: str, ) -> str: if os.path.lexists(path): st = _lstat_strict(path) if stat.S_ISLNK(st.st_mode): _die(symlink_msg, 1) if not stat.S_ISDIR(st.st_mode): _die(f"{invariant_msg}-not-dir", 1) if not _test_mode() and (st.st_uid != 0 or st.st_gid != 0): _die(invariant_msg, 1) if stat.S_IMODE(st.st_mode) != mode: os.chmod(path, mode, follow_symlinks=False) st = os.lstat(path) if stat.S_ISLNK(st.st_mode) or not stat.S_ISDIR(st.st_mode): _die(symlink_msg, 1) if not _test_mode(): if st.st_uid != 0 or st.st_gid != 0 or stat.S_IMODE(st.st_mode) != mode: _die(invariant_msg, 1) return path os.makedirs(path, mode=mode) st = os.lstat(path) if stat.S_ISLNK(st.st_mode): _die(symlink_msg, 1) if not stat.S_ISDIR(st.st_mode): _die(f"{invariant_msg}-not-dir", 1) if stat.S_IMODE(st.st_mode) != mode: os.chmod(path, mode, follow_symlinks=False) st = os.lstat(path) if not _test_mode(): if st.st_uid != 0 or st.st_gid != 0 or stat.S_IMODE(st.st_mode) != mode: _die(invariant_msg, 1) return path def ensure_control_base() -> str: return _ensure_root_control_dir( seat_control_base(), CONTROL_BASE_MODE, symlink_msg="control-base-symlink", invariant_msg="control-base-invariant", ) def ensure_control_dir(seat_id: str) -> str: ensure_control_base() return _ensure_root_control_dir( control_dir_for_seat(seat_id), CONTROL_DIR_MODE, symlink_msg="control-dir-symlink", invariant_msg="control-dir-invariant", ) def atomic_root_control_write(seat_id: str, name: str, content: str, *, mode: int) -> str: control = ensure_control_dir(seat_id) final = os.path.join(control, name) if os.path.lexists(final): st = os.lstat(final) if stat.S_ISLNK(st.st_mode): _die(f"control-symlink:{final}", 1) if not _test_mode() and (st.st_uid != 0 or st.st_gid != 0): _die(f"control-owner:{final}", 1) dir_fd = os.open(control, os.O_RDONLY | os.O_DIRECTORY | _O_NOFOLLOW) try: tmp = f".{name}.{os.getpid()}.tmp" flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | _O_NOFOLLOW try: fd = os.open(tmp, flags, mode, dir_fd=dir_fd) except OSError as exc: if exc.errno in (errno.EEXIST, errno.ELOOP): _die(f"control-temp-create:{final}:{exc.strerror}", 1) raise try: with os.fdopen(fd, "w", encoding="utf-8") as fh: fh.write(content) fh.flush() os.fchmod(fh.fileno(), mode) os.fsync(fh.fileno()) except OSError: try: os.unlink(tmp, dir_fd=dir_fd) except OSError: pass raise os.replace(tmp, name, src_dir_fd=dir_fd, dst_dir_fd=dir_fd) os.fsync(dir_fd) finally: os.close(dir_fd) st = os.lstat(final) if stat.S_ISLNK(st.st_mode) or not stat.S_ISREG(st.st_mode): _die(f"control-not-regular:{final}", 1) if _test_mode(): return final if st.st_uid != 0 or st.st_gid != 0 or stat.S_IMODE(st.st_mode) != mode: _die(f"control-finalize:{final}", 1) return final def read_owner_marker(seat_id: str) -> str: path = owner_marker_path(seat_id) if not os.path.isfile(path): _die("implementer-owner-marker-missing", 1) st = os.lstat(path) if stat.S_ISLNK(st.st_mode) or not stat.S_ISREG(st.st_mode): _die("implementer-owner-marker-invalid", 1) if not _test_mode() and (st.st_uid != 0 or st.st_gid != 0): _die("implementer-owner-marker-invalid", 1) with open(path, encoding="utf-8") as fh: return fh.read().strip() def write_owner_marker(seat_id: str, operator_user: str) -> None: atomic_root_control_write(seat_id, "owner", operator_user, mode=CONTROL_OWNER_MARKER_MODE) def pw_comment_matches(username: str, operator_user: str) -> bool: try: pw = pwd.getpwnam(username) except KeyError: return False return pw.pw_gecos == f"{OWNER_MARKER_PREFIX}{operator_user}" def validate_implementer_passwd(pw: pwd.struct_passwd, operator_user: str, seat_id: str) -> None: want_user = implementer_username_for_seat(seat_id) if pw.pw_name != want_user: _die(f"implementer-username-mismatch:{pw.pw_name}", 1) if pw.pw_uid == 0: _die("implementer-uid-root", 1) if pw.pw_shell != NOLOGIN_SHELL: _die("implementer-shell-invalid", 1) want_home = implementer_home_for_seat(seat_id) if pw.pw_dir != want_home: _die("implementer-home-mismatch", 1) if pw.pw_gecos != f"{OWNER_MARKER_PREFIX}{operator_user}": _die("implementer-gecos-mismatch", 1) def ensure_implementer_user(seat_id: str, operator_user: str) -> tuple[str, int, int]: username = implementer_username_for_seat(seat_id) home = implementer_home_for_seat(seat_id) marker = f"{OWNER_MARKER_PREFIX}{operator_user}" ensure_control_dir(seat_id) try: pw = pwd.getpwnam(username) validate_implementer_passwd(pw, operator_user, seat_id) validate_implementer_group_membership(username, pw.pw_gid) marker_owner = read_owner_marker(seat_id) if marker_owner != operator_user: _die("implementer-owner-marker-mismatch", 1) return username, pw.pw_uid, pw.pw_gid except KeyError: pass runtime = implementer_runtime_dir(seat_id) os.makedirs(runtime, mode=0o700, exist_ok=True) os.chown(runtime, 0, 0) proc = subprocess.run( [ "useradd", "--system", "--user-group", "--home-dir", home, "--shell", NOLOGIN_SHELL, "--comment", marker, username, ], capture_output=True, text=True, ) if proc.returncode != 0: detail = (proc.stderr or proc.stdout or "").strip() _die(f"implementer-useradd-failed:{detail}", 1) pw = pwd.getpwnam(username) validate_implementer_passwd(pw, operator_user, seat_id) validate_implementer_group_membership(username, pw.pw_gid) write_owner_marker(seat_id, operator_user) os.makedirs(home, mode=0o700, exist_ok=True) os.chown(home, pw.pw_uid, pw.pw_gid) return username, pw.pw_uid, pw.pw_gid def resolve_implementer_for_seat(seat_id: str, operator_user: str) -> tuple[str, int, int]: username = implementer_username_for_seat(seat_id) try: pw = pwd.getpwnam(username) except KeyError: _die("implementer-user-missing", 1) validate_implementer_passwd(pw, operator_user, seat_id) validate_implementer_group_membership(username, pw.pw_gid) marker_owner = read_owner_marker(seat_id) if marker_owner != operator_user: _die("implementer-owner-marker-mismatch", 1) return username, pw.pw_uid, pw.pw_gid def operator_credential_paths(operator_home: str) -> list[str]: return [ f"{operator_home}/.claude/.credentials.json", f"{operator_home}/.ssh", f"{operator_home}/.config/gh", f"{operator_home}/.git-credentials", f"{operator_home}/.gitconfig", ] def verify_implementer_seat_metadata_boundary( implementer_user: str, operator_home: str, seat_id: str, *, sudo: str = "sudo", ) -> None: seat_root = os.path.normpath(f"{operator_home}/seats/{seat_id}") seat_json = os.path.join(seat_root, "seat.json") _verify_implementer_path_boundary( implementer_user, [seat_json], sudo=sudo, require_readable=False, label="seat-metadata", ) if os.path.isdir(seat_root): proc = subprocess.run( [sudo, "-n", "-u", implementer_user, "test", "-w", seat_root], capture_output=True, text=True, ) if proc.returncode == 0: _die(f"seat-root-writable:{seat_root}", 1) def verify_implementer_credential_boundary( implementer_user: str, operator_home: str, *, sudo: str = "sudo", ) -> None: for path in operator_credential_paths(operator_home): if not os.path.exists(path): continue proc = subprocess.run( [sudo, "-n", "-u", implementer_user, "test", "-r", path], capture_output=True, text=True, ) if proc.returncode == 0: _die(f"operator-credential-readable:{path}", 1) def verify_implementer_systemd_absent(implementer_user: str, *, sudo: str = "sudo") -> None: proc = subprocess.run( [sudo, "-n", "-u", implementer_user, "systemctl", "--user", "is-system-running"], capture_output=True, text=True, ) if proc.returncode == 0: _die("user-systemd-available", 1) def _sudo_readable_as_user(path: str, username: str, *, sudo: str = "sudo") -> bool: proc = subprocess.run( [sudo, "-n", "-u", username, "test", "-r", path], capture_output=True, text=True, ) return proc.returncode == 0 def _assert_secure_implementer_runtime_dir( path: str, *, uid: int, gid: int, ) -> None: if not os.path.lexists(path): return st = os.lstat(path) if stat.S_ISLNK(st.st_mode): _die(f"implementer-runtime-symlink:{path}", 1) if not stat.S_ISDIR(st.st_mode): _die(f"implementer-runtime-not-dir:{path}", 1) if st.st_uid != uid or st.st_gid != gid: _die(f"implementer-runtime-ownership-drift:{path}", 1) if stat.S_IMODE(st.st_mode) & 0o077: _die(f"implementer-runtime-insecure-mode:{path}", 1) def _assert_path_unreadable_as_implementer( path: str, implementer_user: str, *, label: str, sudo: str = "sudo", ) -> None: if not os.path.lexists(path): return if _sudo_readable_as_user(path, implementer_user, sudo=sudo): _die(f"{label}:{path}", 1) def _assert_implementer_bus_absent(bus_path: str) -> None: if os.path.lexists(bus_path): _die(f"implementer-bus-present:{bus_path}", 1) def verify_implementer_runtime_bus_absent( implementer_user: str, implementer_uid: int, implementer_gid: int, operator_uid: int | None, *, sudo: str = "sudo", ) -> None: runtime = f"/run/user/{implementer_uid}" _assert_secure_implementer_runtime_dir( runtime, uid=implementer_uid, gid=implementer_gid, ) _assert_implementer_bus_absent(f"{runtime}/bus") if operator_uid is not None: operator_runtime = f"/run/user/{operator_uid}" for path in (operator_runtime, f"{operator_runtime}/bus"): _assert_path_unreadable_as_implementer( path, implementer_user, label="operator-runtime-readable", sudo=sudo, )