secrets-engine/tests/test_task_execution_handoff.py

"""Real local HTTP/producer signatures, synthetic approval/caller only.

These source fixtures never enable Access's runtime-deny policy or admit a real
caller, key, owner grant or native backend. Whole-provider integration is separate.
"""
import copy
from datetime import datetime, timedelta, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
from pathlib import Path
from threading import Thread

import pytest

from secrets_engine.approval_claim import binding_from_check_request, claim_binding_digest
from secrets_engine.approval_consume import _expected_request
from secrets_engine.authorization import request_digest, approval_binding_digest
from secrets_engine.errors import DecisionError
from secrets_engine.execution_contract import ExecutionContract
from secrets_engine.task_execution_handoff import (authorize_task_execution,consume_task_execution,
    TaskExecutionAuthorization)
from tests.authorization_stub import AuthorizationStub
from tests.signature_support import signature_tools, admit_tools, sign_envelope, admit
from tests.test_consume_binding_join import _Cfg,_token,APPROVAL_ID
from tests.test_execution_context import entry,descriptor


def stamp(moment):
    return moment.isoformat()


@pytest.fixture
def lane(tmp_path,signature_tools):
    e=entry();contract=ExecutionContract.parse(descriptor(e))
    cfg=_Cfg(_token(tmp_path),authorization_subject_id='secrets-engine',authorization_subject_type='service')
    admit_tools(cfg,tmp_path,signature_tools)
    original=_expected_request(cfg,e,'execute-ccr',execution_contract=contract)
    stub=AuthorizationStub(approval_id=APPROVAL_ID,package=cfg.authorization_policy_package,version=cfg.authorization_policy_version)
    stub.bind_request(original)
    state={'calls':[],'claim_change':None,'decision_change':None,'response_change':None,'raw_response':None}
    claim=stub.claim();native=binding_from_check_request(original)
    claim['binding'].update(native,digest=claim_binding_digest(**native),human_control=True)
    state['claim']=claim

    class Handler(BaseHTTPRequestHandler):
        def log_message(self,*_args):pass
        def reply(self,payload):
            data=json.dumps(payload).encode() if isinstance(payload,dict) else payload
            self.send_response(200);self.send_header('Content-Type','application/json');self.send_header('Content-Length',str(len(data)));self.end_headers();self.wfile.write(data)
        def do_GET(self):
            state['calls'].append('claim')
            value=copy.deepcopy(state['claim'])
            if state['claim_change']:state['claim_change'](value)
            self.reply(value)
        def do_POST(self):
            submitted=json.loads(self.rfile.read(int(self.headers['Content-Length'])))
            if self.path=='/v1/check':
                state['calls'].append('check')
                envelope=stub.decision(submitted)
                envelope['provenance']['caller']={'mode':'enforce','principal':'system:serviceaccount:secrets-engine:secrets-engine',
                    'audience':'flex-auth','not_after':stamp(datetime.now(timezone.utc)+timedelta(minutes=20))}
                if state['decision_change']:state['decision_change'](envelope)
                self.reply(sign_envelope(envelope,signature_tools))
            else:
                state['calls'].append('consume')
                payload={'approval_id':APPROVAL_ID,'status':'consumed','request_digest':submitted['request_digest'],
                         'decision_id':submitted['decision_id'],'consumed_at':stamp(datetime.now(timezone.utc)),'idempotent':False}
                if state['response_change']:state['response_change'](payload)
                self.reply(state['raw_response'] if state['raw_response'] is not None else payload)
    httpd=ThreadingHTTPServer(('127.0.0.1',0),Handler)
    thread=Thread(target=httpd.serve_forever,kwargs={'poll_interval':0.01},daemon=True);thread.start()
    cfg.approval_url=cfg.pdp_url=f'http://127.0.0.1:{httpd.server_port}'
    try:yield cfg,e,contract,state
    finally:httpd.shutdown();httpd.server_close();thread.join(2)


def prepared(lane):
    cfg,e,contract,state=lane
    return authorize_task_execution(cfg,e,execution_contract=contract)


