"""Kernel namespace binding shared by C01 lifetime owners.

Named namespaces are externally provisioned capabilities, never created from a
request. A name or configuration bit alone is not evidence: open nsfs, verify
its type and compare the actual process namespace with the appointed inode.
Routing/port forwarding remains a separately reviewed deployment requirement.
"""
import fcntl
import os
from pathlib import Path
import stat
import array
import ctypes
import json
import signal
import socket
import struct

NAMES = {'managed': Path('/run/netns/archipelago-managed'),
         'runtime': Path('/run/netns/archipelago-runtime')}
CLONE_NEWNET = 0x40000000
NS_GET_NSTYPE = 0xb703
NS_GET_USERNS = 0xb701


def identity(fd):
    if fcntl.ioctl(fd, NS_GET_NSTYPE) != CLONE_NEWNET:
        raise ValueError('appointed object is not a network namespace')
    meta = os.fstat(fd)
    return meta.st_dev, meta.st_ino


def appointed(role):
    path = NAMES[role]
    for parent in reversed(path.parents):
        meta = parent.lstat()
        if not stat.S_ISDIR(meta.st_mode) or meta.st_uid != 0 or meta.st_mode & 0o022:
            raise ValueError('mutable network namespace ancestry')
    fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC)
    try:
        meta = os.fstat(fd)
        if not stat.S_ISREG(meta.st_mode) or meta.st_mode & 0o022:
            raise ValueError('mutable network namespace capability')
        result = identity(fd)
        # nsfs inodes are owned by initial-namespace root, which can appear as
        # overflow UID inside a disposable user namespace. Authenticate kernel
        # ownership using NS_GET_USERNS, not an invented UID translation.
        userns = fcntl.ioctl(fd, NS_GET_USERNS)
        try:
            actual, expected = os.fstat(userns), os.stat('/proc/1/ns/user')
            if (actual.st_dev, actual.st_ino) != (expected.st_dev, expected.st_ino):
                raise ValueError('network namespace belongs to an unapproved user namespace')
        finally:
            os.close(userns)
        return result
    finally:
        os.close(fd)


def process_namespace(pid):
    fd = os.open(f'/proc/{pid}/ns/net', os.O_RDONLY | os.O_CLOEXEC)
    try:
        return identity(fd)
    finally:
        os.close(fd)


def require_private(pid, role):
    expected = appointed(role)
    if expected == process_namespace(1):
        raise ValueError('host network/abstract IPC namespace cannot be admitted')
    if expected != process_namespace(pid):
        raise ValueError('process differs from appointed network namespace')
    return expected


def enter_host_worker(libc):
    """Place only the forked root host worker in the actual PID1 network realm.

    The destination is supervisor-selected, never an application-supplied name.
    Existing connected control descriptors survive setns; new host-tool sockets
    must reach host services rather than the managed application's loopback.
    """
    if os.geteuid() != 0:
        raise ValueError('host worker namespace transition requires root')
    require_private(os.getpid(), 'managed')
    fd = os.open('/proc/1/ns/net', os.O_RDONLY | os.O_CLOEXEC)
    try:
        expected = identity(fd)
        if libc.setns(fd, CLONE_NEWNET) != 0:
            raise ValueError('root host-operation network namespace unavailable')
        if process_namespace(os.getpid()) != expected:
            raise ValueError('host worker network namespace transition mismatch')
    finally:
        os.close(fd)


