#!/usr/bin/python3
"""Durable FD-pinned public OS staging on an existing protected data inode.

Root-owned coordinator state stays on the OS filesystem. No directory is created
through the app-owned data ancestry. Mount recovery authenticates original and
alias inodes; a pathname swap, unknown partial object or foreign mount refuses.
"""
import argparse
import ctypes
import fcntl
import hashlib
import json
import os
from pathlib import Path
import signal
import socket
import stat
import time

from d09_store import atomic, read, secure
import os_update_custody as custody

LIBC=ctypes.CDLL(None,use_errno=True)
MS_BIND=4096
MS_PRIVATE=1<<18
CONFIG=Path('/etc/archipelago-os-custody.json')
STATE=Path('/var/lib/archipelago-delivery-coordinator')

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

def identity(fd):
    m=os.fstat(fd)
    return [m.st_dev,m.st_ino,m.st_uid,m.st_gid,stat.S_IMODE(m.st_mode)]

def pin_source(path,expected):
    path=Path(path)
    require(path.is_absolute() and '..' not in path.parts,'canonical original support path required')
    fd=os.open('/',os.O_PATH|os.O_DIRECTORY|os.O_CLOEXEC)
    try:
        for component in path.parts[1:]:
            child=os.open(component,os.O_PATH|os.O_DIRECTORY|os.O_NOFOLLOW|os.O_CLOEXEC,dir_fd=fd)
            os.close(fd);fd=child
        require(identity(fd)==expected and expected[2:4]==[0,0] and not expected[4]&0o022,
                'original support inode/custody differs')
        return fd
    except BaseException:os.close(fd);raise

def mounted(path):
    # All production destination paths are fixed ASCII without escaped fields.
    return sum(row.split()[4]==str(path) for row in Path('/proc/self/mountinfo').read_text().splitlines())

