372 lines
16 KiB
Python
372 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""Real child/cgroup failures, only inside this fixture's delegated systemd unit."""
|
|
import errno
|
|
import os
|
|
from pathlib import Path
|
|
import signal
|
|
import resource
|
|
import select
|
|
import socket
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / 'scripts'))
|
|
from c01_children import Children, RoleGroup, drop_credentials, process_group_path
|
|
import c01_children as lifecycle
|
|
|
|
assert os.geteuid() == 0
|
|
unit = Path(process_group_path(os.getpid())).parent.name
|
|
assert unit.startswith('opencode-c01-children-') and unit.endswith('.service')
|
|
assert os.environ.get('ARCHY_C01_CHILDREN_FIXTURE') == '1'
|
|
|
|
|
|
def settle(owner, close=True):
|
|
deadline = time.monotonic() + 5
|
|
while owner.children:
|
|
assert not owner.poll(), 'child lifecycle reported a kernel failure'
|
|
assert time.monotonic() < deadline, 'fixture children did not retire'
|
|
time.sleep(0.005)
|
|
if close:
|
|
owner.close()
|
|
|
|
|
|
def waiting(control):
|
|
parent = os.getppid()
|
|
drop_credentials(65534, 65534, parent)
|
|
control.send(b'ready')
|
|
while True:
|
|
signal.pause()
|
|
|
|
|
|
def exercise(name, action):
|
|
owner = Children()
|
|
try:
|
|
action(owner)
|
|
settle(owner)
|
|
assert not any(owner.root.glob('c01-*'))
|
|
print('PASS child ownership:', name, flush=True)
|
|
except BaseException:
|
|
owner.request_stop()
|
|
settle(owner)
|
|
raise
|
|
|
|
|
|
with tempfile.TemporaryDirectory(prefix='c01-children-') as temporary:
|
|
root = Path(temporary); root.chmod(0o777)
|
|
|
|
def ordinary(owner):
|
|
def child(control):
|
|
drop_credentials(65534, 65534, os.getppid())
|
|
assert os.getresuid() == (65534, 65534, 65534)
|
|
control.send(b'dropped')
|
|
record = owner.spawn('application', child)
|
|
record.channel.settimeout(2)
|
|
assert record.channel.recv(32) == b'dropped'
|
|
assert record.pidfd is not None
|
|
exercise('permanent credential drop and actual bounded role cgroup', ordinary)
|
|
|
|
def escaped_descendant(owner):
|
|
def child(control):
|
|
drop_credentials(65534, 65534, os.getppid())
|
|
parent, ready = socket.socketpair()
|
|
descendant = os.fork()
|
|
if descendant == 0:
|
|
parent.close()
|
|
os.setsid()
|
|
grandchild = os.fork()
|
|
if grandchild:
|
|
os._exit(0)
|
|
os.setsid()
|
|
ready.send(str(os.getpid()).encode())
|
|
while True:
|
|
signal.pause()
|
|
ready.close(); escaped = parent.recv(32); parent.close()
|
|
assert os.waitpid(descendant, 0)[0] == descendant
|
|
control.send(escaped)
|
|
record = owner.spawn('application', child)
|
|
record.channel.settimeout(2)
|
|
descendant = int(record.channel.recv(32))
|
|
# Do not use PID liveness after reaping as an identity assertion.
|
|
handle = os.pidfd_open(descendant)
|
|
try:
|
|
settle(owner, close=False)
|
|
try:
|
|
os.waitpid(descendant, os.WNOHANG)
|
|
except ChildProcessError:
|
|
pass
|
|
else:
|
|
raise AssertionError('escaped descendant was not reaped')
|
|
finally:
|
|
os.close(handle)
|
|
exercise('double-fork setsid descendant killed and reaped after leader exit', escaped_descendant)
|
|
|
|
for stage in ('pidfd', 'cgroup'):
|
|
def admission_failure(owner, stage=stage):
|
|
marker = root / (stage + '-body-ran')
|
|
def forbidden(control):
|
|
marker.write_text('ran')
|
|
manager = patch.object(os, 'pidfd_open', side_effect=OSError(errno.EMFILE, 'fixture')) if stage == 'pidfd' else \
|
|
patch.object(RoleGroup, 'attach', side_effect=OSError(errno.EPERM, 'fixture'))
|
|
with manager:
|
|
try:
|
|
owner.spawn('host-io', forbidden)
|
|
except OSError:
|
|
pass
|
|
else:
|
|
raise AssertionError('injected admission failure was accepted')
|
|
assert owner.stopping and len(owner.children) == 1
|
|
assert next(iter(owner.children.values())).pid is not None
|
|
settle(owner, close=False)
|
|
assert not marker.exists(), 'unregistered child crossed its execution barrier'
|
|
exercise(stage + ' failure retains and retires the unadmitted direct child', admission_failure)
|
|
|
|
def interrupted_registration(owner):
|
|
original = os.pidfd_open
|
|
def interrupted(pid):
|
|
result = original(pid)
|
|
os.kill(os.getpid(), signal.SIGTERM)
|
|
return result
|
|
with patch.object(os, 'pidfd_open', side_effect=interrupted):
|
|
owner.spawn('authority', waiting)
|
|
assert owner.stopping and len(owner.children) == 1
|
|
exercise('termination during registration cannot escape ownership', interrupted_registration)
|
|
|
|
def retained_waitability(owner):
|
|
record = owner.spawn('host-io', waiting)
|
|
record.channel.settimeout(2); assert record.channel.recv(32) == b'ready'
|
|
owner.request_stop()
|
|
# Model an unreapable observation, not an actual D-state filesystem.
|
|
with patch.object(os, 'waitid', return_value=None):
|
|
assert owner.poll() == []
|
|
assert owner.children['host-io'] is record
|
|
try:
|
|
owner.spawn('administrator', waiting)
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
raise AssertionError('replacement created over retained worker')
|
|
exercise('unreapable observation retains role and rejects replacement', retained_waitability)
|
|
|
|
def channel_failure(owner):
|
|
original = socket.socketpair
|
|
first = original(socket.AF_UNIX, socket.SOCK_SEQPACKET)
|
|
with patch.object(socket, 'socketpair', side_effect=[first, OSError(errno.EMFILE, 'fixture')]):
|
|
try:
|
|
owner.spawn('administrator', waiting)
|
|
except OSError:
|
|
pass
|
|
else:
|
|
raise AssertionError('socket allocation failure was accepted')
|
|
assert not owner.children and all(channel.fileno() == -1 for channel in first)
|
|
assert not (owner.root / 'c01-administrator').exists()
|
|
exercise('pre-fork channel allocation failure releases only owned resources', channel_failure)
|
|
|
|
def live_orphans(owner):
|
|
def child(control):
|
|
drop_credentials(65534, 65534, os.getppid())
|
|
for _ in range(32): # More than this role's entire eight-task budget.
|
|
middle = os.fork()
|
|
if middle == 0:
|
|
orphan = os.fork()
|
|
if orphan == 0:
|
|
os._exit(0)
|
|
os._exit(0)
|
|
os.waitpid(middle, 0)
|
|
control.send(b'round')
|
|
assert control.recv(4) == b'next'
|
|
record = owner.spawn('host-io', child)
|
|
record.channel.settimeout(2)
|
|
for _ in range(32):
|
|
assert record.channel.recv(8) == b'round'
|
|
deadline = time.monotonic() + 2
|
|
while True:
|
|
assert not owner.poll()
|
|
assert record.status is None, 'live leader was prematurely retired'
|
|
if (record.group.path / 'pids.current').read_text().strip() == '1':
|
|
break
|
|
assert time.monotonic() < deadline, 'completed orphan consumed the live role PID budget'
|
|
time.sleep(0.005)
|
|
record.channel.send(b'next')
|
|
exercise('live leader reclaims completed adopted orphans beyond PID budget', live_orphans)
|
|
|
|
def waiting_tree(control):
|
|
drop_credentials(65534, 65534, os.getppid())
|
|
parent, ready = socket.socketpair()
|
|
descendant = os.fork()
|
|
if descendant == 0:
|
|
parent.close(); os.setsid(); ready.send(b'ready')
|
|
while True:
|
|
signal.pause()
|
|
ready.close(); assert parent.recv(5) == b'ready'; parent.close()
|
|
control.send(str(descendant).encode())
|
|
while True:
|
|
signal.pause()
|
|
|
|
def group_kill_failure(owner):
|
|
record = owner.spawn('authority', waiting_tree)
|
|
record.channel.settimeout(2); record.channel.recv(32)
|
|
observer = select.poll(); observer.register(record.pidfd, select.POLLIN)
|
|
with patch.object(RoleGroup, 'kill', side_effect=OSError(errno.EIO, 'fixture')):
|
|
owner.request_stop()
|
|
assert owner.poll(), 'group termination error was hidden'
|
|
assert observer.poll(2000), 'group kill error suppressed pidfd leader signal'
|
|
assert owner.children['authority'] is record and not record.group.empty()
|
|
try:
|
|
owner.spawn('authority', waiting)
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
raise AssertionError('replacement escaped a retained populated group')
|
|
settle(owner, close=False)
|
|
exercise('group kill failure still signals leader and retains descendants', group_kill_failure)
|
|
|
|
def exhausted_descriptors(owner):
|
|
record = owner.spawn('authority', waiting_tree)
|
|
record.channel.settimeout(2); descendant = int(record.channel.recv(32))
|
|
descendant_handle = os.pidfd_open(descendant)
|
|
observer = select.poll()
|
|
observer.register(record.pidfd, select.POLLIN); observer.register(descendant_handle, select.POLLIN)
|
|
original = resource.getrlimit(resource.RLIMIT_NOFILE)
|
|
ceiling = max(int(name) for name in os.listdir('/proc/self/fd')) + 16
|
|
held = []
|
|
try:
|
|
resource.setrlimit(resource.RLIMIT_NOFILE, (ceiling, original[1]))
|
|
while True:
|
|
try:
|
|
held.append(os.open('/dev/null', os.O_RDONLY | os.O_CLOEXEC))
|
|
except OSError as error:
|
|
assert error.errno == errno.EMFILE
|
|
break
|
|
owner.request_stop()
|
|
owner.poll() # proc scans can fail; reserved termination FDs cannot.
|
|
exited = set()
|
|
deadline = time.monotonic() + 2
|
|
while len(exited) < 2:
|
|
exited.update(fd for fd, _ in observer.poll(20))
|
|
assert time.monotonic() < deadline, 'EMFILE prevented descendant/leader termination'
|
|
assert owner.children['authority'] is record, 'unknown retirement was discarded'
|
|
finally:
|
|
for fd in held:
|
|
os.close(fd)
|
|
resource.setrlimit(resource.RLIMIT_NOFILE, original)
|
|
os.close(descendant_handle)
|
|
exercise('real descriptor exhaustion preserves whole-group termination', exhausted_descriptors)
|
|
|
|
def cleanup_retry(owner):
|
|
record = owner.spawn('administrator', waiting)
|
|
record.channel.settimeout(2); assert record.channel.recv(32) == b'ready'
|
|
owner.request_stop()
|
|
with patch.object(RoleGroup, 'close', side_effect=OSError(errno.EBUSY, 'fixture')):
|
|
deadline = time.monotonic() + 2
|
|
while not record.reaped:
|
|
owner.poll()
|
|
assert time.monotonic() < deadline
|
|
time.sleep(0.005)
|
|
assert owner.children['administrator'] is record and record.pid is None and record.pidfd is None
|
|
assert record.channel.fileno() == -1
|
|
try:
|
|
owner.close()
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
raise AssertionError('owner closed over pending cgroup retirement')
|
|
reused = os.open('/dev/null', os.O_RDONLY | os.O_CLOEXEC)
|
|
try:
|
|
with patch.object(os, 'kill', side_effect=AssertionError('released PID signalled')), \
|
|
patch.object(signal, 'pidfd_send_signal', side_effect=AssertionError('retired pidfd signalled')), \
|
|
patch.object(os, 'waitpid', side_effect=AssertionError('released PID waited')):
|
|
assert owner.poll() == []
|
|
os.fstat(reused)
|
|
assert not owner.children and not record.group.path.exists()
|
|
finally:
|
|
os.close(reused)
|
|
exercise('post-reap cleanup remains owned and retries without stale PID or FD reuse', cleanup_retry)
|
|
|
|
def replenished_observations(owner):
|
|
record = owner.spawn('host-io', waiting)
|
|
record.channel.settimeout(2); assert record.channel.recv(32) == b'ready'
|
|
snapshots = 0
|
|
def replenished(path, *args, **kwargs):
|
|
nonlocal snapshots
|
|
assert str(path).endswith('/children')
|
|
snapshots += 1
|
|
assert snapshots <= 64, 'live adoption drain never yielded to control'
|
|
return str(1_000_000 + snapshots)
|
|
# Deterministic non-quiescing kernel-observation seam. No fake PID is
|
|
# ever signaled/waited on: both wait calls are explicitly substituted.
|
|
with patch.object(Path, 'read_text', replenished), \
|
|
patch.object(lifecycle, 'process_group_path', return_value=record.group.kernel_path), \
|
|
patch.object(os, 'waitid', return_value=object()), patch.object(os, 'waitpid', return_value=(1, 0)) as reap:
|
|
assert owner._reap_adopted(record) is False
|
|
assert snapshots < 64 and reap.call_count > 0
|
|
owner.request_stop(); reap.reset_mock()
|
|
assert owner._reap_adopted(record) is False
|
|
reap.assert_not_called()
|
|
exercise('non-quiescing adoption observation yields and honors stop', replenished_observations)
|
|
|
|
def continuous_orphans(owner):
|
|
def producer(control):
|
|
drop_credentials(65534, 65534, os.getppid())
|
|
control.send(b'ready')
|
|
while True:
|
|
try:
|
|
middle = os.fork()
|
|
except BlockingIOError:
|
|
time.sleep(0.001)
|
|
continue
|
|
if middle == 0:
|
|
try:
|
|
orphan = os.fork()
|
|
if orphan == 0:
|
|
os._exit(0)
|
|
finally:
|
|
os._exit(0)
|
|
os.waitpid(middle, 0)
|
|
record = owner.spawn('host-io', producer)
|
|
record.channel.settimeout(2); assert record.channel.recv(32) == b'ready'
|
|
for _ in range(16):
|
|
assert not owner.poll()
|
|
time.sleep(0.005)
|
|
os.kill(os.getpid(), signal.SIGTERM)
|
|
assert owner.stopping
|
|
exercise('continuous real orphan producer remains stoppable', continuous_orphans)
|
|
|
|
def zombie_behind_live_children(owner):
|
|
def producer(control):
|
|
drop_credentials(65534, 65534, os.getppid())
|
|
for index in range(33):
|
|
parent, ready = socket.socketpair()
|
|
middle = os.fork()
|
|
if middle == 0:
|
|
parent.close()
|
|
orphan = os.fork()
|
|
if orphan == 0:
|
|
ready.send(b'ready')
|
|
if index == 32:
|
|
os._exit(0)
|
|
while True:
|
|
signal.pause()
|
|
os._exit(0)
|
|
ready.close(); assert parent.recv(5) == b'ready'; parent.close()
|
|
os.waitpid(middle, 0)
|
|
control.send(b'constructed')
|
|
while True:
|
|
signal.pause()
|
|
record = owner.spawn('application', producer)
|
|
record.channel.settimeout(5); assert record.channel.recv(32) == b'constructed'
|
|
adopted = [int(text) for text in Path(f'/proc/self/task/{os.getpid()}/children').read_text().split()
|
|
if int(text) != record.pid]
|
|
assert len(adopted) == 33
|
|
assert all(os.waitid(os.P_PID, pid, os.WEXITED | os.WNOHANG | os.WNOWAIT) is None for pid in adopted[:32])
|
|
deadline = time.monotonic() + 2
|
|
while os.waitid(os.P_PID, adopted[-1], os.WEXITED | os.WNOHANG | os.WNOWAIT) is None:
|
|
assert time.monotonic() < deadline
|
|
time.sleep(0.005)
|
|
for _ in range(4):
|
|
assert owner.poll() == []
|
|
assert (record.group.path / 'pids.current').read_text().strip() == '33', 'live prefix starved the waitable orphan'
|
|
owner.request_stop()
|
|
exercise('32 live adopted children cannot starve a later zombie', zombie_behind_live_children)
|