#!/usr/bin/python3
"""Owned disposable realm. Real policy/kernel tests; activation seam is explicit.

Requires private /etc,/home,/run,/var/lib,/var/spool and PID/net namespaces.
Nothing here may run against the installed account database or service bus.
"""
import ast
import ctypes
import json
import os
from pathlib import Path
import select
import shutil
import signal
import socket
import subprocess
import sys
import time

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import c01_access as access
import c01_access_policy as policy_module
import c01_access_kernel as kernel
from c01_access_client import AccessLease
from c01_children import Children
from access_owner_regressions import generation_race, enclosing_retirement, ssh_route, endpoint_retirement_guards

LIBC = ctypes.CDLL(None, use_errno=True)
CHECKS = []


def passed(text):
    CHECKS.append(text)
    print('PASS:', text, flush=True)


def require(value, text='fixture assertion failed'):
    if not value:
        raise AssertionError(text)


def wait(check, seconds=15):
    deadline = time.monotonic() + seconds
    while time.monotonic() < deadline:
        value = check()
        if value:
            return value
        time.sleep(.025)
    raise AssertionError('bounded fixture wait expired')


def rejected(body, text):
    try:
        body()
    except (ValueError, OSError):
        passed(text)
        return
    raise AssertionError('unexpected acceptance: ' + text)


def write(path, data, mode=0o600, uid=0):
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(data if isinstance(data, bytes) else data.encode())
    path.chmod(mode)
    os.chown(path, uid, uid)


def dump(path, data):
    write(path, json.dumps(data, separators=(',', ':')))


def own_dir(path, uid=0):
    path = Path(path)
    path.mkdir(parents=True, exist_ok=True)
    path.chmod(0o700)
    os.chown(path, uid, uid)


def cgroup(path, subgroup=False):
    path.mkdir()
    if subgroup:
        (path / 'cgroup.subtree_control').write_text('+cpu +memory +pids')
        (path / 'supervisor').mkdir()
    return path


def fork_legacy(group):
    read_fd, write_fd = os.pipe()
    pid = os.fork()
    if pid == 0:
        try:
            os.close(read_fd)
            (group / 'cgroup.procs').write_text(str(os.getpid()))
            os.setgroups([27])
            os.setresgid(1000, 1000, 1000)
            os.setresuid(1000, 1000, 1000)
            cached_key = Path('/var/lib/legacy/nostr_secret').read_bytes()
            require(cached_key == b'FIXTURE-NOT-A-REAL-SIGNING-KEY')
            os.write(write_fd, b'R')
            while True:
                signal.pause()
        except BaseException:
            import traceback
            traceback.print_exc()
            os._exit(111)
    os.close(write_fd)
    require(select.select([read_fd], [], [], 3)[0] and os.read(read_fd, 1) == b'R', 'legacy process barrier')
    os.close(read_fd)
    return pid


