"""Finite keeper + exact retained-container lifetime for runtime socket owners.

No Podman mutation, container-name lookup or application-supplied commands/paths.
The only D-Bus use is authenticated reads of appointed user-unit properties.
"""
import hashlib,importlib.util,json,os,re,select,socket,sqlite3,stat,struct,subprocess,sys,threading,time
from pathlib import Path

PREFIX='/user.slice/user-1000.slice/user@1000.service'
DATABASE='file:/var/lib/archipelago/containers/storage/db.sql?mode=ro'
NODE_LABEL=r'[a-z0-9](?:[a-z0-9-]{0,62}[a-z0-9])?'
ROOTLESS_FORWARDER=Path('/usr/libexec/podman/rootlessport')

def require(value,reason):
    if not value:raise ValueError(reason)

def same_process(before,after,thread_net=False):
    # Go's namespace helpers can move the thread represented by /proc/PID/ns/net
    # without changing the process or any listening socket's creation namespace.
    if thread_net:
        before={k:v for k,v in before.items() if k!='net'}
        after={k:v for k,v in after.items() if k!='net'}
    return before==after

def validate(value,group):
    require(type(value) is dict and set(value)=={'kind','unit','container_id','image_id','config_sha256','scope_kind','payload_principal'},'retained producer fields')
    require(value['kind']=='retained-container-v1','retained producer kind')
    for key in ('container_id','image_id','config_sha256'):require(type(value[key]) is str and re.fullmatch('[0-9a-f]{64}',value[key]),'retained identity digest')
    require(re.fullmatch(r'archipelago-retained-'+NODE_LABEL+'-'+value['container_id']+r'\.service',value['unit']),'retained unit identity')
    require(group==PREFIX+'/app.slice/'+value['unit'],'retained stable service cgroup')
    require(value['scope_kind'] in ('split-payload','bound-libpod-scope'),'retained cgroup mode')
    principal=value['payload_principal']
    require(type(principal) is list and len(principal)==2 and all(type(n) is int and 0<n<2**32-1 for n in principal),'measured payload principal')

def payload_group(value,group):
    return (group+'/libpod-payload-'+value['container_id'] if value['scope_kind']=='split-payload'
            else PREFIX+'/user.slice/libpod-'+value['container_id']+'.scope/container')

def load_gio():
    if 'gi' not in sys.modules:
        source=Path('/usr/lib/python3/dist-packages/gi/__init__.py')
        for path in (source,*source.parents):
            m=path.lstat();require(m.st_uid==0 and not m.st_mode&0o022 and not stat.S_ISLNK(m.st_mode),'protected GI module')
        spec=importlib.util.spec_from_file_location('gi',source,submodule_search_locations=[str(source.parent)])
        module=importlib.util.module_from_spec(spec);sys.modules['gi']=module;spec.loader.exec_module(module)
    from gi.repository import Gio,GLib
    return Gio,GLib

class Manager:
    def __init__(self):
        self.Gio,self.GLib=load_gio();self.connection=None;self.pidfd=None
        try:
            cancel=self.Gio.Cancellable();timer=threading.Timer(1,cancel.cancel);timer.start()
            try:self.connection=self.Gio.DBusConnection.new_for_address_sync('unix:path=/run/user/1000/systemd/private',self.Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT,None,cancel)
            finally:timer.cancel();timer.join()
            with socket.fromfd(self.connection.get_stream().get_socket().get_fd(),socket.AF_UNIX,socket.SOCK_STREAM) as channel:
                pid,uid,gid=struct.unpack('3i',channel.getsockopt(socket.SOL_SOCKET,socket.SO_PEERCRED,12))
                self.pidfd=channel.getsockopt(socket.SOL_SOCKET,77);os.set_inheritable(self.pidfd,False)
                self.pid=pid
            result=subprocess.run(['/usr/bin/systemctl','show','user@1000.service','--property=MainPID','--value'],capture_output=True,text=True,timeout=1,check=True)
            require((pid,uid,gid)==(int(result.stdout.strip()),1000,1000),'user-manager peer identity')
        except Exception:
            self.close();raise ValueError('authenticated user-manager observation unavailable') from None
    def unit(self,name):
        try:return self.read_unit(name)
        except Exception:raise ValueError('appointed user-unit observation unavailable') from None
    def read_unit(self,name):
        require(not select.select([self.pidfd],[],[],0)[0],'user-manager lifetime lost')
        def call(path,interface,method,parameters):
            return self.connection.call_sync(None,path,interface,method,parameters,None,self.Gio.DBusCallFlags.NONE,500,None).unpack()[0]
        path=call('/org/freedesktop/systemd1','org.freedesktop.systemd1.Manager','GetUnit',self.GLib.Variant('(s)',(name,)))
        unit=call(path,'org.freedesktop.DBus.Properties','GetAll',self.GLib.Variant('(s)',('org.freedesktop.systemd1.Unit',)))
        service=call(path,'org.freedesktop.DBus.Properties','GetAll',self.GLib.Variant('(s)',('org.freedesktop.systemd1.Service',)))
        invocation=unit['InvocationID']
        if not isinstance(invocation,str):invocation=bytes(invocation).hex()
        return {'Id':unit['Id'],'MainPID':service['MainPID'],'ControlGroup':service['ControlGroup'],
                'ActiveState':unit['ActiveState'],'InvocationID':invocation}
    def close(self):
        if self.pidfd is not None:os.close(self.pidfd);self.pidfd=None
        if self.connection is not None:self.connection.close_sync(None);self.connection=None