def test_exact_preparation_then_one_strict_cas(lane):
    cfg,e,contract,state=lane
    result=prepared(lane)
    assert state['calls']==['claim','check']
    handoff=result.as_owner_handoff()
    assert set(handoff)=={'execution_contract','approval_id','request_digest','decision_id','native_binding','approval_binding_digest',
                         'submitted_request_digest','decision_not_before','decision_expires_at','approval_expires_at'}
    assert handoff['execution_contract']==contract.as_dict()
    assert handoff['native_binding']==binding_from_check_request(_expected_request(cfg,e,'execute-ccr',execution_contract=contract))
    changed=result.as_owner_handoff();changed['execution_contract']['ccr_id']='changed'
    assert result.as_owner_handoff()==handoff
    with pytest.raises(AttributeError):result._handoff='changed'
    receipt=consume_task_execution(cfg,result)
    assert receipt.approval_id==handoff['approval_id'] and receipt.request_digest==handoff['request_digest']
    assert receipt.decision_id==handoff['decision_id'] and receipt.idempotent is False
    assert state['calls']==['claim','check','consume']
    with pytest.raises(DecisionError):consume_task_execution(cfg,result)
    assert state['calls'].count('consume')==1


@pytest.mark.parametrize('field',['action','actor','principal','purpose','target','digest'])
def test_changed_native_claim_refuses_before_cas(lane,field):
    cfg,e,contract,state=lane
    def change(claim):claim['binding'][field]={'id':'other'} if field=='target' else 'other'
    state['claim_change']=change
    with pytest.raises(DecisionError):prepared(lane)
    assert 'consume' not in state['calls']


@pytest.mark.parametrize('field',['mode','principal','audience','not_after'])
def test_signed_wrong_caller_refuses_before_cas(lane,field):
    cfg,e,contract,state=lane
    state['decision_change']=lambda envelope:envelope['provenance']['caller'].__setitem__(field,'wrong')
    with pytest.raises(DecisionError):prepared(lane)
    assert 'consume' not in state['calls']


@pytest.mark.parametrize('field',['execution_actor','execution_grant_ccr_id','grant_immutable_sha256','ccr_id','backend_origin'])
def test_signed_changed_descriptor_refuses_before_cas(lane,field):
    cfg,e,contract,state=lane
    def change(envelope):envelope['binding']['context']['execution_contract'][field]='wrong'
    state['decision_change']=change
    with pytest.raises(DecisionError):prepared(lane)
    assert 'consume' not in state['calls']


@pytest.mark.parametrize('field',['subject','resource','request_digest','submitted_request_digest','approval_binding_digest'])
def test_signed_changed_request_correspondence_refuses_before_cas(lane,field):
    cfg,e,contract,state=lane
    def change(envelope):
        envelope['binding'][field]={'id':'other'} if field in ('subject','resource') else 'sha256:'+'0'*64
    state['decision_change']=change
    with pytest.raises(DecisionError):prepared(lane)
    assert 'consume' not in state['calls']


@pytest.mark.parametrize('field',['approval_id','decision_id','request_digest','status','consumed_at','idempotent'])
def test_cas_correspondence_never_coerces_or_falls_back(lane,field):
    cfg,e,contract,state=lane;result=prepared(lane)
    state['response_change']=lambda payload:payload.__setitem__(field,'wrong')
    with pytest.raises(DecisionError):consume_task_execution(cfg,result)
    with pytest.raises(DecisionError):consume_task_execution(cfg,result)
    assert state['calls'].count('consume')==1


@pytest.mark.parametrize('value',[True,1,0,'false',None])
def test_nonfresh_or_nonboolean_idempotence_refuses(lane,value):
    cfg,e,contract,state=lane;result=prepared(lane)
    state['response_change']=lambda payload:payload.__setitem__('idempotent',value)
    with pytest.raises(DecisionError):consume_task_execution(cfg,result)


@pytest.mark.parametrize('body',[b'{"status":"consumed","status":"consumed"}',b'{"status":NaN}',b'['*2000+b'0'+b']'*2000])
def test_duplicate_nonfinite_or_deep_consume_body_refuses(lane,body):
    cfg,e,contract,state=lane;result=prepared(lane)
    state['raw_response']=body
    with pytest.raises(DecisionError):consume_task_execution(cfg,result)


