#!/usr/bin/env python3
"""Disposable process integration; no installed state or trusted-time claim."""
import hashlib
import json
import os
from pathlib import Path
import shutil
import signal
import socket
import subprocess
import sys
import time
import traceback
from types import SimpleNamespace

sys.path.insert(0, str(Path(__file__).resolve().parents[2] / 'scripts'))
import c01_supervisor as supervisor
import c01_host_watch as watch
import c01_views as views
from c01_fixture_process import run_bounded

assert os.geteuid() == 0 and os.environ.get('ARCHY_C01_SUPERVISOR_FIXTURE') == '1'
assert sys.argv[1:] in ([], ['--python-only'])
assert Path(supervisor.lifecycle.process_group_path(os.getpid())).parent.name.startswith('opencode-c01-supervisor-')
for directory in ('/run', '/var/lib/archipelago', '/var/lib/archipelago-c01', '/etc/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), directory

root = Path('/var/lib/archipelago-c01/fixture')
root.mkdir(mode=0o700)


def refused(body, text=None):
    try:
        body()
    except (OSError, ValueError) as error:
        if text: assert text in str(error), str(error)
    else:
        raise AssertionError('unexpected admission')


def owner(name):
    return supervisor.Supervisor(supervisor.LaunchLock(root / (name + '.lock')))


def await_status(record, wanted):
    record.channel.settimeout(5)
    assert record.channel.recv(128) == wanted


def done(record):
    print('PASS ' + record, flush=True)


refused(lambda: supervisor.admit({'old_writers_excluded': True, 'migration_complete': True}, None, None),
        'C01_ADMISSION_UNAVAILABLE')
done('persisted-exclusion-booleans-cannot-admit')

assert supervisor.runtime_workspace_environment({'home': '/home/preserved-runtime'}) == {
    'ARCHIPELAGO_RUNTIME_WORKSPACE': '/home/preserved-runtime/Work'}
assert supervisor.BASE / 'manager' != Path('/home/preserved-runtime/Work')
done('runtime-workspace-contract-is-independent-of-private-manager-home')

key = root / 'key'
key.write_bytes(bytes(range(1, 33)))
key.chmod(0o600)
fd = supervisor.root_capability(key, 32)
identity = supervisor.key_binding(fd)
assert os.lseek(fd, 0, os.SEEK_CUR) == 0
os.link(key, root / 'key-alias')
refused(lambda: supervisor.root_capability(key, 32), 'invalid root capability')
(root / 'key-alias').unlink()
done('key-capability-no-parent-read-and-hardlink-denial')

managed = root / 'manager'; managed.mkdir(mode=0o700)
directory = views.pin_directory(managed)
identity_dir = managed / 'identity'; identity_dir.mkdir(mode=0o700)
secret = identity_dir / 'nostr_secret'; secret.write_bytes(b'synthetic-identity'); secret.chmod(0o600)
assert supervisor.private_identity(directory, 0) == (secret.stat().st_dev, secret.stat().st_ino)
os.link(secret, root / 'secret-alias')
refused(lambda: supervisor.private_identity(directory, 0), 'secret/alias')
(root / 'secret-alias').unlink()
identity_dir.rename(managed / 'original')
identity_dir.symlink_to(managed / 'original')
refused(lambda: supervisor.private_identity(directory, 0))
os.close(directory)
done('identity-symlink-and-external-hardlink-denied')

owned = owner('keyless')
try:
    owned.children.reserve('administrator')
    reservation = owned.children.reserved['administrator']
    reserved_inode = reservation[0].path.stat().st_ino
    refused(lambda: owned.children.reserve('administrator'), 'role already')
    def child(control):
        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) != (identity['device'], identity['inode'])
        supervisor.lifecycle.drop_credentials(60002, 60002, os.getppid())
        control.send(b'KEYLESS')
        supervisor.child_wait(control, b'RELEASE', supervisor.now() + 5_000_000_000)
    record = owned.children.spawn('administrator', child)
    assert not owned.children.reserved and record.group.path.stat().st_ino == reserved_inode
    time.sleep(0.2)  # Force queued status to survive delayed parent scheduling.
    owned.wait(record, b'KEYLESS', 60002, 60002, supervisor.now() + 5_000_000_000)
    assert os.lseek(fd, 0, os.SEEK_CUR) == 0
    supervisor.packet(record.channel, b'RELEASE')
    end = time.monotonic() + 3
    while 'administrator' not in owned.children.results and time.monotonic() < end:
        owned.tick(); time.sleep(0.005)
    assert owned.children.results.get('administrator') == 0
