"""Root C01 child ownership: bounded cgroups, pidfds and nonblocking retirement. Requires a systemd delegated unit with DelegateSubgroup=supervisor. Only this unit's newly created fixed role subgroups are managed. Application containers belong to a separate runtime service and must never be placed in these groups. """ import ctypes from dataclasses import dataclass import os from pathlib import Path import signal import socket import stat LIBC = ctypes.CDLL(None, use_errno=True) ROLES = {'application': (2 * 1024**3, 256), 'authority': (128 * 1024**2, 16), 'host-io': (64 * 1024**2, 8), 'administrator': (128 * 1024**2, 8), 'collector': (128 * 1024**2, 8), 'tor-owner': (128 * 1024**2, 16), 'admission': (64 * 1024**2, 8)} STOP_SIGNALS = {signal.SIGTERM, signal.SIGINT} def require(condition, message): if not condition: raise ValueError(message) def process_group_path(pid): lines = Path(f'/proc/{pid}/cgroup').read_text().splitlines() require(len(lines) == 1 and lines[0].startswith('0::/'), 'unified process cgroup required') return lines[0][3:] def drop_credentials(uid, gid, parent): require(type(uid) is int and type(gid) is int and 0 < uid < 2**32 - 1 and 0 < gid < 2**32 - 1, 'dedicated nonroot credentials required') require(LIBC.prctl(38, 1, 0, 0, 0) == 0, 'C01 no-new-privileges unavailable') os.setgroups([]) os.setresgid(gid, gid, gid) os.setresuid(uid, uid, uid) # systemd may retain inheritable/ambient setup capabilities (and keep-caps). # UID change alone is not the empty-capability contract the native roles # verify. Drop every capability set explicitly before any native exec. class Header(ctypes.Structure): _fields_=[('version',ctypes.c_uint32),('pid',ctypes.c_int)] class Caps(ctypes.Structure): _fields_=[('effective',ctypes.c_uint32),('permitted',ctypes.c_uint32),('inheritable',ctypes.c_uint32)] header=Header(0x20080522,0);caps=(Caps*2)() require(LIBC.capset(ctypes.byref(header),caps)==0 and LIBC.prctl(47,4,0,0,0)==0, 'role capability drop failed') require(LIBC.capget(ctypes.byref(header),caps)==0 and all(not (c.effective|c.permitted|c.inheritable) for c in caps),'role retained capabilities') # Credential changes clear PDEATHSIG. Restore/check it before child work. require(LIBC.prctl(4, 0, 0, 0, 0) == 0 and LIBC.prctl(1, signal.SIGKILL, 0, 0, 0) == 0 and os.getppid() == parent, 'C01 dropped child lifetime failed') class RoleGroup: def __init__(self, root, role): require(role in ROLES, 'unknown protected role') self.path = root / ('c01-' + role) self.path.mkdir(mode=0o700) # Never adopt a preexisting group or its tasks. self.kernel_path = '/' + str(self.path.relative_to('/sys/fs/cgroup')) self.kill_fd, self.events_fd = None, None self.removed = False memory, tasks = ROLES[role] try: (self.path / 'memory.max').write_text(str(memory)) (self.path / 'memory.swap.max').write_text('0') (self.path / 'pids.max').write_text(str(tasks)) (self.path / 'cpu.max').write_text('100000 100000') require((self.path / 'cgroup.kill').is_file(), 'cgroup kill unavailable') # Reserve termination/observation capacity before admitting a child. # EMFILE later must not prevent revoking its entire descendant tree. self.kill_fd = os.open(self.path / 'cgroup.kill', os.O_WRONLY | os.O_CLOEXEC | os.O_NOFOLLOW) self.events_fd = os.open(self.path / 'cgroup.events', os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW) except BaseException: for fd in (self.kill_fd, self.events_fd): if fd is not None: os.close(fd) self.path.rmdir() raise def attach(self, pid): (self.path / 'cgroup.procs').write_text(str(pid)) require(process_group_path(pid) == self.kernel_path, 'child cgroup admission failed') def kill(self): # Kills descendants even after setsid/double-fork. The path is a newly # owned, root-only kernel cgroup, never an application-selected filename. require(not self.removed and self.kill_fd is not None, 'role termination capability retired') os.lseek(self.kill_fd, 0, os.SEEK_SET) require(os.write(self.kill_fd, b'1') == 1, 'role termination write incomplete') def empty(self): if self.removed: return True os.lseek(self.events_fd, 0, os.SEEK_SET) raw = os.read(self.events_fd, 4096) require(len(raw) < 4096, 'role cgroup observation oversized') values = dict(line.split() for line in raw.decode().splitlines()) return values['populated'] == '0' def close(self): if not self.removed: require(self.empty(), 'cannot retire populated role cgroup') self.path.rmdir() self.removed = True for field in ('kill_fd', 'events_fd'): fd = getattr(self, field) if fd is not None: # Linux close releases the descriptor even on an interrupted # close. Never retry an old descriptor number after release. setattr(self, field, None) os.close(fd) @dataclass class Child: role: str group: RoleGroup channel: socket.socket pid: int = None pidfd: int = None status: object = None reaped: bool = False exit_code: int = None class Children: def __init__(self): require(os.getresuid() == (0, 0, 0), 'C01 child owner requires root') require(len(list(Path('/proc/self/task').iterdir())) == 1, 'C01 child owner must be single-threaded') path = Path('/sys/fs/cgroup') / process_group_path(os.getpid()).lstrip('/') require(path.name == 'supervisor', 'systemd DelegateSubgroup=supervisor required') self.root = path.parent metadata = self.root.stat() require(stat.S_ISDIR(metadata.st_mode) and metadata.st_uid == 0 and not metadata.st_mode & 0o022, 'unprotected delegated cgroup') controllers = set((self.root / 'cgroup.subtree_control').read_text().split()) if not {'memory', 'pids', 'cpu'} <= controllers: available = set((self.root / 'cgroup.controllers').read_text().split()) require({'memory', 'pids', 'cpu'} <= available and (self.root / 'cgroup.type').read_text().strip() == 'domain' and not (self.root / 'cgroup.procs').read_text().strip(), 'delegated unit root must be an empty domain') (self.root / 'cgroup.subtree_control').write_text('+memory +pids +cpu') controllers = set((self.root / 'cgroup.subtree_control').read_text().split()) require({'memory', 'pids', 'cpu'} <= controllers, 'required delegated controllers absent') self.children, self.results, self.reserved = {}, {}, {} self.stopping = False self.prior_handlers = {value: signal.getsignal(value) for value in (*STOP_SIGNALS, signal.SIGCHLD)} self.prior_subreaper = ctypes.c_int() require(LIBC.prctl(37, ctypes.byref(self.prior_subreaper), 0, 0, 0) == 0 and LIBC.prctl(36, 1, 0, 0, 0) == 0, 'C01 subreaper unavailable') for value in STOP_SIGNALS: signal.signal(value, self.request_stop) signal.signal(signal.SIGCHLD, signal.SIG_DFL) def request_stop(self, *_): # No exception can interrupt fork registration or bookkeeping/reaping. self.stopping = True def reserve(self, role): """Prestage an empty role and all fork/control FDs before fresh NTS. A reservation owns no PID. The same owner retires it on setup failure; spawn consumes it once and otherwise retains its existing semantics. """ require(not self.stopping and role not in self.children and role not in self.results and role not in self.reserved, 'role already owned/retired or supervisor stopping') group = RoleGroup(self.root, role) channels = [] try: parent, child = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) channels.extend((parent, child)) parent.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1) # Grants may be queued before the child reaches its native entry. # Capture the real sender credentials at enqueue time on both ends. if role == 'tor-owner': child.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1) gate_parent, gate_child = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) channels.extend((gate_parent, gate_child)) except BaseException: for channel in channels: channel.close() group.close() raise self.reserved[role] = (group, parent, child, gate_parent, gate_child) def spawn(self, role, body, keep=()): require(not self.stopping and role not in self.children and role not in self.results, 'role already owned/retired or supervisor stopping') if role not in self.reserved: self.reserve(role) group, parent, child, gate_parent, gate_child = self.reserved.pop(role) record = Child(role, group, parent) self.children[role] = record expected_parent = os.getpid() previous_mask = signal.pthread_sigmask(signal.SIG_BLOCK, STOP_SIGNALS) try: pid = os.fork() if pid == 0: try: require(LIBC.prctl(1, signal.SIGKILL, 0, 0, 0) == 0 and os.getppid() == expected_parent, 'C01 child parent changed') signal.set_wakeup_fd(-1) for value in STOP_SIGNALS: signal.signal(value, signal.SIG_DFL) retained = {child.fileno(), gate_child.fileno(), *keep} for name in os.listdir('/proc/self/fd'): fd = int(name) if fd > 2 and fd not in retained: try: os.close(fd) except OSError: pass signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask) gate_child.settimeout(10) require(gate_child.recv(4) == b'RUN', 'C01 child registration barrier refused') gate_child.close() body(child) os._exit(0) except BaseException as error: # Never print exception text/arguments: setup can hold keys. # Source locations and errno let operators diagnose the # first real refusal without logging private material. try: import traceback locations=','.join(Path(frame.filename).name+':'+str(frame.lineno) for frame in traceback.extract_tb(error.__traceback__)) os.write(2,('C01_CHILD_SETUP_REFUSED role='+role+' type='+type(error).__name__+ ' errno='+str(getattr(error,'errno',None))+' locations='+locations+'\n').encode()) except BaseException:pass os._exit(111) # Register the still-unreaped direct PID before any fallible pidfd or # cgroup admission. Failure retains it for the same owner's cleanup. record.pid = pid child.close(); gate_child.close() record.pidfd = os.pidfd_open(pid) group.attach(pid) require(gate_parent.send(b'RUN') == 3, 'C01 child start barrier failed') return record except BaseException as error: self.stopping = True if record.pid is None: parent.close(); group.close(); del self.children[role] else: cleanup = self._kill(record) if cleanup: error.add_note('C01 child retirement pending: ' + ', '.join(cleanup)) raise finally: child.close(); gate_child.close(); gate_parent.close() signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask) def rearm_collector(self): """Reuse only a successfully reaped/empty/removed collector slot. _finish records success only AFTER pidfd/cgroup/adopted-child retirement. This re-arms resource ownership, never an old time authorization or PID. """ require(not self.stopping and 'collector' not in self.children and 'collector' not in self.reserved and self.results.get('collector') == 0, 'collector lifetime not successfully retired') del self.results['collector'] @staticmethod def _kill(record): errors = [] try: record.group.kill() except (OSError, ValueError) as error: errors.append('group:' + type(error).__name__) if record.reaped: return errors try: if record.pidfd is not None: signal.pidfd_send_signal(record.pidfd, signal.SIGKILL) elif record.pid is not None: # pidfd creation failed, but this direct PID is still unreaped # and cannot have been reused. Never signal after reaping it. os.kill(record.pid, signal.SIGKILL) except ProcessLookupError: pass except OSError as error: errors.append('leader:' + type(error).__name__) return errors def _reap_adopted(self, record, final=False): if self.stopping and not final: return False leaders = {item.pid for item in self.children.values() if not item.reaped} # Inspect one snapshot, bounded by this role's enforced pids.max. Looking # at only an arbitrary prefix would starve zombies behind live children. # A producer cannot extend this snapshot by replenishing during the scan. descendants = [int(text) for text in Path(f'/proc/self/task/{os.getpid()}/children').read_text().split() if int(text) not in leaders and process_group_path(int(text)) == record.group.kernel_path] if not descendants: return True require(len(descendants) < ROLES[record.role][1], 'role exceeded its enforced task budget') for pid in descendants: if self.stopping and not final: break status = os.waitid(os.P_PID, pid, os.WEXITED | os.WNOHANG | os.WNOWAIT) if status is not None: os.waitpid(pid, 0) # Even a fully reaped snapshot requires a fresh later observation. The # final post-empty scan is finite too and retains the leader until clean. return False def _finish(self, role, record): require(record.reaped, 'cannot finish an unreaped role') if record.pidfd is not None: fd, record.pidfd = record.pidfd, None os.close(fd) record.channel.close() record.group.close() # Keep the cleanup record until every owned resource is retired. A retry # of this branch never signals or waits on the already released PID. self.results[role] = record.exit_code del self.children[role] def poll(self): """Retire without blocking; keep every unreaped role/slot registered.""" errors = [] if self.stopping: for role, (group, *channels) in list(self.reserved.items()): try: for channel in channels: channel.close() group.close() del self.reserved[role] except (OSError, ValueError) as error: errors.append((role, type(error).__name__)) for role, record in list(self.children.items()): try: if record.reaped: self._finish(role, record) continue if self.stopping: errors.extend((role, error) for error in self._kill(record)) # Completed orphans consume pids.max until reaped. Reclaim them # while the leader is still alive, not only during shutdown. try: self._reap_adopted(record) except (OSError, ValueError) as error: self.stopping = True errors.append((role, type(error).__name__)) if record.status is None: record.status = os.waitid(os.P_PID, record.pid, os.WEXITED | os.WNOHANG | os.WNOWAIT) if record.status is None: continue errors.extend((role, error) for error in self._kill(record)) if not record.group.empty(): continue # Check adoption AFTER the cgroup becomes empty, not from an # earlier snapshot while a last descendant was still exiting. # Re-read after each batch so double-fork generations settle too. if not self._reap_adopted(record, final=True): continue done, status = os.waitpid(record.pid, os.WNOHANG) require(done == record.pid, 'C01 leader reap ownership changed') record.reaped = True record.pid = None # WNOWAIT above pins the exited leader until this sole reaper. record.exit_code = os.waitstatus_to_exitcode(status) self._finish(role, record) except (OSError, ValueError) as error: self.stopping = True errors.append((role, type(error).__name__)) if errors: self.stopping = True return errors def close(self): require(not self.children and not self.reserved, 'unreaped/unretired C01 roles remain owned') require(not Path(f'/proc/self/task/{os.getpid()}/children').read_text().strip(), 'unreaped adopted C01 descendants remain') for value, handler in self.prior_handlers.items(): signal.signal(value, handler) require(LIBC.prctl(36, self.prior_subreaper.value, 0, 0, 0) == 0, 'restore subreaper state failed')