railiance-platform/scripts/create-t03-approval-requests.py

"""Silent, single-attempt creation of the three operator-reviewed T03 requests."""
import base64,json,os,stat,subprocess,time
from pathlib import Path
from datetime import datetime,timezone,timedelta
from urllib.request import Request,build_opener,ProxyHandler,HTTPRedirectHandler
from urllib.parse import urlencode
from urllib.error import HTTPError
import jwt
ROOT=Path('/home/worsch/railiance-platform')
RECEIPT=ROOT/'docs/evidence/2026-09-14-t03-native-approval-requests.json'
POLICY='workload-kv-read-secrets-engine-requester-client'
KV='platform/data/workloads/secrets-engine/approval-requester'
ISSUER='https://kc.coulomb.social'
class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self,*args,**kwargs): return None

def http(url,*,body=None,headers=None):
    req=Request(url,data=body,headers=headers or {})
    try:
        with build_opener(ProxyHandler({}),NoRedirect()).open(req,timeout=20) as response:
            content=response.read(1048577);status=response.status
    except HTTPError as error:
        status=error.code;content=error.read(1048577);error.close()
    if len(content)>1048576:raise ValueError('response_too_large')
    return status,json.loads(content)

def bao(*args):
    p=subprocess.run(['bao',*args],capture_output=True,text=True,timeout=20)
    if p.returncode:raise ValueError('metadata_failed')
    return json.loads(p.stdout)

def check_identity(d):
    policies=set(d.get('policies',[]))|set(d.get('identity_policies',[]))
    if POLICY not in policies or policies-{POLICY,'default'} or not d.get('entity_id') or not 0<d.get('ttl',0)<=900:
        raise ValueError('reader_identity_failed')

def requests():
    records=json.loads(Path('/home/worsch/flex-auth/examples/informed-decision-t03/records.json').read_text())
    evaluations=json.loads(Path('/home/worsch/secrets-engine/docs/evidence/2026-09-14-t03-native-pdp-inputs.json').read_text())
    if {r['action'] for r in records.values()}!={'apply','verify','exec'} or len(records)!=3:raise ValueError('exact_three_records_required')
    result=[]
    for memo,record in records.items():
        d=evaluations[record['action']]
        if d['provenance']['policy_package_digest']!='sha256:bd11c5fe77ce6439c65fea225ad6b71d2110efc5e7b5bc9b499c59cd0a53b8b4' or d['binding']['subject']['id']!='secrets-engine' or record['binding']['actor']!='secrets-engine':raise ValueError('native_evaluator_binding_drift')
        digest='sha256:'+__import__('hashlib').sha256(json.dumps(record['binding'],sort_keys=True,separators=(',',':'),ensure_ascii=False).encode()).hexdigest()
        if digest!=record['binding_digest']:raise ValueError('native_binding_drift')
        result.append((memo,record,d['binding']['request_digest']))
    return result

def main(receipt):
    prepared=requests()
    if Path.home().parent.name!='.warden-attended-login' or os.getenv('BAO_TOKEN') or os.getenv('VAULT_TOKEN'):raise ValueError('attended_reader_required')
    check_identity(bao('token','lookup','-format=json')['data'])
    for path,expected in [(KV,['read']),('platform/data/workloads/secrets-engine/approval-client',['deny']),('platform/metadata/workloads/secrets-engine',['deny'])]:
        if bao('token','capabilities','-format=json',path)!=expected:raise ValueError('reader_scope_failed')
    helper=Path.home()/'.vault-token';info=helper.lstat()
    if not stat.S_ISREG(info.st_mode) or stat.S_IMODE(info.st_mode)!=0o600 or info.st_uid!=os.getuid():raise ValueError('private_helper_required')
    status,data=http('https://bao.coulomb.social/v1/'+KV+'?version=1',headers={'X-Vault-Token':helper.read_text().strip()})
    if status!=200 or data['data']['metadata']['version']!=1:raise ValueError('requester_delivery_failed')
    secret=data['data']['data']['CLIENT_SECRET'];del data
    def exchange(scope,credential=secret):
        auth=base64.b64encode(('secrets-engine-requester:'+credential).encode()).decode()
        return http(ISSUER+'/token',body=urlencode({'grant_type':'client_credentials','scope':scope}).encode(),headers={'Authorization':'Basic '+auth,'Content-Type':'application/x-www-form-urlencoded'})
    status,tokens=exchange('approval:create')
    if status!=200:raise ValueError('requester_exchange_failed')
    token=tokens['access_token'];status,jwks=http(ISSUER+'/jwks')
    if status!=200:raise ValueError('jwks_failed')
    header=jwt.get_unverified_header(token)
    keys=[key for key in jwks['keys'] if key['kid']==header.get('kid')]
    if header.get('alg')!='RS256' or len(keys)!=1:raise ValueError('signing_key_failed')
    claims=jwt.decode(token,jwt.PyJWK.from_dict(keys[0]).key,algorithms=['RS256'],issuer=ISSUER,audience='approval-engine',options={'strict_aud':True,'require':['sub','iat','exp','iss','aud']})
    expected={'sub':'secrets-engine','tenant':'tenant:platform','principal_type':'service','scope':'approval:create','roles':['secrets-engine-requester'],'groups':[]}
    if any(claims.get(k)!=v for k,v in expected.items()) or claims['exp']-claims['iat']!=900:raise ValueError('requester_claims_failed')
    for scope in ('approval:approve','approval:consume','approval:read'):
        if exchange(scope)[0]!=400:raise ValueError('excess_scope_not_refused')
    if exchange('approval:create','invalid-synthetic-credential')[0]!=401:raise ValueError('wrong_secret_not_refused')
    receipt.update(phase='requester_verified',signature_verified=True,excess_scopes_refused=True,wrong_secret_refused=True,reader_scope_verified=True)
    for memo,record,pdp_digest in prepared:
        now=datetime.now(timezone.utc)
        body={'id':record['approval_id'],'binding':record['binding'],'validity':{'not_before':now.isoformat(),'expires_at':(now+timedelta(hours=24)).isoformat()},'required_count':1,'human_control':True,'pdp_path':True,'pdp_digest':pdp_digest}
        receipt['phase']='create_attempt:'+record['action'];RECEIPT.write_text(json.dumps(receipt,indent=2)+'\n')
        status,result=http('http://127.0.0.1:18281/v1/approvals',body=json.dumps(body).encode(),headers={'Authorization':'Bearer '+token,'Content-Type':'application/json'})
        if status!=201 or result['status']!='requested' or result['entries'] or result['binding']['digest']!=record['binding_digest'] or result['binding']['human_control'] is not True or result['binding']['pdp_digest']!=pdp_digest:raise ValueError('request_creation_requires_reconciliation')
        receipt['requests'].append({'memo_id':memo.removeprefix('memo:'),'action':record['action'],'approval':result})
        RECEIPT.write_text(json.dumps(receipt,indent=2)+'\n')
    receipt.update(status='created',phase='three_unapproved_requests_created',human_entries_created=False,approvals_consumed=False)

if __name__=='__main__':
    if RECEIPT.exists():raise SystemExit(1)
    receipt={'observed_at':datetime.now(timezone.utc).isoformat(),'status':'failed','phase':'preflight','requests':[],'credential_values_emitted':False}
    try:main(receipt)
    except Exception:raise SystemExit(1) from None
    finally:RECEIPT.write_text(json.dumps(receipt,indent=2)+'\n')