384 lines
18 KiB
Python
384 lines
18 KiB
Python
"""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')
|