"""Embedded C01 orchestration; native /proc/self/exe remains the service MainPID. Production admission requires live exclusion/runtime/route owners and a measured conditional asset/service contract. There is no persisted 'migration done' bit, script selector, test environment bypass, automatic provisioning, or replacement of an unreaped role. The dependency-ready launch engine below is also exercised in disposable tests. """ import array import fcntl import hashlib import json import os from pathlib import Path import pwd import select import signal import socket import stat import struct import time import c01_children as lifecycle import c01_host_watch as host_watch import c01_network_settings_owner as network_settings import c01_network_settings_watch as network_settings_watch import c01_tor import c01_radio import c01_nts as nts import c01_recovery as recovery import c01_views as views import c01_runtime as runtime_service import c01_runtime_view import c01_access_client import c01_network import c01_network_client import c01_network_owner import c01_d09 import c01_power import default_login_migration # This separate role has its own fixed cgroup budget; HostOwner's limits and # two-second operation contract are not repurposed for NM activation. lifecycle.ROLES['network-settings'] = (128 * 1024**2, 16) CONFIG = Path('/etc/archipelago-c01/launch.json') EPOCH_CONFIG = Path('/etc/archipelago-c01/epoch.json') LOCK = Path('/var/lib/archipelago-c01/launch.lock') SETUP_NS = 120_000_000_000 BASE = Path('/run/archipelago-c01') LIBC = lifecycle.LIBC require = lifecycle.require now = host_watch.now class AdmissionUnavailable(ValueError): pass def close_except(keep): for name in os.listdir('/proc/self/fd'): fd = int(name) if fd > 2 and fd not in keep: try: os.close(fd) except OSError: pass def root_ancestry(path): path = Path(path) require(path.is_absolute() and '..' not in path.parts, 'absolute protected path required') for parent in path.parents: meta = parent.lstat() require(stat.S_ISDIR(meta.st_mode) and meta.st_uid == 0 and not meta.st_mode & 0o022, 'unprotected input ancestry') def root_capability(path, size=None): """Validate a storage-key FD WITHOUT reading a byte into the parent heap.""" root_ancestry(path) fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC | os.O_NONBLOCK) try: meta = os.fstat(fd) require(stat.S_ISREG(meta.st_mode) and meta.st_uid == 0 and meta.st_nlink == 1 and stat.S_IMODE(meta.st_mode) in (0o400, 0o600) and (size is None or meta.st_size == size), 'invalid root capability') return fd except BaseException: os.close(fd) raise def key_binding(fd): meta = os.fstat(fd) require(os.lseek(fd, 0, os.SEEK_CUR) == 0, 'parent consumed a storage key') return {'device': meta.st_dev, 'inode': meta.st_ino} def exact_object(pairs): value = {} for key, item in pairs: require(key not in value, 'duplicate supervisor input') value[key] = item return value def root_json(path): fd = root_capability(path) try: before = os.fstat(fd) require(0 < before.st_size <= 16384, 'oversized supervisor input') raw = os.read(fd, 16385) after = os.fstat(fd) fields = ('st_dev', 'st_ino', 'st_uid', 'st_gid', 'st_mode', 'st_nlink', 'st_size', 'st_mtime_ns', 'st_ctime_ns') require(len(raw) == before.st_size and all(getattr(before, k) == getattr(after, k) for k in fields), 'supervisor input changed') return json.loads(raw, object_pairs_hook=exact_object), hashlib.sha256(raw).hexdigest() finally: os.close(fd) class LaunchLock: """Never unlink: one root-owned inode across failures and service restarts.""" def __init__(self, path=LOCK): root_ancestry(path) self.fd = os.open(path, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_CLOEXEC, 0o600) try: meta = os.fstat(self.fd) require(stat.S_ISREG(meta.st_mode) and meta.st_uid == 0 and stat.S_IMODE(meta.st_mode) == 0o600 and meta.st_nlink == 1, 'unprotected launch lock') fcntl.flock(self.fd, fcntl.LOCK_EX | fcntl.LOCK_NB) named = os.stat(path, follow_symlinks=False) require((meta.st_dev, meta.st_ino) == (named.st_dev, named.st_ino), 'launch lock replaced') except BaseException: os.close(self.fd) raise def close(self): fd, self.fd = self.fd, None if fd is not None: os.close(fd) def bind_group(self, root): """A crash releases flock, not the obligation to inspect retained tasks. Persist only a kernel ownership locator (never an exclusion attestation). A same-boot previous role group blocks even if empty: only its original reaper or explicit offline recovery may remove that uncertainty. """ os.lseek(self.fd, 0, os.SEEK_SET) raw = os.read(self.fd, 4097) require(len(raw) <= 4096, 'launch ownership locator oversized') boot = recovery.boot_id() if raw: previous = json.loads(raw, object_pairs_hook=exact_object) require(set(previous) == {'boot', 'group'}, 'invalid launch ownership locator') if previous['boot'] == boot: prior = Path(previous['group']) require(prior.is_relative_to('/sys/fs/cgroup') and '..' not in prior.parts, 'invalid prior cgroup locator') require(not any((prior / ('c01-' + role)).exists() for role in lifecycle.ROLES), 'prior protected role retirement remains unresolved') for directory in (prior, prior / 'supervisor'): if directory.exists(): tasks = (directory / 'cgroup.procs').read_text().split() require(all(int(pid) == os.getpid() for pid in tasks), 'prior supervisor still active') require(not any((root / ('c01-' + role)).exists() for role in lifecycle.ROLES), 'delegated role ownership predates this launch') raw = json.dumps({'boot': boot, 'group': str(root)}, separators=(',', ':')).encode() os.lseek(self.fd, 0, os.SEEK_SET) require(os.write(self.fd, raw) == len(raw), 'launch ownership locator write failed') os.ftruncate(self.fd, len(raw)) os.fsync(self.fd) def principal(name): account = pwd.getpwnam(name) require(account.pw_uid > 0 and account.pw_gid > 0 and account.pw_shell in ('/usr/sbin/nologin', '/sbin/nologin', '/bin/false') and os.getgrouplist(name, account.pw_gid) == [account.pw_gid], 'managed principal must be non-login and have no supplementary access') require(len([p for p in pwd.getpwall() if p.pw_uid == account.pw_uid]) == 1, 'managed UID has another account alias') for path in Path('/proc').iterdir(): if not path.name.isdecimal(): continue try: status = dict(line.split(':', 1) for line in (path / 'status').read_text().splitlines() if ':' in line) except FileNotFoundError: continue require(account.pw_uid not in map(int, status['Uid'].split()), 'managed principal already active') return account def private_identity(directory, uid): """Descriptor-relative no-follow checks; no secret content enters Python.""" identity = os.open('identity', os.O_PATH | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC, dir_fd=directory) try: meta = os.fstat(identity) require(meta.st_uid == uid and stat.S_IMODE(meta.st_mode) == 0o700, 'private identity directory owner/mode mismatch') secret = os.open('nostr_secret', os.O_PATH | os.O_NOFOLLOW | os.O_CLOEXEC, dir_fd=identity) try: meta = os.fstat(secret) require(stat.S_ISREG(meta.st_mode) and meta.st_uid == uid and meta.st_nlink == 1 and stat.S_IMODE(meta.st_mode) == 0o600 and 0 < meta.st_size <= 256, 'private identity secret/alias mismatch') return meta.st_dev, meta.st_ino finally: os.close(secret) finally: os.close(identity) def load_configuration(): value, _ = root_json(CONFIG) fields = {'version', 'application_user', 'authority_user', 'manager_backing', 'snapshot_backing', 'authority_backing', 'runtime_data_dir', 'runtime_uid', 'runtime_gid', 'snapshot_key', 'authority_key', 'binding', 'platform_profile', 'platform_profile_sha256', 'clock_domain', 'host_ip', 'proxy_path', 'native_sha256'} if value.get('version') in (4,5): fields.update(('epoch_sha256','power_sha256')) if value.get('version') == 5:fields.add('admission_sha256') require(type(value) is dict and set(value) == fields and type(value['version']) is int and value['version'] in (3, 4, 5), 'unsupported supervisor configuration') recovery.binding_bytes(value['binding']) require(type(value['clock_domain']) is str and len(value['clock_domain']) == 32 and all(c in '0123456789abcdef' for c in value['clock_domain']) and value['clock_domain'] != '0' * 32, 'qualified clock domain required') app, authority = principal(value['application_user']), principal(value['authority_user']) require(len({app.pw_uid, authority.pw_uid, value['runtime_uid'], 0}) == 4 and len({app.pw_gid, authority.pw_gid, value['runtime_gid'], 0}) == 4, 'manager/runtime/authority credentials overlap') profile, digest = root_json(value['platform_profile']) require(digest == value['platform_profile_sha256'], 'platform qualification digest mismatch') qualification = recovery.PlatformQualification(**profile) # Structural/profile byte binding precedes owner acquisition. Timing # qualification remains mandatory immediately before admission, but must # not stand in for measuring the independent exclusion/runtime/route graph. return value, app, authority, qualification def load_epoch_configuration(value): require(value['version'] in (4,5), 'epoch-bound launch configuration required') epoch, digest = root_json(EPOCH_CONFIG) require(digest == value['epoch_sha256'] and set(epoch) == {'version','backing','key','delegation'} and type(epoch['version']) is int and epoch['version'] == 1, 'epoch configuration binding') require(Path(epoch['backing']).parent == Path(value['authority_backing']).parent and epoch['backing'] != value['authority_backing'], 'epoch must share exclusive authority custody') delegation = epoch['delegation'] require({'id','allow_wake','max_awake'} <= set(delegation) <= {'id','allow_wake','max_awake','characterized','allow_refresh'}, 'epoch delegation fields') recovery.EpochIssuerDelegation(bytes(delegation['id']), delegation['allow_wake'], delegation.get('allow_refresh', False)).validate() if delegation.get('characterized') is not None: recovery.validate_characterized_model(delegation['characterized']) maximum = delegation['max_awake'] require(set(maximum) == {'secs','nanos'} and type(maximum['secs']) is int and 0 <= maximum['secs'] < 2**64 and type(maximum['nanos']) is int and 0 <= maximum['nanos'] < 1_000_000_000, 'epoch awake duration encoding') require(maximum['secs'] or maximum['nanos'], 'epoch awake appointment missing') return epoch def admit(value, app, authority, owner=None): """Require live exclusion/runtime/routes and a measured conditional contract. Account flags and persisted readiness booleans cannot replace these owners. """ if owner is None or not isinstance(owner.access_lease, c01_access_client.AccessLease) or not isinstance( owner.runtime_lease, c01_runtime_view.RuntimeLease): raise AdmissionUnavailable('C01_ADMISSION_UNAVAILABLE: live access/runtime owners required') owner.access_lease.check() owner.runtime_lease.check() require(isinstance(owner.network_lease, c01_network_client.NetworkLease), 'live network owner required') owner.network_lease.check() c01_network.require_private(os.getpid(), 'managed') import c01_admission require(value['version']==5 and isinstance(value.get('admission_sha256'),str) and len(value['admission_sha256'])==64 and all(c in '0123456789abcdef' for c in value['admission_sha256']), 'protected conditional admission appointment required') contract,contract_digest=root_json(c01_admission.POLICIES/(value['admission_sha256']+'.json')) require(contract_digest==value['admission_sha256'],'admission policy bytes') c01_admission.validate(contract,value) owner.required_routes=contract['required_routes'] policy=c01_network_owner.Policy() try: require(policy.value['version'] in (2,3,4),'owned runtime egress required') management=policy.routes.get('management') require(management is not None and management['source']=='host' and management['target']=='managed' and management['port']==5678 and management['target_uid']==app.pw_uid, 'required management route appointment') require(all(name in policy.routes for name in owner.required_routes),'required route absent from policy') prestart_routes=[name for name in owner.required_routes if policy.routes[name]['target']!='managed'] policy.check() finally:policy.close() owner.network_lease.require_targets(prestart_routes) # A separate, bounded root role measures the policy/assets and actual # service dispositions. Its authenticated progress is held through READY # and all normal/clock-only operation. No stored flag grants admission. owner.admission_hold=c01_admission.Hold(owner,value,close_except,packet) owner.admission_hold.check() print('C01_CONDITIONAL_ADMISSION_HELD policy='+value['admission_sha256'],flush=True) def acquire_admission_leases(owner, value, app, authority): """Acquire and retain real capabilities before considering native launch. Partial acquisition is owned by Supervisor.retire, including all failure paths; never return a detached success boolean or discard a live lease. """ require(owner.access_lease is None and owner.runtime_lease is None and owner.network_lease is None, 'admission capability already acquired') c01_network.require_private(os.getpid(), 'managed') same_value, launch_digest = root_json(CONFIG) require(same_value == value, 'launch configuration changed during admission') owner.access_lease = c01_access_client.AccessLease(launch_digest) runtime, runtime_digest = runtime_service.load_config() require((runtime['manager_uid'], runtime['manager_gid']) == (app.pw_uid, app.pw_gid) and (runtime['authority_uid'], runtime['authority_gid']) == (authority.pw_uid, authority.pw_gid) and (runtime['runtime_uid'], runtime['runtime_gid']) == (value['runtime_uid'], value['runtime_gid']) and runtime['runtime_data_dir'] == value['runtime_data_dir'], 'runtime service differs from admitted principals/data root') require({str(Path(value[name + '_backing']).parent) for name in ('manager', 'snapshot', 'authority')} <= set(runtime['private_parents']), 'runtime view omits private backing exclusions') owner.runtime_parameters = (app.pw_uid, app.pw_gid, value['runtime_uid'], value['runtime_gid'], runtime_digest) owner.runtime_workspace = runtime_workspace_environment(runtime) owner.runtime_lease = c01_runtime_view.RuntimeLease(*owner.runtime_parameters) owner.network_lease = c01_network_client.NetworkLease(app.pw_uid, value['runtime_uid']) require(root_json(CONFIG)[0] == value and runtime_service.load_config() == (runtime, runtime_digest), 'configuration changed while acquiring lifetime capabilities') owner.access_lease.check() owner.runtime_lease.check() def peer(channel): return struct.unpack('3i', channel.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)) def packet(channel, payload): if not isinstance(payload, bytes): payload = json.dumps(payload, separators=(',', ':')).encode() require(0 < len(payload) <= 8192, 'supervisor packet bound') require(channel.send(payload, socket.MSG_DONTWAIT | socket.MSG_NOSIGNAL) == len(payload), 'supervisor packet send uncertain') def child_wait(channel, expected, deadline): while now() < deadline: ready, _, _ = select.select([channel], [], [], 0.05) if ready: raw, ancillary, flags, _ = channel.recvmsg(8193, 0, socket.MSG_DONTWAIT) require(raw == expected and not ancillary and not flags and now() < deadline, 'supervisor setup barrier refused') return raise ValueError('supervisor setup deadline expired') def native(executable, role, descriptors, environment): copies = {slot: fcntl.fcntl(fd, fcntl.F_DUPFD_CLOEXEC, 32) for slot, fd in descriptors.items()} executable = fcntl.fcntl(executable, fcntl.F_DUPFD_CLOEXEC, 32) for slot, fd in copies.items(): os.dup2(fd, slot, inheritable=True) close_except(set(copies) | {executable}) os.execve(executable, ['archipelago', role], environment) def collect_and_authorize(control, qualification, binding, domain, authority_peer, address, deadline, epoch=None, context=None): """Keyless root role, including all blocking collector/helper work. MainPID owns this entire cgroup BEFORE execution. A stuck chronyc spawn/read/ reap cannot block MainPID: it kills this role and retains the leader plus all descendants until the qualified Children owner actually observes retirement. PreparedRecovery never crosses IPC: its actual root collector parent forks the permanently dropped administrator only after collecting the observation. """ require(os.getresuid() == (0, 0, 0) and os.getresgid() == (0, 0, 0), 'collector requires root') gate_parent, gate_child = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) try: packet(control, b'COLLECTING') policy = nts.Policy(frozenset({'time.cloudflare.com', 'nts.netnod.se'})) delegation = None managed=os.open('/proc/self/ns/net',os.O_RDONLY|os.O_CLOEXEC) expected=c01_network.identity(managed) try: # Collect in the appointed NTS producer realm rather than relying # on cross-namespace datagram reply routing. Only this already-owned # keyless role joins it; application/authority never inherit it. c01_network.enter_host_worker(LIBC) if epoch is not None: delegation = recovery.EpochIssuerDelegation(bytes(epoch['delegation']['id']), epoch['delegation']['allow_wake'], epoch['delegation'].get('allow_refresh', False)) maximum = epoch['delegation']['max_awake'] approval = recovery.prepare_epoch(policy, qualification, binding, context, delegation, maximum['secs'] * 1_000_000_000 + maximum['nanos'], deadline) else: approval = recovery.prepare(policy, qualification, binding, domain) finally: restored=LIBC.setns(managed,c01_network.CLONE_NEWNET)==0 os.close(managed) require(restored and c01_network.require_private(os.getpid(),'managed')==expected, 'collector must return to the appointed authority realm') parent = os.getpid() previous_mask = signal.pthread_sigmask(signal.SIG_BLOCK, lifecycle.STOP_SIGNALS) try: pid = os.fork() if pid == 0: try: require(LIBC.prctl(1, signal.SIGKILL, 0, 0, 0) == 0 and os.getppid() == parent, 'collector administrator parent changed') close_except({gate_child.fileno()}) signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask) lifecycle.drop_credentials(authority_peer['uid'], authority_peer['gid'], parent) child_wait(gate_child, b'AUTHORIZE', deadline) gate_child.close() channel = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) channel.connect(address) if delegation is not None: approval.send_epoch_once(channel, recovery.Peer(**authority_peer), delegation) else: approval.send_once(channel, recovery.Peer(**authority_peer)) os._exit(0) except BaseException: os._exit(111) # Register the direct child before any fallible reporting. The root # owner already owns its inherited cgroup even if this process dies. admin_pid = pid finally: signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask) gate_child.close() packet(control, {'pid': admin_pid, 'uid': authority_peer['uid'], 'gid': authority_peer['gid']}) child_wait(control, b'RELAYED', deadline) packet(gate_parent, b'AUTHORIZE') gate_parent.close() while now() < deadline: reaped, status = os.waitpid(admin_pid, os.WNOHANG) if reaped: require(reaped == admin_pid and os.waitstatus_to_exitcode(status) == 0, 'recovery administrator failed') packet(control, b'ADMIN_SENT') # Do not exit/close a queued status before MainPID consumes it. # Authority acceptance is independently required by MainPID. child_wait(control, b'RELEASE', deadline) return time.sleep(0.01) raise ValueError('collector administrator setup expired') finally: gate_parent.close() gate_child.close() # No blocking cleanup wait: any failure exits the leader; MainPID owns # the still-unreaped descendant set, including uninterruptible helpers. def start_collector(owner, qualification, binding, domain, authority_peer, address, deadline, epoch=None, context=None): # No storage-key descriptors or other parent's path capabilities are kept. # Call only after authenticated STORE_READY; prepare() is never called by # the root observer itself, including validation and helper construction. return owner.children.spawn('collector', lambda control: collect_and_authorize( control, qualification, binding, domain, authority_peer, address, deadline, epoch, context)) class Supervisor: """One owner through RUNNING -> STOPPING -> reaped, with no restart branch.""" def __init__(self, lock): self.lock = lock self.children = lifecycle.Children() try: lock.bind_group(self.children.root) except BaseException: self.children.close() raise self.watch = None self.network_settings_watch = None self.tor_watch = None self.runtime_lease = None self.access_lease = None self.network_lease = None self.runtime_parameters = None self.runtime_workspace = None self.failure = None self.completed_one_shots = {'administrator'} self.notifier = None self.power_lease = None self.power_channel = None self.power_state = None self.power_epoch_revoked = False self.authority_record = None self.authority_credentials = None self.power_command = b'SLEEP_REVOKE' self.clock_sequence = 0 self.prior_power_signal = signal.signal(signal.SIGUSR1, lambda *_: None) self.wake_read, self.wake_write = os.pipe2(os.O_NONBLOCK | os.O_CLOEXEC) self.prior_wakeup = signal.set_wakeup_fd(self.wake_write) def notify(self, message): if self.notifier is not None: require(self.notifier.send(message, socket.MSG_DONTWAIT | socket.MSG_NOSIGNAL) == len(message), 'service notification failed') def stop(self, reason): if self.failure is None: try:os.write(2,('C01_SUPERVISOR_STOP reason='+str(reason)+'\n').encode()) except OSError:pass self.failure = self.failure or reason self.children.request_stop() if self.watch: self.watch.revoke(reason) if self.network_settings_watch: self.network_settings_watch.revoke(reason) if self.tor_watch: self.tor_watch.revoke(reason) def tick(self): hold=getattr(self,'admission_hold',None) if hold is not None and not self.children.stopping: try:hold.check() except (OSError,ValueError) as error: try:os.write(2,('C01_ADMISSION_REVOKED type='+type(error).__name__+' reason='+str(error)+'\n').encode()) except OSError:pass self.stop('admission lifetime lost') if self.power_lease is not None and not self.children.stopping: try: request = self.power_lease.request() if request is not None and self.power_state is None: self.power_state = request self.power_channel = self.power_lease.connect(request) if self.authority_record is None: self.stop('power requested before epoch authority exists') else: packet(self.authority_record.channel, self.power_command) self.power_deadline = now() + 3_000_000_000 if self.power_state is not None and not self.children.stopping: channel = self.authority_record.channel if select.select([channel], [], [], 0)[0]: raw, anc, flags, _ = channel.recvmsg(8193, socket.CMSG_SPACE(12), socket.MSG_CMSG_CLOEXEC) require(flags & ~socket.MSG_CMSG_CLOEXEC == 0 and anc == [(socket.SOL_SOCKET, socket.SCM_CREDENTIALS, struct.pack('3i', *self.authority_credentials))], 'power epoch acknowledgement credentials') self.power_epoch_revoked = raw == b'EPOCH_REVOKED' self.stop('power caller retirement') elif now() >= self.power_deadline: self.stop('power epoch revocation uncertain') except (OSError, ValueError): self.stop('power authority lost') if not self.children.stopping: for lease, reason in ((self.access_lease, 'access lease lost'), (self.runtime_lease, 'runtime lease lost'), (self.network_lease, 'network lease lost')): if lease is not None: try: lease.check() except (OSError, ValueError): self.stop(reason) if self.watch and not self.children.stopping and not self.watch.receive(): self.stop('host worker failed') if self.network_settings_watch and not self.children.stopping and not self.network_settings_watch.receive(): self.stop('network settings worker failed') if self.tor_watch and not self.children.stopping and not self.tor_watch.receive(): self.stop('Tor lifecycle worker failed') errors = self.children.poll() if errors: self.stop('child retirement error') for role, code in self.children.results.items(): if role not in self.completed_one_shots or code != 0: self.stop('supervised role exited') if self.children.stopping and self.watch: self.watch.revoke('supervisor stopping') if self.children.stopping and self.network_settings_watch: self.network_settings_watch.revoke('supervisor stopping') if self.children.stopping and self.tor_watch: self.tor_watch.revoke('supervisor stopping') if self.watch and 'host-io' in self.children.results: self.watch.release_reaped() if self.network_settings_watch and 'network-settings' in self.children.results: self.network_settings_watch.release_reaped() if self.tor_watch and 'tor-owner' in self.children.results: self.tor_watch.release_reaped() return not self.children.stopping def receive(self, record, uid, gid, deadline): while now() < deadline and self.tick(): require(record.role in self.children.children and record.pid is not None and record.channel.fileno() >= 0, 'child exited before setup status') ready, _, _ = select.select([record.channel, self.wake_read], [], [], 0.05) if self.wake_read in ready: os.read(self.wake_read, 4096) if record.channel in ready: raw, anc, flags, _ = record.channel.recvmsg(8193, socket.CMSG_SPACE(12), socket.MSG_CMSG_CLOEXEC) for level, kind, data in anc: if (level, kind) == (socket.SOL_SOCKET, socket.SCM_RIGHTS): rights = array.array('i') rights.frombytes(data[:len(data) - len(data) % rights.itemsize]) for fd in rights: os.close(fd) require(flags & ~socket.MSG_CMSG_CLOEXEC == 0 and anc == [(socket.SOL_SOCKET, socket.SCM_CREDENTIALS, struct.pack('3i', record.pid, uid, gid))] and 0 < len(raw) <= 8192 and now() < deadline, 'child setup status refused') return raw raise ValueError('supervisor stopped or setup expired') def wait(self, record, expected, uid, gid, deadline): require(self.receive(record, uid, gid, deadline) == expected, 'child setup status refused') def relay_collector(self, collector, authority, uid, gid, deadline): self.wait(collector, b'COLLECTING', 0, 0, deadline) admin = json.loads(self.receive(collector, 0, 0, deadline), object_pairs_hook=exact_object) require(type(admin) is dict and set(admin) == {'pid', 'uid', 'gid'} and type(admin['pid']) is int and admin['pid'] > 1 and admin['pid'] not in (os.getpid(), collector.pid, authority.pid) and type(admin['uid']) is int and type(admin['gid']) is int and (admin['uid'], admin['gid']) == (uid, gid), 'collector administrator identity refused') # Only exact late peer credentials cross this channel, never time data. packet(authority.channel, admin) packet(collector.channel, b'RELAYED') return admin def retire(self): self.stop('retiring') # Deliberately no deadline that releases the lock over a D-state worker. # Service TimeoutStopSec=infinity and Restart=no preserve this owner. while True: try: self.tick() if not self.children.children and not self.children.reserved: self.children.close() break ready, _, _ = select.select([self.wake_read], [], [], 0.05) if ready: os.read(self.wake_read, 4096) except (OSError, ValueError): self.children.request_stop() time.sleep(0.05) signal.set_wakeup_fd(self.prior_wakeup) os.close(self.wake_read) os.close(self.wake_write) if self.runtime_lease is not None: self.runtime_lease.close() if self.access_lease is not None: self.access_lease.close() if self.network_lease is not None: self.network_lease.close() self.lock.close() if self.power_channel is not None: try: packet(self.power_channel, {'nonce':self.power_state['nonce'], 'epoch_revoked':self.power_epoch_revoked,'callers_reaped':True}) except (OSError, ValueError): # The power owner retains Draining and cannot permit sleep from # a lost reply. All children are already reaped; finish closing # our capabilities without manufacturing a second receipt. self.failure = 'power retirement receipt delivery uncertain' finally: self.power_channel.close() if self.power_lease is not None: self.power_lease.close() signal.signal(signal.SIGUSR1, self.prior_power_signal) if self.notifier is not None: self.notifier.close() class Prepared: """Prestage all paths, pinned keys, mounts and payloads before live NTS.""" def __init__(self, value): self.fds = [] self.value = value try: self.epoch = load_epoch_configuration(value) self.executable = self.own(os.open('/proc/self/exe', os.O_RDONLY | os.O_CLOEXEC)) meta = os.fstat(self.executable) require(stat.S_ISREG(meta.st_mode) and meta.st_uid == 0 and not meta.st_mode & 0o022, 'native MainPID is not root protected') with open(f'/proc/self/fd/{self.executable}', 'rb') as stream: require(hashlib.file_digest(stream, 'sha256').hexdigest() == value['native_sha256'], 'native artifact differs from provisioning') self.keys = {role: self.own(root_capability(value[name], 32)) for role, name in (('application', 'snapshot_key'), ('authority', 'authority_key'))} self.keys['epoch'] = self.own(root_capability(self.epoch['key'], 32)) require(len({tuple(key_binding(fd).values()) for fd in self.keys.values()}) == 3, 'epoch/authority/application key alias') require(key_binding(self.keys['application']) != key_binding(self.keys['authority']), 'storage keys alias the same inode') self.backings = {name: self.own(views.require_private_backing(value[name + '_backing'])) for name in ('manager', 'snapshot', 'authority')} self.backings['epoch'] = self.own(views.require_private_backing(self.epoch['backing'])) paths = [Path(value[name + '_backing']) for name in ('manager','snapshot','authority')] paths.append(Path(self.epoch['backing'])) runtime = Path(value['runtime_data_dir']) paths.append(runtime) require(len(set(paths)) == 5 and all(not a.is_relative_to(b) for a in paths for b in paths if a != b), 'private and runtime roots overlap') for name, fd in self.backings.items(): require(views.identity(fd)[2:] == (value['runtime_uid'], value['runtime_gid'], 0o700), 'private backing must preserve original runtime ownership and mode') private_identity(self.backings['manager'], value['runtime_uid']) self.backing_sources = {name: views.PinnedSource.capture(fd, self.epoch['backing'] if name == 'epoch' else value[name + '_backing']) for name, fd in self.backings.items()} root_ancestry(runtime) self.runtime = self.own(os.open(runtime, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC)) require(runtime.resolve(strict=True) == runtime, 'runtime root must be canonical') self.runtime_source = views.PinnedSource.capture(self.runtime, runtime) root_ancestry(value['proxy_path']) self.proxy = self.own(os.open(value['proxy_path'], os.O_PATH | os.O_NOFOLLOW | os.O_CLOEXEC)) meta = os.fstat(self.proxy) require(stat.S_ISREG(meta.st_mode) and meta.st_uid == 0 and meta.st_nlink == 1 and not meta.st_mode & 0o6022 and meta.st_mode & 0o005 == 0o005, 'runtime proxy not protected') self.proxy_source = views.PinnedSource.capture(self.proxy, value['proxy_path']) except BaseException: self.close() raise def own(self, fd): self.fds.append(fd) return fd def close(self): while self.fds: os.close(self.fds.pop()) def role_views(self, account, role, runtime_view=None): mounts = views.PrivateMounts() # Filesystem Unix sockets (user/system buses, Podman, terminal sockets) # are absent. Abstract Unix IPC additionally requires admission's private # routed network namespace; pathname hiding alone never attests that. for target in ('/run', '/home', '/root', '/tmp', '/var/tmp'): mounts.cover(target, 0o755 if target == '/run' else 0o700) BASE.mkdir(mode=0o755) BASE.chmod(0o755) # UMask=0077 must not hide the role's owned views. if role == 'authority': # The common binding names the manager root, but authority gets no # manager capability. A root-owned empty placeholder satisfies the # native canonical-path check without exposing manager state. (BASE / 'manager').mkdir(mode=0o700) chosen = ('manager', 'snapshot') if role == 'application' else ('authority', 'epoch') for name in chosen: destination = BASE / name destination.mkdir(mode=0o700) mounts.idmapped_directory(self.backing_sources[name], destination, self.value['runtime_uid'], account.pw_uid, self.value['runtime_gid'], account.pw_gid) if role == 'application': mounts.idmapped_directory(self.runtime_source, self.value['runtime_data_dir'], self.value['runtime_uid'], account.pw_uid, self.value['runtime_gid'], account.pw_gid) install_proxy_aliases(self.proxy_source) if runtime_view is not None: runtime_view.install() else: mounts.cover(self.value['runtime_data_dir'], 0o700) # Backing parents remain root-private; do not unhide them for role UIDs. def install_proxy_aliases(proxy): """Root-created, read-only exact-inode aliases in the private /run mount. PATH is convenience only: native AppArmor/Landlock/seccomp enforce execution. Production application setup additionally installs its independently acquired runtime socket capability. A view-only fixture without it remains fail-closed. """ require(isinstance(proxy, views.PinnedSource) and not proxy.directory and proxy.namespace != views.namespace_identity(), 'pre-namespace pinned proxy locator required') require(not proxy.checked_protections() & os.ST_NOEXEC, 'proxy source is noexec') directory = BASE / 'bin' directory.mkdir(mode=0o755) directory.chmod(0o755) target = directory / 'archipelago-runtime-proxy' target.touch(mode=0o555, exist_ok=False) local = proxy.reopen() try: require(not os.fstatvfs(local).f_flag & os.ST_NOEXEC, 'proxy locator mount is noexec') require(views.LIBC.mount(f'/proc/self/fd/{local}'.encode(), os.fsencode(target), None, 4096, None) == 0, 'pinned proxy bind failed') finally: os.close(local) require(views.LIBC.mount(None, os.fsencode(target), None, 4096 | 32 | 1 | 2 | 4, None) == 0, 'read-only proxy bind failed') proxy.checked_protections() before, after = os.fstat(proxy.fd), target.stat() require((before.st_dev, before.st_ino) == (after.st_dev, after.st_ino), 'proxy bind identity mismatch') require(os.statvfs(target).f_flag & os.ST_RDONLY and not os.statvfs(target).f_flag & os.ST_NOEXEC, 'proxy mount protection mismatch') for command in ('podman', 'systemctl', 'journalctl', 'sh', 'bash', 'tmux', 'loginctl', 'mkdir', 'rm', 'cp', 'mv', 'cat', 'test', 'which', 'curl', 'git', 'df', 'du'): (directory / command).symlink_to('archipelago-runtime-proxy') def runtime_workspace_environment(runtime): # A pathname in the separately qualified runtime view, never manager HOME. # The handler validates/creates it through the proxy in that actual context. return {'ARCHIPELAGO_RUNTIME_WORKSPACE': str(Path(runtime['home']) / 'Work')} def launch(owner, prepared, app, authority, qualification): """Internal engine. Caller must hold enforced admission for its entire life.""" value = prepared.value require(owner.access_lease is not None and owner.runtime_lease is not None and owner.network_lease is not None and owner.runtime_parameters is not None and owner.runtime_workspace is not None and owner.tick(), 'live access/runtime capabilities required before launch') parent, children = os.getpid(), owner.children setup_end = now() + SETUP_NS authority_address = '\0archy-c01-authority-' + os.urandom(16).hex() host_address = '\0archy-c01-host-' + os.urandom(16).hex() listener = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) listener.bind(host_address) listener.listen(1) listener.settimeout(1) settings_address = '\0archy-c01-network-settings-' + os.urandom(16).hex() settings_listener = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) settings_listener.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1) settings_listener.bind(settings_address) settings_listener.listen(1) settings_listener.settimeout(1) journal = network_settings.open_journal() tor_listener = c01_tor.listener() radio_channel = c01_radio.connect() radio_pid = None tor_address = tor_listener.getsockname() tor_dirs = c01_tor.capabilities(prepared.runtime) for fd in tor_dirs: prepared.own(fd) env = {'PATH': str(BASE / 'bin'), 'HOME': str(BASE / 'manager'), 'LANG': 'C.UTF-8', 'ARCHIPELAGO_DATA_DIR': str(BASE / 'manager'), 'ARCHIPELAGO_RUNTIME_DATA_DIR': value['runtime_data_dir'], 'ARCHIPELAGO_HOST_IP': value['host_ip'], 'ARCHIPELAGO_BIND': '127.0.0.1:5678', 'ARCHIPELAGO_USE_QUADLET_BACKENDS': 'true'} env.update(owner.runtime_workspace) if 'INVOCATION_ID' in os.environ: env['INVOCATION_ID'] = os.environ['INVOCATION_ID'] def authority_child(control): prepared.role_views(authority, 'authority') keep = {prepared.executable, prepared.keys['authority'], prepared.keys['epoch'], control.fileno()} close_except(keep) # In particular, NO application key survives the drop. lifecycle.drop_credentials(authority.pw_uid, authority.pw_gid, parent) server = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) server.bind(authority_address) server.listen(2) server.settimeout(120) packet(control, b'LISTENING') connection, _ = server.accept() require(peer(connection)[1:] == (app.pw_uid, app.pw_gid), 'authority app credentials refused') packet(control, b'CONNECTED') child_wait(control, b'EXEC', setup_end) native(prepared.executable, 'c01-authority-v4', {3: control.fileno(), 4: connection.fileno(), 5: server.fileno(), 9: prepared.keys['authority'], 13: prepared.keys['epoch']}, env) def application_child(control): control.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1) # Never let parent/child drain a fork-shared renewal channel. Acquire the # child's socket-mount capability independently before covering /run. runtime_view = c01_runtime_view.RuntimeLease(*owner.runtime_parameters) try: prepared.role_views(app, 'application', runtime_view) finally: runtime_view.close() close_except({control.fileno(), prepared.executable, prepared.keys['application'], prepared.proxy} | ({radio_channel.fileno()} if radio_channel is not None else set())) lifecycle.drop_credentials(app.pw_uid, app.pw_gid, parent) clock = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) clock.connect(authority_address) host = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) host.connect(host_address) settings = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) settings.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1) settings.connect(settings_address) tor = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) tor.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1) tor.connect(tor_address) packet(control, b'CONNECTED') d09 = c01_d09.receive_child(control, parent, setup_end) c01_d09.wait_exec(control, parent, setup_end) native(prepared.executable, 'c01-managed-v7' if radio_channel is not None else 'c01-managed-v6', {3: control.fileno(), 4: clock.fileno(), 6: host.fileno(), 8: prepared.proxy, 9: prepared.keys['application'], 10: d09, 11: settings.fileno(), 12: tor.fileno(), **({14: radio_channel.fileno()} if radio_channel is not None else {})}, env) try: authority_child_record = children.spawn('authority', authority_child, prepared.fds) owner.authority_record = authority_child_record owner.authority_credentials = (authority_child_record.pid, authority.pw_uid, authority.pw_gid) owner.wait(authority_child_record, b'LISTENING', authority.pw_uid, authority.pw_gid, setup_end) application = children.spawn('application', application_child, c01_radio.application_fds(prepared.fds, radio_channel)) owner.wait(application, b'CONNECTED', app.pw_uid, app.pw_gid, setup_end) if radio_channel is not None: radio_pid = c01_radio.bind(radio_channel, application.pid, app.pw_uid, app.pw_gid) radio_channel.close() d09_session, d09_fd = c01_d09.register(owner, application.pid, setup_end) try: c01_d09.send_child(application.channel, d09_session, d09_fd) finally: os.close(d09_fd) owner.wait(authority_child_record, b'CONNECTED', authority.pw_uid, authority.pw_gid, setup_end) accepted, _ = listener.accept() require(peer(accepted) == (application.pid, app.pw_uid, app.pw_gid), 'host app peer mismatch') app_peer = {'pid': application.pid, 'uid': app.pw_uid, 'gid': app.pw_gid} authority_peer = {'pid': authority_child_record.pid, 'uid': authority.pw_uid, 'gid': authority.pw_gid} tor_accepted, _ = tor_listener.accept() tor_worker = c01_tor.spawn(owner, prepared.executable, tor_accepted, app_peer, tor_dirs, native, packet, env) settings_accepted, _ = settings_listener.accept() require(peer(settings_accepted) == (application.pid, app.pw_uid, app.pw_gid), 'network settings app peer mismatch') settings_accepted.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1) def settings_child(control): control.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1) require(LIBC.prctl(38, 1, 0, 0, 0) == 0, 'network settings NNP unavailable') c01_network.enter_host_worker(LIBC) packet(control, b'NETWORK_CHANNEL') native(prepared.executable, 'network-settings-owner-v1', {3: control.fileno(), 4: settings_accepted.fileno(), 5: journal}, env) settings_worker = children.spawn('network-settings', settings_child, (settings_accepted.fileno(), journal, prepared.executable)) owner.wait(settings_worker, b'NETWORK_CHANNEL', 0, 0, setup_end) owner.network_settings_watch = network_settings_watch.Watch(settings_worker.channel, settings_accepted, settings_worker.pid) journal_meta = os.fstat(journal) packet(settings_worker.channel, {'version': 1, 'process_id': settings_worker.pid, 'application': app_peer, 'journal_device': journal_meta.st_dev, 'journal_inode': journal_meta.st_ino}) def host_child(control): require(LIBC.prctl(38, 1, 0, 0, 0) == 0, 'host NNP unavailable') c01_network.enter_host_worker(LIBC) native(prepared.executable, 'host-access-owner', {3: control.fileno(), 4: accepted.fileno(), 5: prepared.runtime}, env) worker = children.spawn('host-io', host_child, (accepted.fileno(), prepared.runtime, prepared.executable)) owner.watch = host_watch.HostWatch(worker.channel, accepted, worker.pid, now() + host_watch.SETUP_LIMIT_NS) runtime_meta = os.fstat(prepared.runtime) packet(worker.channel, {'version': 2, 'process_id': worker.pid, 'application': app_peer, 'data_device': runtime_meta.st_dev, 'data_inode': runtime_meta.st_ino}) # Validate the actual finite role handshakes before releasing application # execution. Clock authorization below still requires a fresh barrier. for watch,label in ((owner.watch,'host'),(owner.network_settings_watch,'network'),(owner.tor_watch,'Tor')): while watch.state=='starting' and now()