"""Root-only live runtime lease and detached socket-mount capability.

Capture BEFORE covering /run. The root broker clones its exact socket mount;
move_mount later installs that detached mount without a foreign /proc/fd pathname
or reopening a now-covered locator. No saved readiness bit is accepted.
"""
import ctypes
import os
from pathlib import Path
import platform
import select
import stat

import c01_runtime as wire

LIBC = ctypes.CDLL(None, use_errno=True)
require = wire.require


class RuntimeLease:
    def __init__(self, manager_uid, manager_gid, runtime_uid, runtime_gid, digest):
        require(os.getresuid() == (0, 0, 0), 'root runtime preparation required')
        require(platform.machine() in ('x86_64', 'aarch64'), 'runtime mount syscall ABI not qualified')
        self.channel = None
        self.path_fd = self.pidfd = self.tree = None
        try:
            self.channel = wire.root_connect(wire.PUBLIC / 'readiness.sock')
            value, rights = wire.receive(self.channel, 3)
            self.path_fd, self.pidfd, self.tree = rights
            fields = {'version', 'configuration_sha256', 'pid', 'uid', 'gid', 'cgroup', 'mount_namespace',
                      'listener_device', 'listener_inode', 'manager_uid', 'manager_gid', 'alive_until_boottime_ns'}
            require(type(value) is dict and set(value) == fields and value['version'] == 2 and
                    (value['manager_uid'], value['manager_gid'], value['uid'], value['gid']) ==
                    (manager_uid, manager_gid, runtime_uid, runtime_gid) and
                    value['configuration_sha256'] == digest, 'runtime live identity binding mismatch')
            meta = os.fstat(self.path_fd)
            require(stat.S_ISSOCK(meta.st_mode) and meta.st_uid == 0 and meta.st_gid == manager_gid and
                    stat.S_IMODE(meta.st_mode) == 0o660 and
                    (meta.st_dev, meta.st_ino) == (value['listener_device'], value['listener_inode']),
                    'runtime listener capability mismatch')
            # Prove this is a pidfd for the reported process, not merely an FD
            # that happens to be non-readable. /proc fdinfo is kernel-authored.
            info = dict(row.split(':', 1) for row in Path(f'/proc/self/fdinfo/{self.pidfd}').read_text().splitlines() if ':' in row)
            require(int(info['Pid']) == value['pid'] and value['pid'] > 1, 'runtime pidfd mismatch')
            self.identity = value
            self.expiry = value['alive_until_boottime_ns']
            self.channel.setblocking(False)
            self.check()
            # Readiness v2 supplies the producer-created detached mount. Never
            # reopen a foreign proc-fd path or fall back to a pathname bind.
            tree_meta = os.fstat(self.tree)
            require((tree_meta.st_dev, tree_meta.st_ino, tree_meta.st_mode, tree_meta.st_uid, tree_meta.st_gid) ==
                    (meta.st_dev, meta.st_ino, meta.st_mode, meta.st_uid, meta.st_gid),
                    'detached runtime socket grant mismatch')
        except BaseException:
            self.close()
            raise

    def check(self):
        require(not select.select([self.pidfd], [], [], 0)[0], 'runtime worker exited')
        # Absolute BOOTTIME deadlines prevent buffered, pre-pause messages from
        # reviving stale readiness. Bounded drain, no blocking manager syscall.
        for _ in range(16):
            if not select.select([self.channel], [], [], 0)[0]:
                break
            message, _ = wire.receive(self.channel)
            require(type(message) is dict and set(message) == {'alive_until_boottime_ns'},
                    'runtime renewal schema')
            expiry = message['alive_until_boottime_ns']
            require(type(expiry) is int and expiry >= self.expiry, 'runtime renewal regressed')
            self.expiry = expiry
        now = wire.runtime_owner.now()
        require(type(self.expiry) is int and now < self.expiry <= now + 3_000_000_000,
                'runtime live lease expired')

    def install(self):
        self.check()
        target = wire.PUBLIC
        created=False
        try:target.mkdir(mode=0o755);created=True
        except FileExistsError:pass
        wire.protected(target, directory=True)
        if created:target.chmod(0o755)
        require(stat.S_IMODE(target.stat().st_mode)==0o755,'runtime socket directory traversal differs')
        path = target / 'runtime.sock'
        fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC | os.O_NOFOLLOW, 0o600)
        os.close(fd)
        # move_mount(2), MOVE_MOUNT_F_EMPTY_PATH, fixed target only.
        require(LIBC.syscall(429, self.tree, ctypes.c_char_p(b''), -100,
                            ctypes.c_char_p(os.fsencode(path)), 4) == 0, 'runtime socket mount failed')
        meta = path.stat()
        require((meta.st_dev, meta.st_ino) == (self.identity['listener_device'], self.identity['listener_inode']),
                'installed runtime socket identity changed')
        self.check()

    def close(self):
        if self.channel is not None:
            self.channel.close(); self.channel = None
        for name in ('path_fd', 'pidfd', 'tree'):
            fd = getattr(self, name, None)
            if fd is not None and fd >= 0:
                os.close(fd)
            setattr(self, name, None)