"""Kernel retirement and ongoing credential/view evidence for the access owner.

Only root-reviewed cgroup/inode inventories can be terminated. No kill-by-UID,
PID lookup followed by an unpinned signal, service discovery, or shell execution.
"""
import errno
import os
from pathlib import Path
import select
import stat
import time
import c01_network

from c01_access_policy import require, canonical, open_path, identity, protected, STATE, atomic

CGROUP = Path('/sys/fs/cgroup')


def now():
    return time.clock_gettime_ns(time.CLOCK_BOOTTIME)


def process(pid):
    root = Path('/proc') / str(pid)
    raw = (root / 'stat').read_text()
    start = int(raw[raw.rindex(')') + 2:].split()[19])
    status = dict(row.split(':', 1) for row in (root / 'status').read_text().splitlines() if ':' in row)
    group = (root / 'cgroup').read_text().splitlines()
    require(len(group) == 1 and group[0].startswith('0::/'), 'unified process hierarchy required')
    return {'pid': pid, 'ppid': int(status['PPid']), 'start': start, 'uids': list(map(int, status['Uid'].split())),
            'gids': list(map(int, status['Gid'].split())), 'groups': list(map(int, status['Groups'].split())),
            'caps': {key: int(status[key], 16) for key in ('CapEff', 'CapPrm', 'CapInh', 'CapAmb')},
            'nnp': int(status['NoNewPrivs']), 'group': str(CGROUP / group[0][3:].lstrip('/')),
            'namespaces': {kind: os.stat(root / 'ns' / kind).st_ino for kind in ('mnt', 'net', 'user')}}


def all_processes():
    for path in Path('/proc').iterdir():
        if path.name.isdecimal():
            try:
                yield process(int(path.name))
            except (FileNotFoundError, ProcessLookupError):
                continue


def members(path):
    result = set()
    directories = [path]
    while directories:
        directory = directories.pop()
        result.update(map(int, (directory / 'cgroup.procs').read_text().split()))
        require(len(result) <= 8192, 'retirement cohort task bound')
        directories.extend(p for p in directory.iterdir() if p.is_dir() and not p.is_symlink())
    return result


def enclosing_cohort(spec, runtime):
    """Only the original UID's exact slice may enclose a delegated old manager.

    Root owns the migration boundary; the reviewed PID/start inventory authorizes
    the entire slice, including any old sessions. Never infer a kill inventory.
    The coordinator must remove the emptied old slice before recreating it.
    """
    base = {'path', 'device', 'inode', 'members'}
    require(set(spec) in (base, base | {'replacement'}), 'cohort inventory schema')
    if 'replacement' not in spec:
        return False
    group = CGROUP / runtime['user_cgroup'].lstrip('/')
    require(spec['replacement'] == 'runtime-user-slice' and runtime['runtime_uid'] == 1000 and
            group == CGROUP / 'user.slice/user-1000.slice/user@1000.service' and
            canonical(spec['path']) == group.parent, 'unsupported enclosing cohort replacement')
    return True


def migration_boundary(path):
    protected(path, directory=True)
    # A delegated descendant cannot migrate out without write access to the
    # common ancestor's cgroup.procs. Pin every enclosing migration boundary.
    for directory in (path, *path.parents):
        if not directory.is_relative_to(CGROUP):
            break
        for name in ('cgroup.procs', 'cgroup.threads'):
            protected(directory / name)


def replacement_tree(spec, runtime):
    path = canonical(spec['path'])
    if not path.exists():
        return
    meta = path.lstat()
    require((meta.st_dev, meta.st_ino) != (spec['device'], spec['inode']),
            'legacy cgroup retirement not completed by original owner')
    group = CGROUP / runtime['user_cgroup'].lstrip('/')
    if enclosing_cohort(spec, runtime):
        migration_boundary(path)
        require(not (path / 'cgroup.procs').read_text().strip(), 'replacement slice has direct tasks')
        for child in path.iterdir():
            require(not child.is_symlink(), 'replacement slice alias')
            if child.is_dir():
                require(child == group, 'unapproved replacement slice subtree')
    else:
        require(path == group, 'unapproved legacy restart hierarchy reappeared')


