221 lines
11 KiB
Python
221 lines
11 KiB
Python
#!/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')
|