def stored(value):
    try:
        db=sqlite3.connect(DATABASE,uri=True,timeout=.25)
        try:
            db.execute('PRAGMA query_only=ON');db.execute('BEGIN')
            c=db.execute('SELECT JSON FROM ContainerConfig WHERE ID=?',(value['container_id'],)).fetchone()
            s=db.execute('SELECT JSON FROM ContainerState WHERE ID=?',(value['container_id'],)).fetchone()
            require(c is not None and s is not None,'retained container missing')
            config,state=json.loads(c[0]),json.loads(s[0])
            digest=hashlib.sha256(json.dumps(config,sort_keys=True,separators=(',',':'),ensure_ascii=False).encode()).hexdigest()
            require(digest==value['config_sha256'] and config.get('rootfsImageID')==value['image_id'],'retained complete config/image changed')
            require(state.get('state')==3 and type(state.get('pid')) is int and state['pid']>1,'retained container not running')
            if value['scope_kind']=='bound-libpod-scope':require(config.get('cgroupsMode')=='enabled' and config.get('cgroupParent')=='user.slice','retained external scope changed')
            else:require(config.get('cgroupsMode')=='split' and not config.get('cgroupParent'),'retained split scope changed')
            return state['pid']
        finally:db.close()
    except sqlite3.Error:raise ValueError('retained runtime metadata unavailable') from None

