railiance-platform/tests/test_task_execution_mediator.py

"""Whole source-server/native fixture proof, never production owner acceptance."""
from contextlib import contextmanager
from dataclasses import replace
import importlib
import json
from pathlib import Path
import sys
from types import SimpleNamespace
import uuid

import pytest
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts'))
from test_task_execution_native_fixture import owner,bao_dev,compiler
from task_execution_mediator import ConfirmationConfig,run_native_fixture_task
from task_execution_journal import Refused
from task_execution_source_servers import approval_server,reviewer_server,access_server,http,private
from secrets_engine.approval_consume import _request_purpose
from secrets_engine.approval_claim import binding_from_check_request
from secrets_engine.authorization import build_action_request
from secrets_engine.decision_check import check_decision
from secrets_engine.execution_contract import ExecutionContract


@contextmanager
def whole_fixture(owner,tmp_path,*,actual_policy=False,confirmation_read=True):
    native,journal,binding,call,compiler,descriptor=owner
    snapshot={path:compiler.read(Path(path)) for path in compiler.pins}
    entry=compiler._catalogue_entry(snapshot)
    approval_id=str(uuid.uuid4())
    entry=replace(entry,approval={**entry.approval,'authorization_id':approval_id})
    request=build_action_request(entry,'execute-ccr',subject_id='secrets-engine',subject_type='service',
                                 purpose=_request_purpose(entry),execution_contract=ExecutionContract.parse(descriptor))
    with reviewer_server(tmp_path) as reviewer, access_server(tmp_path,descriptor,request,reviewer,actual_policy=actual_policy) as access, approval_server(tmp_path) as approval:
        issue=access.offline_envelope
        body={'id':approval_id,'binding':binding_from_check_request(issue['binding']),
              'pdp_digest':issue['binding']['request_digest'],'pdp_path':True,'human_control':True,
              'validity':{'not_before':'2026-01-01T00:00:00+00:00',
                          'expires_at':reviewer_expiry()}}
        status,obj=http(approval.origin,'POST','/v1/approvals',body,approval.service_token)
        assert status==201
        assert http(approval.origin,'POST','/v1/approvals/'+approval_id+'/entries',{},approval.human_token)[0]==200
        approval_token=private(tmp_path/'approval.token',approval.service_token)
        pdp_token=private(tmp_path/'pdp.token',reviewer.caller_token)
        confirmation_token=private(tmp_path/'confirmation.token',approval.service_token if confirmation_read else approval.no_read_token)
        cfg=SimpleNamespace(approval_url=approval.origin,approval_token_file=approval_token,
                            approval_client_secret_file=None,pdp_url=access.origin,pdp_token_file=pdp_token,
                            pdp_trust_file=access.trustfile,pdp_verifier=access.verifier,
                            authorization_subject_id='secrets-engine',authorization_subject_type='service',
                            authorization_policy_package=access.package,authorization_policy_version=access.version,
                            clock_trust_file=None)
        yield SimpleNamespace(native=native,journal=journal,call=call,compiler=compiler,
                              descriptor=descriptor,entry=entry,cfg=cfg,approval=approval,
                              confirmation=ConfirmationConfig(approval.origin,confirmation_token),
                              approval_id=approval_id,request=request,reviewer=reviewer)


def reviewer_expiry():
    from datetime import datetime,timedelta,timezone
    return (datetime.now(timezone.utc)+timedelta(minutes=10)).isoformat()


def invoke(fixture):
    return run_native_fixture_task(compiler=fixture.compiler,journal=fixture.journal,
                plan_request=fixture.native.meta['request'],owner_descriptor=fixture.native.admission.descriptor,
                secrets_cfg=fixture.cfg,entry=fixture.entry,confirmation=fixture.confirmation,
                native_admission=fixture.native.admission)


def test_real_source_servers_signed_fixture_allow_and_actual_task_policy_deny(owner,tmp_path):
    # These two Checks test real source servers and independently installed keys;
    # they are not a fabricated DecisionEnvelope or activation of a real policy.
    for actual_policy in [False,True]:
        directory=tmp_path/('actual-policy' if actual_policy else 'fixture-policy');directory.mkdir()
        with whole_fixture(owner,directory,actual_policy=actual_policy) as fixture:
            status,claim=http(fixture.approval.origin,'GET','/v1/approvals/'+fixture.approval_id+'/claim',token=fixture.approval.service_token)
            assert status==200
            request=json.loads(json.dumps(fixture.request));request['context']['approval']=claim
            envelope=check_decision(base_url=fixture.cfg.pdp_url,token_file=fixture.cfg.pdp_token_file,request=request,
                                    trust_file=fixture.cfg.pdp_trust_file,verifier=fixture.cfg.pdp_verifier)
            assert envelope['effect']==('deny' if actual_policy else 'allow')
            assert envelope['provenance']['caller']['mode']=='enforce'
            assert all(fixture.reviewer.calls)
            assert fixture.approval.engine.get(fixture.approval_id).status=='approved'
            assert fixture.call('GET','sys/policies/acl/'+fixture.native.policy_name)[0]==404


