#!/usr/bin/env python3 """Actual non-test binary, real distinct UIDs and launch sockets, disposable mounts. Invoked only by the ignored Rust fixture through test-backend-isolated.sh. No time assertion, wallet, live supervisor, or installation authority is created. """ import array import fcntl import json import os from pathlib import Path import signal import shutil import socket import subprocess import sys import tempfile import time assert os.geteuid() == 0 and os.getppid() == 1 assert os.environ.get('ARCHY_TEST_ISOLATED') == '1' assert os.environ.get('ARCHY_TEST_C01_STARTUP') == '1' # The fixture must not populate a real installed payload/config directory. for directory in ('/opt/archipelago', '/etc/archipelago'): fields = [line.split() for line in Path('/proc/self/mountinfo').read_text().splitlines()] assert any(row[4] == directory and row[row.index('-') + 1] == 'tmpfs' for row in fields), directory binary = Path(sys.argv[1]).resolve(strict=True) APP, PEER = 65534, 65533 def drop(uid): os.setgroups([]) os.setresgid(uid, uid, uid) os.setresuid(uid, uid, uid) def independent_endpoint(): parent, child = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) pid = os.fork() if pid == 0: try: parent.close() drop(PEER) retained, passed = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) child.sendmsg([b'endpoint'], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, array.array('i', [passed.fileno()]))]) passed.close() child.recv(1) # Keep the actual peer alive until the case is finished. retained.close() finally: os._exit(0) child.close() parent.settimeout(5) _, ancillary, flags, _ = parent.recvmsg(32, socket.CMSG_SPACE(array.array('i').itemsize)) assert not flags and len(ancillary) == 1 level, kind, raw = ancillary[0] assert (level, kind) == (socket.SOL_SOCKET, socket.SCM_RIGHTS) received = array.array('i'); received.frombytes(raw) assert len(received) == 1 return pid, parent, received[0] def wait(pid, read_fd): os.set_blocking(read_fd, False) output = bytearray() deadline = time.monotonic() + 10 try: while True: try: output.extend(os.read(read_fd, 65536)) except BlockingIOError: pass assert len(output) <= 1024 * 1024, 'startup output exceeded fixture bound' done, status = os.waitpid(pid, os.WNOHANG) if done: # The terminated program has no authorized child work. Retire # any stray descendants before collecting the final pipe bytes. try: os.killpg(pid, signal.SIGKILL) except ProcessLookupError: pass while True: try: part = os.read(read_fd, 65536) except BlockingIOError: break if not part: break output.extend(part) return os.waitstatus_to_exitcode(status), output.decode(errors='replace') assert time.monotonic() < deadline, 'startup did not refuse within the fixture deadline' time.sleep(0.01) except BaseException: try: os.killpg(pid, signal.SIGKILL) except ProcessLookupError: pass os.waitpid(pid, 0) raise finally: os.close(read_fd) with tempfile.TemporaryDirectory(prefix='c01-startup-') as temporary: root = Path(temporary) root.chmod(0o755) executable = root / 'archipelago' shutil.copyfile(binary, executable) executable.chmod(0o755) proxy = root / 'archipelago-runtime-proxy' with proxy.open('xb') as output: subprocess.run([str(executable), 'export-runtime-proxy'], stdout=output, check=True, timeout=5) proxy.chmod(0o555) tools = root / 'bin'; tools.mkdir() # Detect forbidden scheduling as well as actual privileged writes. This # stub never invokes sudo/systemd or changes installed files. sudo = tools / 'sudo' sudo.write_text('#!/bin/sh\n: > "$C01_FIXTURE_MARKER"\nexit 0\n') sudo.chmod(0o755) web = Path('/opt/archipelago/web-ui'); web.mkdir(mode=0o700) cases = [ # Older ABI refusal is not evidence for later storage/path/proxy checks. # Current FD10/11/12 probes live in c01-normal-systemd.py. ('current-missing-fd12', 'c01-managed-v6', False, 'required Tor lifecycle FD12 missing'), ('radio-v7-missing-fd14', 'c01-managed-v7', False, 'required v7 radio FD14 missing'), ('obsolete-v3', 'c01-managed-v3', False, 'obsolete protected application ABI'), ('obsolete-v4', 'c01-managed-v4', False, 'obsolete protected application ABI'), ('obsolete-v5', 'c01-managed-v5', False, 'obsolete protected application ABI'), ('invalid-admin', 'c01-authority-v3', True, 'C01 authority peer identity mismatch'), ('legacy-application', 'c01-managed-v2', False, 'unknown command or option'), ('legacy-authority', 'c01-authority', False, 'unknown command or option'), ('unknown-command', 'c01-managed-typo', False, 'unknown command or option'), ('version', '--version', False, 'archipelago '), ] for name, selector, authenticated, message in cases: case = root / name; case.mkdir(mode=0o755) data = case / 'data'; data.mkdir(mode=0o700) runtime_data = case / 'runtime'; runtime_data.mkdir(mode=0o700); os.chown(runtime_data, APP, APP) store = case / 'store'; store.mkdir(mode=0o700) scratch = case / 'scratch'; scratch.mkdir(mode=0o700) os.chown(data, PEER if selector == 'c01-authority-v3' else APP, PEER if selector == 'c01-authority-v3' else APP) for path in (store, scratch): os.chown(path, APP, APP) marker = scratch / 'privileged-startup-ran' configured = scratch / 'must-not-be-created' peer_pid, peer_control, peer_fd = independent_endpoint() grant_parent, grant_child = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) admin_parent, admin_child = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) reader, writer = os.pipe() key_path = case / 'storage.key' if name == 'invalid-key-capability': os.mkfifo(key_path, 0o600) else: key_path.write_bytes(bytes([7] * 32)); key_path.chmod(0o600) key_fd = os.open(key_path, os.O_RDONLY | os.O_NONBLOCK | os.O_CLOEXEC) key_metadata = os.fstat(key_fd) if name == 'invalid-proxy': fifo = case / 'proxy-fifo'; os.mkfifo(fifo, 0o600) proxy_fd = os.open(fifo, os.O_PATH | os.O_CLOEXEC) else: proxy_fd = os.open(proxy, os.O_PATH | os.O_CLOEXEC) pid = os.fork() if pid == 0: try: os.setsid() os.dup2(writer, 1); os.dup2(writer, 2) copies = [fcntl.fcntl(fd, fcntl.F_DUPFD_CLOEXEC, 20) for fd in (grant_child.fileno(), peer_fd, admin_child.fileno(), admin_child.fileno(), proxy_fd, key_fd)] # Preserve slots before closing unrelated inherited descriptors. for slot, fd in zip((3, 4, 5, 6, 8, 9), copies): if authenticated and (slot != 5 or selector == 'c01-authority-v3') and (slot not in (6, 8) or selector == 'c01-managed-v3'): os.dup2(fd, slot, inheritable=True) else: os.close(slot) os.close(7) os.closerange(10, 1048576) drop(APP) env = {'PATH': str(tools) + ':/usr/bin:/bin', 'HOME': str(scratch), 'ARCHIPELAGO_DATA_DIR': str(data if name == 'runtime-bound-path' else configured), 'ARCHIPELAGO_RUNTIME_DATA_DIR': str(configured if name == 'runtime-bound-path' else runtime_data), 'ARCHIPELAGO_HOST_IP': '127.0.0.1', 'ARCHIPELAGO_BIND': '127.0.0.1:0', 'C01_FIXTURE_MARKER': str(marker), 'RUST_MIN_STACK': '16777216'} os.execve(executable, [str(executable), selector], env) except BaseException: os._exit(111) os.close(writer) if name != 'invalid-proxy': grant_child.close() admin_child.close(); os.close(peer_fd); os.close(proxy_fd) if authenticated: grant_parent.send(json.dumps({'version': 3, 'role': 'authority' if selector == 'c01-authority-v3' else 'application', 'process_id': pid, 'data_dir': str(data), 'store_dir': str(store), 'runtime_data_dir': None if name == 'missing-runtime-binding' else str(runtime_data), 'binding': {'local_identity': [1] * 32, 'incarnation': [2] * 32, 'reader_schema': 2, 'writer_schema': 2}, 'key_file': {'device': key_metadata.st_dev, 'inode': key_metadata.st_ino}, 'peer': {'pid': peer_pid, 'uid': PEER, 'gid': PEER}, 'admin': {'pid': os.getpid(), 'uid': APP, 'gid': APP} if selector == 'c01-authority-v3' else None, 'provision': selector == 'c01-authority-v3', 'old_writers_excluded': True}).encode()) try: code, output = wait(pid, reader) assert (code == 0) == (selector == '--version'), (name, code, output) assert message in output, (name, output) assert not configured.exists(), (name, 'created an unadmitted data directory') assert not marker.exists(), (name, 'scheduled privileged startup') assert list(store.iterdir()) == [], (name, 'mutated authority state before channel admission') if name == 'invalid-proxy': # Retained duplicate of the receiving endpoint proves the # private grant remained queued, and its inherited key capability # has not been read before executable admission. grant_child.setblocking(False) assert json.loads(grant_child.recv(8192))['key_file'] == {'device': key_metadata.st_dev, 'inode': key_metadata.st_ino} assert os.lseek(key_fd, 0, os.SEEK_CUR) == 0 print('PASS C01 startup:', name) finally: grant_parent.close(); grant_child.close(); admin_parent.close(); peer_control.close() os.close(key_fd) os.waitpid(peer_pid, 0) # The same explicitly isolated namespace also checks the positive recovery # handoff against the non-test authority executable and distinct real UIDs. import runpy runpy.run_path(str(Path(__file__).with_name('c01-recovery.py')), run_name='__main__') runpy.run_path(str(Path(__file__).with_name('host-access.py')), run_name='__main__')