"""Owner-derived metadata pilot compiler. Source-only; never authorizes or dispatches.

OwnerConfig is installed by the trusted owner, never deserialized from a request.
Six fixed calls on two structure objects. Recompile immediately before each call.
Canonical PDP request digest is opaque: the authenticated mediator verifies it.
"""
from __future__ import annotations
import hashlib
import json
import os
from pathlib import Path
import re
import stat
import types
from dataclasses import dataclass
from uuid import UUID
import yaml

class Refused(Exception): pass

def need(value, reason):
    if not value: raise Refused(reason)

def encoded(value):
    return json.dumps(value,sort_keys=True,separators=(',', ':'),ensure_ascii=True,allow_nan=False).encode()

def content_hash(raw): return hashlib.sha256(raw).hexdigest()
def tagged(raw): return 'sha256:'+content_hash(raw)
def digest_ok(value): return type(value) is str and re.fullmatch(r'sha256:[0-9a-f]{64}',value) is not None

class UniqueLoader(yaml.SafeLoader): pass
def unique(loader,node):
    result={}
    for key,val in node.value:
        k=loader.construct_object(key)
        need(k not in result,'duplicate_owner_key')
        result[k]=loader.construct_object(val)
    return result
UniqueLoader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,unique)

TARGET_KEYS=('kind','org','repo','mount','path','fields','mount_management','consumers',
             'delivery_modes','delivery_auth','delivery_config','auth_capability','workload_delivery')
REQUEST_KEYS={'action','kind','catalog_id','operation_id','request_digest','ccr_sha256','catalog_sha256','operations'}

@dataclass(frozen=True)
class OwnerConfig:
    platform_repo: Path
    secrets_repo: Path
    ccr_filename: str
    catalog_filename: str
    catalog_directory: Path
    # Exact path/hash pairs, configured independently by the owner. No request paths.
    source_pins: tuple[tuple[str,str], ...]

@dataclass(frozen=True)
class Plan:
    serialized: bytes
    @property
    def content_digest(self): return content_hash(self.serialized)  # Journal uses bare content hash.
    def metadata(self):
        try: return json.loads(self.serialized)
        except Exception: raise Refused('invalid_plan') from None

@dataclass(frozen=True)
class Call:
    ordinal: int
    kind: str
    method: str
    path: str
    payload: bytes
    @property
    def payload_sha256(self): return tagged(self.payload)
    @property
    def call_digest(self):
        return content_hash(encoded({'ordinal':self.ordinal,'kind':self.kind,'method':self.method,
                                     'path':self.path,'payload_sha256':self.payload_sha256}))

