#!/usr/bin/python3
"""Real namespace/systemd prerequisite lanes, never a migration/ready receipt.

No access transition is fabricated. Native admission still refuses unsupported
host operations. This fixture tests the implementable network/lifetime boundary
and a real PAM-less rootless user manager, Podman and persistent tmux separately.
"""
import hashlib
import json
import os
from pathlib import Path
import pwd
import shutil
import socket
import subprocess
import sys
import tarfile
import time
import types

assert os.geteuid() == 0 and socket.gethostname() == 'c01-prerequisite-qualification'
assert Path('/run/.containerenv').exists() and os.environ.get('C01_DISPOSABLE_SYSTEMD') == '1'
PACKAGE = Path('/opt/c01-fixture')
for directory, _, files in os.walk(PACKAGE):
    for p in [Path(directory), *(Path(directory) / f for f in files)]:
        mode=p.stat().st_mode & 0o7777
        if mode & 0o755 != mode:p.chmod(mode & 0o755)
sys.path.insert(0, str(PACKAGE / 'scripts'))
import c01_network as network
import c01_access_kernel as kernel
import c01_supervisor as supervisor


def run(*args, check=True):
    result = subprocess.run(args, capture_output=True, text=True, timeout=60)
    if check and result.returncode:
        raise AssertionError((args, result.returncode, result.stdout, result.stderr))
    return result


def write(path, value, mode=0o644):
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(value.encode() if isinstance(value, str) else value)
    path.chmod(mode)


def wait(predicate, label):
    for _ in range(600):
        result = predicate()
        if result:
            return result
        time.sleep(.1)
    raise AssertionError('timeout: ' + label)


def refused(body):
    try:
        body()
    except (ValueError, OSError):
        return
    raise AssertionError('unexpected boundary acceptance')


def own(path):
    Path(path).mkdir(parents=True, exist_ok=True, mode=0o700)
    os.chown(path, 1000, 1000)


ROOT = Path('/var/lib/c01-fixture')
assert not ROOT.exists(), 'fresh disposable overlay required'
(ROOT / 'receipts').mkdir(parents=True, mode=0o755)
run('tar', '-xf', str(PACKAGE/'tmux-runtime.tar'), '-C', str(PACKAGE))
write('/usr/bin/tmux', '#!/bin/sh\nexec /opt/c01-fixture/tmux-runtime/ld-linux-x86-64.so.2 --library-path /opt/c01-fixture/tmux-runtime /opt/c01-fixture/tmux-runtime/tmux "$@"\n', 0o755)
write('/etc/archipelago-c01/runtime-tmux.conf', (PACKAGE/'image-recipe/configs/archipelago-runtime-tmux.conf').read_bytes())
run('systemctl', 'stop', 'archipelago.service', check=False)
run('systemctl', 'stop', 'user@1000.service', check=False)
for role in ('managed', 'runtime'):
    run('ip', 'netns', 'add', 'archipelago-' + role)
    run('nsenter', '--net=/run/netns/archipelago-' + role, '--', 'ip', 'link', 'set', 'lo', 'up')
refused(lambda: network.require_private(os.getpid(), 'managed'))
print('namespace metadata:', [(role, network.NAMES[role].stat()) for role in network.NAMES], flush=True)
assert network.appointed('managed') != network.appointed('runtime')
print('PASS real nsfs binding: host and wrong/private namespaces refuse', flush=True)

