#!/usr/bin/python3 """C01 root access/retirement owner; fixed source, fixed policy, finite protocol. `transition` is an OFFLINE guarded operation, never exposed on the socket. `serve` issues revocable capabilities only with current kernel/policy evidence. An interrupted transition or owner lifetime leaves a persistent restart fence. Neither CLI installs itself, starts the application, nor bypasses admit(). """ import array import errno import fcntl import json import os from pathlib import Path import select import signal import socket import stat import struct import subprocess import sys import time import types # The -I entry never searches a writable directory or loads adjacent bytecode. if __name__ == '__main__': for name in ('c01_network', 'c01_access_policy', 'c01_migration_layout', 'c01_access_generation', 'c01_access_kernel', 'c01_children'): 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('access 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 c01_access_policy as policy_module import c01_access_kernel as kernel import c01_children as children_module import c01_access_generation require = policy_module.require BASE = Path('/run/archipelago-c01') SOCKET = BASE / 'runtime-access.sock' MAX_PACKET = 8192 CHECK_NS = 5_000_000_000 WATCH_PACKET = struct.Struct('!8sQQQ') OPERATIONS = ['freeze-capture', 'mask-stop-activation', 'replace-policy', 'kill-observe-empty', 'reload-ssh-restart-polkit', 'quarantine-specific-copies', 'verify-effective-policy-and-preservation'] ENV = {'PATH': '/usr/sbin:/usr/bin:/sbin:/bin', 'LC_ALL': 'C', 'HOME': '/root'} def send(channel, value): raw = json.dumps(value, separators=(',', ':')).encode() require(len(raw) <= MAX_PACKET and channel.send(raw, socket.MSG_DONTWAIT | socket.MSG_NOSIGNAL) == len(raw), 'access send failed') def receive(channel, credentials=None): raw, ancillary, flags, _ = channel.recvmsg(MAX_PACKET + 1, socket.CMSG_SPACE(12) + socket.CMSG_SPACE(64), socket.MSG_DONTWAIT | socket.MSG_CMSG_CLOEXEC) peers, unwanted = [], False for level, kind, data in ancillary: 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) unwanted = True elif (level, kind) == (socket.SOL_SOCKET, socket.SCM_CREDENTIALS) and len(data) == 12: peers.append(struct.unpack('3i', data)) else: unwanted = True require(raw, 'access channel closed before protocol packet') require(0 < len(raw) <= MAX_PACKET and not flags & ~socket.MSG_CMSG_CLOEXEC and not unwanted and peers == ([] if credentials is None else [credentials]), 'access framing/credentials refused') return json.loads(raw, object_pairs_hook=policy_module.exact) def peer(channel): return struct.unpack('3i', channel.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)) def revoke(channel): # shutdown revokes aliases too, including any accidental in-process duplicate. try: channel.shutdown(socket.SHUT_RDWR) except OSError: pass channel.close() def notify_ready(): 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(b'READY=1\nSTATUS=Live access policy checked; guarded lease listener available') def fixed_command(argv): """Only called by the bounded, cgroup-owned root checker/transition child. No caller chooses argv. A stuck spawn/read/reap strands that helper, not the MainPID. MainPID revokes and keeps its lock and Children registry until reaped. """ import resource import tempfile # Regular bounded output avoids pipe-EOF descendants and unbounded buffering. with tempfile.TemporaryFile() as output: result = subprocess.run(argv, stdin=subprocess.DEVNULL, stdout=output, stderr=subprocess.DEVNULL, env=ENV, timeout=5, check=False, preexec_fn=lambda: resource.setrlimit(resource.RLIMIT_FSIZE, (1024 * 1024, 1024 * 1024))) output.seek(0) raw = output.read(1024 * 1024 + 1) require(len(raw) <= 1024 * 1024, 'policy probe output bound') return result.returncode, raw.decode() def semantic_checks(policy): import pwd v = policy.value shadow = {row.split(':')[0]: row.split(':')[1] for row in policy_module.read(Path('/etc/shadow'), True).decode().splitlines()} operator = pwd.getpwnam(v['operator']['name']) require((operator.pw_uid, operator.pw_gid) == (1001, 1001) and operator.pw_shell not in ('/sbin/nologin', '/usr/sbin/nologin', '/bin/false') and ((bool(shadow[operator.pw_name]) and not shadow[operator.pw_name].startswith(('!', '*'))) or v.get('operator_sudo_authentication') == 'preserve-verified-noninteractive'), 'separate operator access not preserved') require(fixed_command(['/usr/sbin/visudo', '-c'])[0] == 0, 'installed sudo policy is invalid') nss = policy_module.read(Path('/etc/nsswitch.conf')).decode().splitlines() sudo_sources = [row.split(':', 1)[1].split('#', 1)[0].split() for row in nss if row.split(':', 1)[0].strip() == 'sudoers'] require(not sudo_sources or sudo_sources == [['files']], 'external sudo policy source refused') # Exact sudoers bytes plus syntax success distinguish an intended denial # from a missing/broken sudo installation. NSS/plugin inventory is pinned. for account in v['accounts'].values(): entry = pwd.getpwnam(account['name']) require((entry.pw_uid, entry.pw_gid) == (account['uid'], account['gid']) and entry.pw_shell in ('/sbin/nologin', '/usr/sbin/nologin', '/bin/false') and os.getgrouplist(entry.pw_name, entry.pw_gid) == [entry.pw_gid] and sum(p.pw_uid == entry.pw_uid for p in pwd.getpwall()) == 1 and shadow[entry.pw_name].startswith('!'), 'managed login/credential policy changed') code, listing = fixed_command(['/usr/bin/sudo', '-n', '-l', '-U', entry.pw_name]) # Some sudo versions return zero for a successfully produced *empty* # listing. Require the exact C-locale no-permission result, not status # alone (and never interpret an authentication/plugin failure as denial). denial = f'User {entry.pw_name} is not allowed to run sudo on {os.uname().nodename}.' require(code in (0, 1) and listing.strip() == denial, 'managed principal sudo denial unproved') ssh_checks(policy) # A rule returning YES before the C01 rule would bypass it. Refuse every # lexically earlier/equal competing rule in both polkit search roots. rule = '00-archipelago-c01.rules' for directory in ('/etc/polkit-1/rules.d', '/run/polkit-1/rules.d', '/usr/local/share/polkit-1/rules.d', '/usr/share/polkit-1/rules.d'): if Path(directory).exists(): policy_module.protected(Path(directory), directory=True) for path in Path(directory).glob('*.rules'): policy_module.tree_digest(path) require(path.name > rule or path == policy_module.TARGETS['polkit'], 'earlier polkit rule can grant access') for unit in v['restart_units']: path = Path('/etc/systemd/system') / unit policy_module.protected(path.parent, directory=True) require(path.is_symlink() and os.readlink(path) == '/dev/null', 'legacy activation path is not masked') # An allow file takes precedence over a deny file. Jobs already in spools # must be explicitly retired, not merely prevented from new submission. for path in (Path('/etc/cron.allow'), Path('/etc/at.allow')): if path.exists(): require(not set(policy.names) & set(policy_module.read(path).decode().splitlines()), 'job allowlist grants access') for name in policy.names: require(not any((Path(root) / name).exists() for root in ('/var/spool/cron', '/var/spool/cron/crontabs')), 'managed crontab remains active') for directory in (Path('/var/spool/at'), Path('/var/spool/cron/atjobs')): if directory.exists(): managed = {item['uid'] for item in v['accounts'].values()} require(not any(path.lstat().st_uid in managed for path in directory.iterdir()), 'managed queued at job remains active') # Subordinate IDs and original paths are preserved without chown/copy. for file, key in (('/etc/subuid', 'subuids'), ('/etc/subgid', 'subgids')): rows = [row.split(':') for row in policy_module.read(Path(file)).decode().splitlines()] ranges = [[int(row[1]), int(row[2])] for row in rows if row[0] == policy.runtime['runtime_user']] require(ranges == policy.runtime[key], 'original subordinate allocation changed') def ssh_effective(text): # OpenSSH emits list-valued settings (including DenyUsers/AllowUsers) as # repeated rows. Never discard earlier entries with dict(row.split(...)). result = {} for row in text.splitlines(): if ' ' in row: name, value = row.split(' ', 1) result.setdefault(name, []).append(value) return result def ssh_checks(policy, staged=False): """Global-only finite policy plus actual sshd parsing, including remote IPs. The byte-preserved baseline is the coordinator's verified administrator route. This checks configuration preservation, not an SSH authentication UAT. """ target = policy_module.STAGED / 'sshd' if staged else policy_module.TARGETS['sshd'] post = policy_module.read(target) policy_module.ssh_baseline(post, policy.names, policy.value['files']['sshd']['before']) require(fixed_command(['/usr/sbin/sshd', '-t', '-f', str(target)])[0] == 0, 'staged/current SSH syntax refused') for address in ('127.0.0.1', '198.51.100.23', '2001:db8::23'): for name in [*policy.names, policy.value['operator']['name']]: context = f'user={name},host=admin.example,addr={address}' code, text = fixed_command(['/usr/sbin/sshd', '-T', '-f', str(target), '-C', context]) effective = ssh_effective(text) denied = [name for row in effective.get('denyusers', []) for name in row.split()] require(code == 0 and denied == policy.names, 'global effective SSH denial differs') if staged and name == policy.value['operator']['name']: code, old_text = fixed_command(['/usr/sbin/sshd', '-T', '-f', str(policy_module.TARGETS['sshd']), '-C', context]) old = ssh_effective(old_text) require(code == 0 and {k: v for k, v in effective.items() if k != 'denyusers'} == {k: v for k, v in old.items() if k != 'denyusers'}, 'administrator SSH route changed') def completed_history(value, policy): require(value['state'] in ('retired', 'owner-active') and value['configuration_sha256'] == policy.digest, 'no completed operation history to requalify') evidence = value['evidence'] require(type(evidence) is dict and evidence.get('operations') == OPERATIONS and type(evidence.get('writers')) is list and type(evidence.get('copies')) is list, 'incomplete operation provenance') expected = {(int(pid), start, spec['path']) for spec in policy.value['cohorts'] for pid, start in spec['members'].items()} writers = evidence['writers'] require(len(writers) == len(expected) and all(any( item['pid'] == pid and item['start'] == start and Path(item['group']).is_relative_to(Path(group)) for item in writers) for pid, start, group in expected), 'retirement writer provenance differs') copies = [{'index': index, 'operation': 'quarantined-known-copy' if item['disposition'] == 'quarantine-known-copy' else 'verified-exclusive-backing', 'identity': item['identity']} for index, item in enumerate(policy.value['copies'])] require(evidence['copies'] == copies, 'retirement copy provenance differs') if policy.value['version']==3: plan=policy.value['migration_layout'] expected_layout={'source_identity':plan['source_identity'],'target_identity':plan['target_identity'], 'moved':[{k:item[k] for k in ('name','identity','mode_after','tree_metadata_sha256')} for item in plan['entries'] if item['disposition']=='private']} require(evidence.get('layout')==expected_layout,'missing exact layout preservation evidence') class Journal: """One persistent inode/lock and a conservative crash fence, never a grant.""" def __init__(self, policy): self.policy = policy policy_module.protected(policy_module.STATE, private=True, directory=True) self.path = policy_module.STATE / 'journal.json' self.lock = policy_module.STATE / 'owner.lock' self.fd = os.open(self.lock, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_CLOEXEC, 0o600) try: meta = os.fstat(self.fd) require(meta.st_uid == 0 and stat.S_IMODE(meta.st_mode) == 0o600 and meta.st_nlink == 1 and stat.S_ISREG(meta.st_mode), 'unprotected access lock') fcntl.flock(self.fd, fcntl.LOCK_EX | fcntl.LOCK_NB) require(self.lock.lstat().st_ino == meta.st_ino, 'access lock replaced') except BaseException: os.close(self.fd) raise self.boot = Path('/proc/sys/kernel/random/boot_id').read_text().strip() def record(self, state, evidence=None): owner = kernel.process(os.getpid()) policy_module.atomic(self.path, json.dumps({'version': 1, 'state': state, 'configuration_sha256': self.policy.digest, 'boot': self.boot, 'owner': {'pid': os.getpid(), 'start': owner['start'], 'group': owner['group']}, 'evidence': evidence}, separators=(',', ':')).encode()) def begin_transition(self): require(not self.path.exists(), 'prior access transition/lifetime requires offline reconciliation') self.record('transition-pending') def begin_serving(self): value, _ = policy_module.json_read(self.path) completed_history(value, self.policy) require(value['state'] == 'retired' and value['configuration_sha256'] == self.policy.digest and value['boot'] == self.boot, 'no reconciled same-boot retirement; cached state cannot admit') self.record('owner-active', value['evidence']) def pin_endpoint(self): value, _ = policy_module.json_read(self.path) meta = SOCKET.lstat() self.record('owner-active', {**value['evidence'], 'endpoint': policy_module.identity(meta)}) def close(self): os.close(self.fd) def preflight(policy): policy.preserved() policy.preflight_files() if policy.value['version']==3: import c01_migration_layout c01_migration_layout.preflight(policy) for unit in policy.value['restart_units']: path = Path('/etc/systemd/system') / unit policy_module.protected(path.parent, directory=True) require(not os.path.lexists(path), 'unit mask preflight conflict: ' + unit) ssh_checks(policy, staged=True) require(fixed_command(['/usr/sbin/visudo', '-c', '-f', str(policy_module.STAGED / 'sudoers')])[0] == 0, 'staged sudo syntax refused') cohorts = [] try: paths = [policy_module.canonical(spec['path']) for spec in policy.value['cohorts']] require(all(not a.is_relative_to(b) and not b.is_relative_to(a) for index, a in enumerate(paths) for b in paths[index + 1:]), 'overlapping cohort inventories') for spec in policy.value['cohorts']: cohorts.append(kernel.Cohort(spec, policy.runtime)) managed = {account['uid'] for account in policy.value['accounts'].values()} classified = {int(pid) for spec in policy.value['cohorts'] for pid in spec['members']} require(all(not managed.intersection(item['uids']) or item['pid'] in classified for item in kernel.all_processes()), 'unclassified preflight credential-bearing process') finally: for cohort in cohorts: cohort.close() def retirement(policy, progress, checkpoint=None): """Guarded finite transition. Caller retains the helper's full cgroup.""" cohorts = [] try: preflight(policy) for spec in policy.value['cohorts']: cohorts.append(kernel.Cohort(spec, policy.runtime)) for cohort in cohorts: cohort.freeze() deadline = kernel.now() + CHECK_NS while not all(c.events().get('frozen') == '1' for c in cohorts): require(kernel.now() < deadline, 'legacy cohort freeze incomplete') time.sleep(.02) writers = [item for cohort in cohorts for item in cohort.capture()] classified = {item['pid'] for item in writers} managed = {item['uid'] for item in policy.value['accounts'].values()} require(all(not managed.intersection(item['uids']) or item['pid'] in classified for item in kernel.all_processes()), 'unclassified old credential-bearing process') progress('frozen-inventoried-cohorts') # Masks are persistent and precede policy replacement and old-key custody. for unit in policy.value['restart_units']: path = Path('/etc/systemd/system') / unit policy_module.protected(path.parent, directory=True) require(not os.path.lexists(path), 'unit mask requires reviewed removal of existing override') os.symlink('/dev/null', path) require(fixed_command(['/usr/bin/systemctl', 'daemon-reload'])[0] == 0, 'mask reload failed') # Stop requests cancel queued activations; do not wait for process exit. require(fixed_command(['/usr/bin/systemctl', '--no-block', 'stop', *policy.value['restart_units']])[0] == 0, 'legacy activation stop failed') progress('masked-and-stopped-activation') policy.install_files() progress('installed-exact-access-policy') # File parsing alone does not retire a daemon's old policy/cache. These # fixed service operations are part of the guarded transaction. Existing # operator SSH sessions are preserved by reload (never ssh restart). require(fixed_command(['/usr/bin/systemctl', 'reload', 'ssh.service'])[0] == 0, 'SSH policy reload failed') require(fixed_command(['/usr/bin/systemctl', 'restart', 'polkit.service'])[0] == 0, 'polkit policy/cache retirement failed') progress('reloaded-ssh-retired-polkit-cache') for cohort in cohorts: cohort.kill() while not all(c.retired() for c in cohorts): # No false completion deadline: MainPID owns this helper and its # timeout fences admission; unresolved legacy cohorts stay journaled. time.sleep(.02) progress('kernel-observed-old-writer-exit') layout=None if policy.value['version']==3: require(checkpoint is not None,'layout migration needs a durable retired-writer checkpoint') checkpoint('writers-retired',{'writers':writers}) import c01_migration_layout layout=c01_migration_layout.apply_retired(policy) copies = kernel.quarantine(policy) policy.verify_files() policy.verify_trees() policy.preserved() semantic_checks(policy) kernel.verify_private(policy) require(all(not managed.intersection(item['uids']) for item in kernel.all_processes()), 'legacy credential cohort respawned during transition') evidence={'writers': writers, 'copies': copies, 'operations': OPERATIONS} if layout is not None:evidence['layout']=layout return evidence finally: # Never thaw surviving old writers or restore old privileges on failure. for cohort in cohorts: cohort.close() def check_current(policy, supervisor): for label, operation in (('files', policy.verify_files), ('trees', policy.verify_trees), ('preserved', policy.preserved), ('semantics', lambda: semantic_checks(policy)), ('private', lambda: kernel.verify_private(policy))): started=kernel.now() operation() elapsed=kernel.now()-started if elapsed > 500_000_000: print('C01_ACCESS_SLOW_CHECK: '+label+' '+str(elapsed),flush=True) # A reused same inode could still contain an unresolved original cohort. # Even an empty old group must be retired by its systemd owner before serve. for spec in policy.value['cohorts']: kernel.replacement_tree(spec, policy.runtime) retired = False active = supervisor if supervisor is not None and supervisor.get('retiring'): active = None group = Path(policy.value['supervisor_cgroup']) # Keep the reserved lifetime until BOTH direct owner exit and complete # kernel role retirement. No adoption of an empty-but-unremoved role. retired = (not group.exists() or (not kernel.members(group) and not any((group / ('c01-' + role)).exists() for role in children_module.ROLES))) elif supervisor is not None: try: current = kernel.process(supervisor['pid']) except (FileNotFoundError, ProcessLookupError): # MainPID's retained pidfd still owns this identity. A clean owner # exit may race this check; no remaining managed process is allowed. active = None else: require(current['start'] == supervisor['start'] and current['uids'] == [0] * 4 and current['gids'] == [0] * 4 and Path(current['group']).is_relative_to(Path(policy.value['supervisor_cgroup'])), 'exclusive supervisor changed') kernel.verify_supervisor(current, policy) # A reserved candidate has no authority to launch roles until publication. # Thus an existing orphan cannot acquire legitimacy by asking for a grant. evidence = kernel.verify_processes(policy, None if active and active.get('pending') else active) return {'runtime_processes': len(evidence), 'supervisor_retired': retired} def checker(control, policy): # This process is forked BEFORE public listeners/leases exist. Inheritance # cannot keep a public provider endpoint alive after MainPID death. while True: ready, _, _ = select.select([control], [], [], 1) if not ready: continue request = receive(control) require(set(request) == {'challenge', 'supervisor', 'generation', 'access_generation_head'}, 'private checker protocol') observed = kernel.now() try: policy.generations.refresh(request['access_generation_head']) evidence = check_current(policy, request['supervisor']) except (OSError, ValueError, KeyError) as error: # Fixed policy/kernel diagnostics only; no file content or credentials. send(control, {'challenge': request['challenge'], 'generation': request['generation'], 'failed': type(error).__name__, 'reason': str(error)[:256]}) continue # An obsolete generation cannot poison the next lifecycle. send(control, {'challenge': request['challenge'], 'generation': request['generation'], 'observed': observed, **evidence}) def watch_packet(kind, generation, deadline, token=0): return WATCH_PACKET.pack(kind, generation, deadline, token) def watch_evidence(raw, generation, evidence_deadline, stamp): require(len(raw) == WATCH_PACKET.size, 'access observer packet size') kind, current, expiry, token = WATCH_PACKET.unpack(raw) require(kind in (b'PULSE\0\0\0', b'HOLD\0\0\0\0') and current >= generation and evidence_deadline <= expiry and stamp < expiry <= stamp + CHECK_NS, 'access observer evidence expired/regressed') return kind.rstrip(b'\0'), current, expiry, token def watchdog(control, parent_pidfd): """Independent BOOTTIME observer; no public socket exists at its fork. The frozen runtime-access-v1 protocol has no renewal messages. MainPID sends an explicit revocation alias BEFORE each grant. This observer can shutdown those aliases even if MainPID is stuck in an uninterruptible kernel syscall; it does not rely on SIGKILL completion. PDEATHSIG kills this no-I/O/no-fork observer on MainPID death, so aliases cannot conceal a dead provider. """ deadline = kernel.now() + CHECK_NS evidence_deadline = 0 generation = 0 held = set() try: while kernel.now() < deadline: ready, _, _ = select.select([control, *held], [], [], .05) if kernel.now() >= deadline: break # A queued renewal cannot resurrect already expired evidence. if control in ready: raw, ancillary, flags, _ = control.recvmsg(WATCH_PACKET.size, socket.CMSG_SPACE(16 * 4), socket.MSG_CMSG_CLOEXEC) rights, unwanted = [], 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) % values.itemsize]) rights.extend(values) else: unwanted = True try: kind, generation, evidence_deadline, token = watch_evidence(raw, generation, evidence_deadline, kernel.now()) require(not unwanted and not flags & ~socket.MSG_CMSG_CLOEXEC and ((kind == b'PULSE' and not rights and token == 0) or (kind == b'HOLD' and len(rights) == 1 and token > 0)), 'access observer control lost') if rights: channel = socket.socket(fileno=rights.pop()) held.add(channel) require(len(held) <= 32, 'access revocation alias bound') deadline = min(evidence_deadline, kernel.now() + CHECK_NS) if kind == b'PULSE': send(control, {'alive': kernel.now()}) else: send(control, {'held': token}) finally: for fd in rights: revoke(socket.socket(fileno=fd)) for channel in list(held): if channel in ready: revoke(channel) held.remove(channel) for channel in held: revoke(channel) held.clear() finally: for channel in held: revoke(channel) signal.pidfd_send_signal(parent_pidfd, signal.SIGKILL) def start_watchdog(children): fd = os.pidfd_open(os.getpid()) try: return children.spawn('authority', lambda control: watchdog(control, fd), keep=(fd,)) finally: os.close(fd) class Leases: """Authenticated challenge/lifetime state; no persisted admission Boolean.""" def __init__(self, policy, guard): self.policy = policy self.guard = guard self.clients = {} self.nonces = set() self.supervisor = None self.retiring = None self.last_observed = 0 self.generation = 0 self.observed_generation = 0 self.next_grant = 0 self.failed = False # No fork is needed once serving. Defense in depth for a future helper: # close aliases in the child; shutdown would incorrectly revoke parent. os.register_at_fork(after_in_child=self.close_in_child) def close_in_child(self): for channel, entry in self.clients.items(): channel.close() os.close(entry['pidfd']) if self.retiring is not None: os.close(self.retiring['pidfd']) def fresh(self): return not self.failed and 0 <= kernel.now() - self.last_observed < CHECK_NS def accept(self, channel, request): pid, uid, gid = peer(channel) require((uid, gid) == (0, 0) and pid > 1 and self.fresh(), 'access peer/current evidence refused') require(type(request) is dict and set(request) == {'protocol', 'configuration_sha256', 'nonce'} and policy_module.hex_digest(request['nonce']) and request['nonce'] not in self.nonces and len(self.nonces) < 4096 and len(self.clients) < 16, 'access challenge bound/replay') protocol = request['protocol'] require(protocol in ('runtime-access-v1', 'supervisor-access-v1'), 'unknown access protocol') role = 'broker' if protocol == 'runtime-access-v1' else 'supervisor' if role == 'supervisor': require(self.policy.generations.value['pending'] is None, 'launch generation transition pending') digest = self.policy.value['runtime_sha256'] if role == 'broker' else self.policy.generations.value['active_launch_sha256'] require(request['configuration_sha256'] == digest, 'access configuration digest mismatch') pidfd = os.pidfd_open(pid) try: item = kernel.process(pid) require(item['uids'] == [0] * 4 and item['gids'] == [0] * 4 and Path(item['group']).is_relative_to(Path(self.policy.value[role + '_cgroup'])) and not select.select([pidfd], [], [], 0)[0], 'root service identity mismatch') require(not any(entry['role'] == role for entry in self.clients.values()), 'exclusive access role already held') require(role != 'supervisor' or self.retiring is None, 'prior supervisor retirement remains owned') self.nonces.add(request['nonce']) self.clients[channel] = {'pidfd': pidfd, 'role': role, 'pid': pid, 'start': item['start']} if role == 'supervisor': self.generation += 1 self.supervisor = {'pid': pid, 'start': item['start'], 'pending': True} self.clients[channel]['request'] = request else: self.publish(channel, request) except BaseException: if channel not in self.clients: os.close(pidfd) raise def publish(self, channel, request): require(self.fresh(), 'expired access grant') self.next_grant += 1 packet = watch_packet(b'HOLD', self.observed_generation, self.last_observed + CHECK_NS, self.next_grant) require(self.guard.sendmsg([packet], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, array.array('i', [channel.fileno()]))], socket.MSG_DONTWAIT | socket.MSG_NOSIGNAL) == len(packet), 'independent access revocation capability delivery failed') self.clients[channel]['grant'] = (self.next_grant, request) def confirm(self, token): require(type(token) is int, 'invalid watchdog handoff acknowledgment') for channel, entry in self.clients.items(): if entry.get('grant', (None,))[0] == token: require(self.fresh() and not select.select([channel, entry['pidfd']], [], [], 0)[0], 'expired/departed candidate before acknowledged grant') _, request = entry.pop('grant') send(channel, request) return # A disconnected candidate may have retired between HOLD and its ack. require(0 < token <= self.next_grant, 'unknown watchdog handoff acknowledgment') def checked(self, response, checked_supervisor): if response['generation'] != self.generation: return False if 'failed' in response: raise ValueError('access checker refused: ' + response['failed'] + ': ' + response['reason']) require(not self.last_observed or self.fresh(), 'prior evidence expired before checker delivery') self.last_observed = response['observed'] self.observed_generation = response['generation'] self.settle_supervisor(response['supervisor_retired'], checked_supervisor) for channel, entry in self.clients.items(): if 'request' in entry: require(checked_supervisor == self.supervisor and not select.select( [channel, entry['pidfd']], [], [], 0)[0], 'candidate supervisor departed before grant') self.publish(channel, entry.pop('request')) self.supervisor = {'pid': entry['pid'], 'start': entry['start']} # The candidate scan is consumed before handoff. Subsequent # scans use active even if the watchdog ack is still pending; # no older pending scan may fail after the grant is published. self.generation += 1 return True def poll(self): require(self.fresh(), 'access evidence deadline expired') for channel, entry in list(self.clients.items()): if select.select([channel, entry['pidfd']], [], [], 0)[0]: if entry['role'] != 'supervisor': raise ValueError('runtime access consumer exited/disconnected/sent unsolicited data') # Normal manager retirement must not take persistent runtime # services with it. Any remaining managed role is still rejected # by the checker; no replacement is issued over retained groups. revoke(channel) del self.clients[channel] self.retiring = entry self.generation += 1 self.supervisor = {'pid': entry['pid'], 'start': entry['start'], 'retiring': True} def settle_supervisor(self, observed_retired, checked_supervisor): if (self.retiring is not None and observed_retired and checked_supervisor == self.supervisor and select.select([self.retiring['pidfd']], [], [], 0)[0]): os.close(self.retiring['pidfd']) self.retiring = None self.supervisor = None self.generation += 1 def stop(self): self.failed = True for channel, entry in self.clients.items(): revoke(channel) os.close(entry['pidfd']) self.clients.clear() if self.retiring is not None: os.close(self.retiring['pidfd']) self.retiring = None self.supervisor = None def serve(policy, journal, children): journal.begin_serving() # Crash fence before any checker or public capability. record = children.spawn('host-io', lambda control: checker(control, policy)) guard = start_watchdog(children) generation_worker = children.spawn('application', lambda control: c01_access_generation.worker( control, policy.generations, policy.value['supervisor_cgroup'], send, receive)) generation_pending = None record.channel.setblocking(False) leases = Leases(policy, guard.channel) listener = None pending = {} challenge = None checked_supervisor = None checked_generation = None deadline = kernel.now() + CHECK_NS startup_deadline = deadline pulse = 0 guard_observed = kernel.now() try: while not children.stopping: children.poll() require('host-io' in children.children and 'authority' in children.children, 'access checker/watchdog exited') stamp = kernel.now() if stamp - pulse >= 250_000_000: packet = watch_packet(b'PULSE', leases.observed_generation, leases.last_observed + CHECK_NS if leases.last_observed else startup_deadline) require(guard.channel.send(packet, socket.MSG_DONTWAIT | socket.MSG_NOSIGNAL) == len(packet), 'access watchdog delivery failed') pulse = stamp if challenge is None and (generation_pending is None or not generation_pending[1]): challenge = os.urandom(32).hex() checked_supervisor = leases.supervisor checked_generation = leases.generation send(record.channel, {'challenge': challenge, 'supervisor': checked_supervisor, 'generation': checked_generation, 'access_generation_head':policy.generations.anchor}) deadline = stamp + CHECK_NS require(stamp < deadline, 'access checker stalled') require(stamp - guard_observed < CHECK_NS, 'independent access observer stalled') ready, _, _ = select.select([record.channel, guard.channel, generation_worker.channel, *pending, *([] if listener is None else [listener])], [], [], .05) if generation_worker.channel in ready: response=receive(generation_worker.channel,(generation_worker.pid,0,0)) require(generation_pending is not None,'unsolicited generation result') channel,mutating=generation_pending generation_pending=None if set(response)=={'result','head','state'}: require(set(response['state'])=={'active_launch_sha256','active_workspace_digest','pending'},'generation snapshot') policy.generations.anchor=response['head'] policy.generations.value.update(response['state']) if leases.supervisor is not None or leases.retiring is not None: response['result']['quiescent']=False try: send(channel,response['result']) except OSError: pass # lost ack is inspected by exact record, never repeated else: require(response=={'failed':True},'generation helper outcome') channel.close() if mutating: leases.generation+=1 if guard.channel in ready: response = receive(guard.channel, (guard.pid, 0, 0)) if set(response) == {'held'}: leases.confirm(response['held']) else: require(set(response) == {'alive'} and type(response['alive']) is int and 0 <= kernel.now() - response['alive'] < CHECK_NS, 'access observer heartbeat refused') guard_observed = response['alive'] if record.channel in ready: response = receive(record.channel, (record.pid, 0, 0)) require(set(response) in ({'challenge', 'generation', 'failed', 'reason'}, {'challenge', 'generation', 'observed', 'runtime_processes', 'supervisor_retired'}) and response['challenge'] == challenge and response['generation'] == checked_generation, 'access checker generation refused') if 'observed' in response: require(type(response['observed']) is int and 0 <= kernel.now() - response['observed'] < CHECK_NS, 'access checker evidence refused') accepted = leases.checked(response, checked_supervisor) challenge = None if listener is None and accepted: policy_module.protected(BASE, directory=True) listener = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET | socket.SOCK_CLOEXEC) listener.bind(str(SOCKET)) # Only offline requalification may retire a stale socket. os.chmod(SOCKET, 0o600) journal.pin_endpoint() listener.listen(16) listener.setblocking(False) notify_ready() if listener is not None: leases.poll() if listener in ready: channel, _ = listener.accept() channel.setblocking(False) if len(pending) >= 16 or peer(channel)[1:] != (0, 0): channel.close() else: pending[channel] = kernel.now() + 1_000_000_000 for channel, expiry in list(pending.items()): if kernel.now() >= expiry: channel.close() del pending[channel] elif channel in ready: try: request=receive(channel) require(type(request) is dict, 'access request object required') if request.get('protocol')=='d09-access-generation-v1': pid,uid,gid=peer(channel) current=kernel.process(pid) code,main=fixed_command(['/usr/bin/systemctl','show','archipelago-d09-lifecycle.service','-p','MainPID','--value']) require(code==0 and main.strip()==str(pid) and (uid,gid)==(0,0) and current['uids']==[0]*4 and current['group']=='/sys/fs/cgroup/system.slice/archipelago-d09-lifecycle.service', 'generation actor must be current root D09 MainPID') mutating=request.get('method')!='inspect' require(generation_pending is None and (not mutating or (leases.supervisor is None and leases.retiring is None)), 'generation handler busy or management lifetime not retired') if mutating: leases.generation+=1 # old checker results cannot poison this transition send(generation_worker.channel,request) generation_pending=(channel,mutating) else: leases.accept(channel, request) except (OSError, ValueError, KeyError, TypeError): if channel in leases.clients: raise # An uncertain delivered grant fences the whole owner. channel.close() del pending[channel] finally: if generation_pending is not None: generation_pending[0].close() leases.stop() for channel in pending: channel.close() if listener is not None: listener.close() children.request_stop() # The journal remains owner-active if storage stalls here. Either state # refuses a restart. Do NOT perform blocking journal I/O in this observer. def resume_retired_layout(policy,journal): require(policy.value['version']==3,'explicit layout migration required') old,_=policy_module.json_read(journal.path) require(old['state']=='writers-retired' and old['configuration_sha256']==policy.digest and old['boot']==journal.boot, 'no same-boot authenticated retired-writer checkpoint') writers=old['evidence']['writers'] expected={(int(pid),start,spec['path']) for spec in policy.value['cohorts'] for pid,start in spec['members'].items()} require(len(writers)==len(expected) and all(any(item['pid']==pid and item['start']==start and Path(item['group']).is_relative_to(Path(group)) for item in writers) for pid,start,group in expected), 'retired-writer checkpoint differs from approved inventory') policy.verify_files();policy.verify_trees();policy.preserved();semantic_checks(policy);ssh_checks(policy) managed={a['uid'] for a in policy.value['accounts'].values()} require(all(not managed.intersection(item['uids']) for item in kernel.all_processes()),'migration writers remain') for spec in policy.value['cohorts']:kernel.replacement_tree(spec,policy.runtime) import c01_migration_layout layout=c01_migration_layout.apply_retired(policy,resume=True) copies=kernel.quarantine(policy,resume=True) kernel.verify_private(policy) require(all(not managed.intersection(item['uids']) for item in kernel.all_processes()),'migration writer appeared') return {'writers':writers,'copies':copies,'layout':layout,'operations':OPERATIONS} def transition(policy, journal, children, resume=False): require(resume or not journal.path.exists(), 'prior access transition/lifetime requires offline reconciliation') def body(control): try: send(control, {'progress': 'preflight-started'}) if resume: evidence=resume_retired_layout(policy,journal) else: preflight(policy) # Deterministic conflicts leave no pending migration. journal.begin_transition() send(control, {'progress': 'intent-durable'}) evidence = retirement(policy, lambda step: send(control, {'progress': step}),journal.record) except (OSError, ValueError) as error: # Fixed root-only checks contain no credential contents. Retain a # bounded diagnostic instead of losing the first failure as EOF. # The pending journal and all retirement obligations remain intact. import traceback frame = traceback.extract_tb(error.__traceback__)[-1] location = f'{Path(frame.filename).name}:{frame.lineno}:{frame.name}' send(control, {'failure': type(error).__name__, 'reason': (location + ': ' + str(error))[:256]}) require(control.recv(8) == b'RELEASE', 'failed transition release refused') raise # Potentially large process inventory stays in the protected journal; # never fit/truncate it into a protocol packet or log secret contents. journal.record('operations-verified', evidence) send(control, {'complete': True}) # Keep the direct child alive until MainPID consumes the completion. require(control.recv(8) == b'RELEASE', 'transition release refused') record = children.spawn('collector', body, keep=(journal.fd,)) deadline = kernel.now() + 30_000_000_000 completed = False while not children.stopping: children.poll() require('collector' in children.children, 'transition helper exited before completion') require(kernel.now() < deadline, 'transition deadline expired; retirement remains fenced') ready, _, _ = select.select([record.channel], [], [], .05) if ready: packet = receive(record.channel, (record.pid, 0, 0)) if set(packet) == {'failure', 'reason'}: require(packet['failure'] in ('OSError', 'PermissionError', 'FileNotFoundError', 'ProcessLookupError', 'ValueError') and type(packet['reason']) is str and len(packet['reason']) <= 256, 'invalid transition diagnostic') record.channel.send(b'RELEASE') raise ValueError('access transition refused: ' + packet['reason']) if packet == {'complete': True}: record.channel.send(b'RELEASE') completed = True break require(set(packet) == {'progress'} and packet['progress'] in { 'preflight-started', 'intent-durable', 'frozen-inventoried-cohorts', 'masked-and-stopped-activation', 'installed-exact-access-policy', 'reloaded-ssh-retired-polkit-cache', 'kernel-observed-old-writer-exit'}, 'transition progress refused') print('C01_ACCESS_TRANSITION: ' + packet['progress'], flush=True) require(completed, 'transition interrupted') while children.children: children.poll() require(not children.stopping and kernel.now() < deadline, 'transition helper retirement unresolved') time.sleep(.02) require(children.results.get('collector') == 0, 'transition helper failed') value, _ = policy_module.json_read(journal.path) require(value['state'] == 'operations-verified', 'missing verified transition operations') journal.record('retired', value['evidence']) def retire_endpoint(evidence): """Offline only, after all former lifetimes retire under the persistent lock.""" policy_module.protected(BASE, directory=True) if not os.path.lexists(SOCKET): return meta = SOCKET.lstat() require(stat.S_ISSOCK(meta.st_mode) and (meta.st_uid, meta.st_gid) == (0, 0) and stat.S_IMODE(meta.st_mode) == 0o600 and meta.st_nlink == 1 and policy_module.identity(meta) == evidence.get('endpoint'), 'unowned stale access endpoint') with socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET | socket.SOCK_NONBLOCK) as probe: try: probe.connect(str(SOCKET)) except OSError as error: require(error.errno == errno.ECONNREFUSED, 'access endpoint retirement not proved') else: raise ValueError('former access listener remains active') current = SOCKET.lstat() require(policy_module.identity(current) == policy_module.identity(meta) and stat.S_ISSOCK(current.st_mode) and current.st_nlink == 1 and (current.st_mtime_ns, current.st_ctime_ns) == (meta.st_mtime_ns, meta.st_ctime_ns), 'stale access endpoint changed') SOCKET.unlink() def requalify(policy, journal, children): """Offline, fully quiescent owner restart; never a remote reset/unfence RPC. A partial migration cannot use this path. Existing runtime apps must already be stopped by their owning coordinator; this operation kills no process and does not restore old credentials, start apps, or change desired-state data. """ previous, _ = policy_module.json_read(journal.path) completed_history(previous, policy) require(previous['state'] == 'owner-active' or previous['boot'] != journal.boot, 'same-boot completed retirement already eligible') prior = previous['owner'] prior_group = policy_module.canonical(prior['group']) require(prior_group.is_relative_to(kernel.CGROUP) and prior_group.name == 'supervisor', 'invalid former owner hierarchy') if previous['boot'] == journal.boot: try: item = kernel.process(prior['pid']) except (FileNotFoundError, ProcessLookupError): pass else: require(item['start'] != prior['start'], 'former access owner remains alive') if prior_group.parent.exists(): require(not any((prior_group.parent / ('c01-' + role)).exists() for role in children_module.ROLES) and kernel.members(prior_group.parent) <= {os.getpid()}, 'former access helper retirement unresolved') for field in ('broker_cgroup', 'supervisor_cgroup'): group = Path(policy.value[field]) require(not group.exists() or (not kernel.members(group) and not any((group / ('c01-' + role)).exists() for role in children_module.ROLES)), 'consumer retirement unresolved') managed = {a['uid'] for a in policy.value['accounts'].values()} require(all(not managed.intersection(item['uids']) for item in kernel.all_processes()), 'offline requalification requires quiescent managed/runtime principals') # Fence before I/O helper work; interruption needs explicit offline review, # never an automatic retry that trusts the prior journal state. journal.record('requalification-pending', previous['evidence']) def body(control): check_current(policy, None) require(all(not managed.intersection(item['uids']) for item in kernel.all_processes()), 'managed process appeared during requalification') send(control, {'verified': True}) require(control.recv(8) == b'RELEASE', 'requalification release refused') record = children.spawn('collector', body) deadline = kernel.now() + 30_000_000_000 while True: children.poll() require(not children.stopping and 'collector' in children.children and kernel.now() < deadline, 'requalification helper failed/stalled') if select.select([record.channel], [], [], .05)[0]: require(receive(record.channel, (record.pid, 0, 0)) == {'verified': True}, 'requalification evidence refused') record.channel.send(b'RELEASE') break while children.children: children.poll() require(not children.stopping and kernel.now() < deadline, 'requalification helper retirement unresolved') time.sleep(.02) require(children.results.get('collector') == 0, 'requalification helper failed') retire_endpoint(previous['evidence']) evidence = {**previous['evidence'], 'requalification': {'prior_boot': previous['boot'], 'former_owner': prior, 'operation': 'quiescent-kernel-and-current-policy-reverified'}} journal.record('retired', evidence) def main(): require(len(sys.argv) == 2 and sys.argv[1] in ('serve', 'transition', 'requalify', 'resume-layout'), 'fixed access operation required') require(os.getresuid() == (0, 0, 0) and os.getresgid() == (0, 0, 0) and len(list(Path('/proc/self/task').iterdir())) == 1, 'single-threaded root owner required') os.umask(0o077) require(children_module.LIBC.prctl(4, 0, 0, 0, 0) == 0, 'owner dumpability protection unavailable') policy = policy_module.Policy() journal = Journal(policy) children = None try: children = children_module.Children() require(not any((children.root / ('c01-' + role)).exists() for role in children_module.ROLES), 'prior access helper cohort remains unresolved') # A crash can leave a prior checker/transition helper (including D-state) # behind. Children never adopts an existing role group. Journal blocks too. if sys.argv[1] == 'transition': transition(policy, journal, children) elif sys.argv[1] == 'resume-layout': transition(policy,journal,children,resume=True) elif sys.argv[1] == 'requalify': requalify(policy, journal, children) else: serve(policy, journal, children) finally: if children is not None: children.request_stop() while children.children or children.reserved: children.poll() time.sleep(.05) children.close() journal.close() # Only after owned helper/descendants really retire. if __name__ == '__main__': main()