class Cohort:
    def __init__(self, spec, runtime=None):
        enclosing = enclosing_cohort(spec, runtime or {})
        self.path = canonical(spec['path'])
        require(self.path.is_relative_to(CGROUP) and len(self.path.parts) >= 5,
                'cannot own a global hierarchy')
        require(not Path(process(os.getpid())['group']).is_relative_to(self.path), 'cannot retire own service')
        meta = protected(self.path, directory=True)
        if enclosing:
            migration_boundary(self.path)
        require((meta.st_dev, meta.st_ino) == (spec['device'], spec['inode']), 'cohort identity replaced')
        self.fd = open_path(self.path, True)
        self.kill_fd = self.events_fd = self.freeze_fd = None
        self.pidfds = {}
        self.spec = spec
        try:
            require((os.fstat(self.fd).st_dev, os.fstat(self.fd).st_ino) == (spec['device'], spec['inode']),
                    'opened cohort identity changed')
            self.kill_fd = os.open('cgroup.kill', os.O_WRONLY | os.O_CLOEXEC, dir_fd=self.fd)
            self.events_fd = os.open('cgroup.events', os.O_RDONLY | os.O_CLOEXEC, dir_fd=self.fd)
            self.freeze_fd = os.open('cgroup.freeze', os.O_WRONLY | os.O_CLOEXEC, dir_fd=self.fd)
            for fd in (self.kill_fd, self.freeze_fd):
                control = os.fstat(fd)
                require(control.st_uid == 0 and not control.st_mode & 0o022,
                        'mutable cohort retirement control')
            self.preflight()
        except BaseException:
            self.close()
            raise

    def write(self, fd, value):
        require((self.path.stat().st_dev, self.path.stat().st_ino) ==
                (self.spec['device'], self.spec['inode']), 'cohort locator changed')
        os.lseek(fd, 0, 0)
        require(os.write(fd, value) == len(value), 'cohort control write failed')

    def events(self):
        os.lseek(self.events_fd, 0, 0)
        raw = os.read(self.events_fd, 4096)
        require(len(raw) < 4096, 'cohort events bound')
        return dict(row.split() for row in raw.decode().splitlines())

    def freeze(self):
        self.write(self.freeze_fd, b'1')

    def preflight(self):
        expected = {int(pid): start for pid, start in self.spec['members'].items()}
        require(len(expected) == len(self.spec['members']) and members(self.path) == set(expected),
                'cohort membership differs from guarded inventory')
        for pid, start in expected.items():
            item = process(pid)
            require(item['start'] == start and Path(item['group']).is_relative_to(self.path),
                    'cohort preflight identity changed')

    def capture(self):
        require(self.events().get('frozen') == '1', 'cohort not frozen')
        expected = {int(pid): start for pid, start in self.spec['members'].items()}
        require(members(self.path) == set(expected), 'cohort membership differs from guarded inventory')
        evidence = []
        for pid, start in expected.items():
            fd = os.pidfd_open(pid)
            self.pidfds[pid] = fd
            item = process(pid)
            require(item['start'] == start and Path(item['group']).is_relative_to(self.path) and
                    not select.select([fd], [], [], 0)[0], 'old writer changed or exited during capture')
            evidence.append(item)
        return evidence

    def kill(self):
        # PID1 may complete the requested stop and remove the original group
        # before our explicit kill. Never address a replacement at that name.
        # An unlinked retained kernel directory PLUS every captured pidfd exit
        # is retirement evidence; a missing pathname alone is not.
        if self.retired():
            return
        try:
            self.write(self.kill_fd, b'1')
        except OSError:
            if not self.retired():
                raise

    def retired(self):
        # pidfd exit plus empty kernel hierarchy: no potentially signing task is
        # left. These are not our direct children; systemd/their parent reaps them.
        captured = len(self.pidfds) == len(self.spec['members'])
        exited = captured and all(select.select([fd], [], [], 0)[0] for fd in self.pidfds.values())
        if self.unlinked():
            return exited
        try:
            return self.events().get('populated') == '0' and exited
        except OSError:
            if self.unlinked():
                return exited
            raise

    def unlinked(self):
        # kernfs may retain a cached nonzero link count, or return ENODEV from
        # getattr/read after PID1 removes a cgroup. procfs's exact '(deleted)'
        # suffix on OUR still-open directory is kernel dentry evidence, unlike
        # absence of an unpinned pathname. Keep every pidfd until retirement too.
        try:
            if os.fstat(self.fd).st_nlink == 0:
                return True
        except OSError as error:
            if error.errno not in (errno.ENOENT, errno.ENODEV):
                raise
        return os.readlink(f'/proc/self/fd/{self.fd}') == str(self.path) + ' (deleted)'

    def close(self):
        for fd in [self.kill_fd, self.events_fd, self.freeze_fd, *self.pidfds.values(), self.fd]:
            if fd is not None:
                os.close(fd)
        self.kill_fd = self.events_fd = self.freeze_fd = self.fd = None
        self.pidfds = {}