def make_policy(root, legacy_group, legacy_pid):
    for path in ('/etc/archipelago-c01/access-policy', '/etc/polkit-1/rules.d', '/etc/systemd/system',
                 '/etc/ssh', '/etc/pam.d', '/etc/cron.d', '/var/spool/cron', '/run/sshd',
                 '/run/archipelago-c01', '/var/lib/archipelago-c01/access',
                 '/var/lib/archipelago-c01/access/quarantine'):
        own_dir(path)
    Path('/run/archipelago-c01').chmod(0o755)
    Path('/etc/ssh').chmod(0o755)
    own_dir('/home/archipelago-admin', 1001)
    for path in ('/home/archipelago', '/var/lib/runtime/graphroot', '/var/lib/runtime/data', '/var/lib/runtime/home'):
        own_dir(path, 1000)
    for role in ('manager', 'snapshot', 'authority'):
        own_dir('/var/lib/private-' + role)
        own_dir('/var/lib/private-' + role + '/data', 61001 if role != 'authority' else 61002)
    accounts = [('root', 0, '/root', '/bin/bash'), ('archipelago', 1000, '/home/archipelago', '/bin/bash'),
                ('archipelago-admin', 1001, '/home/archipelago-admin', '/bin/bash'),
                ('c01-manager', 61001, '/var/empty', '/usr/sbin/nologin'),
                ('c01-authority', 61002, '/var/empty', '/usr/sbin/nologin'),
                ('sshd', 61004, '/var/empty', '/usr/sbin/nologin')]
    old = {}
    old['passwd'] = ''.join(f'{name}:x:{uid}:{uid}::{home}:{shell}\n' for name, uid, home, shell in accounts).encode()
    old['shadow'] = ''.join(f'{name}:$6$fixture$not-a-real-password:20000:0:99999:7:::\n'
                            for name, *_ in accounts).encode()
    old['group'] = (''.join(f'{name}:x:{uid}:\n' for name, uid, *_ in accounts) + 'sudo:x:27:archipelago\n').encode()
    old['gshadow'] = (''.join(f'{name}:!::\n' for name, *_ in accounts) + 'sudo:!::archipelago\n').encode()
    old['sudoers'] = b'root ALL=(ALL:ALL) ALL\narchipelago ALL=(ALL:ALL) NOPASSWD: ALL\n'
    old['sshd'] = (b'HostKey /etc/ssh/fixture_host\nUsePAM no\nPort 22222\nListenAddress 127.0.0.1\n'
                   b'AuthorizedKeysFile /etc/ssh/authorized_%u\nPasswordAuthentication no\n'
                   b'KbdInteractiveAuthentication no\nAllowTcpForwarding yes\n')
    for name, raw in old.items():
        write(policy_module.TARGETS[name], raw, 0o600 if name in ('shadow', 'gshadow') else (0o440 if name == 'sudoers' else 0o644))
    write('/etc/nsswitch.conf', 'passwd: files\ngroup: files\nshadow: files\nsudoers: files\n', 0o644)
    write('/etc/sudo.conf', '', 0o644)
    write('/etc/pam.d/sudo', 'auth required pam_permit.so\naccount required pam_permit.so\nsession required pam_permit.so\n', 0o644)
    write('/etc/subuid', 'archipelago:200000:65536\n', 0o644)
    write('/etc/subgid', 'archipelago:200000:65536\n', 0o644)
    subprocess.run(['/usr/bin/ssh-keygen', '-q', '-t', 'ed25519', '-N', '', '-f', '/etc/ssh/fixture_host'], check=True, timeout=10)
    names = ['archipelago', 'c01-manager', 'c01-authority']
    subprocess.run(['/usr/bin/ssh-keygen', '-q', '-t', 'ed25519', '-N', '', '-f', '/run/fixture_client'],
                   check=True, timeout=10)
    for name in [*names, 'archipelago-admin']:
        write('/etc/ssh/authorized_' + name, Path('/run/fixture_client.pub').read_bytes(), 0o644)
    future = dict(old)
    future['passwd'] = ''.join(f'{name}:x:{uid}:{uid}::{home}:{"/usr/sbin/nologin" if name in names else shell}\n'
                               for name, uid, home, shell in accounts).encode()
    future['shadow'] = ''.join(f'{name}:{"!" if name in names else ""}$6$fixture$not-a-real-password:20000:0:99999:7:::\n'
                               for name, *_ in accounts).encode()
    future['group'] = old['group'].replace(b'sudo:x:27:archipelago', b'sudo:x:27:')
    future['gshadow'] = old['gshadow'].replace(b'sudo:!::archipelago', b'sudo:!::')
    future['sudoers'] = policy_module.sudo_policy('archipelago-admin')
    future['polkit'] = policy_module.polkit_policy(names)
    future['sshd'] = policy_module.ssh_policy(old['sshd'], names)
    future['cron-deny'] = future['at-deny'] = '\n'.join(names).encode() + b'\n'
    for name, raw in future.items():
        write(policy_module.STAGED / name, raw)
    runtime = {'runtime_user': 'archipelago', 'runtime_uid': 1000, 'runtime_gid': 1000,
               'manager_uid': 61001, 'manager_gid': 61001, 'authority_uid': 61002, 'authority_gid': 61002,
               'operator_uid': 1001, 'operator_gid': 1001, 'home': '/home/archipelago',
               'home_backing': '/var/lib/runtime/home', 'runtime_dir': '/run/user/1000',
               'runtime_data_dir': '/var/lib/runtime/data', 'graphroot': '/var/lib/runtime/graphroot',
               'subuids': [[200000, 65536]], 'subgids': [[200000, 65536]],
               'user_cgroup': str((root / 'runtime.service').relative_to('/sys/fs/cgroup')).join(('/', '')),
               'private_parents': ['/var/lib/private-manager', '/var/lib/private-snapshot', '/var/lib/private-authority']}
    launch = {'application_user': 'c01-manager', 'authority_user': 'c01-authority',
              'runtime_uid': 1000, 'runtime_gid': 1000, 'runtime_data_dir': '/var/lib/runtime/data',
              'manager_backing': '/var/lib/private-manager/data', 'snapshot_backing': '/var/lib/private-snapshot/data',
              'authority_backing': '/var/lib/private-authority/data'}
    dump(policy_module.RUNTIME, runtime)
    dump(policy_module.LAUNCH, launch)
    masks = ['legacy-c01.service', 'user@61001.service', 'user@61002.service']
    # Compute the reviewed desired policy-tree hashes using ONLY the private
    # realm, then restore preimages so production transition must perform writes.
    for name, raw in future.items():
        write(policy_module.TARGETS[name], raw, 0o600 if name in ('shadow', 'gshadow') else (0o440 if name == 'sudoers' else 0o644))
    for unit in masks:
        (Path('/etc/systemd/system') / unit).symlink_to('/dev/null')
    trees = {path: policy_module.tree_digest(Path(path)) for path in (
        '/etc/sudo.conf', '/etc/nsswitch.conf', '/etc/pam.d', '/etc/ssh', '/etc/polkit-1',
        '/usr/share/polkit-1/rules.d', '/etc/systemd/system', '/etc/cron.d', '/var/spool/cron', '/etc/subuid', '/etc/subgid')}
    for name, path in policy_module.TARGETS.items():
        if name in old:
            write(path, old[name], 0o600 if name in ('shadow', 'gshadow') else (0o440 if name == 'sudoers' else 0o644))
        else:
            path.unlink()
    for unit in masks:
        (Path('/etc/systemd/system') / unit).unlink()
    meta = legacy_group.stat()
    source = Path('/var/lib/legacy/nostr_secret')
    manifest = {'version': 1, 'authorization': 'one-qualified-guarded-framework-migration',
        'inventory_scope': 'complete-known-writers-and-copies-no-unresolved-signers',
        'runtime_sha256': policy_module.sha(policy_module.read(policy_module.RUNTIME)),
        'launch_sha256': policy_module.sha(policy_module.read(policy_module.LAUNCH)),
        'operator': {'name': 'archipelago-admin', 'uid': 1001, 'gid': 1001},
        'accounts': {role: {'name': name, 'uid': uid, 'gid': uid} for role, name, uid in (
            ('runtime', 'archipelago', 1000), ('manager', 'c01-manager', 61001), ('authority', 'c01-authority', 61002))},
        'files': {name: {'before': policy_module.sha(old[name]) if name in old else None,
                         'after': policy_module.sha(raw)} for name, raw in future.items()},
        'trees': trees,
        'cohorts': [{'path': str(legacy_group), 'device': meta.st_dev, 'inode': meta.st_ino,
                    'members': {str(legacy_pid): kernel.process(legacy_pid)['start']}}],
        'copies': [{'source': str(source), 'identity': policy_module.identity(source.stat()), 'disposition': 'quarantine-known-copy'}],
        'preserved': [{'path': runtime[key], 'identity': policy_module.identity(Path(runtime[key]).stat())}
                      for key in ('home', 'graphroot', 'runtime_data_dir')],
        'restart_units': masks, 'broker_cgroup': str(root / 'broker.service'),
        'supervisor_cgroup': str(root / 'manager.service')}
    dump(policy_module.CONFIG, manifest)
    return policy_module.Policy()


