secrets-engine/tests/test_task_approval_preparation.py

"""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']