def quarantine(policy, resume=False):
    """Move specifically inventoried copies; do not read, overwrite or erase keys.

    All old credential-bearing cohorts must already have exited. A successful
    rename is only disposition of this inode, never proof about unknown copies.
    Cross-filesystem moves refuse; the coordinator must provision same-filesystem
    private custody or resolve that case before authorizing a transition.
    """
    destination = STATE / 'quarantine'
    protected(destination, private=True, directory=True)
    evidence = []
    seen = set()
    for index, item in enumerate(policy.value['copies']):
        fields={'source', 'identity', 'disposition'}
        if policy.value['version']==3 and 'custody' in item:fields.add('custody')
        require(set(item) == fields and
                item['disposition'] in ('quarantine-known-copy', 'retained-exclusive-backing'),
                'unresolved/copied/external signing authority blocks admission')
        source = canonical(item['source'])
        require(source not in seen, 'duplicate signing-copy locator')
        seen.add(source)
        if resume and item['disposition']=='quarantine-known-copy':
            import c01_migration_layout
            target=c01_migration_layout.custody(policy,index,item)
            if not os.path.lexists(source):
                protected(target.parent,private=True,directory=True)
                fd=open_path(target)
                try:
                    meta=os.fstat(fd)
                    require(identity(meta)==item['identity'] and stat.S_ISREG(meta.st_mode) and meta.st_nlink==1,
                            'partial quarantine custody differs')
                finally:os.close(fd)
                evidence.append({'index':index,'operation':'quarantined-known-copy','identity':identity(meta)})
                continue
        fd = open_path(source)
        parent = open_path(source.parent, True)
        try:
            meta = os.fstat(fd)
            require(identity(meta) == item['identity'] and stat.S_ISREG(meta.st_mode) and meta.st_nlink == 1,
                    'known signing-copy metadata/alias differs')
            if item['disposition'] == 'retained-exclusive-backing':
                allowed = [Path(policy.launch[k]) for k in ('manager_backing', 'authority_backing')]
                require(any(source.is_relative_to(root) for root in allowed), 'copy outside exclusive backing')
                for root in allowed:
                    protected(root.parent, private=True, directory=True)
                evidence.append({'index': index, 'operation': 'verified-exclusive-backing', 'identity': identity(meta)})
            else:
                require(os.stat(source.name, dir_fd=parent, follow_symlinks=False) == meta, 'copy locator changed')
                if policy.value['version']==3:
                    import c01_migration_layout
                    target=c01_migration_layout.custody(policy,index,item)
                else:target = destination / str(index)
                protected(target.parent,private=True,directory=True)
                require(target.parent.stat().st_dev==meta.st_dev,'quarantine requires same-filesystem private custody')
                require(not target.exists() and not target.is_symlink(), 'copy custody already exists')
                if policy.value['version']==3:
                    target_fd=c01_migration_layout.custody_directory(policy,index,item)
                    source_fd=c01_migration_layout.readable(source.parent)
                    try:
                        if c01_migration_layout.LIBC.renameat2(source_fd,os.fsencode(source.name),
                                target_fd,os.fsencode(target.name),1):
                            code=c01_migration_layout.ctypes.get_errno();raise OSError(code,os.strerror(code))
                        os.fsync(source_fd);os.fsync(target_fd)
                    finally:os.close(source_fd);os.close(target_fd)
                else:os.rename(source.name, target, src_dir_fd=parent)
                require(identity(target.lstat()) == item['identity'], 'copy custody differs')
                # Root-private ancestor is the access control; chmod does NOT
                # revoke old FDs or caches. Process retirement occurred first.
                evidence.append({'index': index, 'operation': 'quarantined-known-copy', 'identity': identity(meta)})
        finally:
            os.close(parent)
            os.close(fd)
    return evidence


