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

Reject advisory policy envelopes before lifecycle effects

Browse source 
Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
This commit is contained in:
tegwick 2026-10-11 02:22:07 +02:00
parent caab9a90ec
commit 1e0df2f4a3
13 changed files with 2854 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

63
tests/test_lifecycle_policy_component.py
View file

@ -54,3 +54,66 @@ def test_initial_audience_collaborates_across_revisions_with_real_pdp(workspace,
reopened.retrieve(foreign, memo.id)
assert calls and all(c['known'] and c['audiences'] == ['flex-auth'] for c in calls)
assert (fixture / 'pin.json').read_bytes() == pins
@pytest.mark.parametrize("stale_field", ["version", "package_digest"])
def test_stale_consumer_pin_fails_closed_until_exact_owner_repin(workspace, tmp_path, signing_key, stale_field):
binary = os.environ.get('INFD_FLEX_AUTH_BINARY')
source = os.environ.get('INFD_FLEX_AUTH_SOURCE')
if not binary or not source:
pytest.skip('set INFD_FLEX_AUTH_BINARY and INFD_FLEX_AUTH_SOURCE for candidate owner policy')
fixture = Path(source) / 'examples/informed-decision-lifecycle'
with flex_component(binary, tmp_path, signing_key, fixture=fixture) as (pdp, calls):
repo, memo = workspace
app = LifecycleService(repo, pdp)
hub = Caller(Attribution('tenant:platform', Actor('system:serviceaccount:state-hub:state-hub', ActorKind.SERVICE)),
time.time() + 600, dumps({'principal_type_source': 'authentication-derived',
'tenant_source': 'registration-supplied', 'authentication_audience': 'informed-decision-lifecycle',
'workload_uid': 'synthetic-verified-uid'}))
current = getattr(pdp, stale_field)
setattr(pdp, stale_field, 'v1' if stale_field == 'version' else
'sha256:cee6a79f74e81d0d389a2c2e2451a92be3434ea9dfcee4986e8c7e6cfdbb1827')
with pytest.raises(LifecycleError) as refused:
app.register(hub, memo, key='repin-proof')
assert refused.value.status == 503 and refused.value.code == 'invalid_decision'
with repo.store._connection() as db:
assert db.execute('SELECT count(*) FROM memos').fetchone()[0] == 0
assert db.execute('SELECT count(*) FROM lifecycle_events').fetchone()[0] == 0
setattr(pdp, stale_field, current)
app.register(hub, memo, key='repin-proof')
assert app.retrieve(hub, memo.id) == memo
assert calls and all(c['known'] and c['audiences'] == ['flex-auth'] for c in calls)
def test_real_enforcement_response_with_readback_purpose_cannot_authorize(workspace, tmp_path, signing_key):
binary = os.environ.get('INFD_FLEX_AUTH_BINARY')
source = os.environ.get('INFD_FLEX_AUTH_SOURCE')
if not binary or not source:
pytest.skip('set INFD_FLEX_AUTH_BINARY and INFD_FLEX_AUTH_SOURCE for candidate owner policy')
fixture = Path(source) / 'examples/informed-decision-lifecycle'
with flex_component(binary, tmp_path, signing_key, fixture=fixture) as (pdp, calls):
repo, memo = workspace
app = LifecycleService(repo, pdp)
hub = Caller(Attribution('tenant:platform', Actor('system:serviceaccount:state-hub:state-hub', ActorKind.SERVICE)),
time.time() + 600, dumps({'principal_type_source': 'authentication-derived',
'tenant_source': 'registration-supplied', 'authentication_audience': 'informed-decision-lifecycle',
'workload_uid': 'synthetic-verified-uid'}))
original = pdp.transport.request
def advisory_response(*args, **kwargs):
status, data = original(*args, **kwargs)
assert status == 200 and data['effect'] == 'allow'
assert data['binding'].get('purpose', '') == ''
# Keep the real request binding, digest, pins and TTL unchanged:
# even an otherwise exact allow cannot cross the advisory boundary.
data['binding']['purpose'] = 'readback'
return status, data
pdp.transport.request = advisory_response
with pytest.raises(LifecycleError) as refused:
app.register(hub, memo, key='advisory-proof')
assert refused.value.status == 503 and refused.value.code == 'invalid_decision'
with repo.store._connection() as db:
assert db.execute('SELECT count(*) FROM memos').fetchone()[0] == 0
assert db.execute('SELECT count(*) FROM lifecycle_events').fetchone()[0] == 0
pdp.transport.request = original
app.register(hub, memo, key='advisory-proof')
assert app.retrieve(hub, memo.id) == memo