"""Supervised single-slot network mutation owner. Never imports runtime code."""
import array
import json
import os
import select
import socket
import stat
import struct
import threading
import time
import c01_network_settings_nm as nm

PROGRESS = struct.Struct('!IIQQ16s')


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


def exact(pairs):
    result = {}
    for key, value in pairs:
        nm.require(key not in result)
        result[key] = value
    return result


def validate_ticket(ticket):
    nm.require(type(ticket) is dict and set(ticket) == {'epoch', 'sequence'})
    nm.require(type(ticket['sequence']) is int and 0 < ticket['sequence'] < 2**64)
    epoch = ticket['epoch']
    nm.require(type(epoch) is list and len(epoch) == 16 and any(epoch) and
               all(type(byte) is int and 0 <= byte <= 255 for byte in epoch))


def receive(channel, identity, limit=4096):
    raw, anc, flags, _ = channel.recvmsg(limit + 1, socket.CMSG_SPACE(12) + socket.CMSG_SPACE(64), socket.MSG_CMSG_CLOEXEC)
    credentials, unexpected = [], False
    for level, kind, data in anc:
        if (level, kind) == (socket.SOL_SOCKET, socket.SCM_RIGHTS):
            rights = array.array('i')
            rights.frombytes(data[:len(data) - len(data) % rights.itemsize])
            for fd in rights:
                os.close(fd)
            unexpected = True
        elif (level, kind) == (socket.SOL_SOCKET, socket.SCM_CREDENTIALS):
            credentials.append(struct.unpack('3i', data))
        else:
            unexpected = True
    nm.require(not unexpected and credentials == [identity] and flags & ~socket.MSG_CMSG_CLOEXEC == 0 and 0 < len(raw) <= limit)
    return json.loads(raw, object_pairs_hook=exact)


def send(channel, value):
    raw = json.dumps(value, separators=(',', ':')).encode()
    nm.require(len(raw) <= 4096 and channel.send(raw, socket.MSG_DONTWAIT | socket.MSG_NOSIGNAL) == len(raw))


def open_journal():
    # Fixed deployment-owned location, no caller path, no aliases in ancestry.
    parent = os.open('/', os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC)
    try:
        for component in ('var', 'lib', 'archipelago-network-settings'):
            if component == 'archipelago-network-settings':
                try:
                    os.mkdir(component, 0o700, dir_fd=parent)
                    os.fsync(parent)
                except FileExistsError:
                    pass
            child = os.open(component, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC, dir_fd=parent)
            meta = os.fstat(child)
            if meta.st_uid != 0 or meta.st_gid != 0 or meta.st_mode & 0o022:
                os.close(child)
                raise ValueError('unprotected network journal')
            os.close(parent)
            parent = child
        return parent
    except BaseException:
        os.close(parent)
        raise


class Journal:
    def __init__(self, fd):
        self.fd = fd
        meta = os.fstat(fd)
        nm.require(stat.S_ISDIR(meta.st_mode) and meta.st_uid == 0 and meta.st_gid == 0 and not meta.st_mode & 0o077)
        self.record = None
        try:
            file = os.open('operation.json', os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC, dir_fd=fd)
        except FileNotFoundError:
            return
        try:
            meta = os.fstat(file)
            nm.require(stat.S_ISREG(meta.st_mode) and meta.st_nlink == 1 and meta.st_uid == 0 and not meta.st_mode & 0o077 and meta.st_size <= 4096)
            self.record = json.loads(os.read(file, 4097), object_pairs_hook=exact)
            nm.require(type(self.record) is dict and set(self.record) == {'ticket', 'phase', 'targets'})
            nm.require(self.record['phase'] in ('Applying', 'Verifying', 'Unknown', 'Succeeded'))
            validate_ticket(self.record['ticket'])
        finally:
            os.close(file)

    def blocked(self):
        return self.record is not None and self.record['phase'] != 'Succeeded'

    def save(self, ticket, phase, targets):
        value = {'ticket': ticket, 'phase': phase, 'targets': targets}
        raw = json.dumps(value, separators=(',', ':')).encode()
        nm.require(len(raw) <= 4096)
        name = 'stage-' + os.urandom(16).hex()
        fd = os.open(name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC, 0o600, dir_fd=self.fd)
        try:
            nm.require(os.write(fd, raw) == len(raw))
            os.fsync(fd)
        finally:
            os.close(fd)
        os.rename(name, 'operation.json', src_dir_fd=self.fd, dst_dir_fd=self.fd)
        os.fsync(self.fd)
        self.record = value