def verify_copies(policy):
    for index, item in enumerate(policy.value['copies']):
        if item['disposition'] == 'quarantine-known-copy':
            require(not os.path.lexists(item['source']), 'legacy signing copy reappeared')
            if policy.value['version']==3:
                import c01_migration_layout
                path=c01_migration_layout.custody(policy,index,item)
            else:path = STATE / 'quarantine' / str(index)
            protected(path.parent, private=True, directory=True)
        else:
            require(item['disposition'] == 'retained-exclusive-backing', 'unresolved signing copy')
            path = Path(item['source'])
        fd = open_path(path)
        try:
            require(identity(os.fstat(fd)) == item['identity'] and os.fstat(fd).st_nlink == 1,
                    'signing-copy custody changed')
        finally:
            os.close(fd)


def verify_private(policy):
    for key in ('manager_backing', 'snapshot_backing', 'authority_backing'):
        root = Path(policy.launch[key])
        protected(root.parent, private=True, directory=True)
        fd = open_path(root, True)
        os.close(fd)
    verify_copies(policy)
    if policy.value['version']==3:
        import c01_migration_layout
        c01_migration_layout.verify_live(policy)


class TaskReadRetired(ValueError):
    """Only a missing read through this exact task's proc-directory capability."""
    def __init__(self, task, field):
        super().__init__('pinned task exited during proc read: '+field)
        self.task=task
        self.field=field


class UnobservedTaskExit(ValueError):
    """A pidfd confirmed exit before a bound task observation existed."""


class TaskEvidenceChanged(ValueError):
    def __init__(self, task, fields):
        super().__init__('pinned task credentials/cgroup/namespace incarnation changed: pid='+str(task.pid)+' fields='+','.join(fields))
        self.task=task