def namespace_operation(fd, operation, addresses=None):
    """Fixed private helper operations; the retained owner never changes netns.

    Only namespace/socket capabilities enter this function, never RPC paths or
    argv. The helper inherits no leases/locks/keys and is reaped before return.
    """
    if os.geteuid()!=0 or operation not in ('tcp','udp','tcp6','udp6','uplink','mdns4','mdns6','runtime-addresses','runtime-addresses-check','host-addresses'):
        raise ValueError('fixed root namespace operation required')
    identity(fd)
    if operation.startswith('runtime-addresses'):
        import ipaddress
        if identity(fd)==process_namespace(1) or type(addresses) is not list or not 1<=len(addresses)<=8:raise ValueError('private runtime address operation only')
        for address in addresses:
            ip=ipaddress.IPv4Address(address)
            if str(ip)!=address or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_unspecified:raise ValueError('runtime address scope')
    elif addresses is not None:raise ValueError('unexpected namespace operation arguments')
    parent,child=socket.socketpair(socket.AF_UNIX,socket.SOCK_SEQPACKET|socket.SOCK_CLOEXEC)
    parent.setsockopt(socket.SOL_SOCKET,socket.SO_PASSCRED,1);parent.settimeout(3)
    pid=None;pin=None;reaped=False;rights=[];expected_parent=os.getpid()
    try:
        pid=os.fork()
        if pid==0:
            try:
                parent.close();keep={fd,child.fileno()}
                for entry in os.listdir('/proc/self/fd'):
                    number=int(entry)
                    if number>2 and number not in keep:
                        try:os.close(number)
                        except OSError:pass
                libc=ctypes.CDLL(None,use_errno=True)
                if libc.prctl(1,signal.SIGKILL,0,0,0) or os.getppid()!=expected_parent:os._exit(125)
                if libc.setns(fd,CLONE_NEWNET):raise OSError(ctypes.get_errno(),'owned namespace entry')
                if operation=='host-addresses':
                    import subprocess
                    raw=subprocess.check_output(['/usr/bin/ip','-j','address','show'],timeout=1)
                    if len(raw)>65536:raise ValueError('host interface observation bound')
                    links=json.loads(raw)
                    values=sorted({a['local'] for link in links for a in link.get('addr_info',[]) if a.get('family') in ('inet','inet6') and not a.get('tentative') and not a.get('dadfailed')})
                    if len(values)>64:raise ValueError('host address count bound')
                    fips=[a['local'] for link in links if link['ifname']=='fips0' for a in link.get('addr_info',[]) if a.get('family')=='inet6' and not a.get('tentative') and not a.get('dadfailed')]
                    child.send(json.dumps({'addresses':values,'interfaces':{link['ifname']:link['ifindex'] for link in links},'fips_addresses':fips}).encode())
                elif operation.startswith('runtime-addresses'):
                    import subprocess
                    def current():
                        raw=subprocess.check_output(['/usr/bin/ip','-j','-4','address','show','dev','c01-uplink'],timeout=1)
                        if len(raw)>16384:raise ValueError('runtime interface observation bound')
                        return {a['local'] for link in json.loads(raw) for a in link.get('addr_info',[]) if a.get('family')=='inet'}
                    existing=current()
                    if operation=='runtime-addresses':
                        for address in addresses:
                            if address not in existing:
                                subprocess.run(['/usr/bin/ip','address','add',address+'/32','dev','c01-uplink'],check=True,timeout=1,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)
                    if not set(addresses)<=current():raise ValueError('appointed private runtime addresses absent')
                    child.send(b'ADDRESSES')
                elif operation=='uplink':
                    import runtime_uplink
                    child.send(json.dumps(runtime_uplink.ready()).encode())
                elif operation.startswith('mdns'):
                    import runtime_uplink,c01_network_mdns
                    family=socket.AF_INET6 if operation=='mdns6' else socket.AF_INET
                    with c01_network_mdns.make_socket(family,'c01-uplink',runtime_uplink.ready()[2],private=True) as value:
                        child.sendmsg([b'SOCKET'],[(socket.SOL_SOCKET,socket.SCM_RIGHTS,array.array('i',[value.fileno()]))])
                else:
                    kind=socket.SOCK_STREAM if operation.startswith('tcp') else socket.SOCK_DGRAM
                    with socket.socket(socket.AF_INET6 if operation.endswith('6') else socket.AF_INET,kind|socket.SOCK_CLOEXEC) as value:
                        child.sendmsg([b'SOCKET'],[(socket.SOL_SOCKET,socket.SCM_RIGHTS,array.array('i',[value.fileno()]))])
                os._exit(0)
            except BaseException:os._exit(111)
        child.close();pin=os.pidfd_open(pid)
        raw,anc,flags,_=parent.recvmsg(4097,socket.CMSG_SPACE(12)+socket.CMSG_SPACE(16),socket.MSG_CMSG_CLOEXEC)
        credentials=[];unexpected=False
        for level,kind,data in anc:
            if (level,kind)==(socket.SOL_SOCKET,socket.SCM_RIGHTS):
                values=array.array('i');values.frombytes(data[:len(data)-len(data)%values.itemsize]);rights.extend(values)
            elif (level,kind)==(socket.SOL_SOCKET,socket.SCM_CREDENTIALS):credentials.append(data)
            else:unexpected=True
        if unexpected or flags&~socket.MSG_CMSG_CLOEXEC or not 0<len(raw)<=4096 or credentials!=[struct.pack('3i',pid,0,0)]:
            raise ValueError('namespace helper credentials/framing')
        _,status=os.waitpid(pid,0);reaped=True
        if os.waitstatus_to_exitcode(status):raise ValueError('namespace helper failed')
        if operation.startswith('runtime-addresses'):
            if rights or raw!=b'ADDRESSES':raise ValueError('private address acknowledgement')
            return None
        if operation=='host-addresses':
            value=json.loads(raw)
            if rights or type(value) is not dict or set(value)!={'addresses','interfaces','fips_addresses'} or type(value['addresses']) is not list or len(value['addresses'])>64 or any(type(address) is not str for address in value['addresses']) or type(value['interfaces']) is not dict or len(value['interfaces'])>64 or type(value['fips_addresses']) is not list or len(value['fips_addresses'])>64:raise ValueError('host address observation schema')
            return value
        if operation=='uplink':
            if rights:raise ValueError('unexpected uplink rights')
            value=json.loads(raw)
            if type(value) is not list or len(value)!=3 or not all(type(part) is str for part in value):raise ValueError('uplink observation schema')
            return tuple(value)
        if raw!=b'SOCKET' or len(rights)!=1:raise ValueError('namespace socket capability absent')
        result=socket.socket(fileno=rights.pop());result.set_inheritable(False);return result
    finally:
        if pid and not reaped:
            try:
                if pin is not None:signal.pidfd_send_signal(pin,signal.SIGKILL)
                else:os.kill(pid,signal.SIGKILL) # unreaped direct child cannot be reused
            except ProcessLookupError:pass
            os.waitpid(pid,0)
        for received in rights:os.close(received)
        if pin is not None:os.close(pin)
        parent.close();child.close()