# Appointed supervisor cgroup/netns, real systemd lifetime and abstract IPC.
abstract = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
abstract.bind('\0c01-prerequisite-host-only')
abstract.listen(2)
write(PACKAGE / 'managed-probe.py', '''import json,os,pathlib,socket,sys,time
sys.path.insert(0,'/opt/c01-fixture/scripts')
import c01_network,c01_access_kernel
actual=c01_access_kernel.process(os.getpid())
policy=type('Policy',(),{'value':{'supervisor_cgroup':'/sys/fs/cgroup/system.slice/c01-fixture-managed.service'}})()
c01_access_kernel.verify_supervisor(actual,policy)
s=socket.socket(socket.AF_UNIX,socket.SOCK_STREAM)
try:s.connect('\\0c01-prerequisite-host-only')
except OSError:pass
else:raise AssertionError('host abstract IPC exposed')
pathlib.Path('/var/lib/c01-fixture/receipts/managed.json').write_text(json.dumps(actual))
while True:time.sleep(1)
''')
run('systemd-run', '--unit=c01-fixture-managed', '--property=Type=exec', '--property=Delegate=yes',
    '--property=DelegateSubgroup=supervisor', '--property=NoNewPrivileges=yes',
    '--property=NetworkNamespacePath=/run/netns/archipelago-managed',
    '/usr/bin/python3', '-B', str(PACKAGE / 'managed-probe.py'))
wait(lambda: (ROOT / 'receipts/managed.json').exists(), 'managed namespace proof')
managed = json.loads((ROOT / 'receipts/managed.json').read_text())
print('PASS actual systemd supervisor cgroup/NNP binding and host abstract-IPC denial', flush=True)

# A genuine private TCP route between the two disposable namespaces. This is
# not an invented production ingress/localhost-forwarding policy.
run('ip', 'link', 'add', 'c01-m', 'type', 'veth', 'peer', 'name', 'c01-r')
run('ip', 'link', 'set', 'c01-m', 'netns', 'archipelago-managed')
run('ip', 'link', 'set', 'c01-r', 'netns', 'archipelago-runtime')
for name, interface, address in [('managed', 'c01-m', '198.18.0.1/30'), ('runtime', 'c01-r', '198.18.0.2/30')]:
    run('nsenter', '--net=/run/netns/archipelago-' + name, '--', 'ip', 'address', 'add', address, 'dev', interface)
    run('nsenter', '--net=/run/netns/archipelago-' + name, '--', 'ip', 'link', 'set', interface, 'up')
server = subprocess.Popen(['nsenter', '--net=/run/netns/archipelago-runtime', '--', 'python3', '-c',
    "import socket,pathlib;s=socket.socket();s.bind(('198.18.0.2',5678));s.listen();pathlib.Path('/var/lib/c01-fixture/receipts/route-ready').touch();c,_=s.accept();c.sendall(b'private-route');c.close()"], stdin=subprocess.DEVNULL)
wait(lambda: (ROOT/'receipts/route-ready').exists(), 'private TCP server')
run('nsenter', '--net=/run/netns/archipelago-managed', '--', 'python3', '-c',
    "import socket;s=socket.create_connection(('198.18.0.2',5678),timeout=3);assert s.recv(32)==b'private-route'")
assert server.wait(timeout=5) == 0
print('PASS isolated private route; no host networking or host abstract IPC granted', flush=True)

# Appoint the production egress owner before any user-manager/Podman invocation.
# This uses the outer disposable realm's already bounded usermode WAN, not the
# physical host network namespace or a synthetic default route.
import runpy
for module in ('c01_network','c01_endpoints','c01_uplink','runtime_uplink','c01_network_owner'):
    write(Path('/usr/lib/archipelago')/(module+'.py'),(PACKAGE/'scripts'/(module+'.py')).read_bytes(),0o644)
compose=runpy.run_path(str(PACKAGE/'tests/regression/c01-compose.py'))
compose['network_owner']()
print('PASS production-owned runtime uplink appointed before user manager',flush=True)
network_pid=int(run('systemctl','show','archipelago-c01-network.service','-p','MainPID','--value').stdout)
owner_namespace=network.process_namespace(1)
for _ in range(100):
    assert network.process_namespace(network_pid)==owner_namespace,'owner changed namespace during a probe'
    time.sleep(.01)
print('PASS retained network owner namespace is stable during owned probes',flush=True)
import hashlib
policy_digest=hashlib.sha256(Path('/etc/archipelago-c01/endpoints.json').read_bytes()).hexdigest()
for request in ({'method':'lease','policy_sha256':policy_digest,'nonce':'a'*64},
                {'method':'register','policy_sha256':policy_digest,'name':'management',
                 'pid':os.getpid(),'fd':1,'previous':None}):
    with socket.socket(socket.AF_UNIX,socket.SOCK_SEQPACKET) as caller:
        caller.settimeout(3);caller.connect('/run/archipelago-c01-network/control.sock')
        caller.send(json.dumps(request).encode())
        try: assert caller.recv(65536)==b'', 'unappointed root caller/producer admitted'
        except ConnectionResetError:pass