class Coordinator:
    def __init__(self,config=CONFIG,state=STATE):
        require(os.getresuid()==(0,0,0) and os.getresgid()==(0,0,0),'root custody coordinator required')
        self.config=Path(config);self.state=Path(state);self.pins=[];self.lock=None;self.state_pin=None
        self.raw=read(self.config);self.digest=hashlib.sha256(self.raw).hexdigest()
        self.value=json.loads(self.raw)
        v=self.value
        require(type(v) is dict and set(v)=={'version','node','source','source_identity','alias','os_root','stage_name'},
                'OS custody appointment fields')
        require(type(v['version']) is int and v['version']==1 and v['node']==os.uname().nodename and v['stage_name']=='delivery-os-custody',
                'OS custody node/generation binding')
        require(v['source']=='/var/lib/archipelago/support' and
                v['alias']=='/var/lib/archipelago-delivery-support' and
                v['os_root']=='/var/lib/archipelago-os-update','fixed OS custody locators required')
        expected=v['source_identity']
        require(type(expected) is list and len(expected)==5 and all(type(n) is int and n>=0 for n in expected),
                'source identity schema')
        self.journal=self.state/'mounts.json'
        try:
            self.state_pin,_=custody.mkdir(self.state)
            self.lock=os.open(self.state/'owner.lock',os.O_RDWR|os.O_CREAT|os.O_NOFOLLOW|os.O_CLOEXEC,0o600)
            m=os.fstat(self.lock)
            require(stat.S_ISREG(m.st_mode) and m.st_uid==0 and m.st_nlink==1 and stat.S_IMODE(m.st_mode)==0o600,
                    'coordinator lock custody')
            fcntl.flock(self.lock,fcntl.LOCK_EX|fcntl.LOCK_NB)
            self.lock_identity=identity(self.lock)
            if self.journal.exists():
                self.head=json.loads(read(self.journal))
                require(set(self.head)=={'version','configuration_sha256','sequence','phase','objects'} and
                        type(self.head['version']) is int and self.head['version']==1 and self.head['configuration_sha256']==self.digest and
                        type(self.head['sequence']) is int and self.head['sequence']>=0,
                        'durable custody journal binding')
                require(type(self.head['objects']) is dict and set(self.head['objects'])<=
                        {'support-mountpoint','stage','os-mountpoint'},'durable custody object inventory')
                for name,value in self.head['objects'].items():
                    require(type(value) is list and len(value)==5 and all(type(n) is int and n>=0 for n in value) and
                            value[2:]==[0,0,0o700],'durable root-private object identity')
            else:
                require(set(os.listdir(self.state_pin.fd))=={'owner.lock'},'unreconciled coordinator state')
                self.head={'version':1,'configuration_sha256':self.digest,'sequence':0,'phase':'appointed','objects':{}}
                self.commit('appointed')
        except BaseException:self.close();raise

    def commit(self,phase):
        self.state_pin.check()
        self.head['sequence']+=1;self.head['phase']=phase
        atomic(self.journal,json.dumps(self.head,sort_keys=True,separators=(',',':')).encode())
        self.state_pin.check()

    def directory(self,name,path):
        objects=self.head['objects']
        if name not in objects:
            self.commit('create-'+name+'-intent')
            pin,created=custody.mkdir(path)
            try:
                require(created,'unrecorded existing custody object; explicit reconciliation required')
                objects[name]=identity(pin.fd);self.commit('created-'+name)
            finally:pin.close()
        fd=secure(path,True)
        try:require(identity(fd)==objects[name],'custody directory replaced: '+name)
        finally:os.close(fd)

    def bind(self,name,source_fd,target):
        count=mounted(target)
        expected=identity(source_fd)
        if count:
            require(count==1,'stacked/ambiguous custody mounts')
            fd=secure(target,True)
            try:require(identity(fd)==expected,'foreign custody mount')
            finally:os.close(fd)
        else:
            self.directory(name,target)
            # Open and retain the protected target inode too; mount must not
            # resolve mutable source ancestry after the source pin was captured.
            target_fd=secure(target,True)
            try:
                self.commit('bind-'+name+'-intent')
                if LIBC.mount(os.fsencode('/proc/self/fd/'+str(source_fd)),
                              os.fsencode('/proc/self/fd/'+str(target_fd)),None,MS_BIND,None):
                    raise OSError(ctypes.get_errno(),'bind pinned custody directory')
            finally:os.close(target_fd)
            if LIBC.mount(None,os.fsencode(target),None,MS_PRIVATE,None):
                raise OSError(ctypes.get_errno(),'private custody propagation')
            fd=secure(target,True)
            try:require(identity(fd)==expected,'custody bind postimage differs')
            finally:os.close(fd)
        self.commit('bound-'+name)

    def recover(self):
        custody.refuse_legacy()
        source=pin_source(self.value['source'],self.value['source_identity']);self.pins.append(source)
        alias=Path(self.value['alias']);os_root=Path(self.value['os_root'])
        self.bind('support-mountpoint',source,alias)
        stage=alias/self.value['stage_name']
        self.directory('stage',stage)
        stage_pin=custody.Pin(stage);self.pins.append(stage_pin.fd)
        stage_fd=os.dup(stage_pin.fd);stage_pin.close()
        self.pins[-1]=stage_fd
        self.bind('os-mountpoint',stage_fd,os_root)
        self.check()
        self.commit('ready')
        return self.receipt()

    def check(self):
        require(read(self.config)==self.raw,'OS custody appointment changed')
        self.state_pin.check()
        named=secure(self.state/'owner.lock')
        try:require(identity(named)==self.lock_identity==identity(self.lock),'OS custody owner lock replaced')
        finally:os.close(named)
        require(json.loads(read(self.journal))==self.head,'OS custody journal replaced')
        source=pin_source(self.value['source'],self.value['source_identity'])
        try:
            for path,expected in ((self.value['alias'],identity(source)),
                                  (self.value['os_root'],self.head['objects']['stage'])):
                require(mounted(path)==1,'OS custody mount missing/stacked')
                fd=secure(path,True)
                try:require(identity(fd)==expected,'OS custody mount inode differs')
                finally:os.close(fd)
            fd=secure(Path(self.value['alias'])/self.value['stage_name'],True)
            try:require(identity(fd)==self.head['objects']['stage'],'staging custody replaced')
            finally:os.close(fd)
        finally:os.close(source)

    def receipt(self):
        return {'configuration_sha256':self.digest,'journal_sha256':hashlib.sha256(read(self.journal)).hexdigest(),
                'sequence':self.head['sequence'],'phase':self.head['phase'],'node':self.value['node'],
                'source_identity':self.value['source_identity'],'stage_identity':self.head['objects']['stage'],
                'os_root':self.value['os_root'],'alias':self.value['alias'],'source_ownership_changed':False}

    def close(self):
        for fd in self.pins:os.close(fd)
        self.pins=[]
        if self.lock is not None:os.close(self.lock);self.lock=None
        if self.state_pin is not None:self.state_pin.close();self.state_pin=None

def main():
    p=argparse.ArgumentParser(description=__doc__);p.add_argument('operation',choices=('recover','serve'))
    args=p.parse_args();owner=Coordinator()
    try:
        print(json.dumps(owner.recover(),sort_keys=True),flush=True)
        if args.operation=='serve':
            stopped=False
            def stop(*_):
                nonlocal stopped
                stopped=True
            signal.signal(signal.SIGTERM,stop);signal.signal(signal.SIGINT,stop)
            address=os.environ.get('NOTIFY_SOCKET')
            if address:
                with socket.socket(socket.AF_UNIX,socket.SOCK_DGRAM) as channel:
                    channel.connect('\0'+address[1:] if address.startswith('@') else address)
                    channel.send(b'READY=1\nSTATUS=Root-owned data custody pinned and verified')
            while not stopped:owner.check();time.sleep(.25)
    finally:owner.close()

if __name__=='__main__':main()