"""Fixed two-hop FD10 transfer. Runs on the single-threaded C01 supervisor. All waits keep ticking the existing child/lease owners. No helper thread, caller path, executable selection or update grant is introduced by registration. """ import array import json import os from pathlib import Path import select import socket import stat import struct import time SOCKET = Path('/run/archipelago-d09/c01.sock') LIMIT = 65536 def canonical(value): return json.dumps(value, sort_keys=True, separators=(',', ':'), ensure_ascii=True, allow_nan=False).encode('ascii') def receive(channel, peer, expected_fds, decode=True): raw, ancillary, flags, _ = channel.recvmsg(LIMIT+1, socket.CMSG_SPACE(12)+socket.CMSG_SPACE(64), socket.MSG_CMSG_CLOEXEC) fds, credentials, extra = [], [], False try: for level, kind, value in ancillary: if (level, kind) == (socket.SOL_SOCKET, socket.SCM_RIGHTS): rights = array.array('i'); rights.frombytes(value[:len(value)-len(value)%4]); fds.extend(rights) elif (level, kind) == (socket.SOL_SOCKET, socket.SCM_CREDENTIALS) and len(value) == 12: credentials.append(struct.unpack('3i', value)) else: extra = True if not raw or len(raw)>LIMIT or extra or flags & ~socket.MSG_CMSG_CLOEXEC or credentials != [peer] or len(fds) != expected_fds: raise ValueError('D09 transfer credentials/descriptor count') value = json.loads(raw) if decode else raw if decode and canonical(value) != raw: raise ValueError('D09 transfer noncanonical/duplicate fields') result, fds = fds, [] return value, result finally: for fd in fds: os.close(fd) def start_ticks(pid): raw = Path(f'/proc/{pid}/stat').read_text() return int(raw[raw.rindex(')')+2:].split()[19]) def register(owner, app_pid, deadline): for parent in reversed(SOCKET.parents): meta = parent.lstat() if not stat.S_ISDIR(meta.st_mode) or meta.st_uid != 0 or meta.st_mode & 0o022: raise ValueError('D09 registration ancestry') while owner.tick() and time.clock_gettime_ns(time.CLOCK_BOOTTIME) < deadline: channel = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET | socket.SOCK_CLOEXEC) channel.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1) channel.setblocking(False) try: meta = SOCKET.lstat() if not stat.S_ISSOCK(meta.st_mode) or meta.st_uid != 0 or meta.st_gid != 0 or meta.st_mode & 0o077: raise ValueError('D09 registration socket') channel.connect(str(SOCKET)) break except (FileNotFoundError, ConnectionRefusedError, BlockingIOError): channel.close() time.sleep(.05) else: raise ValueError('D09 registration unavailable/retiring') try: peer = struct.unpack('3i', channel.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)) if peer[0] <= 1 or peer[1:] != (0,0) or peer[0] == os.getpid(): raise ValueError('independent D09 root owner required') group = Path(f'/proc/{peer[0]}/cgroup').read_text().strip() if group != '0::/system.slice/archipelago-d09-lifecycle.service': raise ValueError('D09 owner service identity') owner_pidfd = channel.getsockopt(socket.SOL_SOCKET, 77) # SO_PEERPIDFD, no PID-reuse window os.set_inheritable(owner_pidfd,False) try: ticks = start_ticks(peer[0]) child=owner.children.children['application'] group=child.group.path.stat() pinned=os.fstat(child.group.events_fd) named=(child.group.path/'cgroup.events').stat() if child.pid!=app_pid or (pinned.st_dev,pinned.st_ino)!=(named.st_dev,named.st_ino): raise ValueError('owned application cgroup changed before handoff') raw = canonical({'version':1, 'application_pid':app_pid,'group_device':group.st_dev,'group_inode':group.st_ino}) if channel.send(raw) != len(raw): raise ValueError('D09 registration submit uncertain') while owner.tick() and time.clock_gettime_ns(time.CLOCK_BOOTTIME) < deadline: ready, _, _ = select.select([channel, owner_pidfd], [], [], .05) if owner_pidfd in ready: raise ValueError('D09 owner exited') if channel in ready: metadata, rights = receive(channel, peer, 1) fd = rights[0] try: if metadata.get('owner_pid') != peer[0] or metadata.get('owner_start_ticks') != ticks or metadata.get('supervisor_pid') != os.getpid() or metadata.get('application_pid') != app_pid or metadata.get('application_abi') != 4 or select.select([owner_pidfd],[],[],0)[0]: raise ValueError('D09 session identity') return metadata, fd except BaseException: os.close(fd) raise raise ValueError('D09 registration deadline/retirement') finally: os.close(owner_pidfd) finally: channel.close() def send_child(channel, metadata, fd): raw = canonical({'method':'D09_FD', 'session':metadata}) if channel.sendmsg([raw], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, array.array('i',[fd]))], socket.MSG_DONTWAIT | socket.MSG_NOSIGNAL) != len(raw): raise ValueError('D09 child transfer uncertain') def receive_child(channel, parent, deadline): while time.clock_gettime_ns(time.CLOCK_BOOTTIME) < deadline: if select.select([channel], [], [], .05)[0]: message, rights = receive(channel, (parent,0,0), 1) fd = rights[0] try: if set(message) != {'method','session'} or message['method'] != 'D09_FD': raise ValueError('D09 child transfer envelope') with socket.socket(fileno=os.dup(fd)) as endpoint: peer = struct.unpack('3i', endpoint.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED,12)) if endpoint.getsockopt(socket.SOL_SOCKET,socket.SO_TYPE) != socket.SOCK_SEQPACKET or endpoint.getsockopt(socket.SOL_SOCKET,socket.SO_PASSCRED) != 1 or peer != (message['session']['owner_pid'],0,0): raise ValueError('D09 endpoint creator/type') return fd except BaseException: os.close(fd) raise raise ValueError('D09 child transfer deadline') def wait_exec(channel, parent, deadline): while time.clock_gettime_ns(time.CLOCK_BOOTTIME) < deadline: if select.select([channel], [], [], .05)[0]: raw, _ = receive(channel, (parent,0,0), 0, decode=False) if raw != b'EXEC': raise ValueError('C01 authenticated EXEC barrier') channel.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 0) return raise ValueError('C01 EXEC deadline')