class Compiler:
    def __init__(self, config: OwnerConfig):
        need(type(config) is OwnerConfig,'owner_configuration_required')
        self.config=config
        for root in (config.platform_repo,config.secrets_repo,config.catalog_directory):
            need(type(root) is Path or isinstance(root,Path),'owner_root_invalid')
            need(root.is_absolute() and root.resolve()==root,'owner_root_invalid')
        for name in (config.ccr_filename,config.catalog_filename):
            need(type(name) is str and re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.-]*\.yaml',name),'owner_filename_invalid')
        self.ccr=config.platform_repo/'credential-change-requests'/config.ccr_filename
        self.catalog=config.catalog_directory/config.catalog_filename
        self.validator=config.platform_repo/'scripts/credential-change.py'
        self.catalog_code=config.secrets_repo/'src/secrets_engine/catalog.py'
        self.dependencies=[config.secrets_repo/'src/secrets_engine'/name for name in ('__init__.py','errors.py','redact.py')]
        pins=dict(config.source_pins)
        self.pins=types.MappingProxyType(pins)
        need(len(self.pins)==len(config.source_pins) and all(type(k) is str and digest_ok(v) for k,v in self.pins.items()),'owner_pins_invalid')
        # Compiler source is part of the owner plan, not just generated body hashes.
        required={str(p) for p in [Path(__file__).resolve(),self.ccr,self.catalog,self.validator,self.catalog_code,*self.dependencies]}
        need(required <= set(self.pins),'owner_source_pin_missing')
        extra=set(self.pins)-required
        need(len(extra)==1,'closed_owner_source_set_required')
        artifact=Path(next(iter(extra)))
        need(artifact.parent==config.platform_repo/'openbao/policies'
             and re.fullmatch(r'workload-kv-read-[a-z0-9-]+\.hcl',artifact.name),
             'closed_owner_policy_source_required')

    def read(self,path):
        need(str(path) in self.pins and path.is_absolute() and path.resolve()==path,'owner_source_path_invalid')
        fd=None
        try:
            fd=os.open(path,os.O_RDONLY|os.O_NOFOLLOW|os.O_NONBLOCK)
            before=os.fstat(fd)
            need(stat.S_ISREG(before.st_mode) and before.st_nlink==1 and not before.st_mode & 0o022
                 and 0<before.st_size<=1024*1024,'owner_source_file_invalid')
            pieces=[];size=0
            while size<=1024*1024:
                chunk=os.read(fd,min(4096,1024*1024+1-size))
                if not chunk: break
                size+=len(chunk);pieces.append(chunk)
            after=os.fstat(fd);raw=b''.join(pieces)
            identity=lambda x:(x.st_dev,x.st_ino,x.st_size,x.st_mtime_ns,x.st_ctime_ns,x.st_mode,x.st_nlink)
            need(identity(before)==identity(after) and size==before.st_size and tagged(raw)==self.pins[str(path)],'owner_source_changed')
            return raw
        except (OSError,ValueError): raise Refused('owner_source_unavailable') from None
        finally:
            if fd is not None: os.close(fd)

    def _derive(self):
        # No native operations, CLI applier, subprocess, network or ambient credentials.
        snapshot={p:self.read(Path(p)) for p in self.pins}
        try:
            ccr=yaml.load(snapshot[str(self.ccr)],Loader=UniqueLoader)
            need(type(ccr) is dict and ccr.get('request_type')=='workload-kv-read','unsupported_ccr_kind')
            policy_name=ccr['openbao']['policy_name']
            need(type(ccr['openbao']['kv_path']) is str and re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.-]*(/[A-Za-z0-9][A-Za-z0-9_.-]*)+',ccr['openbao']['kv_path']),'unsafe_logical_kv_path')
            need(type(policy_name) is str and re.fullmatch(r'workload-kv-read-[a-z0-9-]+',policy_name),'unsafe_policy_target')
            policy=self.config.platform_repo/'openbao/policies'/f'{policy_name}.hcl'
            need(ccr['openbao']['policy_file']==str(policy.relative_to(self.config.platform_repo)),'unsafe_policy_source')
            need(str(policy) in snapshot,'owner_policy_pin_missing')
            owner=types.ModuleType('owner_credential_change');owner.__file__=str(self.validator)
            exec(compile(snapshot[str(self.validator)],str(self.validator),'exec'),owner.__dict__)
            # Reuse full existing validator on the pinned parsed bytes; never choose alternate CCR_DIR.
            owner.load_yaml=lambda p: ccr if p==self.ccr else (_ for _ in ()).throw(Refused('unexpected_owner_read'))
            _ccr,errors,_warnings=owner.validate_ccr(self.ccr)
            need(not errors,'owner_ccr_invalid')
            need(ccr['status']=='approved','owner_ccr_not_fresh_approved')
            need(not owner.applier_readiness_blockers(ccr) and not owner.apply_blockers(ccr),'owner_ccr_not_ready')
            need(ccr['openbao']['auth']['method'] in {'oidc','kubernetes'},'unsupported_auth_kind')
            generated=owner.generated_policy_hcl(ccr)
            need(owner.normalized_policy_body(snapshot[str(policy)].decode())==owner.normalized_policy_body(generated),'owner_policy_not_generated')
            # Use the actual consumer catalog validator/defaults, not an alternative broker catalog.
            import secrets_engine.catalog as catalogue
            import secrets_engine.errors as errors_module
            import secrets_engine.redact as redact_module
            import secrets_engine
            for module,p in [(catalogue,self.catalog_code),(secrets_engine,self.dependencies[0]),(errors_module,self.dependencies[1]),(redact_module,self.dependencies[2])]:
                need(Path(module.__file__).resolve()==p,'catalogue_runtime_source_mismatch')
            entry=catalogue.validate_entry(yaml.load(snapshot[str(self.catalog)],Loader=UniqueLoader),source='<owner-pinned-catalogue>')
            need(entry.id==ccr['access_frontdoor']['catalog_id'] and entry.stage=='prod'
                 and ccr['target']['environment']=='production' and entry.kind=='kv'
                 and entry.mount_management=='existing'
                 and entry.delivery_auth_management in {'none','existing'},'catalogue_not_metadata_pilot')
            need(entry.mount==ccr['openbao']['mount'] and entry.path==ccr['openbao']['kv_path'][len(entry.mount)+1:]
                 and sorted(entry.fields)==sorted(ccr['openbao']['fields'])
                 and entry.approval.get('decision_ref')==ccr['id'],'catalogue_ccr_mismatch')
            target={k:getattr(entry,k) for k in TARGET_KEYS}
            auth=ccr['openbao']['auth'];role_path=f"auth/{auth['mount']}/role/{auth['role']}"
            policy_path=f'sys/policies/acl/{policy_name}'
            bodies=[b'',encoded({'policy':generated.rstrip()}),b'',b'',encoded(owner.auth_payload(ccr)),b'']
            paths=[policy_path,policy_path,policy_path,role_path,role_path,role_path]
            methods=['GET','PUT','GET','GET','PUT','GET']
            kinds=['policy_baseline','policy_write','policy_readback','role_baseline','auth_role_write','role_readback']
            calls=tuple(Call(i,kinds[i],methods[i],paths[i],bodies[i]) for i in range(6))
            need(all(re.fullmatch(r'(sys/policies/acl/workload-kv-read-[a-z0-9-]+|auth/(netkingdom|kubernetes)/role/[a-z0-9-]+)',c.path) for c in calls),'unsafe_structure_target')
            # Detect source changes during existing validator/generator reads and catalog validation.
            for p,raw in snapshot.items(): need(self.read(Path(p))==raw,'owner_source_changed')
            return ccr,target,calls
        except Refused: raise
        except BaseException: raise Refused('owner_compilation_refused') from None

    def proposal(self):
        ccr,target,calls=self._derive()
        return {'action':'execute-ccr','kind':'workload-kv-read','catalog_id':ccr['access_frontdoor']['catalog_id'],
                'ccr_sha256':self.pins[str(self.ccr)],'catalog_sha256':tagged(json.dumps(target,sort_keys=True,separators=(',', ':'),ensure_ascii=False,allow_nan=False).encode('utf-8')),
                'catalog_target':json.loads(encoded(target)),
                'operations':[{'method':c.method,'path':c.path,'payload_sha256':c.payload_sha256} for c in calls]}

    def compile(self,request):
        need(type(request) is dict and set(request)==REQUEST_KEYS,'closed_plan_request_required')
        need(digest_ok(request['request_digest']),'canonical_request_digest_required')
        try: need(str(UUID(request['operation_id']))==request['operation_id'],'operation_id_invalid')
        except (ValueError,AttributeError,TypeError): raise Refused('operation_id_invalid') from None
        expected=self.proposal()
        need(all(request[k]==expected[k] for k in REQUEST_KEYS-{'operation_id','request_digest'}),'owner_plan_binding_mismatch')
        return Plan(encoded({'schema':'platform-owner-typed-plan/v1','action':'execute-ccr','operation':'metadata-apply-ccr',
                             'request':request,'catalog_target':expected['catalog_target'],
                             'source_pins':dict(self.pins),'task_token_max_ttl':120,'task_token_max_uses':12,
                             'cleanup_uses_reserved':1}))

    def before_call(self,plan,ordinal):
        need(type(plan) is Plan and type(ordinal) is int and 0<=ordinal<6,'fixed_plan_ordinal_required')
        try: meta=plan.metadata();fresh=self.compile(meta['request'])
        except Refused: raise
        except BaseException: raise Refused('invalid_plan') from None
        need(fresh.serialized==plan.serialized,'immutable_plan_changed')
        # Immediately re-derive owner payload bytes, and reconfirm immutable snapshot after derivation.
        _,_,calls=self._derive();need(self.compile(meta['request']).serialized==plan.serialized,'immutable_plan_changed')
        return calls[ordinal]