"""Live root supervisor lease; pin source listeners through complete retirement."""
import array
import ctypes
import errno
import json
import os
import select
import socket
import struct
import time
from pathlib import Path

from c01_network_owner import CONFIG,CONTROL,Policy,canonical,exact,protected_read,digest,process,require_main,LIBC,listener_address_matches
from c01_endpoints import source_addresses


def receive_listeners(channel,request,total,version,listeners):
    if not 1<=total<=(512 if version==4 else 64):raise ValueError('listener lease bound')
    for offset in range(0,total,64):
        raw,anc,flags,_=channel.recvmsg(8193,socket.CMSG_SPACE(64*4),socket.MSG_CMSG_CLOEXEC)
        unexpected=False;before=len(listeners)
        for level,kind,data in anc:
            if (level,kind)==(socket.SOL_SOCKET,socket.SCM_RIGHTS):
                fds=array.array('i');fds.frombytes(data[:len(data)-len(data)%4]);listeners.extend(fds)
                unexpected |= len(data)%4!=0
            else:unexpected=True
        expected={**request,'offset':offset,'total':total} if version==4 else request
        if unexpected or flags & ~socket.MSG_CMSG_CLOEXEC or len(raw)>8192 or json.loads(raw,object_pairs_hook=exact)!=expected or len(listeners)-before!=min(64,total-offset):
            raise ValueError('network lease binding/rights')


