"""Root-appointed usermode runtime egress, owned by the finite network owner.

No RPC, caller argv, host route/firewall/sysctl mutation or reconnect loop.
The executable, WAN tuple, DNS and target namespace are immutable appointments.
"""
import ctypes
import hashlib
import ipaddress
import os
from pathlib import Path
import re
import select
import signal
import stat
import subprocess
import time

import runtime_uplink
import c01_network

LIBC=ctypes.CDLL(None,use_errno=True)
ROOT=Path('/run/archipelago-c01-network')
NAMESPACE_ROOT=Path('/tmp/archipelago-c01-network-netns')


def validate(value):
    base={'executable','sha256','wan','dns'}
    if type(value) is not dict or set(value) not in (base,base|{'routing','dns_host'},base|{'routing','dns_host','runtime_addresses'}):
        raise ValueError('uplink appointment fields')
    if 'routing' in value:
        if value['routing']!='host-dual-stack' or value['dns_host']!='100.100.100.100' or value['dns']!=['1.1.1.1']:
            raise ValueError('unqualified host-routing/DNS appointment')
    if 'runtime_addresses' in value:
        addresses=value['runtime_addresses']
        if type(addresses) is not list or not 1<=len(addresses)<=8 or len(set(addresses))!=len(addresses):raise ValueError('runtime address bound')
        for address in addresses:
            ip=ipaddress.IPv4Address(address)
            if str(ip)!=address or ip.is_unspecified or ip.is_loopback or ip.is_link_local or ip.is_multicast:raise ValueError('runtime address scope')
    if value['executable'] not in ('/usr/bin/pasta','/usr/bin/pasta.avx2') or not isinstance(value['sha256'],str) or not re.fullmatch('[0-9a-f]{64}',value['sha256']):
        raise ValueError('uplink executable appointment')
    wan=value['wan']
    if type(wan) is not list or len(wan)!=3 or not isinstance(wan[0],str) or not re.fullmatch('[a-zA-Z0-9_.-]{1,15}',wan[0]):
        raise ValueError('uplink WAN tuple')
    for address in wan[1:]:
        ip=ipaddress.IPv4Address(address)
        if ip.is_loopback or ip.is_link_local or ip.is_unspecified or ip.is_multicast:
            raise ValueError('uplink WAN address')
    dns=value['dns']
    if type(dns) is not list or not 1<=len(dns)<=4 or len(set(dns))!=len(dns):
        raise ValueError('uplink DNS appointment')
    for address in dns:
        if not ipaddress.IPv4Address(address).is_global:
            raise ValueError('uplink DNS must be an explicit global IPv4 resolver')
    return value


def arguments(value,target):
    validate(value)
    interface,gateway,source=value['wan']
    argv=[value['executable'],'--foreground','--runas','0:0','--no-netns-quit',
          '--config-net','--no-map-gw','--no-dhcp','--no-dhcpv6','--no-ra',
          '-t','none','-u','none','-T','none','-U','none',
          '--interface',interface,'--address',source,'--gateway',gateway,
          '--ns-ifname','c01-uplink','--netns',str(target)]
    if value.get('routing')=='host-dual-stack':
        # Do not bind outbound sockets to the WAN source/interface: FIPS and
        # MagicDNS require the host kernel's destination-specific selection.
        # no-icmp also suppresses NDP in the installed pasta, breaking IPv6
        # before any TCP socket is created. Keep gateway-to-loopback mapping off.
        argv+=['--dns-forward','1.1.1.1','--dns-host',value['dns_host']]
    else:
        argv+=['-4','--no-ndp','--no-icmp','--outbound-if4',interface,'--outbound',source]
    return argv


