440 lines
21 KiB
Python
440 lines
21 KiB
Python
#!/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()
|