def connect_client(group, digest):
    control, child_control = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET)
    pid = os.fork()
    if pid == 0:
        try:
            control.close()
            (group / 'cgroup.procs').write_text(str(os.getpid()))
            channel = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET)
            channel.settimeout(3)
            channel.connect(str(access.SOCKET))
            request = {'protocol': 'runtime-access-v1', 'configuration_sha256': digest, 'nonce': os.urandom(32).hex()}
            access.send(channel, request)
            require(json.loads(channel.recv(8192)) == request, 'real provider did not grant fresh challenge')
            child_control.send(b'GRANTED')
            channel.settimeout(12)
            require(channel.recv(1) == b'', 'lease did not revoke by EOF')
            child_control.send(b'REVOKED')
            os._exit(0)
        except BaseException:
            import traceback
            traceback.print_exc()
            os._exit(111)
    child_control.close()
    control.settimeout(12)
    require(control.recv(64) == b'GRANTED')
    return pid, control


def runtime_view(policy, group, isolated):
    parent, child = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET)
    pid = os.fork()
    if pid == 0:
        try:
            parent.close()
            (group / 'cgroup.procs').write_text(str(os.getpid()))
            if isolated:
                require(LIBC.unshare(0x00020000 | 0x40000000) == 0, 'fixture mount/net unshare')
                def mount(source, target, kind=None, flags=4096, data=None):
                    require(LIBC.mount(os.fsencode(source) if source else None, os.fsencode(target),
                                       os.fsencode(kind) if kind else None, flags,
                                       os.fsencode(data) if data else None) == 0, 'fixture mount: ' + target)
                mount(None, '/', flags=16384 | (1 << 18))
                mount(policy.runtime['home_backing'], policy.runtime['home'])
                for target in ('/root', '/home/archipelago-admin', '/run/dbus', '/run/systemd/private',
                               '/run/sudo', '/run/polkit-1', *policy.runtime['private_parents']):
                    Path(target).mkdir(parents=True, exist_ok=True)
                    mount('tmpfs', target, 'tmpfs', 2 | 4 | 8, 'mode=000,size=1m')
                write('/run/inert', b'', 0)
                for target in ('sudo', 'pkexec', 'su', 'doas', 'chsh', 'chfn', 'passwd', 'crontab', 'at'):
                    if Path('/usr/bin/' + target).exists():
                        mount('/run/inert', '/usr/bin/' + target)
            os.setgroups([])
            os.setresgid(1000, 1000, 1000)
            os.setresuid(1000, 1000, 1000)
            if isolated:
                require(not os.access('/usr/bin/sudo', os.X_OK), 'sudo executable visible')
                try:
                    os.open('/home/archipelago-admin/credential', os.O_RDONLY)
                except PermissionError:
                    pass
                else:
                    raise AssertionError('operator credential view accessible')
            child.send(b'READY')
            require(child.recv(8) == b'RELEASE')
            os._exit(0)
        except BaseException:
            import traceback
            traceback.print_exc()
            os._exit(111)
    child.close()
    parent.settimeout(5)
    require(parent.recv(16) == b'READY')
    try:
        if isolated:
            require(len(kernel.verify_processes(policy)) == 1)
            passed('real private mount/net/credentials: original UID, curated home, sudo/IPC masks, inaccessible operator backing')
        else:
            rejected(lambda: kernel.verify_processes(policy), 'same-UID runtime in host mount/net view refuses')
    finally:
        parent.send(b'RELEASE')
        parent.close()
        require(os.waitstatus_to_exitcode(os.waitpid(pid, 0)[1]) == 0)


