informed-decision/tools/prepare_metered_memos.py
tegwick 34eda52755 Support exact-ID human review of ordinary approvals and prepare metered memos
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0e387-534d-70e3-ad53-4ea05676db8c
2026-09-27 20:45:28 +02:00

59 lines
5.1 KiB
Python

"""Draft six unsigned memos from the exact native metered-create receipt."""
import argparse
import hashlib
import json
from pathlib import Path
from informed_decision.memo import Memo, BindingSlice, Principal, Scope, BindingLevel, StepKind, PacketItem, Highlight
from informed_decision.records import dumps, memo_from
LANES = {'glas-claude-agent-dev-anthropic': ('provider', True),
'activity-core-metered-worker-token': ('worker', False)}
TEXT = {
'apply': 'Create only the exact data-read policy and AppRole. Token TTL 5 minutes, maximum 15 minutes; single-use SecretIDs last 5 minutes; at most eight token uses. Existing KV values are preserved. This does not run the proof.',
'verify': 'Verify exact data-read access, configured limits, refusal of unrelated paths, metadata, listing and writes, and session revocation. This does not run inference or display credential values.',
'exec': 'Deliver this credential only into the hash-pinned metered owner, together with its separately approved companion. One disposable PROOF.md tool/commit attempt, no publication or automatic retry. Accepted spend memo infd-20260927-b02 limits the attempt to EUR 10 through 2026-09-28T18:20:00Z; daily EUR 20 and total EUR 500 retain prior liabilities. Both lanes must pass their own native consume and delivery gates before the child runs.'}
def prepare(principal, receipt, native, output):
expected = {(lane, action) for lane in LANES for action in TEXT}
rows = receipt['requests']
if receipt['status'] != 'created' or len(rows) != 6 or {(r['catalog'], r['action']) for r in rows} != expected:
raise ValueError('exact six created requests required')
sources = {(r['catalog'], r['action']): r for r in native}
if len(native) != 6 or set(sources) != expected: raise ValueError('exact six native bindings required')
if not principal or any(c.isspace() for c in principal): raise ValueError('live human subject required')
output.mkdir(parents=True, exist_ok=False)
index = {'kind':'informed-decision-bound-sitting','sitting_id':'metered-20260927-activation',
'principal':principal,'submitted':False,'agent_disposition_forbidden':True,'memos':[]}
for row in rows:
lane, action = row['catalog'], row['action']; label, human = LANES[lane]
approval = row['approval']; source = sources[lane, action]; request = source['request']
if approval['status'] != 'requested' or approval['entries'] or approval['binding']['human_control'] is not human:
raise ValueError('unapproved native record with unchanged control required')
binding = {'action':action,'actor':'secrets-engine','principal':'secrets-engine',
'purpose':request['context']['purpose'],'target':request['resource']}
if any(approval['binding'][k] != v for k,v in binding.items()) or approval['binding']['pdp_digest'] != source['decision']['binding']['request_digest']:
raise ValueError('native action binding mismatch')
ident = row['memo_id']; packet_id = ident+'-packet'
question = f'Approve {action} for the metered {label} credential lane?'
brief = TEXT[action]
packet = json.dumps({'workplan':'SECRETS-WP-0009','task':'SECRETS-WP-0009-T03',
'question':question,'consequences':brief,'human_control_required':human,
'reviewer':'Human review; ordinary worker-lane control stays ordinary.',
'approval':approval,'exact_execution_request':request},indent=2)+'\n'
digest = 'sha256:'+hashlib.sha256(packet.encode()).hexdigest()
memo = Memo(id=ident,version=1,question=question,requested_act='approve',
binding_level=BindingLevel.ORGANIZATIONAL,brief=brief,
binding=BindingSlice(principal=Principal(id=principal,kind='person',display_name=principal,role='Metered proof reviewer'),
target=Scope(kind='secret-catalog-lane',id='catalog:'+lane,label=f'Metered {label} credential',environment='prod')),
step_kind=StepKind.APPROVE,packet=(PacketItem(packet_id,'Exact action and consequences',digest),),
highlights=(Highlight(ident+'-scope',packet_id,brief,required_ack=True,severity='critical'),),
approval_id=approval['id'],approval_binding_digest=approval['binding']['digest'])
raw=dumps(memo); assert memo_from(json.loads(raw)).approval_id == approval['id']
(output/(ident+'.memo.json')).write_text(raw+'\n'); (output/(ident+'.packet.json')).write_text(packet)
index['memos'].append({'memo_id':ident,'approval_id':approval['id'],'memo':ident+'.memo.json','packet':ident+'.packet.json'})
(output/'index.json').write_text(json.dumps(index,indent=2)+'\n')
return len(rows)
if __name__ == '__main__':
p=argparse.ArgumentParser(description=__doc__);p.add_argument('--principal',required=True);p.add_argument('--receipt',type=Path,required=True);p.add_argument('--native',type=Path,required=True);p.add_argument('--output',type=Path,required=True);a=p.parse_args()
print('Prepared',prepare(a.principal,json.loads(a.receipt.read_text()),json.loads(a.native.read_text()),a.output),'unsigned memos; no presentations or dispositions.')