coulomb/informed-decision
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Prepare authenticated lifecycle runtime and production release candidate

Browse source 
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a10780-6d41-7952-9718-dfbc1f41236f
This commit is contained in:
tegwick 2026-10-04 21:40:19 +02:00
parent f3b71b1521
commit 4f546ae3b1
41 changed files with 2598 additions and 52 deletions
Download patch file Download diff file Expand all files Collapse all files

28
informed_decision/store.py
View file

@ -259,12 +259,36 @@ class Store:
if row is None or row["outcome"] != "allow" or row["usable_until"] <= time.time():
raise Conflict("a current recorded policy observation is required")
request = json.loads(row["request"])
expected = {"memo_version": memo.version, "approval_id": memo.approval_id,
"approval_binding_digest": memo.approval_binding_digest}
if set(request['context']) - expected.keys():
context = self._intake_context(db, memo)
if context is None:
raise Conflict("intake policy observation has no ready receipt")
expected.update(context)
if (request["action"] != action or request["subject"]["id"] != subject
or request["resource"]["id"] != "memo:" + memo.id
or request["context"] != {"memo_version": memo.version, "approval_id": memo.approval_id,
"approval_binding_digest": memo.approval_binding_digest}):
or request["context"] != expected):
raise Conflict("policy observation does not name this act")
@staticmethod
def _intake_context(db, memo):
row = db.execute("SELECT submitter,metadata,receipt FROM intake_ledger WHERE memo_id=? AND state='ready'",
(memo.id,)).fetchone()
if row is None:
return None
receipt, metadata = json.loads(row['receipt']), json.loads(row['metadata'])
if (receipt['approval_id'] != memo.approval_id
or receipt['approval_binding_digest'] != memo.approval_binding_digest):
raise Conflict('intake receipt does not match memo')
return {'review_group': metadata['review_group'], 'memo_class': metadata['memo_class'],
'intake_submitter': row['submitter'], 'approval_requester': 'informed-decision',
'approval_human_control': True}
def intake_context(self, memo):
with self._connection() as db:
return self._intake_context(db, memo)
def _presentation(self, db, presentation_id):
row = db.execute("SELECT body FROM presentations WHERE id=?", (presentation_id,)).fetchone()
if row is None: