tegwick
4dd004a225
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786
89 lines
5.1 KiB
Python
89 lines
5.1 KiB
Python
"""Signed DENY supplies preapproval facts; it never supplies execution authority."""
|
|
import copy
|
|
import pytest
|
|
from secrets_engine.authorization import request_digest
|
|
from secrets_engine.errors import DecisionError
|
|
from secrets_engine.task_execution_handoff import prepare_task_approval,consume_task_execution
|
|
from tests.test_task_execution_handoff import lane,prepared
|
|
from tests.signature_support import signature_tools
|
|
|
|
|
|
def normalize(envelope):
|
|
envelope['effect']='deny'
|
|
envelope['provenance']['registry_snapshot_digest']='sha256:'+'4'*64
|
|
envelope['binding']['subject']['attributes']={'roles':['Operator'],'display_name':'registry service'}
|
|
envelope['binding']['request_digest']=request_digest(envelope['binding'])
|
|
envelope['binding']['approval_binding_digest']=envelope['binding']['request_digest']
|
|
|
|
|
|
def test_signed_registry_normalization_is_detached_and_cannot_consume(lane):
|
|
cfg,e,contract,state=lane
|
|
state['decision_change']=normalize
|
|
result=prepare_task_approval(cfg,e,execution_contract=contract)
|
|
output=result.as_owner_handoff()
|
|
assert state['calls']==['check']
|
|
assert 'approval' not in output['original_request']['context']
|
|
assert output['approval_binding_digest']!=output['submitted_request_digest']
|
|
assert 'attributes' not in output['original_request']['subject']
|
|
assert output['execution_contract']==contract.as_dict()
|
|
assert output['native_binding']['target']==output['original_request']['resource']
|
|
output['original_request']['context']['purpose']='mutated'
|
|
assert result.as_owner_handoff()['original_request']['context']['purpose']!='mutated'
|
|
with pytest.raises(DecisionError):consume_task_execution(cfg,result)
|
|
assert state['calls']==['check']
|
|
|
|
|
|
@pytest.mark.parametrize('case',['allow','context','descriptor-type','submitted','evaluated','approval-digest','caller','native','future-time','policy'])
|
|
def test_signed_normalization_refuses_changed_correspondence(lane,case):
|
|
cfg,e,contract,state=lane
|
|
def change(envelope):
|
|
normalize(envelope)
|
|
if case=='allow':envelope['effect']='allow'
|
|
elif case=='context':envelope['binding']['context']['purpose']='other'
|
|
elif case=='descriptor-type':envelope['binding']['context']['execution_contract']['backend_budget']['ttl_seconds']=120.0
|
|
elif case=='submitted':envelope['binding']['submitted_request_digest']='sha256:'+'0'*64
|
|
elif case=='evaluated':envelope['binding']['request_digest']='sha256:'+'0'*64
|
|
elif case=='approval-digest':envelope['binding']['approval_binding_digest']='sha256:'+'0'*64
|
|
elif case=='caller':envelope['provenance']['caller']['mode']='disabled'
|
|
elif case=='native':envelope['binding']['resource']['system']='other'
|
|
elif case=='future-time':envelope['provenance']['decision_time']='2099-01-01T00:00:00Z'
|
|
else:envelope['provenance']['policy_version']='other'
|
|
state['decision_change']=change
|
|
with pytest.raises(DecisionError):prepare_task_approval(cfg,e,execution_contract=contract)
|
|
assert 'claim' not in state['calls'] and 'consume' not in state['calls']
|
|
|
|
|
|
def test_signed_context_numeric_kind_cannot_change(lane):
|
|
cfg,e,contract,state=lane
|
|
state['decision_change']=lambda envelope:envelope['binding']['context']['execution_contract']['backend_budget'].__setitem__('ttl_seconds',120.0)
|
|
with pytest.raises(DecisionError):prepared(lane)
|
|
assert 'consume' not in state['calls']
|
|
|
|
|
|
@pytest.mark.parametrize('original,served',[(1,True),(1,1.0),(1.0,1),(False,0)])
|
|
def test_shared_original_target_subset_is_type_sensitive(lane,original,served):
|
|
from datetime import datetime,timedelta,timezone
|
|
from secrets_engine.task_execution_handoff import _task_identity
|
|
from secrets_engine.approval_consume import _expected_request
|
|
cfg,e,contract,state=lane
|
|
request=_expected_request(cfg,e,'execute-ccr',execution_contract=contract)
|
|
request['resource']['attributes']['requested_number']=original
|
|
binding=copy.deepcopy(request);binding['resource']['attributes']['requested_number']=served
|
|
envelope={'binding':binding,'subject':binding['subject'],'resource':binding['resource'],
|
|
'provenance':{'caller':{'mode':'enforce','principal':'system:serviceaccount:secrets-engine:secrets-engine',
|
|
'audience':'flex-auth','not_after':(datetime.now(timezone.utc)+timedelta(minutes=20)).isoformat()}}}
|
|
with pytest.raises(DecisionError):_task_identity(request,envelope,contract.as_dict(),datetime.now(timezone.utc))
|
|
binding['resource']['attributes']['requested_number']=original
|
|
binding['resource']['attributes']['legitimate_extra_registry_fact']=1.0
|
|
_task_identity(request,envelope,contract.as_dict(),datetime.now(timezone.utc))
|
|
|
|
|
|
def test_old_normalization_record_has_no_permission_window(lane):
|
|
cfg,e,contract,state=lane
|
|
def old(envelope):
|
|
normalize(envelope)
|
|
envelope['provenance']['decision_time']='2000-01-01T00:00:00Z'
|
|
state['decision_change']=old
|
|
normalization=prepare_task_approval(cfg,e,execution_contract=contract)
|
|
with pytest.raises(DecisionError):consume_task_execution(cfg,normalization)
|
|
assert state['calls']==['check']
|