@pytest.mark.parametrize('field',['pdp_url','approval_url','authorization_subject_id','authorization_policy_version','pdp_verifier'])
def test_changed_owner_coordinates_refuse_before_cas(lane,field):
    cfg,e,contract,state=lane;result=prepared(lane);setattr(cfg,field,'changed')
    with pytest.raises(DecisionError):consume_task_execution(cfg,result)
    assert 'consume' not in state['calls']


def test_removed_signer_refuses_before_cas(lane):
    cfg,e,contract,state=lane;result=prepared(lane)
    trust=json.loads(cfg.pdp_trust_file.read_text());trust['keys']={'other':next(iter(trust['keys'].values()))}
    cfg.pdp_trust_file.write_text(json.dumps(trust));admit(cfg.pdp_trust_file)
    with pytest.raises(DecisionError):consume_task_execution(cfg,result)
    assert 'consume' not in state['calls']


def test_mapping_boolean_and_public_constructor_are_not_authority(lane):
    cfg,e,contract,state=lane
    for value in [True,{},object()]:
        with pytest.raises(DecisionError):consume_task_execution(cfg,value)
    with pytest.raises(DecisionError):TaskExecutionAuthorization()
    assert not state['calls']


@pytest.mark.parametrize('field',['approval_id','decision_id','consumed_at','idempotent'])
def test_absent_cas_fields_cannot_fall_back(lane,field):
    cfg,e,contract,state=lane;result=prepared(lane)
    state['response_change']=lambda payload:payload.pop(field)
    with pytest.raises(DecisionError):consume_task_execution(cfg,result)
    assert state['calls'].count('consume')==1


def test_expired_preparation_refuses_before_cas(lane,monkeypatch):
    from types import SimpleNamespace
    import secrets_engine.task_execution_handoff as module
    cfg,e,contract,state=lane;result=prepared(lane)
    later=int((datetime.now(timezone.utc)+timedelta(minutes=2)).timestamp()*1_000_000_000)
    monkeypatch.setattr(module,'read_window',lambda cfg:SimpleNamespace(lower_ns=later,upper_ns=later))
    with pytest.raises(DecisionError):consume_task_execution(cfg,result)
    assert 'consume' not in state['calls']


@pytest.mark.parametrize('change',['deny','expired','missing-caller','missing-start'])
def test_signed_decision_missing_authority_refuses_before_cas(lane,change):
    cfg,e,contract,state=lane
    def alter(envelope):
        if change=='deny':envelope['effect']='deny'
        elif change=='expired':envelope['lifetime']['expires_at']=stamp(datetime.now(timezone.utc)-timedelta(seconds=1))
        elif change=='missing-start':envelope['lifetime'].pop('not_before')
        else:envelope['provenance'].pop('caller')
    state['decision_change']=alter
    with pytest.raises(DecisionError):prepared(lane)
    assert 'consume' not in state['calls']


def test_native_binding_equality_does_not_coerce_boolean_enrichment(lane):
    cfg,e,contract,state=lane
    evaluated=_expected_request(cfg,e,'execute-ccr',execution_contract=contract)
    evaluated['resource']['attributes']['synthetic_rank']=1
    native=binding_from_check_request(evaluated)
    state['claim']['binding'].update(copy.deepcopy(native),digest=claim_binding_digest(**native),
                                     pdp_digest=approval_binding_digest(evaluated))
    # Python dict equality treats True == 1. This malformed provider fact must
    # refuse even when its reported native digest names the integer counterpart.
    state['claim']['binding']['target']['attributes']['synthetic_rank']=True
    def enrich(envelope):
        envelope['resource']['attributes']['synthetic_rank']=1
        envelope['binding']['request_digest']=request_digest(envelope['binding'])
        envelope['binding']['approval_binding_digest']=approval_binding_digest(envelope['binding'])
    state['decision_change']=enrich
    with pytest.raises(DecisionError):prepared(lane)
    assert 'consume' not in state['calls']