def test_missing_strict_dependency_never_reads_credentials_or_reserves(monkeypatch):
    original=importlib.import_module
    def missing(name,*args,**kwargs):
        if name=='secrets_engine.task_execution_handoff':raise ImportError()
        return original(name,*args,**kwargs)
    monkeypatch.setattr(importlib,'import_module',missing)
    with pytest.raises(Refused,match='dependency unavailable'):
        run_native_fixture_task(compiler=None,journal=None,plan_request={},owner_descriptor=b'',
                                secrets_cfg=None,entry=None,confirmation=None,native_admission=None)


def test_whole_strict_consume_confirmation_precedes_real_native_effect(owner,tmp_path):
    with whole_fixture(owner,tmp_path) as fixture:
        result=invoke(fixture)
        assert result['native_calls_completed']==6 and result['cleanup_confirmed']
        assert result['native_apply']=='verified'
        assert result['business_state']=='cleanup-confirmed'
        assert len(result['readback_evidence_digest'])==64
        obj=fixture.approval.engine.get(fixture.approval_id)
        assert obj.status=='consumed'
        assert obj.consumed_digest==result['request_digest']
        calls=fixture.journal.db.execute('SELECT ordinal,outcome FROM calls ORDER BY ordinal').fetchall()
        assert calls[:2]==[(0,'complete'),(1,'complete')]
        assert len(calls)==13 # CAS + GET + five native grant calls + six metadata calls
        assert fixture.approval.calls.count(('POST','/v1/approvals/'+fixture.approval_id+'/consume'))==1
        assert fixture.approval.calls.count(('GET','/v1/approvals/'+fixture.approval_id))==1
        assert fixture.call('GET','sys/policies/acl/'+fixture.native.policy_name)[0]==404
        raw=(fixture.journal.directory/'journal.sqlite').read_bytes()
        for token in [fixture.native.admission.owner_token,fixture.native.admission.identity_session.token,
                      fixture.approval.service_token,fixture.reviewer.caller_token]:
            assert token.encode() not in raw
        with pytest.raises(Exception):invoke(fixture)
        assert fixture.approval.calls.count(('POST','/v1/approvals/'+fixture.approval_id+'/consume'))==1


def test_actual_confirmation_read_scope_denial_keeps_native_untouched(owner,tmp_path):
    with whole_fixture(owner,tmp_path,confirmation_read=False) as fixture:
        with pytest.raises(Refused,match='confirmation unconfirmed'):invoke(fixture)
        assert fixture.approval.engine.get(fixture.approval_id).status=='consumed'
        state=fixture.journal.db.execute('SELECT state FROM operations').fetchone()[0]
        assert state=='unknown'
        assert fixture.journal.db.execute("SELECT COUNT(*) FROM calls WHERE outcome='pending'").fetchone()[0]==1
        assert fixture.call('GET','sys/policies/acl/'+fixture.native.policy_name)[0]==404
        policy_path=fixture.compiler.before_call(fixture.native.plan,1).path
        assert fixture.call('GET',policy_path)[0]==404
        with pytest.raises(Exception):invoke(fixture)
        assert fixture.approval.calls.count(('POST','/v1/approvals/'+fixture.approval_id+'/consume'))==1


def test_actual_unadmitted_policy_refuses_before_cas_and_native(owner,tmp_path):
    with whole_fixture(owner,tmp_path,actual_policy=True) as fixture:
        with pytest.raises(Exception):invoke(fixture)
        assert fixture.approval.engine.get(fixture.approval_id).status=='approved'
        assert not any(method=='POST' and path.endswith('/consume') for method,path in fixture.approval.calls)
        assert fixture.journal.db.execute('SELECT COUNT(*) FROM operations').fetchone()[0]==0
        assert fixture.call('GET','sys/policies/acl/'+fixture.native.policy_name)[0]==404


def test_entry_purpose_substitution_refuses_before_authorizer_or_cas(owner,tmp_path):
    with whole_fixture(owner,tmp_path) as fixture:
        fixture.entry=replace(fixture.entry,approval={**fixture.entry.approval,'purpose':'caller-substituted-purpose'})
        before=list(fixture.approval.calls)
        with pytest.raises(Refused,match='approval metadata differs'):invoke(fixture)
        assert fixture.approval.calls==before
        assert fixture.journal.db.execute('SELECT COUNT(*) FROM operations').fetchone()[0]==0
        assert fixture.call('GET','sys/policies/acl/'+fixture.native.policy_name)[0]==404


def test_catalogue_bool_integer_alias_refuses_before_authorizer(owner,tmp_path):
    with whole_fixture(owner,tmp_path) as fixture:
        assert fixture.entry.delivery_auth['secret_id_num_uses']==1
        fixture.entry=replace(fixture.entry,delivery_auth={**fixture.entry.delivery_auth,'secret_id_num_uses':True})
        before=list(fixture.approval.calls)
        with pytest.raises(Refused,match='catalogue does not match'):invoke(fixture)
        assert fixture.approval.calls==before
        assert fixture.journal.db.execute('SELECT COUNT(*) FROM operations').fetchone()[0]==0