d042d468df
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786
353 lines
21 KiB
Python
353 lines
21 KiB
Python
"""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], ...]
|
|
grant_filename: str | None = None
|
|
|
|
@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')]
|
|
self.grant_record=None;self.grant_validator=None
|
|
if config.grant_filename is not None:
|
|
need(type(config.grant_filename) is str and re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.-]*\.yaml',config.grant_filename)
|
|
and config.grant_filename!=config.ccr_filename,'owner_grant_filename_invalid')
|
|
self.grant_record=config.platform_repo/'credential-change-requests'/config.grant_filename
|
|
self.grant_validator=Path(__file__).resolve().parent/'task_execution_grant.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]}
|
|
if self.grant_record is not None:required.update({str(self.grant_record),str(self.grant_validator)})
|
|
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 grant_effects(self,plan,descriptor):
|
|
"""Independently derive the one proposed grant's effectful immutable body."""
|
|
need(type(plan) is Plan and type(descriptor) is dict,'typed_grant_inputs_required')
|
|
meta=plan.metadata();req=meta['request']
|
|
need(self.compile(req).serialized==plan.serialized,'immutable_plan_changed')
|
|
_,_,calls=self._derive()
|
|
budget=descriptor['backend_budget']
|
|
need(type(budget) is dict and set(budget)=={'ttl_seconds','uses'}
|
|
and type(budget['ttl_seconds']) is int and 30<=budget['ttl_seconds']<=120
|
|
and type(budget['uses']) is int and budget['uses']==12,'fixed_grant_budget_required')
|
|
operation_id=req['operation_id'];policy_name='task-exec-'+operation_id
|
|
paths=sorted({call.path for call in calls})
|
|
hcl=''.join('path '+json.dumps(path)+' {\n capabilities = ["read", "create", "update"]\n}\n' for path in paths)
|
|
hcl+='path "auth/token/revoke-self" {\n capabilities = ["update"]\n}\n'
|
|
token_body={'policies':[policy_name],'no_default_policy':True,'renewable':False,
|
|
'ttl':str(budget['ttl_seconds'])+'s','explicit_max_ttl':str(budget['ttl_seconds'])+'s',
|
|
'num_uses':12,'type':'service','display_name':policy_name,'no_parent':False}
|
|
# A grant cannot include its own immutable hash or raw file SHA. The
|
|
# existing descriptor joins it by ID+immutable SHA after this projection.
|
|
descriptor_binding={k:v for k,v in descriptor.items() if k!='grant_immutable_sha256'}
|
|
pins={path:digest for path,digest in self.pins.items() if self.grant_record is None or path!=str(self.grant_record)}
|
|
execution={'action':'execute-ccr','catalog_id':req['catalog_id'],
|
|
'descriptor_binding':descriptor_binding,'source_pins':pins,
|
|
'cleanup':cleanup_specs(operation_id),'cleanup_uses_reserved':1,
|
|
'cleanup_after_business_deadline':True}
|
|
openbao={'policy':{'name':policy_name,'hcl':hcl},'token_create_body':token_body,
|
|
'owner_parent_custody':'platform','identity_policies':[]}
|
|
return execution,openbao
|
|
|
|
def _bind_grant(self,plan,descriptor):
|
|
need(self.grant_record is not None,'owner_grant_source_required')
|
|
# Execute exactly the pinned pure projection source, never a cached module.
|
|
module=types.ModuleType('_owner_task_execution_grant')
|
|
raw=self.read(self.grant_validator)
|
|
exec(compile(raw,str(self.grant_validator),'exec'),module.__dict__)
|
|
record=yaml.load(self.read(self.grant_record),Loader=UniqueLoader)
|
|
execution,openbao=self.grant_effects(plan,descriptor)
|
|
try:
|
|
projection=module.bind(record,grant_id=descriptor['execution_grant_ccr_id'],
|
|
expected_execution=execution,expected_openbao=openbao,
|
|
expected_digest=descriptor['grant_immutable_sha256'])
|
|
except Exception:raise Refused('owner_grant_binding_mismatch') from None
|
|
need(self.read(self.grant_validator)==raw,'owner_grant_source_changed')
|
|
return projection
|
|
|
|
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))
|
|
if self.grant_record is not None:
|
|
bound['execution_grant']=self._bind_grant(plan,descriptor)
|
|
bound['grant_immutable_sha256']=descriptor['grant_immutable_sha256']
|
|
# Without an independently pinned grant this remains preparation only.
|
|
# The native fixture adapter refuses to materialize such an unbound plan.
|
|
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]
|