class Guard:
    def __init__(self,policy,route,actor,process):
        self.fds=[];self.processes=[];self.process=process;self.policy=policy;self.next_check=0
        self.value=policy.manifests[route['projection_sha256']]['producer'];self.group=route['target_cgroup']
        validate(self.value,self.group)
        try:
            if getattr(policy,'retained_manager',None) is None:policy.retained_manager=Manager()
            self.manager=policy.retained_manager
            manager=self.capture(self.manager.pid)
            require(manager['group']==PREFIX+'/init.scope' and manager['uids']==[1000]*4 and manager['gids']==[1000]*4 and list(manager['net'])==policy.value['namespaces']['runtime'],'user manager runtime principal/namespace')
            self.props=self.manager.unit(self.value['unit'])
            require(self.props['Id']==self.value['unit'] and self.props['ControlGroup']==self.group and self.props['ActiveState']=='active' and re.fullmatch('[0-9a-f]{32}',self.props['InvocationID']),'retained keeper unit generation')
            keeper=self.capture(self.props['MainPID'],thread_net=True)
            require(keeper['group']==self.group or keeper['group'].startswith(self.group+'/'),'keeper outside appointed unit')
            require(keeper['uids']==[1000]*4 and keeper['gids']==[1000]*4,'keeper runtime principal')
            self.keeper_argv=[b'/usr/bin/podman',b'start',b'--attach',self.value['container_id'].encode(),b'']
            require(Path(f'/proc/{keeper["pid"]}/cmdline').read_bytes().split(b'\0')==self.keeper_argv,'keeper command differs')
            exe=self.hold(os.open(f'/proc/{keeper["pid"]}/exe',os.O_RDONLY|os.O_CLOEXEC));a=os.fstat(exe);b=Path('/usr/bin/podman').stat()
            require((a.st_dev,a.st_ino)==(b.st_dev,b.st_ino) and b.st_uid==0 and not b.st_mode&0o022,'keeper executable identity')
            self.executable=(a.st_dev,a.st_ino)
            self.payload_pid=stored(self.value);payload=self.capture(self.payload_pid)
            self.payload_net=payload['net']
            self.payload_group=payload_group(self.value,self.group)
            require(payload['group']==self.payload_group and payload['uids']==[self.value['payload_principal'][0]]*4 and payload['gids']==[self.value['payload_principal'][1]]*4,'retained payload identity')
            self.owner_thread_net=actor['group']!=self.payload_group
            # Non-payload publishers are admitted only as the pinned system
            # rootlessport executable under this exact keeper unit.
            if self.owner_thread_net:self.verify_forwarder(actor)
            owner=self.capture(actor['pid'],thread_net=self.owner_thread_net)
            require(same_process(owner,actor,self.owner_thread_net),'socket process changed')
            self.assert_owner(owner)
            self.binding={'unit':self.props,'container_id':self.value['container_id'],'config_sha256':self.value['config_sha256'],
                          'processes':self.processes,'namespaces':[{'pid':n['pid'],'values':n['values']} for n in self.namespace_bindings]}
        except BaseException:self.close();raise
    def hold(self,fd):self.fds.append(fd);return fd
    def capture(self,pid,thread_net=False):
        require(type(pid) is int and pid>1,'retained process ID')
        before=self.process(pid);pin=self.hold(os.pidfd_open(pid))
        if not hasattr(self,'namespace_bindings'):self.namespace_bindings=[]
        ns={}
        for name in (('user','mnt') if thread_net else ('net','user','mnt')):
            fd=self.hold(os.open(f'/proc/{pid}/ns/{name}',os.O_RDONLY|os.O_CLOEXEC));m=os.fstat(fd);ns[name]=(m.st_dev,m.st_ino)
        require(same_process(before,self.process(pid),thread_net) and not select.select([pin],[],[],0)[0],'retained process changed while pinning')
        self.processes.append(before);self.namespace_bindings.append({'pid':pid,'values':ns,'pidfd':pin,'thread_net':thread_net})
        return before
    def assert_owner(self,actor):
        payload=actor['group']==self.payload_group
        require(payload or actor['group']==self.group or actor['group'].startswith(self.group+'/'),'unowned retained socket scope')
        principal=self.value['payload_principal'] if payload else [1000,1000]
        require(actor['uids']==[principal[0]]*4 and actor['gids']==[principal[1]]*4,'retained socket principal/namespace')
        if payload:
            require(list(actor['net'])==self.policy.value['namespaces']['runtime'],'retained socket principal/namespace')
        else:
            require(getattr(self,'forwarder',None) is not None and self.forwarder[0]==actor['pid'],'unverified retained forwarder')
    def verify_forwarder(self,actor):
        for path in ROOTLESS_FORWARDER.parents:
            m=path.lstat();require(stat.S_ISDIR(m.st_mode) and m.st_uid==0 and not m.st_mode&0o022,'protected rootless forwarder ancestry')
        expected=self.hold(os.open(ROOTLESS_FORWARDER,os.O_RDONLY|os.O_CLOEXEC|os.O_NOFOLLOW))
        actual=self.hold(os.open(f'/proc/{actor["pid"]}/exe',os.O_RDONLY|os.O_CLOEXEC))
        a=os.fstat(actual);b=os.fstat(expected)
        require(stat.S_ISREG(b.st_mode) and b.st_uid==0 and not b.st_mode&0o022 and (a.st_dev,a.st_ino)==(b.st_dev,b.st_ino),'retained cross-namespace owner is not the protected rootless forwarder')
        self.forwarder=(actor['pid'],b.st_dev,b.st_ino)
    def check(self,force=False):
        for before,ns in zip(self.processes,self.namespace_bindings):
            require(not select.select([ns['pidfd']],[],[],0)[0] and same_process(before,self.process(before['pid']),ns.get('thread_net',False)),'retained process lifetime lost')
            for name,identity in ns['values'].items():
                m=os.stat(f'/proc/{before["pid"]}/ns/{name}');require((m.st_dev,m.st_ino)==identity,'retained namespace changed')
        forwarder=getattr(self,'forwarder',None)
        if forwarder is not None:
            pid,dev,ino=forwarder;a=os.stat(f'/proc/{pid}/exe');b=ROOTLESS_FORWARDER.lstat()
            require((a.st_dev,a.st_ino)==(b.st_dev,b.st_ino)==(dev,ino) and stat.S_ISREG(b.st_mode) and b.st_uid==0 and not b.st_mode&0o022,'retained rootless forwarder executable changed')
        if force or time.monotonic()>=self.next_check:
            require(self.manager.unit(self.value['unit'])==self.props and stored(self.value)==self.payload_pid,'retained unit/config generation changed')
            m=Path('/usr/bin/podman').stat();actual=Path(f'/proc/{self.props["MainPID"]}/exe').stat()
            require((actual.st_dev,actual.st_ino)==(m.st_dev,m.st_ino)==self.executable and m.st_uid==0 and not m.st_mode&0o022,'keeper executable replaced')
            require(Path(f'/proc/{self.props["MainPID"]}/cmdline').read_bytes().split(b'\0')==self.keeper_argv,'keeper command changed')
            self.next_check=time.monotonic()+.5
    def close(self):
        for fd in reversed(self.fds):os.close(fd)
        self.fds=[]