class PinnedTask:
    def __init__(self, pid):
        self.pid=pid
        try:self.pidfd=os.pidfd_open(pid)
        except ProcessLookupError as error:raise UnobservedTaskExit('task absent before pidfd observation') from error
        self.procfd=None
        self.dependencies=[];self.parents=[];self.task_validations=[]
        try:
            # While this pidfd is live, its numeric PID cannot be reused. Check
            # both sides of opening procfs; never adopt a successor directory.
            if self.exited():raise UnobservedTaskExit('task exited before proc binding')
            procroot=os.open('/proc',os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW|os.O_CLOEXEC)
            try:
                self.bind_pidfd(procroot)
                self.procfd=self.task_read('directory',lambda:os.open(str(pid),os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW|os.O_CLOEXEC,dir_fd=procroot))
                require(os.fstat(self.procfd).st_dev==os.fstat(procroot).st_dev,'foreign proc directory filesystem')
                self.start=self.start_ticks()
                self.bind_pidfd(procroot)
            finally:os.close(procroot)
            if self.exited():raise UnobservedTaskExit('task exited during proc binding')
        except BaseException:self.close();raise

    def exited(self):return bool(select.select([self.pidfd],[],[],0)[0])

    def bind_pidfd(self, procroot):
        # fdinfo renders the retained pidfd's target in THIS procfs PID namespace.
        # This also rejects a proc mount from a different PID namespace; timing
        # and matching start ticks alone are not a binding between the handles.
        fd=os.open('self/fdinfo/'+str(self.pidfd),os.O_RDONLY|os.O_NOFOLLOW|os.O_CLOEXEC,dir_fd=procroot)
        with os.fdopen(fd,'rb') as stream:raw=stream.read(4097)
        require(len(raw)<=4096,'pidfd metadata bound')
        fields=dict(line.split(':',1) for line in raw.decode().splitlines() if ':' in line)
        actual=int(fields['Pid'])
        if actual==-1 and self.exited():raise UnobservedTaskExit('task exited during pidfd binding')
        require(actual==self.pid,'pidfd/proc PID namespace binding differs')

    def task_read(self, field, operation):
        # This catch encloses ONE identified proc operation, never policy,
        # capability validation, filesystem-view inspection or a parent check.
        try:return operation()
        except (FileNotFoundError,ProcessLookupError):
            # Linux can drop nsproxy before publishing whole-process pidfd exit.
            # Wait at most one bounded kernel poll, issuing NO evidence meanwhile.
            # This does not renew any caller/owner/watch deadline. A task still
            # alive after the poll is a fatal unreadable view, never an admission.
            if not self.exited():select.select([self.pidfd],[],[],0.02)
            require(self.exited(),'live pinned task read missing: '+field)
            raise TaskReadRetired(self,field) from None

    def text(self, name):
        require(name in ('stat','status','cgroup'),'fixed proc text field')
        def read():
            fd=os.open(name,os.O_RDONLY|os.O_NOFOLLOW|os.O_CLOEXEC,dir_fd=self.procfd)
            with os.fdopen(fd,'rb') as stream:raw=stream.read(1024*1024+1)
            require(len(raw)<=1024*1024,'proc text bound')
            return raw.decode()
        return self.task_read(name,read)

    def start_ticks(self):
        raw=self.text('stat')
        require(int(raw.split(' ',1)[0])==self.pid,'proc directory PID differs from pidfd')
        return int(raw[raw.rindex(')')+2:].split()[19])

    def namespace(self, kind):
        require(kind in ('mnt','net','user'),'fixed task namespace')
        fd=self.task_read('ns/'+kind,lambda:os.open('ns/'+kind,os.O_RDONLY|os.O_CLOEXEC,dir_fd=self.procfd))
        try:
            if kind=='net':return c01_network.identity(fd)
            m=os.fstat(fd);return m.st_dev,m.st_ino
        finally:os.close(fd)

    def executable(self):
        return Path(self.task_read('exe',lambda:os.readlink('exe',dir_fd=self.procfd)))

    def expect_executable(self, path):
        meta=protected(path) # independent named capability, never task_read
        expected=(meta.st_dev,meta.st_ino)
        def check():
            actual=self.task_read('exe',lambda:os.stat('exe',dir_fd=self.procfd))
            require((actual.st_dev,actual.st_ino)==expected and self.executable()==path,'mapping executable identity changed')
        def independent():
            current=protected(path)
            require((current.st_dev,current.st_ino)==expected,'mapping executable capability changed')
        self.dependencies.append(independent);self.task_validations.append(check)
        check()

    def root(self):
        return self.task_read('root',lambda:os.open('root',os.O_RDONLY|os.O_DIRECTORY|os.O_CLOEXEC,dir_fd=self.procfd))

    def record(self):
        start=self.start_ticks()
        status=dict(row.split(':',1) for row in self.text('status').splitlines() if ':' in row)
        group=self.text('cgroup').splitlines()
        require(len(group)==1 and group[0].startswith('0::/'),'unified process hierarchy required')
        return {'pid':self.pid,'ppid':int(status['PPid']),'start':start,
            'uids':list(map(int,status['Uid'].split())),'gids':list(map(int,status['Gid'].split())),
            'groups':list(map(int,status['Groups'].split())),
            'caps':{key:int(status[key],16) for key in ('CapEff','CapPrm','CapInh','CapAmb')},
            'nnp':int(status['NoNewPrivs']),'group':str(CGROUP/group[0][3:].lstrip('/')),
            'namespaces':{kind:self.namespace(kind)[1] for kind in ('mnt','net','user')}}

    def finish(self, expected):
        if self.exited():return False
        current=self.record()
        require(current['pid']==self.pid and current['start']==self.start,'pinned task PID incarnation changed')
        if current!=expected:
            raise TaskEvidenceChanged(self,sorted(key for key in current if current[key]!=expected[key]))
        return not self.exited()

    def independent(self):
        for check in self.dependencies:check()

    def close(self):
        for parent in self.parents:parent.close()
        self.parents=[]
        if self.procfd is not None:os.close(self.procfd);self.procfd=None
        if self.pidfd is not None:os.close(self.pidfd);self.pidfd=None

    def __enter__(self):return self
    def __exit__(self,*_):self.close()


