"""Bounded command delegation to a SEPARATE unprivileged runtime identity. This is not a root-command service. The root launcher supplies a listening Unix socket, exact manager identity, and the runtime's private home/runtime directory. Existing commands execute with the runtime UID's permissions. The launcher must qualify that account's no-sudo/polkit policy, separation from manager/authority, and private-state DAC/ancestry; nonzero UID alone is not proof of those properties. Root operations remain separately typed. NNP is not imposed here because approved rootless UID-mapping helpers may require their normal setuid execution path. """ import array import ctypes import fcntl from dataclasses import dataclass, field import os from pathlib import Path import selectors import signal import socket import stat import struct import time MAGIC = 0x41524352 MAX_PACKET = 65536 MAX_PENDING = 32 MAX_JOBS = 32 ADMISSION_NS = 2_000_000_000 MAX_REAPS = 64 MAX_CHILD_BYTES = 65536 def require(condition, message): if not condition: raise ValueError(message) def now(): return time.clock_gettime_ns(time.CLOCK_BOOTTIME) def decode(raw): require(0 < len(raw) <= MAX_PACKET, 'runtime packet size') position = 0 def integer(): nonlocal position require(position + 4 <= len(raw), 'truncated runtime integer') value = struct.unpack_from('!I', raw, position)[0] position += 4 return value def string(): nonlocal position size = integer() require(size <= len(raw) - position, 'truncated runtime string') value = raw[position:position + size] position += size require(b'\0' not in value, 'runtime string contains NUL') return value.decode('utf-8', errors='strict') require(integer() == MAGIC and integer() == 1, 'runtime protocol mismatch') argc, envc = integer(), integer() require(1 <= argc <= 256 and envc <= 128, 'runtime vector bounds') cwd = string() require(cwd.startswith('/'), 'runtime cwd must be absolute') argv = [string() for _ in range(argc)] require(argv[0], 'runtime program missing') environment = {} for _ in range(envc): entry = string() key, separator, value = entry.partition('=') require(separator and key and key not in environment, 'invalid runtime environment') environment[key] = value require(position == len(raw), 'extra runtime packet bytes') return argv, environment, cwd def reply(connection, kind, value): try: raw = struct.pack('!III', MAGIC, kind, value) # Only a twelve-byte result; never buffer child stdout/stderr here. require(connection.send(raw, socket.MSG_DONTWAIT | socket.MSG_NOSIGNAL) == len(raw), 'runtime result delivery uncertain') except (OSError, ValueError): pass # No retry: child outcome and response delivery are distinct. def reap_adopted(jobs): """One nonblocking scan, including while leaders are live. Never reap any job leader: its WNOWAIT identity pins cancellation until all fallible cleanup completes. Reap every already-exited adopted child, including promoted persistent work: wait ownership does not follow cgroup migration. Cgroup membership only associates final retirement barriers; live persistent children are neither signalled nor synchronously waited for. """ leaders = {job.pid for job in jobs if job.pid is not None} with open(f'/proc/self/task/{os.getpid()}/children') as stream: children = stream.read(MAX_CHILD_BYTES + 1) require(len(children) <= MAX_CHILD_BYTES, 'adopted child scan bound') pending, reaped = set(), 0 for text in children.split(): child = int(text) if child in leaders: continue try: location = None for job in jobs: if job.pid is None: continue if job.group is None: owned = os.getpgid(child) == job.pid else: if location is None: location = Path(f'/proc/{child}/cgroup').read_text().strip() expected = '0::' + str(job.group.path).removeprefix('/sys/fs/cgroup') owned = location == expected or location.startswith(expected + '/') if not owned: continue pending.add(job.pid) break if reaped < MAX_REAPS: status = os.waitid(os.P_PID, child, os.WEXITED | os.WNOHANG | os.WNOWAIT) if status is not None: os.waitpid(child, os.WNOHANG) reaped += 1 except (ProcessLookupError, FileNotFoundError, ChildProcessError): continue return pending @dataclass class Pending: connection: socket.socket deadline: int payload: bytearray = field(default_factory=bytearray) descriptors: list = field(default_factory=list) first: bool = True def close_descriptors(self): owned, self.descriptors = self.descriptors, [] for fd in owned: try: os.close(fd) except OSError: pass def read(self): require(now() < self.deadline, 'runtime admission expired') data, ancillary, flags, _ = self.connection.recvmsg(MAX_PACKET + 5 - len(self.payload), socket.CMSG_SPACE(16 * array.array('i').itemsize), socket.MSG_DONTWAIT | socket.MSG_CMSG_CLOEXEC) # Own every installed right before checking flags, including malformed # requests. The kernel closes rights that do not fit ancillary storage. unexpected = False received = [] for level, kind, raw in ancillary: if (level, kind) == (socket.SOL_SOCKET, socket.SCM_RIGHTS): values = array.array('i') values.frombytes(raw[:len(raw) - len(raw) % values.itemsize]) received.extend(values) else: unexpected = True self.descriptors.extend(received) require(not unexpected and not flags & ~socket.MSG_CMSG_CLOEXEC and data, 'runtime packet/control truncated or closed') require((self.first and len(received) == 3) or (not self.first and not received), 'runtime requires exactly one stdin/stdout/stderr transfer') self.first = False self.payload.extend(data) require(now() < self.deadline and len(self.payload) <= MAX_PACKET + 4, 'runtime admission bound') if len(self.payload) < 4: return None size = struct.unpack_from('!I', self.payload)[0] require(0 < size <= MAX_PACKET and len(self.payload) <= size + 4, 'runtime frame size') if len(self.payload) != size + 4: return None return decode(bytes(self.payload[4:])) @dataclass class Job: connection: socket.socket pid: int = None cancelled: bool = False status: object = None group: object = None retirement_error: str = None def cancel(self): if self.cancelled: return self.cancelled = True # The child is not reaped/reused until this owner observes completion. # Kill its session's group, not merely a shell that could leave children. if self.pid is not None: self.kill_group() def kill_group(self): if self.group is not None: self.group.kill() # Leader remains an unreaped child, so neither its PID nor its group # identity can be reused while these signals are issued. for target in (-self.pid, self.pid): try: os.kill(target, signal.SIGKILL) except (ProcessLookupError, PermissionError): pass def settle(self, jobs): try: result = self._settle(jobs) self.retirement_error = None return result except (OSError, ValueError) as error: # Failure is local to this reservation. Retain the leader, slot and # group for a later tick, without cancelling unrelated invocations. self.retirement_error = str(error) if self.group is not None: # Release traversal-only descriptors even when failure happened # before close() (e.g. EMFILE opening kill/events/child-list). self.group.release_walk() return False def _settle(self, jobs): if self.pid is None: return self.group is None or self.group.close() if self.cancelled: self.kill_group() # Retry failed cancellation even during final shutdown. if self.status is None: self.status = os.waitid(os.P_PID, self.pid, os.WEXITED | os.WNOHANG | os.WNOWAIT) if self.status is None: return False self.kill_group() if self.group is not None: # Includes double-fork/setsid children. A populated or unreadable # cgroup retains ownership; process-group disappearance is not proof. if not self.group.empty(): return False if self.pid in reap_adopted(jobs): return False # A failed/budget-limited retirement must not release the pinned leader. if self.group is not None and not self.group.close(): return False os.waitpid(self.pid, 0) return True def result(self): if self.pid is None: return 125 # A failed fork still owns its reserved cgroup until close. return self.status.si_status if self.status.si_code == os.CLD_EXITED else -self.status.si_status class RuntimeOwner: def __init__(self, listener, manager_uid, manager_gid, home, runtime_dir, username, groups=None, heartbeat=None): uid, gid = os.geteuid(), os.getegid() require(uid > 0 and gid > 0 and os.getresuid() == (uid, uid, uid) and os.getresgid() == (gid, gid, gid) and set(os.getgroups()) <= {gid}, 'runtime owner requires permanently dropped credentials') require(type(manager_uid) is int and type(manager_gid) is int and manager_uid > 0 and manager_gid > 0 and uid != manager_uid and gid != manager_gid, 'runtime and manager principals must differ') status = dict(line.split(':', 1) for line in Path('/proc/self/status').read_text().splitlines() if ':' in line) require(all(int(status[field], 16) == 0 for field in ('CapEff', 'CapPrm', 'CapInh', 'CapAmb')), 'runtime owner retains capabilities') libc = ctypes.CDLL(None, use_errno=True) require(len(list(Path('/proc/self/task').iterdir())) == 1, 'runtime owner requires a dedicated single-thread process') signal.signal(signal.SIGCHLD, signal.SIG_DFL) require(libc.prctl(4, 0, 0, 0, 0) == 0, 'runtime inspection exclusion failed') require(libc.prctl(36, 1, 0, 0, 0) == 0, 'runtime subreaper unavailable') require(listener.family == socket.AF_UNIX and listener.getsockopt(socket.SOL_SOCKET, socket.SO_TYPE) == socket.SOCK_STREAM and listener.getsockopt(socket.SOL_SOCKET, socket.SO_ACCEPTCONN) == 1 and os.fstat(listener.fileno()).st_uid == 0, 'runtime requires a root-created listening capability') listener.set_inheritable(False); listener.setblocking(False) for directory in (home, runtime_dir): path = Path(directory) meta = path.lstat() require(path.is_absolute() and path.resolve(strict=True) == path and stat.S_ISDIR(meta.st_mode) and meta.st_uid == uid and stat.S_IMODE(meta.st_mode) == 0o700, 'runtime private directory binding') require(type(username) is str and username and all(c.isalnum() or c in '_-' for c in username), 'runtime username') self.listener = listener self.manager = (manager_uid, manager_gid) self.environment = {'HOME': str(home), 'USER': username, 'LOGNAME': username, 'PATH': '/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin', 'XDG_RUNTIME_DIR': str(runtime_dir), 'DBUS_SESSION_BUS_ADDRESS': 'unix:path=' + str(Path(runtime_dir) / 'bus')} self.pending, self.jobs = {}, {} self.stopping = False self.groups = groups self.heartbeat = heartbeat def start(self, pending, request): require(not self.stopping, 'runtime is stopping') require(len(self.jobs) < MAX_JOBS, 'runtime worker limit') argv, environment, cwd = request # A manager's environment must not direct the runtime into the operator's # home/user bus. Application-specific variables remain command inputs. environment = {key: value for key, value in environment.items() if not key.startswith('SUDO_')} environment.update(self.environment) # Reserve the slot before fork. The parent never waits for an exec-error # pipe or performs child cwd/executable/interpreter filesystem access. number = pending.connection.fileno() job = Job(pending.connection) if self.groups is not None: job.group = self.groups.reserve() self.jobs[number] = job parent = os.getpid() try: job.pid = os.fork() except BaseException: job.cancelled = True pending.close_descriptors() # Keep even an unlaunched reservation until bounded retirement. raise if job.pid == 0: try: require(not self.stopping, 'runtime stopped during launch') libc = ctypes.CDLL(None, use_errno=True) require(libc.prctl(1, signal.SIGKILL, 0, 0, 0) == 0 and os.getppid() == parent, 'runtime launch parent changed') os.setsid() if job.group is not None: job.group.enter() copies = [fcntl.fcntl(fd, fcntl.F_DUPFD_CLOEXEC, 20) for fd in pending.descriptors] for target, source in enumerate(copies): os.dup2(source, target, inheritable=True) for item in os.listdir('/proc/self/fd'): fd = int(item) if fd > 2: try: os.close(fd) except OSError: pass for name in ('SIGPIPE', 'SIGXFZ', 'SIGXFSZ'): if hasattr(signal, name): signal.signal(getattr(signal, name), signal.SIG_DFL) signal.pthread_sigmask(signal.SIG_SETMASK, []) os.chdir(cwd) os.execvpe(argv[0], argv, environment) except BaseException: os._exit(125) pending.close_descriptors() def serve(self): try: self._serve() finally: self.stopping = True self.listener.close() for pending in self.pending.values(): pending.close_descriptors(); pending.connection.close() self.pending.clear() for job in self.jobs.values(): try: job.cancel() except (OSError, ValueError): pass # Keep ownership below; inability to kill is not retirement. job.connection.close() # Keep ownership/slots until each child is actually reaped, even # after an exception. The parent must observe this via pidfd, not # block its own lifecycle/control loop waiting on a stuck syscall. while self.jobs: for number, job in list(self.jobs.items()): try: if job.settle(tuple(self.jobs.values())): del self.jobs[number] except (OSError, ValueError): pass # No replacement/release over unreadable or stuck groups. if self.jobs: time.sleep(0.02) def _serve(self): # Signal handlers only set state; they never interrupt resource ownership # transfer or reaping with an exception. def stopping(*_): self.stopping = True signal.signal(signal.SIGTERM, stopping) signal.signal(signal.SIGINT, stopping) signal.pthread_sigmask(signal.SIG_UNBLOCK, [signal.SIGTERM, signal.SIGINT, signal.SIGCHLD]) with selectors.DefaultSelector() as poller: poller.register(self.listener, selectors.EVENT_READ) while not self.stopping or self.pending or self.jobs: if self.heartbeat is not None: self.heartbeat() if self.stopping and self.listener.fileno() >= 0: poller.unregister(self.listener); self.listener.close() for key, _ in poller.select(0.02): connection = key.fileobj if connection is self.listener: client, _ = self.listener.accept() client.set_inheritable(False); client.setblocking(False) pid, uid, gid = struct.unpack('3i', client.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)) if pid <= 0 or (uid, gid) != self.manager or len(self.pending) >= MAX_PENDING or len(self.jobs) >= MAX_JOBS: client.close(); continue pending = Pending(client, now() + ADMISSION_NS) self.pending[client.fileno()] = pending poller.register(client, selectors.EVENT_READ) continue number = connection.fileno() if number in self.jobs: # Any extra control data or EOF revokes this connection; # child input/output uses the separately transferred FDs. try: self.jobs[number].cancel() except (OSError, ValueError) as error: self.jobs[number].retirement_error = str(error) poller.unregister(connection) continue pending = self.pending[number] try: request = pending.read() if request is None: continue self.start(pending, request) del self.pending[number] except (OSError, ValueError, UnicodeError): pending.close_descriptors() reply(connection, 2, 125) poller.unregister(connection) if number not in self.jobs: connection.close() # A failed fork retains both reservation and connection # fd until retirement, so accept cannot reuse its job key. del self.pending[number] for number, pending in list(self.pending.items()): if self.stopping or now() >= pending.deadline: pending.close_descriptors() poller.unregister(pending.connection); pending.connection.close() del self.pending[number] jobs = tuple(self.jobs.values()) try: reap_adopted(jobs) except (OSError, ValueError): pass # Final per-job barrier still refuses unreadable state. for number, job in list(self.jobs.items()): if self.stopping: job.cancelled = True if not job.settle(jobs): continue # Stuck work keeps its slot; never replace it. code = job.result() if not job.cancelled: reply(job.connection, 0 if code >= 0 else 1, code if code >= 0 else -code) try: poller.unregister(job.connection) except (KeyError, ValueError): pass job.connection.close(); del self.jobs[number]