print('PASS real owner refuses unappointed root caller and producer',flush=True)

# Build a tiny local rootfs from container-owned distro objects, no image pulls.
binary = Path('/usr/bin/true')
files = {binary, Path('/bin/true')}
for word in run('ldd', str(binary)).stdout.split():
    if word.startswith('/'):
        files.add(Path(word))
with tarfile.open(PACKAGE / 'probe-rootfs.tar', 'w') as archive:
    for path in sorted(files):
        source = path.resolve()
        info = tarfile.TarInfo(str(path).lstrip('/'))
        info.mode, info.uid, info.gid, info.size = 0o755, 0, 0, source.stat().st_size
        with source.open('rb') as data:
            archive.addfile(info, data)
(PACKAGE / 'probe-rootfs.tar').chmod(0o644)

account = pwd.getpwnam('archipelago')
assert account.pw_uid == account.pw_gid == 1000
# This disposable account inventory is local. Do not let inherited image NSS
# discovery block account retirement or introduce unenumerated remote principals.
write('/etc/nsswitch.conf', 'passwd: files\ngroup: files\nshadow: files\nsudoers: files\nhosts: files dns\nnetworks: files\n')
run('usermod', '-s', '/usr/sbin/nologin', '-G', '', 'archipelago')
run('usermod', '-L', 'archipelago')
history_migration=os.environ.get('C01_HISTORY_MIGRATION')=='1'
subordinate='archipelago:100000:65536\n' if history_migration else 'archipelago:2000:10000\n'
write('/etc/subuid', subordinate)
write('/etc/subgid', subordinate)
node_fixture=PACKAGE/'node-cutover-fixture.json'
if node_fixture.exists():
    node=json.loads(node_fixture.read_text())
    for filename in ('subuid','subgid'):
        assert node['subordinate_maps'][filename][0]=='archipelago:100000:65536'
        write('/etc/'+filename,'\n'.join(node['subordinate_maps'][filename])+'\n')
write('/etc/sudoers', 'root ALL=(ALL:ALL) ALL\n', 0o440)
run('visudo', '-c')
assert run('runuser', '-u', 'archipelago', '--', 'sudo', '-n', 'true', check=False).returncode != 0
# Deliberately writable DAC preimages: namespace denial must do real work.
for path in ('/opt/archipelago/c01-writable-fixture', '/usr/local/bin/c01-writable-fixture'):
    write(path, 'retained old-principal-owned fixture\n', 0o600)
    os.chown(path, 1000, 1000)
home = ROOT / 'home'
for path in (home, home / 'Work', home / '.config', home / '.config/containers',
             home / '.config/systemd', home / '.config/systemd/user', ROOT / 'data', ROOT / 'graphroot',
             home / '.local', home / '.local/state', home / '.local/state/archipelago',
             home / '.local/state/archipelago/sessions'):
    own(path)
for name in ('manager', 'snapshot', 'authority'):
    (ROOT / ('private-' + name)).mkdir(mode=0o700)
if history_migration:(ROOT/'data').chmod(0o755)
write(home / 'Work/operator.txt', 'synthetic retained work\n', 0o600)
os.chown(home / 'Work/operator.txt', 1000, 1000)
preserved = (home / 'Work/operator.txt').stat()
graphroot=ROOT/'data/containers/storage' if history_migration else ROOT/'graphroot'
if history_migration:
    for path in (ROOT/'data/containers',graphroot,home/'.local/share',home/'.local/share/containers'):own(path)
    (home/'.local/share/containers/storage').symlink_to(graphroot)
    os.lchown(home/'.local/share/containers/storage',1000,1000)
driver='overlay' if history_migration else 'vfs'
write(home / '.config/containers/storage.conf',
      '[storage]\ndriver="'+driver+'"\ngraphroot="'+str(graphroot)+'"\nrunroot="/run/user/1000/containers"\n')
