#!/usr/bin/env python3 """Python-only owned collector/fixture regressions. No backend artifact used.""" import json import os from pathlib import Path import signal import socket import subprocess import sys import time from unittest.mock import patch sys.path.insert(0, str(Path(__file__).resolve().parents[2] / 'scripts')) import c01_supervisor as supervisor import c01_host_watch as watch from c01_fixture_process import run_bounded, FixtureTimeout assert os.geteuid() == 0 and os.environ.get('ARCHY_C01_SUPERVISOR_FIXTURE') == '1' assert Path(supervisor.lifecycle.process_group_path(os.getpid())).parent.name.startswith('opencode-c01-supervisor-') for directory in ('/run', '/var/lib/archipelago-c01'): rows = [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 rows) root = Path('/var/lib/archipelago-c01/collector-fixture'); root.mkdir(mode=0o700) key = root / 'storage-key'; key.write_bytes(bytes(range(1, 33))); key.chmod(0o600) keyfd = supervisor.root_capability(key, 32) key_identity = (os.fstat(keyfd).st_dev, os.fstat(keyfd).st_ino) observer_pid = os.getpid() binding = {'local_identity': [1] * 32, 'incarnation': [2] * 32, 'reader_schema': 2, 'writer_schema': 2} qualification = supervisor.recovery.PlatformQualification(os.uname().release, supervisor.recovery.clocksource(), '12' * 32, 5_000_000, 100_000_000, 10) def done(name): print('PASS ' + name, flush=True) def owner(name): return supervisor.Supervisor(supervisor.LaunchLock(root / (name + '.lock'))) def check_keyless_collector(): assert os.getpid() != observer_pid and os.getppid() == observer_pid assert os.getresuid() == (0, 0, 0) for entry in Path('/proc/self/fd').iterdir(): try: meta = os.fstat(int(entry.name)) except OSError: continue assert (meta.st_dev, meta.st_ino) != key_identity def refused(body): try: body() except (OSError, ValueError): return raise AssertionError('unexpected admission') # Exercise the unchanged prepare()/query() path, including its blocking finally # wait, inside the collector. Kill suppression models a helper not yet retired # by the kernel; the outer qualified cgroup owner remains fully operational. owned = owner('stalled-query') client, accepted = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) marker = root / 'helper-wait' real_popen = subprocess.Popen group_kill = None try: app = owned.children.spawn('application', lambda control: time.sleep(60)) authority = owned.children.spawn('authority', lambda control: time.sleep(60)) def host(control): control.send(watch.PACKET.pack(1, watch.READY, 0, 0)) time.sleep(60) worker = owned.children.spawn('host-io', host, keep=(accepted.fileno(),)) owned.watch = watch.HostWatch(worker.channel, accepted, worker.pid, watch.now() + 2_000_000_000) class StalledHelper(real_popen): def __init__(self, *args, **kwargs): # The actual query retains its exact deadline/framing code. Only # the executable and synthetic kernel non-retirement are injected. script = f'import os,signal,time;time.sleep(0.4);os.kill({observer_pid},signal.SIGTERM);time.sleep(60)' super().__init__(['/usr/bin/python3', '-I', '-c', script], **kwargs) def kill(self): pass def wait(self, timeout=None): if timeout is None: marker.write_text(str(self.pid)) return super().wait(timeout=timeout) def blocked_collect(policy): check_keyless_collector() return supervisor.nts.query(('tracking',), supervisor.now() + 100_000_000) with patch.object(supervisor.nts, 'collect', blocked_collect), patch.object(supervisor.nts.subprocess, 'Popen', StalledHelper): collector = supervisor.start_collector(owned, qualification, binding, bytes([7] * 16), {'pid': authority.pid, 'uid': 60002, 'gid': 60002}, '\0unused-collector-fixture', supervisor.now() + 5_000_000_000) group_kill = collector.group.kill collector.group.kill = lambda: None start = time.monotonic() refused(lambda: owned.relay_collector(collector, authority, 60002, 60002, supervisor.now() + 3_000_000_000)) assert time.monotonic() - start < 2, 'root observer did not honor stop during blocking query reap' assert marker.exists(), 'did not enter the actual query blocking-finally path' helper_pid = int(marker.read_text()) assert supervisor.lifecycle.process_group_path(helper_pid) == collector.group.kernel_path end = time.monotonic() + 2 while any(role in owned.children.children for role in ('application', 'authority', 'host-io')) and time.monotonic() < end: owned.tick(); time.sleep(0.005) assert set(owned.children.children) == {'collector'} assert collector.status is not None and not collector.reaped assert not collector.group.empty(), 'synthetic unretired helper disappeared' refused(lambda: supervisor.LaunchLock(root / 'stalled-query.lock')) refused(lambda: supervisor.start_collector(owned, qualification, binding, bytes([7] * 16), {}, '', supervisor.now())) for _ in range(5): os.kill(os.getpid(), signal.SIGTERM) owned.tick() client.settimeout(1); assert client.recv(1) == b'' finally: if group_kill is not None: collector.group.kill = group_kill owned.retire(); client.close() done('blocked-real-query-reap-keeps-observer-responsive-revokes-early-roles-and-retains-lock') assert not Path(f'/proc/{helper_pid}').exists() done('collector-helper-adoption-and-cgroup-retirement-complete-before-lock-release') for stale in (False, True): owned = owner('stale' if stale else 'fresh') address = '\0collector-authority-fixture-' + os.urandom(16).hex() stamp = supervisor.now() - (3_000_000_000 if stale else 0) observation = supervisor.nts.Observation(None, supervisor.recovery.boot_id(), 987_654_321_000, stamp, 0, 100_000_000, stamp) try: def authority_body(control): supervisor.lifecycle.drop_credentials(60002, 60002, os.getppid()) listener = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) listener.bind(address); listener.listen(1); listener.settimeout(5) control.send(b'STORE_READY') control.settimeout(5) admin = json.loads(control.recv(8192)) connection, _ = listener.accept() assert supervisor.peer(connection) == (admin['pid'], 60002, 60002) connection.settimeout(3) if stale: assert connection.recv(8192) == b'' control.send(b'STALE_REFUSED') else: connection.send(bytes([42] * 32)) wire = json.loads(connection.recv(8192)) assert wire['version'] == 4 and wire['challenge'] == [42] * 32 and wire['binding'] == binding assert wire['recovery']['wall'] == supervisor.recovery.duration(observation.utc_ns) assert wire['recovery']['monotonic'] == supervisor.recovery.duration(stamp) assert wire['recovery']['uncertainty'] == supervisor.recovery.duration(observation.observation_error_ns) control.send(b'AUTHORIZED') supervisor.child_wait(control, b'RELEASE', supervisor.now() + 5_000_000_000) authority = owned.children.spawn('authority', authority_body) owned.wait(authority, b'STORE_READY', 60002, 60002, supervisor.now() + 3_000_000_000) def fixture_collect(policy): check_keyless_collector() return observation with patch.object(supervisor.nts, 'collect', fixture_collect): collector = supervisor.start_collector(owned, qualification, binding, bytes([7] * 16), {'pid': authority.pid, 'uid': 60002, 'gid': 60002}, address, supervisor.now() + 5_000_000_000) admin = owned.relay_collector(collector, authority, 60002, 60002, supervisor.now() + 3_000_000_000) assert set(admin) == {'pid', 'uid', 'gid'} # root relay contains no observation if stale: end = time.monotonic() + 3 while not owned.children.stopping and time.monotonic() < end: owned.tick(); time.sleep(0.005) assert owned.children.stopping and owned.children.results.get('collector') == 111 else: owned.wait(authority, b'AUTHORIZED', 60002, 60002, supervisor.now() + 3_000_000_000) # Deliberately delay receipt: collector holds ADMIN_SENT until RELEASE. time.sleep(0.1) owned.wait(collector, b'ADMIN_SENT', 0, 0, supervisor.now() + 3_000_000_000) owned.completed_one_shots.add('collector') supervisor.packet(collector.channel, b'RELEASE') end = time.monotonic() + 2 while 'collector' not in owned.children.results and time.monotonic() < end: owned.tick(); time.sleep(0.005) assert owned.children.results.get('collector') == 0 and not owned.children.stopping assert os.lseek(keyfd, 0, os.SEEK_CUR) == 0 finally: owned.retire() done('stale-inherited-approval-still-refused-at-original-two-second-window' if stale else 'keyless-root-collector-late-admin-inherits-original-approval-and-exact-peer-is-relayed') result = run_bounded(['/usr/bin/python3', '-I', '-c', 'import os; os.write(1,b"x"*1048576); os.write(2,b"y"*1048576)'], seconds=3) assert result.returncode == 0 and len(result.stdout) == len(result.stderr) == 65536 assert result.discarded == 2 * (1048576 - 65536) done('fixture-observer-drains-both-full-pipes-with-bounded-diagnostics') seen = [] try: run_bounded(['/usr/bin/python3', '-I', '-c', 'import os,time;os.write(2,b"synthetic-stall-diagnostic");time.sleep(60)'], seconds=0.2, observe=seen.append) except FixtureTimeout as error: assert error.reaped and b'synthetic-stall-diagnostic' in error.stderr assert seen and not Path(f'/proc/{error.pid}').exists() else: raise AssertionError('fixture timeout failed') done('fixture-deadline-covers-observation-hang-preserves-diagnostics-and-reaps-child') seen = [] def broken_observer(pid): seen.append(pid) raise ValueError('injected observer assertion') try: run_bounded(['/usr/bin/python3', '-I', '-c', 'import time;time.sleep(60)'], observe=broken_observer) except ValueError as error: assert str(error) == 'injected observer assertion' assert any('reaped=True' in note and 'stdout=' in note and 'stderr=' in note for note in error.__notes__) else: raise AssertionError('observer exception disappeared') assert seen and not Path(f'/proc/{seen[0]}').exists() done('fixture-observer-exception-also-terminates-and-reaps-child') os.close(keyfd) assert not Path(f'/proc/self/task/{os.getpid()}/children').read_text().strip() done('python-only-collector-fixtures-leave-no-owned-children')