def _supervisor_attempt(group, digest, block=False, refused=False, nonce=None):
    pid = os.fork()
    if pid == 0:
        try:
            (group / 'cgroup.procs').write_text(str(os.getpid()))
            if nonce is not None:
                channel = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET)
                channel.settimeout(6)
                channel.connect(str(access.SOCKET))
                request = {'protocol': 'supervisor-access-v1', 'configuration_sha256': digest, 'nonce': nonce}
                access.send(channel, request)
                response = channel.recv(8192)
                if response == b'' and not refused:
                    os._exit(75)  # Prior lifecycle may still be checked/retiring.
                require(response == b'' if refused else json.loads(response) == request, 'nonce admission result')
                channel.close()
            else:
                try:
                    lease = AccessLease(digest)
                except ValueError:
                    if not refused:
                        os._exit(75)
                else:
                    require(not refused, 'replacement admitted over unresolved cohort')
                    lease.check()
                    if block:
                        (group / 'c01-authority').mkdir()
                    lease.close()
            os._exit(0)
        except BaseException:
            import traceback
            traceback.print_exc()
            os._exit(111)
    return os.waitstatus_to_exitcode(os.waitpid(pid, 0)[1])


def supervisor_cycle(group, digest, block=False, refused=False, nonce=None):
    # Retirement requires a fresh full policy scan and an empty hierarchy. A
    # sleep is not evidence of that completion. Retry only a refused acquisition
    # with no capability, reaping each root candidate and leaving the hierarchy
    # empty between attempts. Negative/replay cases must refuse on first attempt.
    deadline = time.monotonic() + 12
    while True:
        code = _supervisor_attempt(group, digest, block, refused, nonce)
        if code != 75:
            require(code == 0, 'supervisor lease cycle')
            return
        require(not refused and time.monotonic() < deadline, 'supervisor never became eligible after retirement')
        time.sleep(.2)


def crossboot_retired(policy, journal):
    value, _ = policy_module.json_read(journal.path)
    require(value['state'] == 'retired')
    real_boot = journal.boot
    journal.boot = '00000000-0000-0000-0000-000000000000'
    journal.record('retired', value['evidence'])
    journal.boot = real_boot
    rejected(journal.begin_serving, 'completed retired record from old boot cannot directly admit')
    children = Children()
    try:
        access.requalify(policy, journal, children)
    except ValueError:
        parent = Path(kernel.process(os.getpid())['group']).parent
        members = kernel.members(parent)
        print('RECOVERY-PRIMITIVE', json.dumps({'members': sorted(members),
              'roles': [p.name for p in parent.glob('c01-*')],
              'processes': [{k: item[k] for k in ('pid', 'ppid', 'uids', 'group')}
                            for item in kernel.all_processes() if item['pid'] in members]}), flush=True)
        raise
    children.close()
    require(policy_module.json_read(journal.path)[0]['boot'] == real_boot)


def start_owner(group):
    pid = os.fork()
    if pid == 0:
        (group / 'supervisor/cgroup.procs').write_text(str(os.getpid()))
        os.execve('/usr/bin/python3', ['python3', '-I', '-S', '/run/access-code/c01_access.py', 'serve'],
                  {'PATH': '/usr/sbin:/usr/bin:/bin', 'PYTHONDONTWRITEBYTECODE': '1'})
    fd = os.pidfd_open(pid)
    def accepting():
        require(not select.select([fd], [], [], 0)[0], 'actual source owner exited before listener')
        # Binding/journaling the inode precedes listen(). Existence alone races
        # that intentional crash-safe ordering; probe admission transport itself.
        with socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) as probe:
            probe.settimeout(0.1)
            try:
                probe.connect(str(access.SOCKET))
                return True
            except (FileNotFoundError, ConnectionRefusedError):
                return False
    wait(accepting, 20)
    return pid, fd