finally:
    owned.retire()
os.close(fd)
done('late-administrator-inherits-no-storage-key-fds')

owned = owner('unused-reservation')
owned.children.reserve('administrator')
reserved_path = owned.children.reserved['administrator'][0].path
owned.retire()
assert not reserved_path.exists()
done('unused-precollection-admin-reservation-retired')

owned = owner('retained')
try:
    record = owned.children.spawn('host-io', lambda control: time.sleep(60))
    original_empty = record.group.empty
    record.group.empty = lambda: False  # injected kernel-unretired observation
    owned.stop('injected stall')
    for _ in range(10):
        owned.tick(); time.sleep(0.01)
    assert 'host-io' in owned.children.children
    refused(lambda: supervisor.LaunchLock(root / 'retained.lock'))
    for _ in range(5): os.kill(os.getpid(), signal.SIGTERM)
    assert owned.children.stopping
    record.group.empty = original_empty
finally:
    owned.retire()
reacquired = supervisor.LaunchLock(root / 'retained.lock'); reacquired.close()
done('stall-retains-lock-and-repeated-stop-remains-responsive')

# A new flock owner must not mistake the previous root process's crash for
# completed role retirement, including a different delegated-unit path.
owned = owner('crash-locator')
try:
    record = owned.children.spawn('host-io', lambda control: time.sleep(60))
    other = supervisor.LaunchLock(root / 'second.lock')
    locator = json.dumps({'boot': supervisor.recovery.boot_id(), 'group': str(owned.children.root)}).encode()
    os.write(other.fd, locator)
    refused(lambda: other.bind_group(owned.children.root), 'retirement remains unresolved')
    other.close()
finally:
    owned.retire()
done('crash-locator-rejects-prior-unretired-role')

owned = owner('host-watch')
client, accepted = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET)
try:
    def worker(control):
        control.send(watch.PACKET.pack(1, watch.READY, 0, 0))
        control.send(watch.PACKET.pack(1, watch.STARTED, 1, watch.now() + 300_000_000))
        time.sleep(60)
    record = owned.children.spawn('host-io', worker, keep=(accepted.fileno(),))
    owned.watch = watch.HostWatch(record.channel, accepted, record.pid, watch.now() + 1_000_000_000)
    end = time.monotonic() + 3
    while owned.tick() and time.monotonic() < end: time.sleep(0.01)
    assert owned.children.stopping and owned.watch.failure == 'host worker deadline expired'
    client.settimeout(1); assert client.recv(1) == b''
finally:
    owned.retire(); client.close()
done('host-progress-stall-revokes-actual-application-endpoint')

owned = owner('forged-status')
try:
    record = owned.children.spawn('authority', lambda control: (control.send(b'STORE_READY'), time.sleep(60)))
    refused(lambda: owned.wait(record, b'STORE_READY', 60002, 60002,
                              supervisor.now() + 5_000_000_000), 'status refused')
finally:
    owned.retire()
done('setup-status-requires-actual-role-credentials')

if sys.argv[1:] == ['--python-only']:
    assert not Path(f'/proc/self/task/{os.getpid()}/children').read_text().strip()
    print('Python-only fixture boundary: native embedding/proxy cases NOT executed', flush=True)
    raise SystemExit(0)

# Compose the launcher's actual mount preparation with its lifecycle owner,
# using preserved tmpfs backing inodes, not a mocked view implementation.
prepared = supervisor.Prepared.__new__(supervisor.Prepared)
prepared.fds = []
prepared.value = {'runtime_uid': 1000, 'runtime_gid': 1000, 'runtime_data_dir': '/var/lib/archipelago'}
prepared.backings = {}
prepared.backing_sources = {}
for name in ('manager', 'snapshot', 'authority'):
    path = root / ('backing-' + name); path.mkdir(mode=0o700); os.chown(path, 1000, 1000)
    (path / 'record').write_bytes(name.encode()); (path / 'record').chmod(0o600); os.chown(path / 'record', 1000, 1000)
    prepared.backings[name] = prepared.own(views.require_private_backing(path))
    prepared.backing_sources[name] = views.PinnedSource.capture(prepared.backings[name], path)
runtime = Path('/var/lib/archipelago'); runtime.chmod(0o700); os.chown(runtime, 1000, 1000)
(runtime / 'wallet-fixture').write_bytes(b'preserved'); (runtime / 'wallet-fixture').chmod(0o600)
os.chown(runtime / 'wallet-fixture', 1000, 1000)
prepared.runtime = prepared.own(os.open(runtime, os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC))
prepared.runtime_source = views.PinnedSource.capture(prepared.runtime, runtime)
proxy = root / 'proxy'
with proxy.open('xb') as output:
    subprocess.run(['/opt/fixture/native', 'export-runtime-proxy'], stdout=output, check=True, timeout=5)
proxy.chmod(0o555)
prepared.proxy = prepared.own(os.open(proxy, os.O_PATH | os.O_CLOEXEC))
prepared.proxy_source = views.PinnedSource.capture(prepared.proxy, proxy)
original = {name: views.identity(fd) for name, fd in prepared.backings.items()}
old_namespace = os.stat('/proc/self/ns/mnt').st_ino
host_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
host_socket.bind('/run/forbidden-host-bus')
owned = owner('views')
try:
    # Capture proxy metadata before drop; the protected source path is denied.
    proxy_inode = proxy.stat().st_ino
    def mapped(control):
        try:
            prepared.role_views(SimpleNamespace(pw_uid=60001, pw_gid=60001), 'application')
            supervisor.close_except({control.fileno()})
            supervisor.lifecycle.drop_credentials(60001, 60001, os.getppid())
            assert (supervisor.BASE / 'manager/record').read_bytes() == b'manager'
            assert (runtime / 'wallet-fixture').read_bytes() == b'preserved'
            assert not Path('/run/forbidden-host-bus').exists()
            assert (supervisor.BASE / 'bin/podman').stat().st_ino == proxy_inode
            refused(lambda: (root / 'backing-authority/record').read_bytes())
            refused(lambda: (supervisor.BASE / 'bin/podman').open('wb'))
            control.send(b'VIEWS_READY')
            time.sleep(60)
        except BaseException:
            traceback.print_exc(); raise
    record = owned.children.spawn('application', mapped, keep=prepared.fds)
    owned.wait(record, b'VIEWS_READY', 60001, 60001, supervisor.now() + 15_000_000_000)
finally:
    owned.retire(); host_socket.close()
assert os.stat('/proc/self/ns/mnt').st_ino == old_namespace
assert all(views.identity(fd) == original[name] for name, fd in prepared.backings.items())
assert (runtime / 'wallet-fixture').stat().st_uid == 1000
prepared.close()
done('composed-private-views-proxy-inode-aliases-and-runtime-preservation')

# Use the real embedded payload for both identity and hostile import checks.
binary = root / 'native'; shutil.copyfile('/opt/fixture/native', binary); binary.chmod(0o755)
poison = root / 'poison'; poison.mkdir()
marker = root / 'imported-untrusted-code'
for name in ('sitecustomize', 'c01_supervisor', 'ctypes', 'json'):
    (poison / (name + '.py')).write_text(f'open({str(marker)!r},"w").write("bad")\n')
env = {**os.environ, 'PYTHONPATH': str(poison), 'PYTHONHOME': str(poison), 'PYTHONSTARTUP': str(poison / 'c01_supervisor.py')}
seen_native = False
def sample_native(pid):
    global seen_native
    try:
        actual = Path(f'/proc/{pid}/exe').stat()
        expected = binary.stat()
        seen_native |= (actual.st_dev, actual.st_ino) == (expected.st_dev, expected.st_ino)
    except FileNotFoundError:
        pass
result = run_bounded([str(binary), 'c01-supervisor-v3'], cwd=poison, env=env,
                     seconds=5, observe=sample_native)
output, errors = result.stdout, result.stderr
assert result.returncode != 0 and seen_native and not marker.exists(), errors.decode()
assert b'/etc/archipelago-c01/launch.json' in errors, errors.decode()
assert not output
done('native-mainpid-isolated-embedding-ignores-env-site-and-cwd')

extra = run_bounded([str(binary), 'c01-supervisor-v3', str(poison / 'c01_supervisor.py')], seconds=5)
assert extra.returncode != 0 and b'accepts no arguments' in extra.stderr and not marker.exists()
done('native-supervisor-rejects-arbitrary-root-code-arguments')

assert not Path(f'/proc/self/task/{os.getpid()}/children').read_text().strip()
assert not any((Path('/sys/fs/cgroup') / supervisor.lifecycle.process_group_path(os.getpid()).lstrip('/')).parent.glob('c01-*'))
done('all-owned-children-and-role-cgroups-retired')