class NetworkLease:
    def __init__(self, manager_uid=None, runtime_uid=None):
        self.channel=socket.socket(socket.AF_UNIX,socket.SOCK_SEQPACKET|socket.SOCK_CLOEXEC)
        self.pidfd=None;self.listeners=[];self.policy=None;self.management=None;self.required_targets={}
        self.required_sockets={}
        self.required_generations={};self.required_since={}
        try:
            self.policy=Policy();self.policy_digest=self.policy.sha
            policy=self.policy.value
            self.protocol=policy['version']
            if manager_uid is not None:
                route=self.policy.routes.get('management')
                if route is None or route['target_uid']!=manager_uid:raise ValueError('network manager appointment')
            # Policy has already validated each route's explicit producer
            # principal. The runtime account need not equal a container's
            # subordinate UID; bind the account appointment itself here.
            # Management-only legacy policies have no runtime authority to bind.
            if runtime_uid is not None:
                if self.policy.runtime is None:
                    if any(r['target']=='runtime' for r in policy['routes']):raise ValueError('network runtime appointment')
                elif self.policy.runtime['runtime_uid']!=runtime_uid:raise ValueError('network runtime appointment')
            self.channel.settimeout(3);self.channel.connect(str(CONTROL))
            pid,uid,gid=struct.unpack('3i',self.channel.getsockopt(socket.SOL_SOCKET,socket.SO_PEERCRED,12))
            if (uid,gid)!=(0,0) or process(pid)['group']!='/system.slice/archipelago-c01-network.service':raise PermissionError('network owner identity')
            self.owner= require_main(pid,'archipelago-c01-network.service','/system.slice/archipelago-c01-network.service')
            self.pidfd=self.channel.getsockopt(socket.SOL_SOCKET,77);os.set_inheritable(self.pidfd,False)
            request={'method':'lease','policy_sha256':self.policy_digest,'nonce':os.urandom(32).hex()}
            self.expiry=0;self.start_deadline=time.clock_gettime_ns(time.CLOCK_BOOTTIME)+3_000_000_000
            self.channel.sendall(canonical(request))
            routes=[(route,address) for route in policy['routes'] for address in source_addresses(route)]
            receive_listeners(self.channel,request,len(routes),policy['version'],self.listeners)
            for fd,(route,address) in zip(self.listeners,routes):
                with socket.socket(fileno=os.dup(fd)) as listener:
                    kind=socket.SOCK_STREAM if route['protocol']=='tcp' else socket.SOCK_DGRAM
                    if listener.getsockopt(socket.SOL_SOCKET,socket.SO_TYPE)!=kind or listener.getsockname()[:2]!=(address,route['port']):raise ValueError('network listener capability')
                    with self.policy.socket(route['source'],kind,family=listener.family) as probe:
                        if listener.getsockopt(socket.SOL_SOCKET,71,8)!=probe.getsockopt(socket.SOL_SOCKET,71,8):raise ValueError('network source namespace')
                    if kind==socket.SOCK_STREAM and not listener.getsockopt(socket.SOL_SOCKET,socket.SO_ACCEPTCONN):raise ValueError('network ingress not listening')
            self.channel.setblocking(False);self.check()
        except BaseException:self.close();raise
    def check(self):
        if self.pidfd is None or select.select([self.pidfd],[],[],0)[0]:raise ValueError('network capability lifetime lost')
        for _ in range(16):
            if not select.select([self.channel],[],[],0)[0]:break
            bound=65536 if getattr(self,'protocol',1)==4 else 256
            raw,anc,flags,_=self.channel.recvmsg(bound+1,0)
            if anc or flags or not 0<len(raw)<=bound:raise ValueError('network renewal framing/lifetime lost')
            value=json.loads(raw,object_pairs_hook=exact)
            fields={'alive_until_boottime_ns'}|({'generations'} if getattr(self,'protocol',1)==4 else set())
            if set(value)!=fields:raise ValueError('network renewal schema')
            expiry=value['alive_until_boottime_ns']
            if type(expiry) is not int or expiry<self.expiry:raise ValueError('network renewal regressed')
            self.expiry=expiry
            if getattr(self,'protocol',1)==4:
                generations=value['generations']
                if type(generations) is not dict or set(generations)!=set(self.policy.routes):raise ValueError('network generation inventory')
                for name,expected in self.required_generations.items():
                    if expiry-3_000_000_000>=self.required_since[name] and generations[name]!=expected:
                        raise ValueError('required producer generation lost: '+name)
        stamp=time.clock_gettime_ns(time.CLOCK_BOOTTIME)
        if self.expiry:
            if not stamp<self.expiry<=stamp+3_000_000_000:raise ValueError('network owner progress expired')
        elif stamp>=self.start_deadline:raise ValueError('network initial progress absent')
        if process(self.owner['pid'])!=self.owner:raise ValueError('network owner identity changed')
        self.policy.check()
        for name,(pidfd,expected) in self.required_targets.items():
            if select.select([pidfd],[],[],0)[0] or canonical(process(expected['pid']))!=canonical(expected):
                raise ValueError('required network producer lifetime lost: '+name)
            if name in self.required_sockets:
                number,inode=self.required_sockets[name]
                if os.readlink(f'/proc/{expected["pid"]}/fd/{number}')!=f'socket:[{inode}]':
                    raise ValueError('required network producer descriptor lost: '+name)
    def require_targets(self,names):
        for name in names:
            if name not in self.policy.routes:raise ValueError('required named route absent')
            if name in self.required_targets:continue
            status=self.request({'method':'inspect','name':name})
            expected_fields={'policy_sha256','name','generation','producer'}|({'listener'} if self.policy.value['version']==4 else set())
            if set(status)!=expected_fields or status['name']!=name:
                raise ValueError('required named producer reply binding')
            producer=status.get('producer')
            if type(producer) is not dict or type(producer.get('pid')) is not int:raise ValueError('required named producer absent')
            fd=os.pidfd_open(producer['pid'])
            try:
                if select.select([fd],[],[],0)[0] or canonical(process(producer['pid']))!=canonical(producer):
                    raise ValueError('required named producer changed')
                self.required_targets[name]=(fd,producer)
                if self.policy.value['version']==4:
                    listener=status['listener']
                    if type(listener) is not dict or set(listener)!={'number','inode'} or type(listener['number']) is not int or not 0<=listener['number']<=65535 or type(listener['inode']) is not int or listener['inode']<=0:
                        del self.required_targets[name]
                        raise ValueError('required producer descriptor binding')
                    self.required_sockets[name]=(listener['number'],listener['inode'])
                    if type(status['generation']) is not str or len(status['generation'])!=64:raise ValueError('required producer generation')
                    self.required_generations[name]=status['generation'];self.required_since[name]=time.clock_gettime_ns(time.CLOCK_BOOTTIME)
            except BaseException:os.close(fd);raise
        self.check()
    def request(self, body):
        self.check()
        with socket.socket(socket.AF_UNIX,socket.SOCK_SEQPACKET|socket.SOCK_CLOEXEC) as channel:
            channel.settimeout(3);channel.connect(str(CONTROL))
            if struct.unpack('3i',channel.getsockopt(socket.SOL_SOCKET,socket.SO_PEERCRED,12))!=(self.owner['pid'],0,0):raise ValueError('network owner replaced')
            channel.sendall(canonical({'policy_sha256':self.policy_digest,**body}))
            raw,anc,flags,_=channel.recvmsg(8193,0)
            if anc or flags or not 0<len(raw)<=8192:raise ValueError('network reply framing')
            reply=json.loads(raw,object_pairs_hook=exact)
            if reply.get('policy_sha256')!=self.policy_digest:raise ValueError('network reply policy')
        self.check()
        return reply
    def register_management(self, pid, fd):
        if self.management is not None:raise ValueError('management already registered')
        status=self.request({'method':'inspect','name':'management'})
        if status.get('producer') is not None:raise ValueError('previous management producer active')
        reply=self.request({'method':'register','name':'management','pid':pid,'fd':fd,'previous':status['generation']})
        if set(reply)!={'registered','policy_sha256','generation'} or reply['registered']!='management':raise ValueError('management registration reply')
        self.management=reply['generation']

    def register_managed_listeners(self,pid,names):
        """Discover only this native child's finite appointed listening sockets.

        The owner independently authenticates each producer and pins its socket.
        No connection is redirected to a runtime port when a gate is absent.
        """
        routes={name:self.policy.routes[name] for name in names
                if name!='management' and self.policy.routes[name]['target']=='managed'}
        pending={}
        for name,route in routes.items():
            status=self.request({'method':'inspect','name':name})
            if status.get('producer') is None:pending[name]=(route,status['generation'])
            elif status['producer']['pid']!=pid:raise ValueError('managed producer differs from native child')
        if not pending:return
        entries=list(Path(f'/proc/{pid}/fd').iterdir())
        if len(entries)>4096:raise ValueError('native listener inventory exceeds bound')
        pidfd=os.pidfd_open(pid);matches={name:{} for name in pending}
        try:
            for entry in entries:
                try:
                    if not os.readlink(entry).startswith('socket:['):continue
                except FileNotFoundError:continue
                fd=LIBC.syscall(438,pidfd,int(entry.name),0)
                if fd<0:
                    error=ctypes.get_errno()
                    if error==errno.EBADF:continue
                    raise OSError(error,'native socket inventory unavailable')
                with socket.socket(fileno=fd) as listener:
                    if listener.family not in (socket.AF_INET,socket.AF_INET6):continue
                    kind=listener.getsockopt(socket.SOL_SOCKET,socket.SO_TYPE)
                    if kind==socket.SOCK_STREAM and not listener.getsockopt(socket.SOL_SOCKET,socket.SO_ACCEPTCONN):continue
                    if kind not in (socket.SOCK_STREAM,socket.SOCK_DGRAM):continue
                    for name,(route,_) in pending.items():
                        if kind==(socket.SOCK_STREAM if route['protocol']=='tcp' else socket.SOCK_DGRAM) and listener_address_matches(listener,route):matches[name].setdefault(os.fstat(listener.fileno()).st_ino,int(entry.name))
            if select.select([pidfd],[],[],0)[0]:raise ValueError('native producer exited during inventory')
            for name,numbers in matches.items():
                if len(numbers)>1:raise ValueError('ambiguous native listener')
                if not numbers:continue
                reply=self.request({'method':'register','name':name,'pid':pid,'fd':next(iter(numbers.values())),'previous':pending[name][1]})
                if set(reply)!={'registered','policy_sha256','generation'} or reply['registered']!=name:
                    raise ValueError('native listener registration reply')
        finally:os.close(pidfd)
    def close(self):
        self.channel.close()
        for fd,_ in self.required_targets.values():os.close(fd)
        self.required_targets={}
        self.required_sockets={}
        self.required_generations={};self.required_since={}
        for fd in self.listeners:os.close(fd)
        self.listeners=[]
        if self.pidfd is not None:os.close(self.pidfd);self.pidfd=None
        if self.policy is not None:self.policy.close();self.policy=None