write(home / '.config/containers/containers.conf', '[containers]\ndefault_sysctls=[]\nkeyring=false\n[engine]\ncgroup_manager="cgroupfs"\nevents_logger="file"\n')
write(PACKAGE / 'runtime-probe.py', '''import json,os,pathlib,subprocess,time
def run(*args):
 r=subprocess.run(args,capture_output=True,text=True,timeout=45)
 if r.returncode:raise AssertionError((args,r.returncode,r.stdout,r.stderr))
 return r.stdout
run('podman','import','/opt/c01-fixture/probe-rootfs.tar','localhost/c01-prerequisite:fixture')
run('podman','run','--rm','--network=none','--security-opt=seccomp=unconfined','localhost/c01-prerequisite:fixture','/bin/true')
run('tmux','new-session','-d','-s','retained','-c','/home/archipelago/Work','sleep infinity')
pathlib.Path('/home/archipelago/Work/runtime-ready.json').write_text(json.dumps({'pid':os.getpid(),'podman':True,'tmux':True}))
while True:time.sleep(1)
''')
write(home / '.config/systemd/user/c01-fixture-runtime.service', '''[Unit]
Description=Disposable C01 runtime qualification
Requires=archipelago-runtime-tmux.service archipelago-runtime-podman.service
After=archipelago-runtime-tmux.service archipelago-runtime-podman.service
[Service]
Type=exec
ExecStart=/usr/bin/python3 -B /opt/c01-fixture/runtime-probe.py
Delegate=yes
NoNewPrivileges=no
[Install]
WantedBy=default.target
''')
write(home / '.config/systemd/user/archipelago-runtime-tmux.service',
      (PACKAGE/'image-recipe/configs/archipelago-runtime-tmux.service').read_bytes())
write(home / '.config/systemd/user/archipelago-runtime-podman.service',
      (PACKAGE/'image-recipe/configs/archipelago-runtime-podman.service').read_bytes())
wants = home / '.config/systemd/user/default.target.wants'
own(wants)
(wants / 'c01-fixture-runtime.service').symlink_to('../c01-fixture-runtime.service')
for file, mapping in [('80-archipelago-runtime-user.conf.in', {
    '@ORIGINAL_RUNTIME_UID@':'1000', '@ORIGINAL_HOME@':'/home/archipelago',
    '@SANITIZED_RUNTIME_HOME@':str(home), '@MANAGER_BACKING_PARENT@':str(ROOT/'private-manager'),
    '@SNAPSHOT_BACKING_PARENT@':str(ROOT/'private-snapshot'), '@AUTHORITY_BACKING_PARENT@':str(ROOT/'private-authority')}),
    ('81-c01-runtime-access.conf.in', {'@MANAGER_PARENT@':str(ROOT/'private-manager'),
        '@SNAPSHOT_PARENT@':str(ROOT/'private-snapshot'), '@AUTHORITY_PARENT@':str(ROOT/'private-authority')})]:
    text = (PACKAGE / 'image-recipe/configs' / file).read_text()
    for key, value in mapping.items():text=text.replace(key,value)
    write(Path('/etc/systemd/system/user@1000.service.d') / file.replace('.in',''), text)
Path('/home/archipelago-admin').mkdir(exist_ok=True, mode=0o700)
run('systemctl', 'daemon-reload')
start = run('systemctl', 'start', 'user@1000.service', check=False)
if start.returncode:
    print(run('systemctl', 'status', 'user@1000.service', '--no-pager', check=False).stdout, flush=True)
    print(run('systemctl', 'show', 'user@1000.service', '-p', 'ExecMainStatus', '-p', 'Result', '-p', 'StatusErrno').stdout, flush=True)
    raise AssertionError(start.stderr)
wait(lambda: (home/'Work/runtime-ready.json').exists(), 'real rootless Podman/tmux user service')
runtime = json.loads((home/'Work/runtime-ready.json').read_text())
with kernel.PinnedTask(runtime['pid']) as task:
    evidence = kernel.view_evidence(task, types.SimpleNamespace(runtime={
        'home':'/home/archipelago','home_backing':str(home),
        'private_parents':[str(ROOT/('private-'+r)) for r in ('manager','snapshot','authority')]}))
