#!/usr/bin/python3
"""Actual native Tor worker + production watchdog/systemctl in disposable PID1.

Exercises the finite role independently. Never invokes or replaces admit().
"""
import hashlib
import json
import os
from pathlib import Path
import socket
import subprocess
import sys
import time

BASE = Path('/opt/c01-fixture')
OUT = Path('/var/lib/tor-qualification')
sys.path.insert(0, str(BASE / 'scripts'))
import c01_supervisor as supervisor
import c01_tor


def run(*args, timeout=120):
    p = subprocess.run(args, capture_output=True, text=True, timeout=timeout)
    assert p.returncode == 0, (args, p.returncode, p.stdout, p.stderr)
    return p.stdout


def record(name, value):
    (OUT / name).write_text(json.dumps(value, indent=2) + '\n')


def snapshot():
    paths = list(Path('/var/lib/tor').glob('hidden_service_*'))
    paths += [item for directory in paths for item in directory.iterdir()]
    paths += [Path('/var/lib/archipelago/runtime/lnd/wallet-canary')]
    result = {}
    for path in paths:
        m = path.lstat()
        result[str(path)] = dict(inode=m.st_ino, device=m.st_dev, uid=m.st_uid,
            gid=m.st_gid, mode=m.st_mode,
            sha256=hashlib.sha256(path.read_bytes()).hexdigest() if path.is_file() else None)
    return result


class Session:
    def __init__(self):
        self.lock = supervisor.LaunchLock()
        self.owner = supervisor.Supervisor(self.lock)
        self.listener = c01_tor.listener()
        address = self.listener.getsockname()
        parent = os.getpid()
        def app(control):
            supervisor.lifecycle.drop_credentials(65001, 65001, parent)
            channel = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET)
            channel.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)
            channel.connect(address)
            supervisor.packet(control, b'CONNECTED')
            while True:
                request = control.recv(4096)
                if not request:
                    return
                if request == b'FIXTURE_TRANSFER_SOCKET':
                    import array
                    control.sendmsg([b'SOCKET'], [(socket.SOL_SOCKET, socket.SCM_RIGHTS,
                        array.array('i', [channel.fileno()]))])
                    continue
                channel.send(request)
                reply = channel.recv(4096)
                if not reply:
                    return
                supervisor.packet(control, reply)
        self.app = self.owner.children.spawn('application', app)
        self.owner.wait(self.app, b'CONNECTED', 65001, 65001, supervisor.now()+2_000_000_000)
        accepted, _ = self.listener.accept()
        runtime = os.open('/var/lib/archipelago/runtime', os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC)
        self.dirs = c01_tor.capabilities(runtime)
        os.close(runtime)
        self.executable = os.open('/usr/local/bin/archipelago', os.O_RDONLY | os.O_CLOEXEC)
        self.worker = c01_tor.spawn(self.owner, self.executable, accepted,
            dict(pid=self.app.pid, uid=65001, gid=65001), self.dirs,
            supervisor.native, supervisor.packet, {'PATH':'/usr/bin:/usr/sbin'})
        end = supervisor.now()+2_000_000_000
        while self.owner.tor_watch.state == 'starting' and supervisor.now() < end and self.owner.tick():
            time.sleep(.01)
        assert self.owner.tor_watch.state == 'idle', self.owner.failure
        self.pid = self.worker.pid
        self.group = str(self.worker.group.path)

    def request(self, operation, sequence=None):
        request = 'Status' if sequence is None else {'Apply':dict(sequence=sequence, operation=operation)}
        supervisor.packet(self.app.channel, request)
        return json.loads(self.owner.receive(self.app, 65001, 65001, supervisor.now()+65_000_000_000))

    def close(self):
        self.owner.stop('fixture retirement')
        end = time.monotonic()+10
        while self.owner.children.children and time.monotonic()<end:
            self.owner.tick()
            time.sleep(.01)
        assert not self.owner.children.children, 'unretired Tor/application roles'
        results = dict(self.owner.children.results)
        self.owner.children.close()
        os.close(self.owner.wake_read)
        os.close(self.owner.wake_write)
        import signal
        signal.set_wakeup_fd(self.owner.prior_wakeup)
        self.lock.close()
        self.listener.close()
        for fd in [self.executable, *self.dirs]:
            os.close(fd)
        assert not Path(self.group).exists()
        return results

    def root_sender_refused(self):
        # Deliberately transfer the app endpoint to root. Possessing the socket
        # must not replace exact per-packet nonroot PID/UID authentication.
        import array
        supervisor.packet(self.app.channel, b'FIXTURE_TRANSFER_SOCKET')
        self.app.channel.settimeout(2)
        raw, anc, flags, _ = self.app.channel.recvmsg(64, socket.CMSG_SPACE(12)+socket.CMSG_SPACE(4), socket.MSG_CMSG_CLOEXEC)
        assert raw == b'SOCKET' and not flags & ~socket.MSG_CMSG_CLOEXEC
        rights = array.array('i')
        for level, kind, data in anc:
            if (level,kind) == (socket.SOL_SOCKET,socket.SCM_RIGHTS):
                rights.frombytes(data)
        assert len(rights) == 1
        forged = socket.socket(fileno=rights[0])
        try:
            forged.send(json.dumps({'Apply':dict(sequence=3,operation='Reload')}).encode())
            end = supervisor.now()+2_000_000_000
            while supervisor.now()<end and self.owner.tick():
                time.sleep(.01)
            assert self.owner.tor_watch.state == 'failed', 'root sender was accepted as the app'
        finally:
            forged.close()


