"""Finite, root-appointed serial owner; all tools execute as ordinary runtime. Only startup configuration selects a device, retained HOME and fixed executable digests. Manager requests never contain executable/device/HOME locators. The opened TTY is never reopened, including after unplug. Raw serial descriptors are never granted to managers: bounded IO leases and owned helper groups share it. """ import array import collections import ctypes import fcntl import hashlib import json import os from pathlib import Path import re import resource import select import signal import shutil import socket import stat import struct import termios import time LIMIT = 65536 LIBC = ctypes.CDLL(None, use_errno=True) REGISTRY = None WORK = Path('/var/lib/archipelago-radio-work') class Busy(ValueError): pass def require(ok, detail): if not ok: raise ValueError(detail) def now(): return time.clock_gettime_ns(time.CLOCK_BOOTTIME) def exact(pairs): result = {} for key, value in pairs: require(key not in result, 'duplicate field') result[key] = value return result def receive(channel): raw, ancillary, flags, _ = channel.recvmsg(LIMIT, socket.CMSG_SPACE(12) + socket.CMSG_SPACE(64), socket.MSG_CMSG_CLOEXEC) credentials, rights = [], [] try: for level, kind, data in ancillary: if (level, kind) == (socket.SOL_SOCKET, socket.SCM_CREDENTIALS): credentials.append(struct.unpack('3i', data)) elif (level, kind) == (socket.SOL_SOCKET, socket.SCM_RIGHTS): values = array.array('i'); values.frombytes(data[:len(data)//4*4]); rights.extend(values) else: raise ValueError('unexpected control data') require(raw and not flags & ~socket.MSG_CMSG_CLOEXEC and len(credentials) == 1 and not rights, 'truncated/closed/aliased radio packet') return json.loads(raw, object_pairs_hook=exact), credentials[0] finally: for fd in rights: os.close(fd) def send(channel, value, rights=()): raw = json.dumps(value, separators=(',', ':')).encode() require(len(raw) <= LIMIT, 'radio reply bound') control = [(socket.SOL_SOCKET, socket.SCM_RIGHTS, array.array('i', rights))] if rights else [] require(channel.sendmsg([raw], control, socket.MSG_DONTWAIT | socket.MSG_NOSIGNAL) == len(raw), 'radio reply uncertain') def identity(fd): m = os.fstat(fd) return [m.st_dev, m.st_ino, m.st_rdev, m.st_ctime_ns] def protected(path, directory=False): path = Path(path) require(path.is_absolute() and path.resolve(strict=True) == path, 'canonical appointment required') for parent in path.parents: m = parent.lstat() require(m.st_uid == 0 and not m.st_mode & 0o022, 'mutable appointment ancestry') m = path.lstat() require(m.st_uid == 0 and not m.st_mode & 0o022 and (stat.S_ISDIR(m.st_mode) if directory else stat.S_ISREG(m.st_mode)), 'mutable appointment') return path class Appointment: def __init__(self, fd, path, expected, board): require(stat.S_ISCHR(os.fstat(fd).st_mode) and identity(fd) == expected, 'TTY capability identity differs') require(fcntl.fcntl(fd, fcntl.F_GETFL) & os.O_ACCMODE == os.O_RDWR and os.isatty(fd), 'read/write TTY capability required') require(board in ('heltecv3', 'heltecv4'), 'appointed board required') fcntl.ioctl(fd, termios.TIOCEXCL) # Linux asm-generic ioctl ABI (x86_64/aarch64); Python omits TIOCGEXCL. require(struct.unpack('i', fcntl.ioctl(fd, 0x80045440, struct.pack('i', 0)))[0] == 1, 'TTY exclusion absent') os.set_blocking(fd, False) self.fd, self.path, self.expected, self.board = fd, path, expected, board self.token = os.urandom(32).hex() self.failed = False def check(self): if identity(self.fd) != self.expected: self.failed = True poll = select.poll(); poll.register(self.fd, select.POLLERR | select.POLLHUP) if poll.poll(0): self.failed = True require(not self.failed, 'appointed radio disconnected; new external appointment required') def info(self): self.check() return {'path': self.path, 'identity': self.expected, 'token': self.token, 'board': self.board} class Job: """Root-owned cgroup keeps every helper descendant until actual retirement.""" def __init__(self, parent, argv, fds, environment, uid, gid, executable, input_bytes=b'', enter=lambda: None, admission_end=None): require(REGISTRY is None or not REGISTRY.exists(), 'prior radio reservation requires retirement') self.group = parent / ('radio-' + os.urandom(12).hex()) self.group.mkdir(mode=0o700) self.group_fd = os.open(self.group, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC) require(os.fstat(self.group_fd).st_uid == 0 and stat.S_IMODE(os.fstat(self.group_fd).st_mode) == 0o700, 'radio group custody changed') protected(WORK, directory=True) self.workspace = WORK/self.group.name self.workspace.mkdir(mode=0o700) os.chown(self.workspace, uid, gid) temporary = os.open(self.workspace, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC) work_meta = os.fstat(temporary) self.work_identity = [work_meta.st_dev, work_meta.st_ino] fds = {**fds, 8: temporary} environment = {**environment, 'TMPDIR':'/proc/self/fd/8'} if REGISTRY is not None: record = os.open(REGISTRY, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC, 0o600) with os.fdopen(record, 'w') as stream: m = os.fstat(self.group_fd) json.dump({'path':str(self.group), 'identity':[m.st_dev,m.st_ino], 'work_identity':self.work_identity}, stream) stream.flush(); os.fsync(stream.fileno()) self.log = collections.deque(maxlen=128) self.output = bytearray() self.status = None self.cancelled = False self.began = now() self.limit_ns = (None if argv[1:2] == ['--archy-owned-daemon'] else 30_000_000_000 if len(argv) > 2 and argv[2] in ('capabilities','verify') else 1_800_000_000_000) read, write = os.pipe2(os.O_CLOEXEC | os.O_NONBLOCK) self.read = read stdin = os.memfd_create('radio-input', os.MFD_CLOEXEC) os.write(stdin, input_bytes); os.lseek(stdin, 0, 0) parent_pid = os.getpid() self.pid = os.fork() if self.pid == 0: try: self.control('cgroup.procs', str(os.getpid())) enter() os.setsid() require(LIBC.prctl(1, signal.SIGKILL, 0, 0, 0) == 0, 'helper parent death guard') copies = {target: fcntl.fcntl(source, fcntl.F_DUPFD_CLOEXEC, 32) for target, source in {0: stdin, 1: write, 2: write, **fds}.items()} exe = fcntl.fcntl(executable, fcntl.F_DUPFD_CLOEXEC, 32) for target, source in copies.items(): os.dup2(source, target, inheritable=True) os.set_blocking(1, True) resource.setrlimit(resource.RLIMIT_AS, (1024**3,1024**3)) resource.setrlimit(resource.RLIMIT_FSIZE, (256*1024**2,256*1024**2)) for item in os.listdir('/proc/self/fd'): n = int(item) if n not in {*copies, exe}: try: os.close(n) except OSError: pass os.setgroups([]); os.setgid(gid); os.setuid(uid) require(os.getresuid() == (uid,)*3 and os.getresgid() == (gid,)*3 and not os.getgroups(), 'helper credentials') # setuid clears PDEATHSIG; establish it after the final drop. require(LIBC.prctl(1, signal.SIGKILL, 0, 0, 0) == 0 and os.getppid() == parent_pid, 'helper parent retired during credential drop') require(LIBC.prctl(38, 1, 0, 0, 0) == 0 and LIBC.prctl(4, 0, 0, 0, 0) == 0, 'helper exclusion') status = dict(row.split(':',1) for row in Path('/proc/self/status').read_text().splitlines() if ':' in row) require(all(int(status[key],16) == 0 for key in ('CapEff','CapPrm','CapInh','CapAmb')), 'helper retained capabilities') os.chdir('/') require(admission_end is not None and now() < admission_end, 'late radio helper admission') os.execve(exe, argv, environment) except BaseException: os._exit(125) os.close(write); os.close(stdin); os.close(temporary) def kill(self): self.cancelled = True if os.fstat(self.group_fd).st_nlink != 0: self.control('cgroup.kill', '1') def control(self, name, text=None): fd = os.open(name, (os.O_RDONLY if text is None else os.O_WRONLY) | os.O_NOFOLLOW | os.O_CLOEXEC, dir_fd=self.group_fd) with os.fdopen(fd, 'r' if text is None else 'w') as stream: return stream.read(4096) if text is None else stream.write(text) def tick(self): if self.limit_ns is not None and now()-self.began >= self.limit_ns: self.kill() for _ in range(4): try: data = os.read(self.read, 4096) except BlockingIOError: break if not data: break self.output.extend(data) if len(self.output) > 8192: del self.output[:-8192] if self.status is None: result = os.waitid(os.P_PID, self.pid, os.WEXITED | os.WNOHANG | os.WNOWAIT) if result is not None: self.status = result.si_status if result.si_code == os.CLD_EXITED else -result.si_status if os.fstat(self.group_fd).st_nlink == 0: # Kernel removal of the retained directory requires an empty tree. # This is stronger than a missing pathname (which could be rename). require(self.status is not None, 'removed radio group has live leader') cleanup_workspace(self.group.name, self.work_identity) os.waitpid(self.pid,0); os.close(self.group_fd); os.close(self.read) if REGISTRY is not None: REGISTRY.unlink() return True if self.cancelled or self.status is not None: if self.status == 0 and 'populated 1' in self.control('cgroup.events'): self.status = 125 # A zero-exit leader is not completed writing. self.control('cgroup.kill', '1') if self.status is None or 'populated 1' in self.control('cgroup.events'): return False cleanup_workspace(self.group.name, self.work_identity) os.waitpid(self.pid, 0) before, after = os.fstat(self.group_fd), self.group.lstat() require((before.st_dev, before.st_ino) == (after.st_dev, after.st_ino), 'radio group locator replaced') self.group.rmdir() os.close(self.group_fd) if REGISTRY is not None: REGISTRY.unlink() os.close(self.read) return True def cleanup_workspace(name, expected): require(re.fullmatch('radio-[a-f0-9]{24}',name), 'radio workspace scope') protected(WORK, directory=True) require(stat.S_IMODE(WORK.stat().st_mode)==0o700, 'radio workspace parent is not private') path=WORK/name try: actual=path.lstat() except FileNotFoundError: return # Only root can rename/remove this entry. require(stat.S_ISDIR(actual.st_mode) and [actual.st_dev,actual.st_ino]==expected, 'radio workspace identity changed') require(shutil.rmtree.avoids_symlink_attacks, 'descriptor-relative workspace retirement required') shutil.rmtree(path) class Owner: def __init__(self, appointment, uid, gid, home, tools, groups, custody, manager, enter=lambda: None, heartbeat=lambda: None): require(os.geteuid() == 0 and uid > 0 and gid > 0, 'independent root owner and ordinary runtime required') require(LIBC.prctl(36, 1, 0, 0, 0) == 0, 'radio subreaper') self.radio, self.uid, self.gid = appointment, uid, gid self.home, self.tools, self.groups, self.custody = home, tools, groups, custody require(len(manager)==2 and all(type(n) is int and n>0 for n in manager) and manager[0]!=uid, 'distinct appointed manager required') self.manager=tuple(manager) self.enter, self.heartbeat = enter, heartbeat self.clients = {} self.lease = None self.job = None self.job_result = None self.image = None def clear_image(self): if self.image is not None: os.close(self.image['fd']); self.image = None def add(self, channel, app=None, session=False): require(len(self.clients) < 8, 'radio channel limit') channel.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1) channel.setblocking(False) self.clients[channel.fileno()] = {'channel': channel, 'app': app, 'sequence': 0, 'session': session, 'expiry': now()+120_000_000_000, 'pidfd': os.pidfd_open(app[0]) if app else None} def retire(self, number): state = self.clients.pop(number) state['channel'].close() if state['pidfd'] is not None: os.close(state['pidfd']) if number == self.lease: if self.job is not None: self.job.kill() # lease is retained until populated=0 below else: self.lease = None self.job_result = None self.clear_image() def packet(self, number): state = self.clients[number] channel = state['channel'] require(state['pidfd'] is None or not select.select([state['pidfd']], [], [], 0)[0], 'radio caller incarnation retired') request, peer = receive(channel) require(now() < state['expiry'], 'late queued radio request') if state['app'] is None: require(peer[1:] == (0, 0) and set(request) == {'bind'}, 'root appointment caller required') app = request['bind'] require(type(app) is list and len(app) == 3 and all(type(n) is int and n > 0 for n in app) and tuple(app[1:]) == self.manager, 'appointed application credentials required') status = dict(row.split(':', 1) for row in Path(f'/proc/{app[0]}/status').read_text().splitlines() if ':' in row) require(list(map(int, status['Uid'].split())) == [app[1]]*4 and list(map(int, status['Gid'].split())) == [app[2]]*4 and int(status['NoNewPrivs'])==1 and all(int(status[key],16)==0 for key in ('CapEff','CapPrm','CapInh','CapAmb')), 'application process identity/privileges changed') state['app'] = tuple(app); state['pidfd'] = os.pidfd_open(app[0]) state['expiry'] = 2**63-1 send(channel, {'bound': app}); return require(peer == state['app'], 'radio packet caller differs') require(type(request) is dict and set(request) == {'v', 'seq', 'end', 'op', 'args'} and request['v'] == 1 and type(request['seq']) is int and request['seq'] == state['sequence']+1 and type(request['end']) is int and now() < request['end'] <= now()+2_000_000_000, 'radio request version/sequence/boottime bound') state['sequence'] += 1 self.admission_end = request['end'] state['expiry'] = now()+30_000_000_000 if state['session'] else 2**63-1 op, args = request['op'], request['args'] require(type(args) is dict, 'radio arguments object') rights = [] if op == 'inventory' and not args and not state['session']: value = self.radio.info() elif op == 'rf' and not args and not state['session']: try: value = self.retained_rf() except FileNotFoundError: value = None except (OSError, ValueError): raise Busy('retained-rf-invalid') elif op == 'acquire' and not state['session']: require(args == {'token': self.radio.token}, 'stale radio selection') if self.lease is not None: raise Busy('busy') self.radio.check() owned, grant = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET | socket.SOCK_CLOEXEC) self.add(owned, state['app'], True) self.clients[owned.fileno()]['expiry'] = now()+30_000_000_000 self.lease = owned.fileno(); self.job_result = None rights = [grant.fileno()]; value = self.radio.info() else: require(state['session'] and self.lease == number, 'radio IO requires current exclusive session') if op == 'poll' and not args: value = {'done': self.job is None, 'result': self.job_result, 'output': bytes(self.job.output).decode(errors='replace') if self.job else ''} elif op == 'cancel' and not args: if self.job: self.job.kill() value = True else: require(self.job is None, 'helper still owns radio') self.radio.check() value = self.operation(op, args) if isinstance(value, tuple): value, grant = value rights = [grant.fileno()] require(now() < request['end'], 'radio response expired') try: send(channel, {'v': 1, 'seq': request['seq'], 'value': value}, rights) finally: if rights: grant.close() def operation(self, op, args): require(now() < self.admission_end, 'expired serial operation') fd = self.radio.fd if op == 'daemon': require(set(args) == {'identity', 'display_name', 'ed_pub', 'x_pub', 'rf'}, 'radio daemon schema') require(type(args['identity']) is str and re.fullmatch('[a-f0-9]{128}', args['identity']), 'derived radio identity required') require(type(args['display_name']) is str and len(args['display_name'].encode()) <= 256, 'radio name bound') for field in ('ed_pub', 'x_pub'): require(args[field] is None or (type(args[field]) is str and re.fullmatch('[a-f0-9]{64}', args[field])), 'public identity binding') rf = args['rf'] fields = {'frequency': (137000000,1020000000), 'spreading_factor': (5,12), 'coding_rate':(5,8), 'txpower':(0,22)} require(type(rf) is dict and set(rf) == {*fields, 'bandwidth', 'airtime_limit_short', 'airtime_limit_long'}, 'radio RF schema') for key, (lo, hi) in fields.items(): require(type(rf[key]) is int and lo <= rf[key] <= hi, 'radio RF range') require(type(rf['bandwidth']) is int and rf['bandwidth'] in (7800,10400,15600,20800,31250,41700,62500,125000,250000,500000), 'radio bandwidth') for key in ('airtime_limit_short', 'airtime_limit_long'): require(rf[key] is None or (type(rf[key]) in (int,float) and 0 <= rf[key] <= 100), 'radio airtime range') self.custody() config = os.memfd_create('radio-daemon-config', os.MFD_CLOEXEC | os.MFD_ALLOW_SEALING) own, grant = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM | socket.SOCK_CLOEXEC) try: raw = json.dumps({**args, 'path': self.radio.path}, separators=(',', ':')).encode() require(os.write(config, raw) == len(raw), 'daemon config short write') os.lseek(config, 0, 0) fcntl.fcntl(config, fcntl.F_ADD_SEALS, fcntl.F_SEAL_WRITE | fcntl.F_SEAL_GROW | fcntl.F_SEAL_SHRINK | fcntl.F_SEAL_SEAL) self.job = Job(self.groups, ['/usr/local/bin/archy-reticulum-daemon', '--archy-owned-daemon'], {3: fd, 4: self.home, 5: config, 6: own.fileno()}, {'HOME':'/proc/self/fd/4','PATH':'/usr/bin:/bin','LANG':'C.UTF-8'}, self.uid, self.gid, self.tools['daemon'], enter=self.enter, admission_end=self.admission_end) except BaseException as error: grant.close() raise RuntimeError('radio daemon reservation requires owner retirement') from error finally: os.close(config); own.close() self.job_result = None return True, grant if op == 'image': require(set(args) == {'size', 'sha256'} and type(args['size']) is int and 0 < args['size'] <= 16*1024*1024 and type(args['sha256']) is str and re.fullmatch('[a-f0-9]{64}', args['sha256']), 'image bound/digest') self.clear_image() self.image = {**args, 'written': 0, 'fd': os.memfd_create('radio-image', os.MFD_CLOEXEC | os.MFD_ALLOW_SEALING)} return True if op == 'image-chunk': require(self.image is not None and set(args) == {'data'} and type(args['data']) is str and len(args['data']) <= 8192, 'image chunk bound') data = bytes.fromhex(args['data']) require(0 < len(data) <= self.image['size']-self.image['written'], 'image chunk exceeds appointment') require(os.write(self.image['fd'], data) == len(data), 'image short write') self.image['written'] += len(data) return True if op == 'read': require(set(args) == {'count'} and type(args['count']) is int and 0 < args['count'] <= 4096, 'serial read bound') try: data = os.read(fd, args['count']) except BlockingIOError: data = b'' return data.hex() if op == 'write': require(set(args) == {'data'} and type(args['data']) is str and len(args['data']) <= 8192, 'serial write bound') data = bytes.fromhex(args['data']) try: return os.write(fd, data) except BlockingIOError: return 0 if op == 'configure': require(args == {'baud': 115200}, 'serial configuration not supported') import tty tty.setraw(fd, termios.TCSANOW) attrs = termios.tcgetattr(fd); attrs[4] = attrs[5] = termios.B115200 attrs[2] |= termios.CLOCAL | termios.CREAD termios.tcsetattr(fd, termios.TCSANOW, attrs) return True if op in ('dtr', 'rts'): require(set(args) == {'enabled'} and type(args['enabled']) is bool, 'serial modem flag') bit = termios.TIOCM_DTR if op == 'dtr' else termios.TIOCM_RTS fcntl.ioctl(fd, termios.TIOCMBIS if args['enabled'] else termios.TIOCMBIC, struct.pack('i', bit)) return True if op == 'run': require(set(args) == {'action', 'board', 'band', 'nonce'}, 'radio tool schema') require(args['board'] == self.radio.board and args['band'] in ('1', '2', '3') and re.fullmatch('[a-f0-9]{32}', args['nonce']) is not None, 'radio tool binding') action = args['action'] require(action in ('capabilities', 'install', 'verify', 'write', 'write-slow'), 'fixed radio action required') key_snapshot = self.custody() tool = self.tools['rnode'] argv = ['/usr/local/bin/archy-rnodeconf', '--archy-owned', action, self.radio.path, self.radio.board, args['nonce']] stdin = ((('8' if self.radio.board == 'heltecv3' else '9')+'\n\n'+args['band']+'\ny\n').encode() if action == 'install' else b'') fds = {3: fd, 4: self.home} key_fd = None if action in ('install', 'verify') and key_snapshot is not None: key_fd = os.memfd_create('retained-radio-keys', os.MFD_CLOEXEC | os.MFD_ALLOW_SEALING) raw = json.dumps(key_snapshot, separators=(',', ':')).encode() require(len(raw) <= 524288 and os.write(key_fd, raw) == len(raw), 'radio key snapshot bound') os.lseek(key_fd, 0, 0) fcntl.fcntl(key_fd, fcntl.F_ADD_SEALS, fcntl.F_SEAL_WRITE | fcntl.F_SEAL_GROW | fcntl.F_SEAL_SHRINK | fcntl.F_SEAL_SEAL) fds[7] = key_fd if action in ('write', 'write-slow'): require(self.image is not None and self.image['written'] == self.image['size'], 'complete image required') image = self.image['fd'] seals = fcntl.F_SEAL_WRITE | fcntl.F_SEAL_GROW | fcntl.F_SEAL_SHRINK | fcntl.F_SEAL_SEAL if fcntl.fcntl(image, fcntl.F_GET_SEALS) & seals != seals: fcntl.fcntl(image, fcntl.F_ADD_SEALS, seals) os.lseek(image, 0, 0) with os.fdopen(os.dup(image), 'rb') as stream: require(hashlib.file_digest(stream, 'sha256').hexdigest() == self.image['sha256'], 'image digest mismatch') os.lseek(image, 0, 0) fds[5] = image tool = self.tools['esptool'] argv = ['/usr/local/bin/archy-esptool', '--archy-owned', self.radio.path, '--chip', 'esp32s3', '--port', self.radio.path, *(['--baud','115200'] if action == 'write-slow' else []), 'write_flash', '--erase-all', '0x0', '/proc/self/fd/5'] try: self.job = Job(self.groups, argv, fds, {'HOME': '/proc/self/fd/4', 'PATH': '/usr/bin:/bin', 'LANG': 'C.UTF-8', 'TERM': 'dumb'}, self.uid, self.gid, tool, stdin, self.enter, self.admission_end) except BaseException as error: raise RuntimeError('radio helper reservation requires owner retirement') from error finally: if key_fd is not None: os.close(key_fd) self.job_result = None return True raise ValueError('unknown serial operation') def retained_rf(self): directory = os.open('.archy-reticulum', os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC, dir_fd=self.home) try: fd = os.open('config', os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC | os.O_NONBLOCK, dir_fd=directory) try: require(stat.S_ISREG(os.fstat(fd).st_mode) and os.fstat(fd).st_size <= 65536, 'retained RF config bound') raw = os.read(fd, 65537).decode() finally: os.close(fd) finally: os.close(directory) sections = re.split(r'^\s*\[\[.*?\]\]\s*$', raw, flags=re.M) found = [] mapping = {'frequency':'frequency','bandwidth':'bandwidth','spreadingfactor':'spreading_factor', 'codingrate':'coding_rate','txpower':'txpower','airtime_limit_short':'airtime_limit_short','airtime_limit_long':'airtime_limit_long'} for section in sections: pairs = {} for line in section.splitlines(): key, sep, text = line.partition('=') if sep: key, text = key.strip().lower(), text.strip().strip('"\'') require(key not in pairs, 'ambiguous retained RF settings') pairs[key] = text if pairs.get('type') != 'RNodeInterface' or pairs.get('port') != self.radio.path: continue result = {'enabled': pairs.get('enabled','yes').lower() in ('yes','true','1'), 'port':self.radio.path} for source, target in mapping.items(): if source in pairs: result[target] = float(pairs[source]) if source.startswith('airtime') else int(pairs[source]) require(all(key in result for key in ('frequency','bandwidth','spreading_factor','coding_rate','txpower')), 'incomplete retained RF settings') found.append(result) require(len(found) == 1, 'retained RF selection ambiguous or missing') return found[0] def tick(self, check_lease=True): if check_lease: self.heartbeat() if self.job is not None: if self.job.tick(): self.job_result = {'code': self.job.status, 'output': bytes(self.job.output).decode(errors='replace')} self.job = None try: self.custody() except (OSError, ValueError): self.radio.failed = True self.job_result = {'code': 125, 'output': 'Retained signing/trust custody changed; appointment retired'} if self.lease not in self.clients: self.lease = None self.clear_image() # Reap adopted grandchildren, never the pinned job leader. for text in Path(f'/proc/self/task/{os.getpid()}/children').read_text().split()[:128]: pid = int(text) if self.job is None or pid != self.job.pid: try: os.waitpid(pid, os.WNOHANG) except ChildProcessError: pass for number, state in list(self.clients.items()): dead = state['pidfd'] is not None and bool(select.select([state['pidfd']], [], [], 0)[0]) if dead or now() >= state['expiry']: self.retire(number) def serve(self, listener): listener.setblocking(False) self.stopping = False def stop(*_): self.stopping = True signal.signal(signal.SIGTERM,stop) signal.signal(signal.SIGINT,stop) try: while not self.stopping: self.tick() channels = [listener, *(state['channel'] for state in self.clients.values())] for channel in select.select(channels, [], [], .02)[0]: if channel is listener: client, _ = listener.accept() try: self.add(client) except ValueError: client.close() else: number = channel.fileno() try: self.packet(number) except Busy as error: send(channel, {'v': 1, 'seq': self.clients[number]['sequence'], 'error': str(error)}) except (OSError, ValueError, KeyError, TypeError): self.retire(number) finally: listener.close() for number in list(self.clients): self.retire(number) while self.job is not None: self.tick(False); time.sleep(.02) def main(): global REGISTRY require(os.getresuid() == (0, 0, 0), 'root radio appointment required') REGISTRY = Path('/run/archipelago-radio/job.json') require(not REGISTRY.exists() and not Path('/run/archipelago-radio/control.sock').exists(), 'prior serial lifetime requires retirement') protected(WORK, directory=True) require(stat.S_IMODE(WORK.stat().st_mode)==0o700 and not list(WORK.iterdir()), 'prior radio workspace requires external reconciliation') import c01_runtime import c01_runtime_view runtime, digest = c01_runtime.load_config() lease = c01_runtime_view.RuntimeLease(runtime['manager_uid'], runtime['manager_gid'], runtime['runtime_uid'], runtime['runtime_gid'], digest) namespace_fds = [os.open(f'/proc/{lease.identity["pid"]}/ns/{name}', os.O_RDONLY | os.O_CLOEXEC) for name in ('mnt', 'net')] lease.check() path = protected('/etc/archipelago-c01/radio.json') require(path.stat().st_size <= 16384 and stat.S_IMODE(path.stat().st_mode) == 0o600, 'radio config bound') value = json.loads(path.read_bytes(), object_pairs_hook=exact) require(set(value) == {'version', 'path', 'identity', 'board', 'runtime_uid', 'runtime_gid', 'home', 'home_identity', 'custody', 'rnode_sha256', 'esptool_sha256', 'daemon_sha256'} and value['version'] == 1, 'radio appointment schema') require((value['runtime_uid'], value['runtime_gid']) == (runtime['runtime_uid'], runtime['runtime_gid']), 'radio runtime differs from C01 appointment') require(re.fullmatch('/dev/tty(?:USB|ACM)[0-9]+', value['path']) is not None, 'physical USB TTY appointment required') device = os.open(value['path'], os.O_RDWR | os.O_NOFOLLOW | os.O_NOCTTY | os.O_NONBLOCK | os.O_CLOEXEC) appointment = Appointment(device, value['path'], value['identity'], value['board']) # TIOCEXCL excludes future ordinary opens; reject pre-existing openers too. # Root remains the external appointment authority, never an RPC principal. processes = [p for p in Path('/proc').iterdir() if p.name.isdecimal()] require(len(processes) <= 8192, 'process exclusion inventory bound') for process in processes: if int(process.name) == os.getpid(): continue try: for entry in (process/'fd').iterdir(): try: other = entry.stat() require(not (stat.S_ISCHR(other.st_mode) and other.st_rdev == os.fstat(device).st_rdev), 'radio already held by another process') except FileNotFoundError: pass except FileNotFoundError: pass home_path = Path(value['home']) require(home_path.is_absolute() and home_path.resolve(strict=True)==home_path, 'canonical retained radio HOME required') require(all(not home_path.is_relative_to(Path(parent)) and not Path(parent).is_relative_to(home_path) for parent in runtime['private_parents']), 'radio HOME overlaps C01 private custody') protected(home_path.parent, directory=True) home = os.open(home_path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC) m = os.fstat(home) require([m.st_dev, m.st_ino] == value['home_identity'] and m.st_uid == value['runtime_uid'] and stat.S_IMODE(m.st_mode) == 0o700, 'retained runtime HOME binding differs') if home_path != Path(runtime['home_backing']): # A separate retained radio store is allowed; an operator's entire HOME # is not. Neither path may grant manager/authority/wallet custody. require(set(os.listdir(home)) <= {'.config','.archy-reticulum'}, 'radio-specific HOME contains unrelated data') config_dir=os.open('.config',os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC,dir_fd=home) try: require(set(os.listdir(config_dir)) == {'rnodeconf'}, 'radio-specific config contains unrelated credentials') finally: os.close(config_dir) require(type(value['custody']) is dict and {'.config/rnodeconf/firmware/signing.key', '.config/rnodeconf/firmware/device.key'} <= value['custody'].keys(), 'existing RNode signing key custody required') def custody(): actual = {'.config/rnodeconf/firmware/signing.key', '.config/rnodeconf/firmware/device.key'} trusted = home_path/'.config/rnodeconf/trusted_keys' if trusted.exists(): require(trusted.resolve(strict=True) == trusted, 'trusted key directory alias') actual.update(str(p.relative_to(home_path)) for p in trusted.iterdir() if p.name.endswith('.pubkey')) require(actual == set(value['custody']) and len(actual) <= 32, 'signing/trust inventory changed') snapshot = {} for name, digest in value['custody'].items(): require(name.startswith('.config/rnodeconf/') and '..' not in Path(name).parts and not Path(name).is_absolute(), 'custody name') parent = os.dup(home) try: parts = Path(name).parts for part in parts[:-1]: child = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC, dir_fd=parent) os.close(parent); parent = child key = os.open(parts[-1], os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | os.O_CLOEXEC, dir_fd=parent) try: before = os.fstat(key) require(stat.S_ISREG(before.st_mode) and before.st_size <= 65536, 'retained key object bound') raw = os.read(key,65537) after = os.fstat(key) fields = ('st_dev','st_ino','st_mode','st_uid','st_gid','st_size','st_mtime_ns','st_ctime_ns') require(all(getattr(before,k) == getattr(after,k) for k in fields) and len(raw) == before.st_size and hashlib.sha256(raw).hexdigest() == digest, 'retained RNode key/trust data changed') snapshot[name.removeprefix('.config/rnodeconf/')] = raw.hex() finally: os.close(key) finally: os.close(parent) return snapshot custody() tools = {} for name, executable in (('rnode', 'archy-rnodeconf'), ('esptool', 'archy-esptool'), ('daemon', 'archy-reticulum-daemon')): tool_path = protected('/usr/local/bin/'+executable) tool = os.open(tool_path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC) with os.fdopen(os.dup(tool), 'rb') as stream: require(hashlib.file_digest(stream, 'sha256').hexdigest() == value[name+'_sha256'], 'packaged radio helper differs') os.lseek(tool, 0, 0); tools[name] = tool own_group = Path('/sys/fs/cgroup') / Path('/proc/self/cgroup').read_text().strip().removeprefix('0::/') require(own_group.name == 'archipelago-radio-owner.service', 'dedicated radio owner service required') group = Path(lease.identity['cgroup']) if group.name == 'owner': group = group.parent require(group.name == 'archipelago-runtime-worker.service' and group.is_relative_to(Path('/sys/fs/cgroup')/runtime['user_cgroup'].lstrip('/')), 'radio helpers require the existing exclusive runtime hierarchy') def enter(): require(not select.select([lease.pidfd], [], [], 0)[0] and now() < lease.expiry, 'runtime lease expired before serial exec') require(LIBC.setns(namespace_fds[0], 0x00020000) == 0 and LIBC.setns(namespace_fds[1], 0x40000000) == 0, 'radio helper private runtime views unavailable') last_notify = 0 def heartbeat(): nonlocal last_notify lease.check() if now()-last_notify > 1_000_000_000: c01_runtime.notify(b'WATCHDOG=1'); last_notify = now() directory = protected('/run/archipelago-radio', directory=True) listener = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET | socket.SOCK_CLOEXEC) listener.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1) listener.bind(str(directory/'control.sock')); os.chmod(directory/'control.sock', 0o600); listener.listen(8) c01_runtime.notify(b'READY=1') Owner(appointment, value['runtime_uid'], value['runtime_gid'], home, tools, group, custody, (runtime['manager_uid'],runtime['manager_gid']), enter, heartbeat).serve(listener) def retire_main(): """ExecStopPost recovery, including MainPID SIGKILL. No replacement over a stuck cgroup. The root-only journal pins the one possible helper lifetime. """ require(os.getresuid() == (0,0,0), 'root serial retirement required') directory = protected('/run/archipelago-radio', directory=True) endpoint = directory/'control.sock' if endpoint.exists(): probe = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) probe.settimeout(1) try: try: probe.connect(str(endpoint)) except ConnectionRefusedError: pass else: raise ValueError('live radio owner must retire its own job') finally: probe.close() journal = directory/'job.json' if journal.exists(): import c01_runtime runtime, _ = c01_runtime.load_config() protected(journal) require(journal.stat().st_size <= 4096, 'radio retirement journal bound') value = json.loads(journal.read_bytes(), object_pairs_hook=exact) require(set(value) == {'path','identity','work_identity'}, 'radio retirement schema') group = Path(value['path']) root = Path('/sys/fs/cgroup')/runtime['user_cgroup'].lstrip('/') require(group.is_relative_to(root) and group.parent.name == 'archipelago-runtime-worker.service' and re.fullmatch('radio-[a-f0-9]{24}',group.name) and '..' not in group.parts, 'radio retirement scope') try: fd = os.open(group, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC) except FileNotFoundError: # Absence alone cannot distinguish retirement from a renamed group. raise ValueError('radio lifetime locator missing; external retirement evidence required') else: try: m = os.fstat(fd) require([m.st_dev,m.st_ino] == value['identity'] and m.st_uid == 0, 'radio retirement identity changed') while True: kill = os.open('cgroup.kill',os.O_WRONLY | os.O_CLOEXEC,dir_fd=fd) try: os.write(kill,b'1') finally: os.close(kill) events = os.open('cgroup.events',os.O_RDONLY | os.O_CLOEXEC,dir_fd=fd) try: empty = b'populated 0' in os.read(events,4096) finally: os.close(events) if empty: break time.sleep(.02) actual = group.lstat() require([actual.st_dev,actual.st_ino] == value['identity'], 'radio retirement locator replaced') cleanup_workspace(group.name,value['work_identity']) group.rmdir(); journal.unlink() finally: os.close(fd) if endpoint.exists(): m = endpoint.lstat() require(stat.S_ISSOCK(m.st_mode) and m.st_uid == 0, 'radio endpoint retirement refused') endpoint.unlink()