write(ROOT/'receipts/runtime.json', json.dumps(evidence))
for path in ('/opt/archipelago/c01-writable-fixture', '/usr/local/bin/c01-writable-fixture'):
    before = Path(path).stat()
    result = run('nsenter','-t',str(runtime['pid']),'-m','-n','--','setpriv','--reuid=1000','--regid=1000','--clear-groups','--',
        'python3','-c',"import sys;open(sys.argv[1],'wb').write(b'forbidden')",path,check=False)
    assert result.returncode != 0
    assert Path(path).read_text() == 'retained old-principal-owned fixture\n'
    assert Path(path).stat().st_ino == before.st_ino and Path(path).stat().st_uid == 1000
print('PASS old runtime UID denied package writes despite owning exact original inodes; no data chown', flush=True)
run('systemctl', 'stop', 'c01-fixture-managed.service')
assert Path('/proc/'+str(runtime['pid'])).exists()
run('nsenter','-t',str(runtime['pid']),'-m','-n','--','runuser','-u','archipelago','--',
    'tmux','has-session','-t','retained')
after = (home/'Work/operator.txt').stat()
assert (preserved.st_ino,preserved.st_uid,preserved.st_gid,preserved.st_mode) == (after.st_ino,after.st_uid,after.st_gid,after.st_mode)
assert (home/'Work/operator.txt').read_text() == 'synthetic retained work\n'
print('PASS real PAM-less user manager/rootless Podman/persistent tmux and original synthetic Work inode survive management stop', flush=True)
import select
import signal
owner_pid=int(run('systemctl','show','archipelago-c01-network.service','-p','MainPID','--value').stdout)
children=[int(p) for p in Path(f'/proc/{owner_pid}/task/{owner_pid}/children').read_text().split()]
appointed_exe=Path(json.loads(Path('/etc/archipelago-c01/endpoints.json').read_text())['uplink']['executable']).stat()
egress=[]
for child in children:
    try:actual=Path(f'/proc/{child}/exe').stat()
    except FileNotFoundError:continue
    if (actual.st_dev,actual.st_ino)==(appointed_exe.st_dev,appointed_exe.st_ino):egress.append(child)
assert len(egress)==1, 'exact owned pasta child required'
pin=os.pidfd_open(egress[0])
try:
    signal.pidfd_send_signal(pin,signal.SIGTERM)
    wait(lambda: bool(select.select([pin],[],[],0)[0]), 'egress child exit')
    wait(lambda: run('systemctl','is-active','archipelago-c01-network.service',check=False).stdout.strip()=='failed',
         'network owner fail closed on egress loss')
    wait(lambda: run('systemctl','show','user@1000.service','-p','MainPID','--value').stdout.strip()=='0',
         'runtime hierarchy retired on network owner loss')
    wait(lambda: not Path('/run/user/1000').exists(),'PID1 retired ephemeral runtime metadata')
finally:os.close(pin)
assert not Path('/proc/'+str(runtime['pid'])).exists()
assert (home/'Work/operator.txt').read_text()=='synthetic retained work\n'
print('PASS real pasta loss reaped child, failed network owner and retired runtime hierarchy',flush=True)
# Explicit fixture-owner recovery, not an automatic daemon restart or new grant.
run('systemctl','reset-failed','archipelago-c01-network.service')
run('systemctl','start','archipelago-c01-network.service')
wait(lambda: Path('/run/archipelago-c01-network/control.sock').exists(),'new owned network lifetime')
(home/'Work/runtime-ready.json').unlink()
run('systemctl','start','user@1000.service')
wait(lambda: (home/'Work/runtime-ready.json').exists(),'real Podman runtime bootstrap after explicit recovery')
refused(lambda: supervisor.admit({'migration_complete':True}, None, None))
print('PASS missing live admission/candidate contracts refuse; NOT application readiness', flush=True)