def role_tests():
    OUT.mkdir(mode=0o700, exist_ok=True)
    # Execute the production embedded module table too. The held real launch
    # lock must stop it before acquiring another Children owner or admission.
    guard = supervisor.LaunchLock()
    try:
        embedded = subprocess.run(['/usr/local/bin/archipelago','c01-supervisor-v3'],
            capture_output=True,text=True,timeout=20)
    finally:
        guard.close()
    record('embedded-supervisor.json', dict(exit=embedded.returncode, stderr=embedded.stderr))
    assert embedded.returncode != 0 and 'BlockingIOError' in embedded.stderr and '<embedded:c01_supervisor>' in embedded.stderr, embedded.stderr
    healthy = Path('/etc/tor/torrc').read_text()
    original = snapshot()
    before_pid = run('systemctl','show','tor@default.service','-p','MainPID','--value').strip()
    Path('/etc/tor/torrc').write_text(healthy.replace('SocksPort 9050\n',
        'SocksPort 9050\nSocksPort 192.0.2.88:9050\n'))
    session = Session()
    try:
        record('worker.json', dict(pid=session.pid, group=session.group,
            memory_max=Path(session.group+'/memory.max').read_text(),
            swap_max=Path(session.group+'/memory.swap.max').read_text(),
            tasks_max=Path(session.group+'/pids.max').read_text()))
        assert session.request(None)['sequence'] == 0
        reply = session.request('Heal', 1)
        assert reply['outcome'] == dict(changed=True,reloaded=True,active=True,exported=1), reply
        assert Path('/etc/tor/torrc').read_text() == healthy
        assert run('systemctl','show','tor@default.service','-p','MainPID','--value').strip() == before_pid
        assert Path('/var/lib/archipelago/runtime/tor-hostnames/archipelago').read_bytes() == Path('/var/lib/tor/hidden_service_archipelago/hostname').read_bytes()
        assert snapshot() == original
        record('repair.json', reply)
        # Same durable sequence is a cached completion, not a second reload.
        assert session.request('Heal', 1) == reply
        run('systemctl', 'stop', 'tor@default.service')
        stopped = session.request('Reload', 2)
        assert stopped['outcome']['active'] is False and stopped['outcome']['reloaded'] is False
        assert run('systemctl','show','tor@default.service','-p','ActiveState','--value').strip() == 'inactive'
        assert snapshot() == original
        record('stopped.json', stopped)
    finally:
        record('first-retirement.json', session.close())
    # Fresh direct native child recovers the durable sequence; no new domain.
    session = Session()
    try:
        status = session.request(None)
        assert status['sequence'] == 2 and not status['uncertain']
        record('recovered.json', status)
    finally:
        record('recovery-retirement.json', session.close())
    complete = Path('/var/lib/archipelago-c01/tor/state').read_bytes()
    session = Session()
    try:
        session.root_sender_refused()
        assert Path('/var/lib/archipelago-c01/tor/state').read_bytes() == complete
        assert snapshot() == original
        record('wrong-sender.json', dict(refused=True,journal_preserved=True))
    finally:
        record('wrong-sender-retirement.json', session.close())
    session = Session()
    try:
        try:
            session.request({'Reload':{'path':'/etc/shadow','argv':['start']}},3)
        except (ValueError,OSError,json.JSONDecodeError):
            pass
        else:
            raise AssertionError('nonfinite Tor request accepted')
        assert Path('/var/lib/archipelago-c01/tor/state').read_bytes() == complete
        assert snapshot() == original
        record('nonfinite-request.json',dict(refused=True,journal_preserved=True))
    finally:
        record('nonfinite-retirement.json',session.close())
    run('systemctl', 'start', 'tor@default.service')
    # Make the real PID1 reload job exceed the production sixty-second budget.
    dropin = Path('/etc/systemd/system/tor@default.service.d')
    dropin.mkdir()
    (dropin/'timeout.conf').write_text('[Service]\nExecReload=\nExecReload=/usr/bin/sleep 90\nTimeoutStartSec=120\n')
    run('systemctl','daemon-reload')
    session = Session()
    started = supervisor.now()
    try:
        try:
            session.request('Reload', 3)
        except (ValueError, OSError, json.JSONDecodeError):
            pass
        else:
            raise AssertionError('timed-out mutation reported success')
        elapsed = supervisor.now()-started
        assert 59_000_000_000 <= elapsed <= 65_000_000_000, elapsed
        assert session.owner.tor_watch.state == 'failed'
        intent = json.loads(Path('/var/lib/archipelago-c01/tor/state').read_text())
        assert intent['sequence'] == 3 and intent['outcome'] is None
        assert snapshot() == original
        record('timeout.json', dict(elapsed_ns=elapsed, reason=session.owner.tor_watch.reason,
            journal=intent, service_state=run('systemctl','show','tor@default.service','-p','ActiveState','-p','SubState')))
    finally:
        record('timeout-retirement.json', session.close())
    # No replay after fresh owner startup, even though PID1 may still own the
    # original service job. Reconciliation is external; this fixture stops it.
    session = Session()
    try:
        try:
            status = session.request(None)
            assert status['uncertain'] and status['sequence'] == 3
        except (ValueError, OSError, json.JSONDecodeError):
            assert session.owner.tor_watch.state == 'failed'
        assert json.loads(Path('/var/lib/archipelago-c01/tor/state').read_text()) == intent
    finally:
        record('uncertain-recovery-retirement.json', session.close())
    run('systemctl','stop','tor@default.service')
    assert snapshot() == original
    record('preservation.json', dict(before=original, after=snapshot()))
    record('result.json', dict(exit=0, native_role=True, production_systemctl=True,
        admission_bypassed=False, app_ready=False))


