207 lines
12 KiB
Python
207 lines
12 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 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]
|