railiance-platform/scripts/task_execution_plan.py

"""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 builtins
import hashlib
import json
import os
from pathlib import Path
import re
import stat
import sys
import types
from dataclasses import dataclass
from uuid import UUID, uuid4
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}))


def cleanup_specs(operation_id):
    """Deterministic owner cleanup plan; native handles stay in owner memory."""
    try: need(str(UUID(operation_id)) == operation_id, 'operation_id_invalid')
    except (ValueError, AttributeError, TypeError): raise Refused('operation_id_invalid') from None
    policy_name = 'task-exec-' + operation_id
    specs = []
    for kind, slot, calls in [
        ('task-token-revoke', 'task', [('POST','auth/token/revoke-self'), ('POST','auth/token/lookup-accessor')]),
        ('identity-token-revoke', 'identity', [('POST','auth/token/revoke-self'), ('POST','auth/token/lookup-accessor')]),
        ('grant-policy-delete', 'owner', [('DELETE','sys/policies/acl/'+policy_name), ('GET','sys/policies/acl/'+policy_name)]),
    ]:
        value = {'kind':kind, 'credential_slot':slot, 'operation_id':operation_id,
                 'calls':[{'method':method,'path':path} for method,path in calls]}
        if kind == 'task-token-revoke': value['confirm_not_created_if_no_issue_attempt'] = True
        if kind == 'grant-policy-delete': value['skip_delete_if_no_create_attempt'] = True
        value['call_digest'] = content_hash(encoded(value))
        specs.append(value)
    return specs

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 _catalogue_entry(self, snapshot):
        # Execute the verified bytes, not cached imports or timestamp-valid .pyc
        # files. Private names support dataclass annotation lookup without replacing
        # the application's secrets_engine modules, even during concurrent compiles.
        prefix = '_platform_catalogue_' + uuid4().hex
        modules = {}
        registered = []

        def snapshot_import(name, globals=None, locals=None, fromlist=(), level=0):
            if name == 'secrets_engine' or name.startswith('secrets_engine.'):
                need(level == 0 and name in modules, 'unpinned_catalogue_import')
                module = modules[name]
                if fromlist:
                    need(all(hasattr(module, item) for item in fromlist), 'unpinned_catalogue_import')
                    return module
                return modules['secrets_engine']
            need(level == 0, 'unpinned_catalogue_import')
            return builtins.__import__(name, globals, locals, fromlist, level)

        try:
            sources = [('secrets_engine', self.dependencies[0]),
                       ('secrets_engine.errors', self.dependencies[1]),
                       ('secrets_engine.redact', self.dependencies[2]),
                       ('secrets_engine.catalog', self.catalog_code)]
            for name, path in sources:
                private_name = prefix + name[len('secrets_engine'):]
                module = types.ModuleType(private_name)
                module.__file__ = str(path)
                module.__package__ = prefix
                module.__dict__['__builtins__'] = dict(vars(builtins), __import__=snapshot_import)
                modules[name] = module
                sys.modules[private_name] = module
                registered.append(private_name)
                if name != 'secrets_engine':
                    setattr(modules['secrets_engine'], name.rsplit('.', 1)[1], module)
                exec(compile(snapshot[str(path)], str(path), 'exec'), module.__dict__)
            return modules['secrets_engine.catalog'].validate_entry(
                yaml.load(snapshot[str(self.catalog)], Loader=UniqueLoader),
                source='<owner-pinned-catalogue>')
        finally:
            for name in registered:
                sys.modules.pop(name, None)

    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.
            entry=self._catalogue_entry(snapshot)
            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,'cleanup':cleanup_specs(request['operation_id'])}))

    def bind_execution(self, plan, descriptor, owner_descriptor):
        """Bind exact engine descriptor to owner plan; no authorization is created.

        owner_descriptor must be installed independently by the owner, never copied
        from the request. The engine must separately validate its closed schema and
        full canonical request digest/claim/PDP/consume correspondence.
        """
        need(type(plan) is Plan and type(descriptor) is dict and type(owner_descriptor) is bytes,
             'independent_owner_descriptor_required')
        meta=plan.metadata()
        need(self.compile(meta['request']).serialized == plan.serialized, 'immutable_plan_changed')
        need(encoded(descriptor) == owner_descriptor, 'owner_descriptor_mismatch')
        ccr,_,_=self._derive()
        expected = {'contract_version':'SECRETS-EXEC-20261011-v1',
                    'operation_id':meta['request']['operation_id'],
                    'operation':'metadata-apply-ccr','execution_actor':'service:secrets-engine',
                    'ccr_id':ccr['id'],'catalog_sha256':meta['request']['catalog_sha256'],
                    'operations':meta['request']['operations']}
        need(all(descriptor.get(k)==v for k,v in expected.items()),'execution_descriptor_plan_mismatch')
        bound = dict(meta, execution_descriptor=descriptor,
                     execution_descriptor_sha256=tagged(owner_descriptor))
        # The full descriptor's independently admitted grant immutable digest must
        # cover these cleanup commitments. No grant schema exists here yet, so this
        # source binding cannot itself admit a grant/native production execution.
        return Plan(encoded(bound))

    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'])
            if 'execution_descriptor' in meta:
                fresh=self.bind_execution(fresh,meta['execution_descriptor'],encoded(meta['execution_descriptor']))
        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()
        final=self.compile(meta['request'])
        if 'execution_descriptor' in meta:
            final=self.bind_execution(final,meta['execution_descriptor'],encoded(meta['execution_descriptor']))
        need(final.serialized==plan.serialized,'immutable_plan_changed')
        return calls[ordinal]