449 lines
20 KiB
Python
449 lines
20 KiB
Python
"""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]
|