def network_appointment(role):
    expected=c01_network.appointed(role)
    require(expected!=c01_network.process_namespace(1),'host network/abstract IPC namespace cannot be admitted')
    return expected


def require_task_network(task, role):
    expected=network_appointment(role) # independent errors must propagate
    def revalidate():require(network_appointment(role)==expected,'appointed namespace changed')
    task.dependencies.append(revalidate)
    require(task.namespace('net')==expected,'process differs from appointed network namespace')
    return expected


def inspect_task(task, item, inspect):
    try:
        require(item['pid']==task.pid and item['start']==task.start,'task PID incarnation changed')
        result=inspect()
        for check in task.task_validations:check()
        if task.finish(item):return result
        return {'pid':task.pid,'start':task.start,'retired_pidfd':True}
    except TaskReadRetired as error:
        # A missing read of another task (notably a mapping parent) is NOT
        # evidence that the inspected child may be ignored.
        require(error.task is task and task.exited(),'independent task lifetime lost')
        return {'pid':task.pid,'start':task.start,'retired_pidfd':True}
    finally:
        task.independent() # also on retirement: no independent failure hidden


def view_evidence(task, policy):
    network_appointment('runtime')
    require(isinstance(task,PinnedTask),'retained proc/pidfd capability required')
    item=task.record()
    _,result=stable_task_evidence(task,item,lambda current:live_view_evidence(current,policy,task))
    return result


def stable_task_evidence(task, item, inspect):
    # Discard an incoherent sample and re-run ALL predicates once using a fresh
    # record from the SAME retained task. No PID reopening, cached result, policy
    # failure suppression, deadline reset or unbounded retry is permitted.
    for attempt in range(2):
        try:return item,inspect_task(task,item,lambda:inspect(item))
        except TaskEvidenceChanged as error:
            if error.task is not task or attempt:raise
            try:item=task.record()
            except TaskReadRetired as retired:
                require(retired.task is task and task.exited(),'independent task lifetime lost')
                task.independent()
                return item,{'pid':task.pid,'start':task.start,'retired_pidfd':True}
    raise AssertionError('bounded task sampling exhausted')


def view_parent(root, text):
    """Resolve only directories beneath the retained task root, without aliases."""
    parts=Path(text).parts[1:];require(parts,'nonempty view locator')
    fd=os.dup(root)
    try:
        for component in parts[:-1]:
            require(component not in ('','.','..'),'canonical view component')
            child=os.open(component,os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW|os.O_CLOEXEC,dir_fd=fd)
            os.close(fd);fd=child
        return fd,parts[-1]
    except BaseException:os.close(fd);raise


def view_stat(root, text):
    parent,name=view_parent(root,text)
    try:return os.stat(name,dir_fd=parent,follow_symlinks=False)
    finally:os.close(parent)


