#!/usr/bin/python3 """Fixed root listener broker + separate USER-unit runtime worker. The broker never executes a request or reads private manager state. Its only grant is a root-created command-listening FD to the configured worker identity. Access/retirement is a required live lease from a separate root enforcement owner; there is no saved Boolean that enables this service. See the integration report. """ import array import ctypes import hashlib import json import os import platform from pathlib import Path import pwd import resource import select import signal import socket import stat import struct import subprocess import sys import tempfile import time import types # -I excludes the script directory. Load only the two fixed sibling SOURCE files # after authenticating their lookup chains; do not add an import/bytecode search # path. When embedded by the native supervisor, its fixed bundle supplies them. if __name__ == '__main__': for name in ('c01_network', 'runtime_owner', 'c01_runtime_groups'): path = Path(__file__).absolute().parent / (name + '.py') for item in [*reversed(path.parents), path]: meta = item.lstat() if (meta.st_uid != 0 or meta.st_mode & 0o022 or stat.S_ISLNK(meta.st_mode) or not (stat.S_ISREG(meta.st_mode) if item == path else stat.S_ISDIR(meta.st_mode))): raise ValueError('runtime module lookup is not root protected') module = types.ModuleType(name) module.__file__ = str(path) sys.modules[name] = module exec(compile(path.read_bytes(), str(path), 'exec'), module.__dict__) import runtime_owner import c01_network from c01_runtime_groups import Groups, current_group require = runtime_owner.require BASE = Path('/run/archipelago-runtime') PUBLIC = Path('/run/archipelago-managed') CONFIG = Path('/etc/archipelago-c01/runtime.json') ACCESS = Path('/run/archipelago-c01/runtime-access.sock') LIMIT = 16384 LIBC = ctypes.CDLL(None, use_errno=True) def exact(pairs): value = {} for key, item in pairs: require(key not in value, 'duplicate runtime configuration key') value[key] = item return value def protected(path, directory=False, socket_file=False): path = Path(path) require(path.is_absolute() and '..' not in path.parts, 'absolute protected path required') for parent in reversed(path.parents): meta = parent.lstat() require(stat.S_ISDIR(meta.st_mode) and meta.st_uid == 0 and not meta.st_mode & 0o022, 'runtime input ancestry is mutable') meta = path.lstat() require(meta.st_uid == 0 and not meta.st_mode & (0o007 if socket_file else 0o022) and not stat.S_ISLNK(meta.st_mode), 'runtime input is mutable') if directory: require(stat.S_ISDIR(meta.st_mode), 'runtime directory required') return meta def ranges(path, name): result = [] for row in Path(path).read_text().splitlines(): fields = row.split(':') if fields[0] == name: require(len(fields) == 3, 'invalid subordinate ID allocation') result.append([int(fields[1]), int(fields[2])]) require(result and all(start > 0 and count > 0 for start, count in result), 'missing subordinate ID allocation') return result def load_config(): meta = protected(CONFIG) require(stat.S_ISREG(meta.st_mode) and meta.st_nlink == 1 and stat.S_IMODE(meta.st_mode) == 0o600 and 0 < meta.st_size <= LIMIT, 'runtime config metadata') fd = os.open(CONFIG, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC) try: fields=('st_dev','st_ino','st_mode','st_uid','st_gid','st_nlink','st_size','st_mtime_ns','st_ctime_ns') stable=lambda actual: all(getattr(actual,k)==getattr(meta,k) for k in fields) require(stable(os.fstat(fd)), 'runtime config replaced') raw = os.read(fd, LIMIT + 1) require(len(raw) == meta.st_size and stable(os.fstat(fd)), 'runtime config changed while reading') finally: os.close(fd) value = json.loads(raw, object_pairs_hook=exact) require(type(value) is dict,'runtime configuration object required') fields = {'version', 'runtime_user', 'runtime_uid', 'runtime_gid', 'manager_uid', 'manager_gid', 'authority_uid', 'authority_gid', 'operator_uid', 'operator_gid', 'home', 'home_backing', 'runtime_dir', 'runtime_data_dir', 'graphroot', 'subuids', 'subgids', 'user_cgroup', 'private_parents'} if value.get('version') == 2: fields.add('workspace_helper_sha256') require(type(value.get('workspace_helper_sha256')) is str and len(value['workspace_helper_sha256'])==64 and all(c in '0123456789abcdef' for c in value['workspace_helper_sha256']), 'runtime workspace digest') require(type(value) is dict and set(value) == fields and type(value['version']) is int and value['version'] in (1,2), 'runtime schema') for kind in ('uid', 'gid'): ids = [value[role + '_' + kind] for role in ('runtime', 'manager', 'authority', 'operator')] require(all(type(n) is int and 0 < n < 2**32 - 1 for n in ids) and len(set(ids)) == 4, 'runtime/manager/authority/operator identities overlap') account = pwd.getpwnam(value['runtime_user']) require((account.pw_uid, account.pw_gid, account.pw_dir) == (value['runtime_uid'], value['runtime_gid'], value['home']), 'original runtime identity/home changed') require(account.pw_shell in ('/usr/sbin/nologin', '/sbin/nologin', '/bin/false') and os.getgrouplist(account.pw_name, account.pw_gid) == [account.pw_gid] and sum(p.pw_uid == account.pw_uid for p in pwd.getpwall()) == 1, 'runtime account access not retired') require(ranges('/etc/subuid', account.pw_name) == value['subuids'] and ranges('/etc/subgid', account.pw_name) == value['subgids'], 'original subuid/subgid allocation changed') for key in ('home', 'home_backing', 'runtime_dir', 'runtime_data_dir', 'graphroot'): path = Path(value[key]) require(path.is_absolute() and '..' not in path.parts and str(path) == value[key], 'noncanonical runtime path') require(value['runtime_dir'] == f'/run/user/{account.pw_uid}', 'original user bus path changed') home = Path(value['home_backing']) protected(home.parent, directory=True) require(home.resolve(strict=True) == home and home.stat().st_uid == account.pw_uid and stat.S_IMODE(home.stat().st_mode) == 0o700, 'sanitized runtime home binding') # Home must be provisioned by explicit allowlist; never import operator home. require(not any((home / name).exists() for name in ('.ssh', '.gnupg', '.pki')), 'operator credential directory in runtime home') require(type(value['private_parents']) is list and len(value['private_parents']) >= 2, 'manager/authority private ancestry missing') for parent in value['private_parents']: require(stat.S_IMODE(protected(parent, directory=True).st_mode) == 0o700, 'private backing ancestry is not root-only') require(not Path(value['runtime_data_dir']).is_relative_to(parent) and not Path(value['graphroot']).is_relative_to(parent), 'private/runtime roots overlap') group = Path(value['user_cgroup']) require(group.is_absolute() and '..' not in group.parts and group.name.endswith('.service'), 'runtime user hierarchy locator') return value, hashlib.sha256(raw).hexdigest() def peer(channel): return struct.unpack('3i', channel.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)) def send(channel, value, fds=()): raw = json.dumps(value, separators=(',', ':')).encode() require(len(raw) <= LIMIT, 'runtime control bound') ancillary = [(socket.SOL_SOCKET, socket.SCM_RIGHTS, array.array('i', fds))] if fds else [] require(channel.sendmsg([raw], ancillary, socket.MSG_DONTWAIT | socket.MSG_NOSIGNAL) == len(raw), 'runtime control delivery uncertain') def receive(channel, count=0): raw, ancillary, flags, _ = channel.recvmsg(LIMIT + 1, socket.CMSG_SPACE(16 * 4), socket.MSG_CMSG_CLOEXEC) rights = [] try: unexpected = False for level, kind, data in ancillary: if (level, kind) == (socket.SOL_SOCKET, socket.SCM_RIGHTS): values = array.array('i'); values.frombytes(data[:len(data) - len(data) % 4]) rights.extend(values) else: unexpected = True require(not unexpected and 0 < len(raw) <= LIMIT and not flags & ~socket.MSG_CMSG_CLOEXEC and len(rights) == count, 'runtime control truncated/closed') value = json.loads(raw, object_pairs_hook=exact) return value, rights except BaseException: for fd in rights: os.close(fd) raise def listen(path, gid, kind): protected(path.parent, directory=True) # Refuse a stale socket. Only systemd RuntimeDirectory retirement or explicit # offline recovery can clear a previous broker's uncertain ownership. channel = socket.socket(socket.AF_UNIX, kind | socket.SOCK_CLOEXEC) channel.bind(str(path)) os.chown(path, 0, gid); path.chmod(0o660) channel.listen(16); channel.setblocking(False) return channel def root_connect(path): meta = protected(path, socket_file=True) require(stat.S_ISSOCK(meta.st_mode), 'root control socket required') channel = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET | socket.SOCK_CLOEXEC) try: channel.settimeout(2) channel.connect(str(path)) pid, uid, gid = peer(channel) require(pid > 1 and (uid, gid) == (0, 0), 'root control peer mismatch') return channel except BaseException: channel.close() raise def access_lease(digest): channel = root_connect(ACCESS) nonce = os.urandom(32).hex() request = {'protocol': 'runtime-access-v1', 'configuration_sha256': digest, 'nonce': nonce} send(channel, request) response, _ = receive(channel) require(response == request, 'live runtime access lease refused') channel.setblocking(False) return channel def notify(message): address = os.environ.get('NOTIFY_SOCKET') if address: with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as channel: channel.connect('\0' + address[1:] if address.startswith('@') else address) channel.send(message) def worker_identity(pid, value): c01_network.require_private(pid, 'runtime') status = dict(row.split(':', 1) for row in Path(f'/proc/{pid}/status').read_text().splitlines() if ':' in row) require(list(map(int, status['Uid'].split())) == [value['runtime_uid']] * 4 and list(map(int, status['Gid'].split())) == [value['runtime_gid']] * 4 and set(map(int, status['Groups'].split())) <= {value['runtime_gid']}, 'worker credentials changed') require(all(int(status[key], 16) == 0 for key in ('CapEff', 'CapPrm', 'CapInh', 'CapAmb')), 'worker capabilities retained') group = current_group(pid) parent = Path('/sys/fs/cgroup') / value['user_cgroup'].lstrip('/') require(group.is_relative_to(parent) and (group.name == 'archipelago-runtime-worker.service' or (group.name == 'owner' and group.parent.name == 'archipelago-runtime-worker.service')), 'worker is outside its dedicated USER service') namespace = os.stat(f'/proc/{pid}/ns/mnt') require(namespace.st_ino != os.stat('/proc/self/ns/mnt').st_ino, 'runtime lacks qualified private home view') home = os.stat(f'/proc/{pid}/root{value["home"]}') backing = Path(value['home_backing']).stat() require((home.st_dev, home.st_ino) == (backing.st_dev, backing.st_ino), 'runtime home view changed') return {'pid': pid, 'uid': value['runtime_uid'], 'gid': value['runtime_gid'], 'cgroup': str(group), 'mount_namespace': namespace.st_ino} def broker(): require(os.getresuid() == (0, 0, 0) and os.getresgid() == (0, 0, 0), 'root broker required') require(platform.machine() in ('x86_64', 'aarch64'), 'runtime mount syscall ABI not qualified') value, digest = load_config() lease = access_lease(digest) # No listener is published without the live owner. grant = listen(BASE / 'grant.sock', value['runtime_gid'], socket.SOCK_SEQPACKET) listener = listen(PUBLIC / 'runtime.sock', value['manager_gid'], socket.SOCK_STREAM) readiness = listen(PUBLIC / 'readiness.sock', value['manager_gid'], socket.SOCK_SEQPACKET) listener_path = os.open(PUBLIC / 'runtime.sock', os.O_PATH | os.O_NOFOLLOW | os.O_CLOEXEC) stopping = False def stop(*_): nonlocal stopping stopping = True signal.signal(signal.SIGTERM, stop); signal.signal(signal.SIGINT, stop) worker = None worker_pidfd = None watchers = set() last = 0 challenge = None challenged = 0 identity = None notified = False renewed = 0 try: while not stopping: inputs = [lease, grant, readiness, *watchers] if worker is not None: inputs += [worker, worker_pidfd] ready, _, _ = select.select(inputs, [], [], .05) require(lease not in ready, 'runtime access lease revoked') require(worker_pidfd is None or worker_pidfd not in ready, 'runtime worker exited') stamp = runtime_owner.now() now = stamp / 1_000_000_000 if worker is not None and challenge is None: challenge = os.urandom(16).hex(); challenged = now send(worker, {'ping': challenge}) if worker is not None and worker in ready: packet, _ = receive(worker) require(packet == {'pong': challenge}, 'runtime heartbeat mismatch') identity = worker_identity(peer(worker)[0], value) last = now; challenge = None require(worker is None or now - challenged < 3, 'runtime heartbeat expired') if grant in ready: connection, _ = grant.accept(); connection.setblocking(False) pid, uid, gid = peer(connection) if worker is not None or (uid, gid) != (value['runtime_uid'], value['runtime_gid']): connection.close() else: worker_identity(pid, value) worker_pidfd = os.pidfd_open(pid) send(connection, value, [listener.fileno()]) worker = connection; challenged = now # Startup probes have a separate finite allowance. A worker # sends no pong until actual bus/tmux/Podman probes finish. challenge = os.urandom(16).hex() send(worker, {'ping': challenge}) challenged = now + 30 for connection in list(watchers): if connection in ready: connection.close(); watchers.remove(connection) if last and now - last < .5 and now - renewed >= .5: for connection in list(watchers): try: send(connection, {'alive_until_boottime_ns': stamp + 3_000_000_000}) except (OSError, ValueError): connection.close(); watchers.remove(connection) renewed = now if readiness in ready: connection, _ = readiness.accept(); connection.setblocking(False) pid, uid, gid = peer(connection) if ((uid, gid) not in ((0, 0), (value['manager_uid'], value['manager_gid'])) or last == 0 or now - last >= 3 or len(watchers) >= 16): connection.close(); continue # Live lease: socket EOF/revocation MUST stop manager admission. meta = os.fstat(listener_path) named = (PUBLIC / 'runtime.sock').lstat() require((meta.st_dev, meta.st_ino) == (named.st_dev, named.st_ino), 'runtime socket replaced') tree = None try: # Clone in the namespace that owns the source mount. A # receiver cannot open_tree(CLONE) a foreign O_PATH mount # under the production broker/supervisor sandbox split. tree = LIBC.syscall(428, listener_path, ctypes.c_char_p(b''), 1 | os.O_CLOEXEC | 0x1000) if tree < 0: raise OSError(ctypes.get_errno(), 'runtime detached mount grant unavailable') cloned = os.fstat(tree) require((cloned.st_dev, cloned.st_ino, cloned.st_mode, cloned.st_uid, cloned.st_gid) == (meta.st_dev, meta.st_ino, meta.st_mode, meta.st_uid, meta.st_gid), 'runtime detached grant identity changed') send(connection, {'version': 2, 'configuration_sha256': digest, **identity, 'listener_device': meta.st_dev, 'listener_inode': meta.st_ino, 'manager_uid': value['manager_uid'], 'manager_gid': value['manager_gid'], 'alive_until_boottime_ns': stamp + 3_000_000_000}, [listener_path, worker_pidfd, tree]) except (OSError, ValueError): connection.close() else: watchers.add(connection) finally: if tree is not None and tree >= 0: os.close(tree) if last and not notified: notify(b'READY=1\nSTATUS=Runtime worker identity and live lease observed') notified = True finally: for channel in (lease, grant, listener, readiness, worker, *watchers): if channel is not None: channel.close() if worker_pidfd is not None: os.close(worker_pidfd) os.close(listener_path) def probe(argv, environment): # Runs ONLY as the permanently unprivileged worker, never the root broker. # File output avoids waiting for EOF held by a rootless pause descendant; # the size limit bounds diagnostics without a pipe-reader thread. with tempfile.TemporaryFile() as output: subprocess.run(argv, stdin=subprocess.DEVNULL, stdout=output, stderr=subprocess.DEVNULL, env=environment, timeout=15, check=True, preexec_fn=lambda: resource.setrlimit(resource.RLIMIT_FSIZE, (LIMIT, LIMIT))) output.seek(0) raw = output.read(LIMIT + 1) require(len(raw) <= LIMIT, 'runtime probe output bound') return raw.decode().strip() def worker(): require(os.geteuid() > 0, 'runtime worker cannot run as root') deadline = time.monotonic() + 30 while True: try: channel = root_connect(BASE / 'grant.sock') break except (FileNotFoundError, ConnectionRefusedError): require(time.monotonic() < deadline, 'root runtime broker unavailable') time.sleep(.05) value, rights = receive(channel, 1) listener = socket.socket(fileno=rights[0]) groups = Groups() channel.setblocking(False) last_ping = runtime_owner.now() def heartbeat(): nonlocal last_ping if select.select([channel], [], [], 0)[0]: packet, _ = receive(channel) require(type(packet) is dict and set(packet) == {'ping'} and type(packet['ping']) is str and len(packet['ping']) == 32, 'root heartbeat refused') send(channel, {'pong': packet['ping']}) last_ping = runtime_owner.now() require(runtime_owner.now() - last_ping < 3_000_000_000, 'root runtime heartbeat expired') owner = runtime_owner.RuntimeOwner(listener, value['manager_uid'], value['manager_gid'], value['home'], value['runtime_dir'], value['runtime_user'], groups=groups, heartbeat=heartbeat) environment = {**owner.environment, 'LANG': 'C.UTF-8'} if value['version']==2: helper=Path('/usr/libexec/archipelago-d09-prepare-workspace') protected(helper) require(hashlib.sha256(helper.read_bytes()).hexdigest()==value['workspace_helper_sha256'], 'runtime workspace helper changed') probe(['/bin/bash',str(helper),value['home']],environment) require(probe(['/usr/bin/systemctl', '--user', 'is-active', 'archipelago-runtime-tmux.service'], environment) == 'active', 'persistent tmux service unavailable') # Type=exec means exec succeeded, not that tmux has bound its socket yet. deadline = time.monotonic() + 5 while True: try: probe(['/usr/bin/tmux', 'show-options', '-s'], environment) break except subprocess.CalledProcessError: require(time.monotonic() < deadline, 'persistent tmux socket unavailable') time.sleep(.05) require(probe(['/usr/bin/podman', 'info', '--format', '{{.Store.GraphRoot}}'], environment) == value['graphroot'], 'original Podman graphroot differs') notify(b'READY=1\nSTATUS=Runtime bus, persistent tmux and rootless Podman probes passed') last_ping = runtime_owner.now() try: owner.serve() finally: channel.close() if __name__ == '__main__': require(len(sys.argv) == 2 and sys.argv[1] in ('broker', 'worker'), 'fixed runtime role required') if sys.argv[1] == 'broker': broker() else: worker()