def setup():
    assert os.geteuid() == 0 and Path('/run/.containerenv').exists()
    assert Path('/proc/1/comm').read_text().strip() == 'systemd'
    assert os.environ.get('C01_TOR_PID1') == '1'
    OUT.mkdir(mode=0o700)
    import importlib.util
    spec = importlib.util.spec_from_file_location('native_boundaries', BASE/'tests/regression/c01-normal-systemd.py')
    probes = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(probes)
    probes.EVIDENCE = OUT/'abi'
    probes.EVIDENCE.mkdir()
    for selector, label, expected, slots, passcred in [
        ('c01-managed-v6','missing-fd12','required Tor lifecycle FD12 missing',(),False),
        ('c01-managed-v6','missing-fd11','required network settings FD11 missing',(12,),False),
        ('c01-managed-v6','missing-fd10','required D09 FD10 missing',(11,12),False),
        ('c01-managed-v6','foreign-fd10','D09 FD10 type/passcred',(10,11,12),False),
        ('c01-managed-v6','parent-not-d09','FD10 must name independent D09 creator',(10,11,12),True),
        *[(f'c01-managed-v{v}',f'obsolete-v{v}','obsolete protected application ABI',(),False) for v in (3,4,5)],
    ]:
        probes.native_boundary(selector,label,expected,slots,passcred)
    for path in ['/etc/tor','/var/lib/tor','/var/lib/archipelago-c01/tor',
                 '/var/lib/archipelago/runtime/tor-hostnames','/var/lib/archipelago/runtime/lnd']:
        Path(path).mkdir(parents=True, exist_ok=True)
    # Purely synthetic preserved data, never imported from an installed node.
    assert not list(Path('/var/lib/tor').iterdir()), 'fixture image must not carry Tor identities'
    Path('/var/lib/archipelago/runtime/lnd/wallet-canary').write_bytes(b'synthetic-wallet-no-funds')
    uninstalled = Path('/var/lib/tor/hidden_service_uninstalled')
    uninstalled.mkdir(mode=0o700)
    (uninstalled/'hs_ed25519_secret_key').write_bytes(b'synthetic-retained-private-key')
    config = ('# Auto-generated by Archipelago — do not edit manually\n'
        'User debian-tor\nDataDirectory /var/lib/tor\nDisableNetwork 1\nSocksPort 9050\n'
        'SocksPolicy accept 127.0.0.1/32\nSocksPolicy accept 192.0.2.0/24\nSocksPolicy reject *\n'
        'Log notice stdout\nHiddenServiceDir /var/lib/tor/hidden_service_archipelago\n'
        'HiddenServicePort 80 127.0.0.1:80\n')
    Path('/etc/tor/torrc').write_text(config)
    Path('/etc/systemd/system/tor@default.service').write_text('[Service]\nType=simple\n'
        'ExecStart=/usr/bin/tor -f /etc/tor/torrc\nExecReload=/bin/kill -HUP $MAINPID\n'
        'Restart=no\nNoNewPrivileges=yes\n')
    run('systemctl','daemon-reload')
    run('systemctl','start','tor@default.service')
    end = time.monotonic()+20
    while not Path('/var/lib/tor/hidden_service_archipelago/hostname').exists() and time.monotonic()<end:
        time.sleep(.05)
    assert Path('/var/lib/tor/hidden_service_archipelago/hostname').exists()
    # Tor signal readiness is separate from key creation.
    time.sleep(1)
    run('ip', 'netns', 'add', 'archipelago-managed')
    run('systemd-run','--unit=tor-native-fixture','--wait','--pipe',
        '-p','Delegate=yes','-p','DelegateSubgroup=supervisor','-p','NoNewPrivileges=yes',
        '-p','MemoryMax=512M','-p','MemorySwapMax=0','-p','CPUQuota=200%',
        '-p','NetworkNamespacePath=/run/netns/archipelago-managed',
        'python3','-B',str(Path(__file__)),'roles', timeout=180)
    (OUT/'tor-journal.log').write_text(run('journalctl','-u','tor@default.service','--no-pager'))
    assert 'Received reload signal' in (OUT/'tor-journal.log').read_text()


if __name__ == '__main__':
    if sys.argv[1:] == ['roles']:
        role_tests()
    else:
        setup()