def view_flags(root, text):
    parent,name=view_parent(root,text)
    try:
        fd=os.open(name,os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW|os.O_CLOEXEC,dir_fd=parent)
        try:return os.fstatvfs(fd).f_flag
        finally:os.close(fd)
    finally:os.close(parent)


def live_view_evidence(item, policy, task):
    """Check actual runtime worker/user-manager view, not mount config text.

    Nested rootless container namespaces are separately admitted only beneath the
    qualified user-manager hierarchy; their host authority is bounded by its
    original user namespace maps and protected namespace/IPC launch policy.
    """
    r = policy.runtime
    pid = item['pid']
    require_task_network(task, 'runtime')
    require(item['namespaces']['mnt'] != os.stat('/proc/self/ns/mnt').st_ino and
            item['namespaces']['net'] != os.stat('/proc/self/ns/net').st_ino,
            'runtime requires private mount AND routed private network namespace')
    backing = os.stat(r['home_backing'])
    def backing_check():
        require(identity(os.stat(r['home_backing']))==identity(backing),'runtime home backing changed')
    task.dependencies.append(backing_check)
    root=task.root()
    try:return inspect_runtime_root(root,item,policy,backing)
    finally:os.close(root)


def inspect_runtime_root(root, item, policy, backing):
    r=policy.runtime
    home = view_stat(root,r['home'])
    require((home.st_dev, home.st_ino) == (backing.st_dev, backing.st_ino), 'uncurated runtime home')
    # Inaccessible objects must be missing or a root-owned inaccessible mount.
    # Merely hiding environment variables or checking .ssh at startup is not enough.
    denied = ['/run/dbus/system_bus_socket', '/run/systemd/private', '/run/sudo',
              '/run/polkit-1', '/root', '/home/archipelago-admin',
              '/usr/bin/sudo', '/usr/bin/pkexec', '/usr/bin/su', '/usr/bin/doas',
              '/usr/bin/chsh', '/usr/bin/chfn', '/usr/bin/passwd', '/usr/bin/crontab', '/usr/bin/at']
    denied += r['private_parents']
    for text in denied:
        try:
            meta = view_stat(root,text)
        except FileNotFoundError:
            continue
        require(meta.st_uid == 0 and stat.S_IMODE(meta.st_mode) == 0,
                'runtime privileged IPC/operator path exposed: ' + text)
    for name in ('.ssh','.gnupg','.pki'):
        try:view_stat(root,r['home']+'/'+name)
        except FileNotFoundError:continue
        raise ValueError('operator credential directory exposed')
    for text in ('/opt/archipelago', '/usr/local/bin'):
        require(view_flags(root,text) & os.ST_RDONLY,
                'runtime retains a writable protected package view: ' + text)
    return {'pid': item['pid'], 'start': item['start'], 'namespaces': item['namespaces'], 'home': [home.st_dev, home.st_ino]}


def verify_supervisor(item, policy):
    require(Path(item['group']) == Path(policy.value['supervisor_cgroup']) / 'supervisor' and
            item['uids'] == [0] * 4 and item['gids'] == [0] * 4 and item['nnp'] == 1,
            'supervisor must own the exact delegated lifetime with NNP')
    # Mandatory owner checks never convert exit into benign task retirement.
    network_appointment('managed')
    with PinnedTask(item['pid']) as task:
        require(task.start==item['start'] and task.record()==item,'supervisor incarnation/evidence changed')
        result=require_task_network(task,'managed')
        task.independent()
        require(task.finish(item),'mandatory supervisor exited')
        return result


def verify_processes(policy, supervisor=None):
    r, v = policy.runtime, policy.value
    runtime_group = CGROUP / r['user_cgroup'].lstrip('/')
    observed = []
    host_userns = os.stat('/proc/self/ns/user').st_ino
    # Mandatory appointments are independent of whether any sampled task exits.
    network_appointment('managed');network_appointment('runtime')
    for task,item in pinned_processes({a['uid'] for a in v['accounts'].values()}):
        item,evidence=stable_task_evidence(task,item,lambda current:verify_process_item(current,policy,supervisor,host_userns,runtime_group,task))
        runtime_account=v['accounts'].get('runtime')
        if (evidence is not None and runtime_account is not None and
                item['uids']==[runtime_account['uid']]*4 and item['namespaces']['user']==host_userns):observed.append(evidence)
    return observed