def check_resolver(value):
    if value.get('routing')=='host-dual-stack':
        raw=Path('/etc/resolv.conf').read_text()
        if len(raw)>16384:raise ValueError('host resolver exceeds bound')
        servers=[line.split()[1] for line in raw.splitlines()
                 if line.split() and line.split()[0]=='nameserver' and len(line.split())>=2]
        if value['dns_host'] not in servers:raise ValueError('appointed host DNS is no longer configured')
    if value.get('runtime_addresses'):
        result=subprocess.run(['/usr/bin/ip','-j','-4','address','show'],capture_output=True,check=True,timeout=2)
        if len(result.stdout)>65536:raise ValueError('host address inventory exceeds bound')
        import json
        actual={a['local'] for link in json.loads(result.stdout) for a in link.get('addr_info',[]) if a.get('family')=='inet'}
        if not set(value['runtime_addresses'])<=actual:raise ValueError('runtime bind address is not an actual host address')


class Uplink:
    def __init__(self,policy,value,realm='runtime'):
        if realm not in ('runtime','managed'):raise ValueError('uplink target realm')
        self.realm=realm
        self.policy=policy;self.value=validate(value);self.child=None;self.pidfd=None
        self.exe=None;self.mounted=False;self.next_check=0;self.target=NAMESPACE_ROOT/(realm+'-netns')
        self.resolver=ROOT/('resolv.conf' if realm=='runtime' else 'managed-resolv.conf')
        self.created=[]
        try:
            check_resolver(value)
            if tuple(value['wan'])!=runtime_uplink.ready():raise ValueError('appointed WAN is not current')
            time.sleep(1)
            if tuple(value['wan'])!=runtime_uplink.ready():raise ValueError('WAN did not stabilize')
            path=Path(value['executable'])
            for parent in path.parents:
                m=parent.stat()
                if m.st_uid or m.st_mode&0o022:raise ValueError('mutable uplink executable ancestry')
            if path.lstat().st_uid:raise ValueError('uplink executable alias owner')
            resolved=path.resolve(strict=True)
            for parent in resolved.parents:
                m=parent.stat()
                if m.st_uid or m.st_mode&0o022:raise ValueError('mutable resolved executable ancestry')
            self.exe=os.open(resolved,os.O_RDONLY|os.O_NOFOLLOW|os.O_CLOEXEC)
            m=os.fstat(self.exe)
            if not stat.S_ISREG(m.st_mode) or m.st_uid or m.st_mode&0o022 or m.st_mode&0o6000:
                raise ValueError('mutable uplink executable')
            with os.fdopen(os.dup(self.exe),'rb') as stream:
                if hashlib.file_digest(stream,'sha256').hexdigest()!=value['sha256']:raise ValueError('uplink executable digest')
            self.exe_identity=(m.st_dev,m.st_ino,m.st_size,m.st_mtime_ns,m.st_ctime_ns)
            ROOT.mkdir(mode=0o700,exist_ok=True)
            # The packaged pasta AppArmor profile permits /tmp namespace
            # locators, not arbitrary /run paths. PrivateTmp plus root-only
            # ancestry preserves custody without relaxing the profile.
            NAMESPACE_ROOT.mkdir(mode=0o700,exist_ok=True)
            for directory in (ROOT,NAMESPACE_ROOT):
                meta=directory.lstat()
                if not stat.S_ISDIR(meta.st_mode) or meta.st_uid or meta.st_mode&0o077:raise ValueError('unprotected network-owner directory')
            self.target.touch(mode=0o400,exist_ok=False)
            self.created.append(self.target)
            fd=policy.fds[realm]
            if policy.value['namespaces'][realm]==policy.value['namespaces']['host']:
                raise ValueError('runtime uplink cannot target host namespace')
            if LIBC.mount(f'/proc/self/fd/{fd}'.encode(),str(self.target).encode(),None,4096,None):
                raise OSError(ctypes.get_errno(),'pin runtime uplink namespace')
            self.mounted=True
            if (self.target.stat().st_dev,self.target.stat().st_ino)!=(os.fstat(fd).st_dev,os.fstat(fd).st_ino):
                raise ValueError('runtime namespace bind differs')
            self.dns_bytes=(''.join('nameserver '+a+'\n' for a in value['dns'])+'options timeout:2 attempts:2\n').encode()
            with self.resolver.open('xb') as stream:
                self.created.append(self.resolver);stream.write(self.dns_bytes)
            self.resolver.chmod(0o444)
            parent=os.getpid()
            def child_setup():
                # Kill on owner death, including the fork/prctl race. Popen closes
                # every unrelated descriptor: no lock, control socket or pipe leak.
                if LIBC.prctl(1,signal.SIGKILL,0,0,0) or os.getppid()!=parent:os._exit(125)
            argv=arguments(value,self.target)
            self.child=subprocess.Popen(argv,executable=f'/proc/self/fd/{self.exe}',stdin=subprocess.DEVNULL,close_fds=True,pass_fds=(self.exe,),
                preexec_fn=child_setup,env={'PATH':'/usr/bin:/bin','LANG':'C'})
            self.pidfd=os.pidfd_open(self.child.pid)
            deadline=time.monotonic()+10
            while True:
                self.alive()
                try:
                    self.runtime_ready(check_addresses=False);break
                except (ValueError,subprocess.SubprocessError):
                    if time.monotonic()>=deadline:raise
                    time.sleep(.1)
            if realm=='runtime' and value.get('runtime_addresses'):
                c01_network.namespace_operation(policy.fds[realm],'runtime-addresses',value['runtime_addresses'])
                self.runtime_ready()
        except BaseException:self.close();raise

    def alive(self):
        if self.child.poll() is not None or select.select([self.pidfd],[],[],0)[0]:
            raise ValueError('owned runtime egress exited')
        actual=os.stat(f'/proc/{self.child.pid}/exe')
        if (actual.st_dev,actual.st_ino)!=self.exe_identity[:2]:
            raise ValueError('egress executable changed; appoint final ELF, not an ISA dispatcher')

    def runtime_ready(self,check_addresses=True):
        actual=c01_network.namespace_operation(self.policy.fds[self.realm],'uplink')
        if actual!=('c01-uplink',*self.value['wan'][1:]):raise ValueError('runtime egress route/source differs')
        if check_addresses and self.realm=='runtime' and self.value.get('runtime_addresses'):
            c01_network.namespace_operation(self.policy.fds[self.realm],'runtime-addresses-check',self.value['runtime_addresses'])

    def check(self):
        self.alive()
        if time.monotonic()<self.next_check:return
        self.next_check=time.monotonic()+.5
        check_resolver(self.value)
        if tuple(self.value['wan'])!=runtime_uplink.ready():raise ValueError('appointed WAN changed')
        m=os.stat(self.value['executable'])
        if (m.st_dev,m.st_ino,m.st_size,m.st_mtime_ns,m.st_ctime_ns)!=self.exe_identity:
            raise ValueError('uplink executable replaced')
        if self.resolver.read_bytes()!=self.dns_bytes:raise ValueError('uplink resolver replaced')
        self.runtime_ready()

    def close(self):
        if self.child is not None:
            if self.child.poll() is None:
                self.child.terminate()
                try:self.child.wait(timeout=3)
                except subprocess.TimeoutExpired:self.child.kill();self.child.wait()
            else:self.child.wait()
        if self.pidfd is not None:os.close(self.pidfd);self.pidfd=None
        if self.mounted:
            if LIBC.umount2(str(self.target).encode(),0):raise OSError(ctypes.get_errno(),'runtime namespace unpin')
            self.mounted=False
        for path in self.created:
            if path.exists():path.unlink()
        self.created=[]
        if self.exe is not None:os.close(self.exe);self.exe=None


class Uplinks:
    """One lifetime for both appointed private realms; never target host."""
    def __init__(self,policy,value):
        self.children=[]
        try:
            for realm in ('runtime','managed'):
                self.children.append(Uplink(policy,value,realm))
        except BaseException:self.close();raise

    def check(self):
        for child in self.children:child.check()

    def close(self):
        for child in reversed(self.children):child.close()
        self.children=[]