def main():
    require(os.geteuid() == 0 and os.getpid() == 1 and os.environ.get('C01_ACCESS_FIXTURE') == '1', 'disposable PID realm required')
    mounts = Path('/proc/self/mountinfo').read_text().splitlines()
    for path in ('/etc', '/home', '/run', '/var/lib', '/var/spool'):
        require(any(row.split()[4] == path and ' - tmpfs ' in row for row in mounts), 'private fixture filesystem missing')
    root = Path(kernel.process(os.getpid())['group']).parent
    require(root.name.startswith('c01-access-fixture-') and root.name.endswith('.service'), 'owned fixture unit required')
    (root / 'cgroup.subtree_control').write_text('+cpu +memory +pids')
    # The host-side unshare wrapper shares the outer supervisor cgroup and is
    # rendered as PID 0 in this PID namespace. Give the fixture coordinator its
    # own exact hierarchy, rather than weakening production retirement to ignore
    # invisible tasks. Its helper groups now have the same sole-owner contract
    # as the separate production transition/requalification units.
    coordinator = cgroup(root / 'coordinator.service', True)
    (coordinator / 'supervisor/cgroup.procs').write_text(str(os.getpid()))
    require(kernel.members(coordinator) == {os.getpid()}, 'fixture coordinator shares an unresolved host wrapper')
    passed('fixture coordinator has an exclusive hierarchy; invisible outer unshare task is not exempted by production')
    for path in Path('/opt/source/scripts').glob('c01_access*.py'):
        ast.parse(path.read_text())
    passed('new access sources parse; no compiler/native build')
    write('/run/access-atime', b'public fixture input')
    os.utime('/run/access-atime', ns=(0, Path('/run/access-atime').stat().st_mtime_ns))
    require(policy_module.read(Path('/run/access-atime')) == b'public fixture input')
    passed('deterministic atime advancement does not invalidate stable policy contents/ownership')
    own_dir('/var/lib/legacy', 1000)
    write('/var/lib/legacy/nostr_secret', b'FIXTURE-NOT-A-REAL-SIGNING-KEY', uid=1000)
    legacy_group = cgroup(root / 'legacy.service')
    legacy_pid = fork_legacy(legacy_group)
    policy = make_policy(root, legacy_group, legacy_pid)
    passed('private account/policy manifest validated, distinct operator preserved')
    endpoint_retirement_guards()
    passed('offline socket retirement refuses regular file, live listener and wrong inode; preserves unrelated shared-directory file')
    ssh_route(policy)
    passed('fresh administrator public-key SSH login verifies the private baseline before transition')
    enclosing_retirement(root)
    passed('exact root-owned enclosing slice retires a real delegated UID1000 child; replacement siblings and same inode refuse')
    runtime_bytes = policy_module.read(policy_module.RUNTIME)
    manifest_bytes = policy_module.read(policy_module.CONFIG)
    invalid_runtime = dict(policy.runtime, subuids=[[60000, 65536]])
    dump(policy_module.RUNTIME, invalid_runtime)
    invalid_manifest = dict(policy.value, runtime_sha256=policy_module.sha(policy_module.read(policy_module.RUNTIME)))
    dump(policy_module.CONFIG, invalid_manifest)
    rejected(policy_module.Policy, 'subuid range cannot impersonate manager/authority/operator identities')
    write(policy_module.RUNTIME, runtime_bytes)
    write(policy_module.CONFIG, manifest_bytes)
    old = policy.value['copies'][0]['disposition']
    policy.value['copies'][0]['disposition'] = 'unknown-external-copy'
    rejected(lambda: kernel.quarantine(policy), 'unresolved known/external signing copy refuses; chmod cannot qualify it')
    policy.value['copies'][0]['disposition'] = old
    changed = dict(policy.value['cohorts'][0], inode=0)
    rejected(lambda: kernel.Cohort(changed), 'replaced cgroup inode cannot be retired')
    journal = access.Journal(policy)
    rejected(lambda: access.Journal(policy), 'exclusive root lock blocks concurrent owner')
    for path in (policy_module.STATE, policy_module.STATE / 'quarantine', policy_module.STAGED):
        policy_module.protected(path, private=True, directory=True)
    require(not (legacy_group / 'cgroup.subtree_control').read_text().strip(), 'leaf cohort must admit direct tasks')
    require(subprocess.run(['/usr/sbin/visudo', '-c'], timeout=5).returncode == 0, 'fixture sudo baseline invalid')
    require(subprocess.run(['/usr/sbin/sshd', '-t'], timeout=5).returncode == 0, 'fixture ssh baseline invalid')
    passed('explicit fixture preflight: root-private ancestry, direct-task cgroup, actual sudo/sshd baselines')
    if os.environ.get('C01_ACCESS_CASE') == 'sshparse':
        # Narrow diagnostic after the failed valid preflight. Print only fixture
        # denial tokens, never full SSH/account policy or credentials.
        for address in ('127.0.0.1', '198.51.100.23', '2001:db8::23'):
            code, text = access.fixed_command(['/usr/sbin/sshd', '-T', '-f', str(policy_module.STAGED / 'sshd'),
                                               '-C', f'user=archipelago-admin,host=admin.example,addr={address}'])
            print('SSH-DENIAL-PRIMITIVE', json.dumps({'address': address, 'code': code,
                  'rows': [row for row in text.splitlines() if row.startswith('denyusers ')]}), flush=True)
        (legacy_group / 'cgroup.kill').write_text('1')
        os.waitpid(legacy_pid, 0)
        journal.close()
        return
    if os.environ.get('C01_ACCESS_CASE') == 'preflight':
        # Run the real valid preflight in PID1 first: helper EOF must not hide a
        # deterministic syntax/policy refusal behind a framing diagnostic.
        access.preflight(policy)
        passed('real complete read-only transition preflight succeeds before any helper or policy mutation')
        # Probe effective sudo semantics independently before repeating a whole
        # transition. Only the private fixture sudoers is changed here.
        write('/etc/sudoers', policy_module.sudo_policy('archipelago-admin'), 0o440)
        for name in policy.names + ['archipelago-admin']:
            result = subprocess.run(['/usr/bin/sudo', '-n', '-l', '-U', name], env=access.ENV, timeout=5)
            print('SUDO-PROBE', name, result.returncode, flush=True)
        (legacy_group / 'cgroup.kill').write_text('1')
        os.waitpid(legacy_pid, 0)
        journal.close()
        passed('preflight-only lane; no transition or provider admission attempted')
        return
    calls = []
    original = access.fixed_command
    def command(argv):
        if argv[0] == '/usr/bin/systemctl':
            require(argv == ['/usr/bin/systemctl', 'daemon-reload'] or
                    argv == ['/usr/bin/systemctl', 'reload', 'ssh.service'] or
                    argv == ['/usr/bin/systemctl', 'restart', 'polkit.service'] or
                    argv == ['/usr/bin/systemctl', '--no-block', 'stop', *policy.value['restart_units']],
                    'unexpected activation command')
            calls.append(argv)
            return 0, ''
        return original(argv)
    access.fixed_command = command
    before_files = {name: path.read_bytes() if path.exists() else None for name, path in policy_module.TARGETS.items()}
    for bad_name in policy_module.TARGETS:
        original_before = policy.value['files'][bad_name]['before']
        policy.value['files'][bad_name]['before'] = '0' * 64
        rejected(lambda: access.retirement(policy, lambda step: None), 'preflight rejects stale ' + bad_name)
        policy.value['files'][bad_name]['before'] = original_before
        require(dict(row.split() for row in (legacy_group / 'cgroup.events').read_text().splitlines())['frozen'] == '0')
        require(not calls and not journal.path.exists())
        require(before_files == {name: path.read_bytes() if path.exists() else None
                                 for name, path in policy_module.TARGETS.items()})
    conflict = Path('/etc/systemd/system') / policy.value['restart_units'][-1]
    write(conflict, b'# existing override\n')
    rejected(lambda: access.retirement(policy, lambda step: None), 'last mask conflict refuses before first mask/freeze')
    require(not calls and not journal.path.exists() and
            not (Path('/etc/systemd/system') / policy.value['restart_units'][0]).exists())
    conflict.unlink()
    passed('all finite preimage/mask conflicts refuse without policy mutation, activation commands or frozen cohort')
    original_retirement = access.retirement
    def traced_retirement(*args):
        try:
            return original_retirement(*args)
        except BaseException:
            import traceback
            traceback.print_exc()
            raise
    access.retirement = traced_retirement
    children = Children()
    try:
        access.transition(policy, journal, children)
        pid, status = os.waitpid(legacy_pid, 0)
        require(pid == legacy_pid and os.waitstatus_to_exitcode(status) == -signal.SIGKILL)
        children.close()
    finally:
        access.fixed_command = original
        access.retirement = original_retirement
    require((legacy_group / 'cgroup.events').read_text().split()[1] == '0')
    legacy_group.rmdir()
    value, _ = policy_module.json_read(journal.path)
    require(value['state'] == 'retired' and value['evidence']['writers'][0]['pid'] == legacy_pid)
    require(value['evidence']['copies'][0]['operation'] == 'quarantined-known-copy')
    passed('real freeze + exact PID/start/cgroup capture + cgroup.kill + pidfd exit + direct reap + copy custody')
    passed('actual finite policy replacement; systemctl daemon-reload/stop/ssh-reload/polkit-restart transport is injected')
    policy_module.Policy().verify_files()
    access.semantic_checks(policy)
    ssh_route(policy, denied=True)
    passed('private real sshd preserves fresh admin login and refuses all managed public-key/no-shell forwarding sessions')
    require(original(['/usr/bin/sudo', '-n', '-l', '-U', 'archipelago-admin'])[0] == 0)
    passed('real visudo, sudo -l denies all three managed accounts; distinct administrator sudo retained; real sshd -T DenyUsers')
    require(access.check_current(policy, None) == {'runtime_processes': 0, 'supervisor_retired': False})
    passed('current private backing, known-copy custody, policy trees, account locks and preservation checked independently of journal')
    crossboot_retired(policy, journal)
    passed('completed transition retired provenance requalifies across simulated boot change before first serve (not real reboot)')
    if os.environ.get('C01_ACCESS_CASE') == 'recovery':
        journal.close()
        passed('recovery-only primitive lane; no public provider started')
        return
    runtime_group = cgroup(root / 'runtime.service')
    runtime_view(policy, runtime_group, False)
    runtime_view(policy, runtime_group, True)
    runtime_group.rmdir()
    altered = policy_module.STAGED / 'sudoers'
    saved = altered.read_bytes()
    write(altered, saved + b'archipelago ALL=(ALL:ALL) NOPASSWD: ALL\n')
    rejected(policy_module.Policy, 'staged policy digest mutation refuses')
    write(altered, saved)
    alias = Path('/run/policy-alias')
    alias.symlink_to('/etc/pam.d')
    rejected(lambda: policy_module.tree_digest(alias), 'policy directory alias cannot hide descendants')
    own_dir('/run/policy-links')
    write('/run/policy-links/target', b'ExecStart=/first\n')
    write('/run/policy-links/other', b'ExecStart=/first\n')
    middle = Path('/run/policy-links/middle')
    middle.symlink_to('target')
    top = Path('/run/policy-links/top')
    top.symlink_to('middle')
    digest = policy_module.tree_digest(top)
    write('/run/policy-links/target', b'ExecStart=/changed\n')
    require(policy_module.tree_digest(top) != digest)
    write('/run/policy-links/target', b'ExecStart=/first\n')
    require(policy_module.tree_digest(top) == digest)
    middle.unlink()
    middle.symlink_to('other')
    require(policy_module.tree_digest(top) != digest)
    passed('policy alias hash covers target bytes and intermediate retarget with unchanged original link')
    own_dir('/run/policy-links/real/subdir')
    own_dir('/run/policy-links/plain')
    write('/run/policy-links/payload', b'decoy')
    write('/run/policy-links/real/payload', b'effective')
    hop = Path('/run/policy-links/hop')
    hop.symlink_to('real/subdir')
    parent_link = Path('/run/policy-links/parent-link')
    parent_link.symlink_to('hop/../payload')
    require(parent_link.read_bytes() == b'effective', 'kernel alias/parent counterexample not reproduced')
    rejected(lambda: policy_module.tree_digest(parent_link), 'unchecked alias before parent traversal refuses')
    parent_link.unlink()
    parent_link.symlink_to('plain/../payload')
    digest = policy_module.tree_digest(parent_link)
    require(parent_link.read_bytes() == b'decoy')
    write('/run/policy-links/payload', b'changed-decoy')
    require(policy_module.tree_digest(parent_link) != digest)
    parent_link.unlink()
    parent_link.symlink_to('real/subdir/../payload')
    digest = policy_module.tree_digest(parent_link)
    write('/run/policy-links/real/payload', b'changed-effective')
    require(policy_module.tree_digest(parent_link) != digest)
    passed('raw alias/.. refuses before normalization; protected parent-relative targets hash actual loader bytes')
    journal.close()
    # Install fixed SOURCE only into the private realm, then invoke the actual
    # isolated production entry. No fixture method replaces its checker/leases.
    own_dir('/run/access-code')
    for name in ('c01_access', 'c01_access_policy', 'c01_access_kernel', 'c01_children'):
        write('/run/access-code/' + name + '.py', (Path('/opt/source/scripts') / (name + '.py')).read_bytes())
    owner_group = cgroup(root / 'owner.service', True)
    broker_group = cgroup(root / 'broker.service')
    manager_group = cgroup(root / 'manager.service')
    generation_race(policy, manager_group)
    passed('held no-supervisor scan cannot publish candidate; generation check then real dropped manager/authority children; orphan rejection retained')
    owner_pid, owner_pidfd = start_owner(owner_group)
    passed('actual fixed-source serve entry starts only after real checker evidence')
    wrong = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET)
    wrong.settimeout(2)
    wrong.connect(str(access.SOCKET))
    access.send(wrong, {'protocol': 'runtime-access-v1', 'configuration_sha256': '0' * 64, 'nonce': 'a' * 64})
    require(wrong.recv(1) == b'')
    wrong.close()
    passed('wrong digest/root service peer refused over actual socket')
    client_pid, control = connect_client(broker_group, policy.value['runtime_sha256'])
    passed('fresh nonce+digest lease issued to actual root broker cgroup; held open')
    supervisor_cycle(manager_group, policy.value['launch_sha256'], block=True)
    time.sleep(.6)
    supervisor_cycle(manager_group, policy.value['launch_sha256'], refused=True)
    (manager_group / 'c01-authority').rmdir()
    time.sleep(.6)
    supervisor_cycle(manager_group, policy.value['launch_sha256'])
    require(not select.select([control], [], [], 0)[0], 'manager retirement revoked persistent runtime')
    passed('real supervisor client pidfd lease; clean supervisor retirement/reacquisition preserves runtime lease')
    passed('empty-but-unretired authority cgroup blocks replacement until actual group retirement')
    time.sleep(.6)
    nonce = os.urandom(32).hex()
    supervisor_cycle(manager_group, policy.value['launch_sha256'], nonce=nonce)
    time.sleep(.6)
    supervisor_cycle(manager_group, policy.value['launch_sha256'], refused=True, nonce=nonce)
    passed('saved nonce replay refused after prior supervisor lifetime has actually retired')
    mode = os.environ.get('C01_ACCESS_CASE', 'drift')
    expiry_started = None
    if mode == 'expiry':
        checker_pid, = [int(pid) for pid in Path(f'/proc/{owner_pid}/task/{owner_pid}/children').read_text().split()
                        if kernel.process(int(pid))['group'] == str(owner_group / 'c01-host-io')]
        checker_fd = os.pidfd_open(checker_pid)
        signal.pidfd_send_signal(checker_fd, signal.SIGSTOP)
        os.close(checker_fd)
        expiry_started = time.monotonic()
        # MainPID continues pulsing while its last evidence ages; then it stalls
        # near expiry. Pulses must not buy another five seconds of capability.
        time.sleep(4.0)
        require(not select.select([owner_pidfd], [], [], 0)[0], 'owner died before expiry interleaving')
        signal.pidfd_send_signal(owner_pidfd, signal.SIGSTOP)
    elif mode == 'stall':
        signal.pidfd_send_signal(owner_pidfd, signal.SIGSTOP)
    else:
        write('/etc/sudoers', b'root ALL=(ALL:ALL) ALL\narchipelago ALL=(ALL:ALL) NOPASSWD: ALL\n', 0o440)
    require(control.recv(64) == b'REVOKED')
    if expiry_started is not None:
        require(time.monotonic() - expiry_started < 5.5, 'heartbeat extended stale policy evidence')
        passed('checker stall followed by late MainPID stall revokes by absolute evidence deadline, not last pulse plus five seconds')
    control.close()
    require(os.waitstatus_to_exitcode(os.waitpid(client_pid, 0)[1]) == 0)
    wait(lambda: select.select([owner_pidfd], [], [], 0)[0], 12)
    code = os.waitstatus_to_exitcode(os.waitpid(owner_pid, 0)[1])
    os.close(owner_pidfd)
    require(code == -signal.SIGKILL if mode in ('stall', 'expiry') else code != 0, 'owner must fail closed')
    passed('independent watchdog kills SIGSTOP owner; endpoint EOF cannot be hidden by helper copies' if mode in ('stall', 'expiry')
           else 'effective policy drift revokes actual lease and retires owner/checker cohorts')
    # A recorded retired Boolean never overrides active/failure ownership.
    policy = policy_module.Policy()
    journal = access.Journal(policy)
    rejected(journal.begin_serving, 'owner-active crash fence prevents restart over unresolved lifetime, even with retired operations')
    # The explicit requalification API must also refuse unresolved owned groups;
    # it cannot simply rewrite the crash fence. Cleanup below is fixture-only,
    # after kernel-empty evidence and reaping in this private PID namespace.
    if mode in ('stall', 'expiry'):
        rejected(lambda: access.requalify(policy, journal, None), 'offline requalification refuses unresolved former helper cgroups')
        wait(lambda: not kernel.members(owner_group))
        while True:
            try:
                if os.waitpid(-1, os.WNOHANG)[0] == 0:
                    break
            except ChildProcessError:
                break
        for path in owner_group.glob('c01-*'):
            require(not kernel.members(path), 'fixture cleanup cannot remove populated cohort')
            path.rmdir()
    else:
        write('/etc/sudoers', policy_module.read(policy_module.STAGED / 'sudoers'), 0o440)
    children = Children()
    access.requalify(policy, journal, children)
    children.close()
    value, _ = policy_module.json_read(journal.path)
    require(value['state'] == 'retired' and value['evidence']['requalification']['operation'] ==
            'quiescent-kernel-and-current-policy-reverified')
    passed('explicit offline requalification rechecks current policy, copy custody and quiescent kernel ownership before new lifetime')
    require(not access.SOCKET.exists(), 'requalification left stale socket pathname')
    crossboot_retired(policy, journal)
    passed('completed requalification retired provenance requalifies across simulated boot change before admission (not real reboot)')
    journal.close()
    owner_pid, owner_pidfd = start_owner(owner_group)
    client_pid, control = connect_client(broker_group, policy.value['runtime_sha256'])
    supervisor_cycle(manager_group, policy.value['launch_sha256'])
    signal.pidfd_send_signal(owner_pidfd, signal.SIGTERM)
    require(control.recv(64) == b'REVOKED')
    control.close()
    require(os.waitstatus_to_exitcode(os.waitpid(client_pid, 0)[1]) == 0)
    wait(lambda: select.select([owner_pidfd], [], [], 0)[0], 12)
    require(os.waitstatus_to_exitcode(os.waitpid(owner_pid, 0)[1]) == 0)
    os.close(owner_pidfd)
    passed('second actual fixed-source serve after requalification binds, grants fresh broker/supervisor leases and revokes on clean stop')
    require(not Path('/var/lib/legacy/nostr_secret').exists())
    passed('no legacy key path reappeared; only fixture inode held in root-private custody')
    print(json.dumps({'checks': CHECKS, 'case': mode, 'systemd_activation': 'injected finite transport only',
                      'polkit_daemon': 'not exercised; exact generated rule/order/inventory verified',
                      'production_runtime_view': 'isolated representative process only; not real user manager/Podman composition',
                      'whole_service_acceptance': False}), flush=True)


if __name__ == '__main__':
    main()