def pinned_processes(uids):
    for path in Path('/proc').iterdir():
        if not path.name.isdecimal():continue
        try:task=PinnedTask(int(path.name))
        except (TaskReadRetired,UnobservedTaskExit):continue # no observation published
        try:
            status=dict(row.split(':',1) for row in task.text('status').splitlines() if ':' in row)
            if not uids.intersection(map(int,status['Uid'].split())):continue
            item=task.record()
            require(item['start']==task.start,'task changed during snapshot')
            if not task.exited():yield task,item
        except TaskReadRetired as error:
            require(error.task is task and task.exited(),'unbound task read disappearance')
        finally:task.close()


def verify_process_item(item, policy, supervisor, host_userns, runtime_group, task):
    """All per-task procfs reads run beneath the same pinned-exit guard.

    This includes managed-role namespace inspection and mapping-helper ancestry,
    not just the ordinary runtime view. A missing path while the captured task
    is still alive continues to revoke the lease.
    """
    v=policy.value
    for role, account in v['accounts'].items():
        if account['uid'] not in item['uids']:
            continue
        if role == 'runtime' and item['uids'] != [account['uid']] * 4:
            # Rootless mappings require these two fixed setuid helpers. An
            # arbitrary root executable or changed argv cannot substitute:
            # inspect the actual executable inode and its unprivileged
            # parent, namespace and owned runtime hierarchy.
            executable = task.executable()
            require(executable in (Path('/usr/bin/newuidmap'), Path('/usr/bin/newgidmap')),
                    'unapproved privileged runtime helper')
            task.expect_executable(executable)
            require(str(executable) in policy.value['trees'], 'mapping helper not in reviewed policy inventory')
            try:
                parent_task=PinnedTask(item['ppid']);task.parents.append(parent_task)
                parent=parent_task.record()
            except (ProcessLookupError,TaskReadRetired,UnobservedTaskExit) as error:
                raise ValueError('mapping parent lifetime unavailable') from error
            require(parent['uids'] == [account['uid']] * 4 and not any(parent['caps'].values()) and
                    item['uids'] == [account['uid'], 0, 0, 0] and
                    item['gids'] == [account['gid']] * 4 and set(item['groups']) <= {account['gid']} and
                    item['namespaces'] == parent['namespaces'] and
                    Path(item['group']).is_relative_to(runtime_group) and
                    Path(parent['group']).is_relative_to(runtime_group), 'mapping helper provenance refused')
            def parent_check():
                try:
                    parent_task.independent()
                    require(parent_task.finish(parent),'mapping parent lifetime lost')
                except TaskReadRetired as error:raise ValueError('mapping parent lifetime lost') from error
            task.dependencies.append(parent_check)
            try:live_view_evidence(parent,policy,parent_task)
            except TaskReadRetired as error:raise ValueError('mapping parent view lost') from error
            continue
        require(item['uids'] == [account['uid']] * 4 and item['gids'] == [account['gid']] * 4 and
                set(item['groups']) <= {account['gid']}, 'managed process retained credentials')
        group = Path(item['group'])
        if role == 'runtime':
            require(group.is_relative_to(runtime_group), 'runtime process outside exclusive user manager')
            if item['namespaces']['user'] == host_userns:
                require(not any(item['caps'].values()), 'runtime retained host capabilities')
                return live_view_evidence(item, policy, task)
        else:
            require(supervisor is not None and group.is_relative_to(Path(v['supervisor_cgroup'])) and
                    not any(item['caps'].values()) and item['nnp'] == 1,
                     'managed role lacks live exclusive supervisor ownership')
            require_task_network(task, 'managed')
    return None