"""Disposable-only kernel regressions, called after access_owner_fixture guards. Never invoke against installed accounts/cgroups. The checker barrier is injected only in a forked fixture child; production has no barrier or admission bypass. """ import array import ctypes import json import os from pathlib import Path import select import signal import socket import subprocess import time import c01_access as access import c01_access_kernel as kernel def endpoint_retirement_guards(): from unittest.mock import patch assert os.getpid() == 1 and os.environ.get('C01_ACCESS_FIXTURE') == '1' path = access.BASE / 'fixture-stale.sock' unrelated = access.BASE / 'fixture-preserved' unrelated.write_bytes(b'preserve exact unrelated object') def refused(evidence): try: access.retire_endpoint(evidence) except ValueError: assert path.exists() else: raise AssertionError('unowned/live endpoint retired') with patch.object(access, 'SOCKET', path): path.write_bytes(b'not a socket') path.chmod(0o600) refused({'endpoint': access.policy_module.identity(path.lstat())}) path.unlink() listener = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) listener.bind(str(path)) path.chmod(0o600) listener.listen(1) identity = access.policy_module.identity(path.lstat()) refused({'endpoint': identity}) listener.close() wrong = identity.copy() wrong[1] += 1 refused({'endpoint': wrong}) access.retire_endpoint({'endpoint': identity}) assert not path.exists() and unrelated.read_bytes() == b'preserve exact unrelated object' unrelated.unlink() def ssh_route(policy, denied=False): """Real private sshd + fresh admin key login and no-shell managed attempts. Loopback transport only; remote-context semantics are checked separately by sshd -T and the global-only grammar. This is not a production reload test. """ assert os.getpid() == 1 and os.environ.get('C01_ACCESS_FIXTURE') == '1' argv = ['/usr/bin/ssh', '-F', '/dev/null', '-i', '/run/fixture_client', '-p', '22222', '-oBatchMode=yes', '-oIdentitiesOnly=yes', '-oStrictHostKeyChecking=no', '-oUserKnownHostsFile=/dev/null', '-oConnectTimeout=2'] with Path('/run/fixture-sshd.log').open('wb') as log: daemon = subprocess.Popen(['/usr/sbin/sshd', '-D', '-e', '-f', '/etc/ssh/sshd_config'], stdin=subprocess.DEVNULL, stdout=log, stderr=log) try: deadline = time.monotonic() + 3 while True: assert daemon.poll() is None and time.monotonic() < deadline with socket.socket() as probe: if probe.connect_ex(('127.0.0.1', 22222)) == 0: break time.sleep(.025) result = subprocess.run([*argv, 'archipelago-admin@127.0.0.1', '/usr/bin/true'], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=log, timeout=5) assert result.returncode == 0, 'fresh verified administrator key route failed' if denied: for name in policy.names: result = subprocess.run([*argv, '-N', '-oExitOnForwardFailure=yes', '-L127.0.0.1:22223:127.0.0.1:22222', name + '@127.0.0.1'], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=log, timeout=3) assert result.returncode == 255, 'managed public-key/no-shell SSH session admitted' deadline = time.monotonic() + 3 descendants = Path(f'/proc/{daemon.pid}/task/{daemon.pid}/children') while descendants.read_text().strip(): assert time.monotonic() < deadline, 'private sshd session retirement incomplete' time.sleep(.025) finally: daemon.terminate() try: daemon.wait(timeout=3) except subprocess.TimeoutExpired: daemon.kill() daemon.wait(timeout=3) def generation_race(policy, group): assert os.getpid() == 1 and os.environ.get('C01_ACCESS_FIXTURE') == '1' control, checker_side = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) barrier, checker_barrier = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) checker_pid = os.fork() if checker_pid == 0: control.close() barrier.close() original = access.check_current def held_check(value, supervisor): first = held_check.first held_check.first = False if first: checker_barrier.send(b'SCANNING') assert checker_barrier.recv(16) == b'RELEASE' return original(value, supervisor) held_check.first = True access.check_current = held_check access.checker(checker_side, policy) os._exit(112) checker_side.close() checker_barrier.close() guard, observer = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) leases = access.Leases(policy, guard) leases.last_observed = kernel.now() candidate_control, child_control = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) path = Path('/run/access-generation.sock') listener = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) listener.bind(str(path)) listener.listen(1) candidate = None try: access.send(control, {'challenge': 'old', 'supervisor': None, 'generation': 0}) barrier.settimeout(3) assert barrier.recv(16) == b'SCANNING' candidate = os.fork() if candidate == 0: try: listener.close() control.close() barrier.close() guard.close() observer.close() candidate_control.close() (group / 'cgroup.procs').write_text(str(os.getpid())) channel = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) channel.settimeout(5) channel.connect(str(path)) request = {'protocol': 'supervisor-access-v1', 'configuration_sha256': policy.value['launch_sha256'], 'nonce': os.urandom(32).hex()} access.send(channel, request) assert json.loads(channel.recv(8192)) == request roles = [] for role, slot in (('manager', 'application'), ('authority', 'authority')): target = group / ('c01-' + slot) target.mkdir() read_fd, write_fd = os.pipe() child = os.fork() if child == 0: channel.close() # Dropped roles never retain the capability. child_control.close() os.close(read_fd) (target / 'cgroup.procs').write_text(str(os.getpid())) assert ctypes.CDLL(None).prctl(38, 1, 0, 0, 0) == 0 account = policy.value['accounts'][role] os.setgroups([]) os.setresgid(*([account['gid']] * 3)) os.setresuid(*([account['uid']] * 3)) os.write(write_fd, b'R') while True: signal.pause() os.close(write_fd) assert select.select([read_fd], [], [], 2)[0] and os.read(read_fd, 1) == b'R' os.close(read_fd) roles.append((child, target)) child_control.send(b'ROLES') assert child_control.recv(16) == b'RELEASE' for child, target in roles: os.kill(child, signal.SIGKILL) os.waitpid(child, 0) target.rmdir() channel.close() os._exit(0) except BaseException: import traceback traceback.print_exc() os._exit(111) child_control.close() channel, _ = listener.accept() assert select.select([channel], [], [], 2)[0] leases.accept(channel, access.receive(channel)) assert leases.supervisor['pending'] and leases.generation == 1 assert not select.select([candidate_control], [], [], .1)[0], 'grant escaped before checker generation' barrier.send(b'RELEASE') assert select.select([control], [], [], 3)[0] old = access.receive(control) assert not leases.checked(old, None) assert not select.select([candidate_control], [], [], .1)[0], 'obsolete scan published candidate' reserved = leases.supervisor access.send(control, {'challenge': 'candidate', 'supervisor': reserved, 'generation': leases.generation}) assert select.select([control], [], [], 3)[0] assert leases.checked(access.receive(control), reserved) assert not select.select([candidate_control], [], [], .1)[0], 'grant escaped before watchdog custody ack' raw, ancillary, flags, _ = observer.recvmsg(access.WATCH_PACKET.size, socket.CMSG_SPACE(4)) kind, generation, expiry, token = access.watch_evidence(raw, 0, 0, kernel.now()) assert kind == b'HOLD' and generation == 1 and not flags for level, kind, data in ancillary: assert (level, kind) == (socket.SOL_SOCKET, socket.SCM_RIGHTS) rights = array.array('i') rights.frombytes(data) for fd in rights: os.close(fd) leases.confirm(token) candidate_control.settimeout(3) assert candidate_control.recv(16) == b'ROLES' kernel.verify_processes(policy, leases.supervisor) try: kernel.verify_processes(policy, None) except ValueError: pass else: raise AssertionError('orphan role children accepted without supervisor') active = leases.supervisor access.send(control, {'challenge': 'active', 'supervisor': active, 'generation': leases.generation}) assert select.select([control], [], [], 3)[0] assert leases.checked(access.receive(control), active) assert not select.select([channel], [], [], 0)[0] candidate_control.send(b'RELEASE') assert os.waitstatus_to_exitcode(os.waitpid(candidate, 0)[1]) == 0 candidate = None leases.poll() retiring = leases.supervisor access.send(control, {'challenge': 'retired', 'supervisor': retiring, 'generation': leases.generation}) assert select.select([control], [], [], 3)[0] assert leases.checked(access.receive(control), retiring) assert leases.supervisor is None and leases.retiring is None finally: leases.stop() if candidate is not None: (group / 'cgroup.kill').write_text('1') os.waitpid(candidate, 0) os.kill(checker_pid, signal.SIGKILL) os.waitpid(checker_pid, 0) for channel in (control, barrier, guard, observer, candidate_control, listener): channel.close() path.unlink() def enclosing_retirement(root): """Real delegated descendant, root-owned enclosing freeze/kill capabilities. Remap only the cgroup mount prefix in this test to the fixture's own subtree. This does not qualify systemd's real user@.service deletion/recreation jobs. """ from unittest.mock import patch assert os.getpid() == 1 and os.environ.get('C01_ACCESS_FIXTURE') == '1' mount = root / 'enclosing' mount.mkdir() parent = mount / 'user.slice' parent.mkdir() enclosing = parent / 'user-1000.slice' enclosing.mkdir() delegated = enclosing / 'user@1000.service' delegated.mkdir() os.chown(delegated, 1000, 1000) for name in ('cgroup.procs', 'cgroup.threads', 'cgroup.subtree_control'): os.chown(delegated / name, 1000, 1000) reader, writer = os.pipe() pid = os.fork() if pid == 0: os.close(reader) (delegated / 'cgroup.procs').write_text(str(os.getpid())) os.setgroups([]) os.setresgid(1000, 1000, 1000) os.setresuid(1000, 1000, 1000) os.write(writer, b'R') while True: signal.pause() os.close(writer) assert select.select([reader], [], [], 2)[0] and os.read(reader, 1) == b'R' os.close(reader) original_process, original_mount = kernel.process, kernel.CGROUP start = original_process(pid)['start'] meta = enclosing.stat() spec = {'path': str(enclosing), 'device': meta.st_dev, 'inode': meta.st_ino, 'members': {str(pid): start}, 'replacement': 'runtime-user-slice'} runtime = {'runtime_uid': 1000, 'user_cgroup': '/user.slice/user-1000.slice/user@1000.service'} def real_process(number): item = original_process(number) item['group'] = str(original_mount / Path(item['group']).relative_to(mount)) return item with patch.object(kernel, 'CGROUP', mount), patch.object(kernel, 'process', real_process): cohort = kernel.Cohort(spec, runtime) try: cohort.freeze() expiry = time.monotonic() + 2 while cohort.events().get('frozen') != '1': assert time.monotonic() < expiry time.sleep(.01) cohort.capture() cohort.kill() expiry = time.monotonic() + 2 while not cohort.retired(): assert time.monotonic() < expiry time.sleep(.01) os.waitpid(pid, 0) finally: cohort.close() try: kernel.replacement_tree(spec, runtime) except ValueError: pass else: raise AssertionError('same original slice inode accepted') delegated.rmdir() enclosing.rmdir() enclosing.mkdir() delegated.mkdir() os.chown(delegated, 1000, 1000) kernel.replacement_tree(spec, runtime) unexpected = enclosing / 'session-1.scope' unexpected.mkdir() try: kernel.replacement_tree(spec, runtime) except ValueError: pass else: raise AssertionError('unreviewed replacement session accepted') unexpected.rmdir() delegated.rmdir() enclosing.rmdir() parent.rmdir() mount.rmdir()