class Owner:
    def __init__(self, journal, factory=nm.NetworkManager):
        self.journal, self.factory = journal, factory
        self.epoch = list(os.urandom(16))
        self.counter = 0
        self.receipt = None
        self.end = 0
        self.thread = None
        self.lock = threading.Lock()
        self.revoked = False

    def submit(self, mutation):
        nm.validate(mutation)
        with self.lock:
            nm.require(not self.revoked)
            nm.require(self.thread is None or not self.thread.is_alive(), 'network slot busy')
            self.counter += 1
            ticket = {'epoch': self.epoch, 'sequence': self.counter}
            self.receipt = {'ticket': ticket, 'phase': 'Rejected' if self.journal.blocked() else 'Accepted'}
            self.end = now() + 45_000_000_000
            if self.receipt['phase'] == 'Accepted':
                self.thread = threading.Thread(target=self.work, args=(mutation, ticket), daemon=True)
                self.thread.start()
            return dict(self.receipt)

    def work(self, mutation, ticket):
        effect, targets = False, []
        def before(paths):
            nonlocal effect, targets
            with self.lock:
                nm.require(not self.revoked and now() < self.end)
                targets = paths
                # Fence memory before disk: uncertain persistence must freeze too.
                effect = True
                self.journal.save(ticket, 'Applying', targets)
                self.receipt['phase'] = 'Applying'
        def verifying():
            with self.lock:
                nm.require(not self.revoked and now() < self.end)
                self.receipt['phase'] = 'Verifying'
        try:
            self.factory().execute(mutation, self.end / 1e9, before, verifying)
            with self.lock:
                nm.require(not self.revoked and now() < self.end and self.receipt['phase'] == 'Verifying')
                self.journal.save(ticket, 'Succeeded', targets)
                self.receipt['phase'] = 'Succeeded'
        except BaseException:
            with self.lock:
                self.receipt['phase'] = 'Unknown' if effect or self.revoked else 'Rejected'
                if effect:
                    self.revoked = True  # Cannot clear uncertainty in this lifetime.
                    try:
                        self.journal.save(ticket, 'Unknown', targets)
                    except BaseException:
                        pass  # Original write-ahead record remains the restart fence.
        finally:
            if 'Wifi' in mutation:
                mutation['Wifi']['password'] = ''

    def status(self, ticket):
        validate_ticket(ticket)
        with self.lock:
            nm.require(self.receipt is not None and ticket == self.receipt['ticket'])
            if now() >= self.end and self.receipt['phase'] in ('Accepted', 'Applying', 'Verifying'):
                self.revoked = True
                self.receipt['phase'] = 'Unknown'
            return dict(self.receipt)


def main():
    nm.require(os.getresuid() == (0, 0, 0) and os.getresgid() == (0, 0, 0))
    parent = os.getppid()
    nm.require(parent > 1)
    control, app = socket.socket(fileno=3), socket.socket(fileno=4)
    for channel in (control, app):
        nm.require(channel.getsockopt(socket.SOL_SOCKET, socket.SO_TYPE) == socket.SOCK_SEQPACKET)
        channel.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)
        channel.setblocking(False)
        channel.set_inheritable(False)
    nm.require(struct.unpack('3i', control.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)) == (parent, 0, 0))
    nm.require(select.select([control], [], [], 2)[0])
    grant = receive(control, (parent, 0, 0))
    nm.require(set(grant) == {'version', 'process_id', 'application', 'journal_device', 'journal_inode'} and type(grant['version']) is int and grant['version'] == 1 and grant['process_id'] == os.getpid())
    peer = grant['application']
    nm.require(set(peer) == {'pid', 'uid', 'gid'} and all(type(peer[k]) is int and peer[k] > 0 for k in peer))
    identity = (peer['pid'], peer['uid'], peer['gid'])
    nm.require(struct.unpack('3i', app.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)) == identity)
    meta = os.fstat(5)
    nm.require((meta.st_dev, meta.st_ino) == (grant['journal_device'], grant['journal_inode']))
    owner = Owner(Journal(5))
    def progress(event, sequence=0, deadline=0):
        raw = PROGRESS.pack(1, event, sequence, deadline, bytes(owner.epoch))
        nm.require(control.send(raw, socket.MSG_DONTWAIT | socket.MSG_NOSIGNAL) == len(raw))
    progress(1)
    exchange, active, heartbeat = 0, False, now()
    while os.getppid() == parent:
        if select.select([control], [], [], 0)[0]:
            raise ValueError('network owner revoked')
        if active:
            state = owner.status(owner.receipt['ticket'])
            nm.require(now() < owner.end, 'network owner deadline')
            if state['phase'] in ('Succeeded', 'Rejected', 'Unknown') and not owner.thread.is_alive():
                progress(4, owner.counter, owner.end)
                active = False
            elif now() >= heartbeat:
                progress(3, owner.counter, owner.end)
                heartbeat = now() + 500_000_000
        elif now() >= heartbeat:
            # Idle liveness also bounds a stall before STARTED can be emitted.
            progress(5)
            heartbeat = now() + 500_000_000
        if not select.select([app], [], [], .05)[0]:
            continue
        request = receive(app, identity)
        nm.require(type(request) is dict and set(request) == {'version', 'sequence', 'operation'} and type(request['sequence']) is int and 0 < request['sequence'] < 2**64 and request['sequence'] == exchange + 1 and type(request['version']) is int and request['version'] == 1)
        operation = request['operation']
        nm.require(type(operation) is dict and len(operation) == 1)
        if 'Submit' in operation:
            result = owner.submit(operation['Submit'])
            if result['phase'] == 'Accepted':
                progress(2, owner.counter, owner.end)
                active = True
                heartbeat = now() + 500_000_000
        elif 'Status' in operation:
            result = owner.status(operation['Status'])
        else:
            raise ValueError('unknown network request')
        exchange += 1
        send(app, {'version': 1, 'sequence': exchange, 'receipt': result})
    raise ValueError('network parent lost')