secrets-engine/tests/test_decision_signature.py

"""Producer-signed native replay and real consumer boundary regression checks."""
import copy
import json
from pathlib import Path
import subprocess
from types import SimpleNamespace

import pytest

from secrets_engine import approval_consume
from secrets_engine.authorization import validate_decision_envelope
from secrets_engine.decision_check import check_decision
from secrets_engine.decision_signature import load_trust
from secrets_engine.errors import DecisionError
from tests.test_decision_transport import server, credential, clear_proxies

FIXTURE = Path(__file__).parent / 'fixtures/pdp-signature/decision_rotate_signed.json'
PUBLIC = 'IVL40Zt5HSRFMkLhXy6rbLfP-ntqXtMAl5YOBpiB2xI'

from tests.signature_support import signature_tools, admit

@pytest.fixture(scope='module')
def verifier(signature_tools):
    return signature_tools/'verify'

@pytest.fixture
def trust_file(tmp_path):
    envelope = json.loads(FIXTURE.read_bytes())
    provenance = envelope['provenance']
    trust = {k:provenance[k] for k in ('evaluator','mode','policy_package','policy_version','policy_package_digest')}
    trust.update(schema='secrets-engine.pdp-trust.v1', keys={'testdata-ed25519':PUBLIC})
    path = tmp_path/'trust.json'; path.write_text(json.dumps(trust)); return admit(path)


def checked(body, credential, trust_file, verifier):
    with server(body=body) as (origin, received):
        result = check_decision(base_url=origin, token_file=credential, request={},
                                trust_file=trust_file, verifier=verifier)
        assert received == [True]
        return result


def test_genuine_native_signature_at_consumer_boundary(credential, trust_file, verifier):
    result = checked(FIXTURE.read_bytes(),credential,trust_file,verifier)
    assert result['effect']=='allow'
    assert result['binding']==json.loads(FIXTURE.read_bytes())['binding']

@pytest.mark.parametrize('change', ['allow_forgery','tampered','unknown_key','unsigned','absent','algorithm','unknown_field','duplicate','null','case_alias'])
def test_response_authentication_refuses(credential, trust_file, verifier, change):
    env=json.loads(FIXTURE.read_bytes())
    if change=='allow_forgery': env['signature']['value']='A'*86
    elif change=='tampered': env['binding']['action']='destroy'
    elif change=='unknown_key': env['signature']['kid']='new-key'
    elif change=='unsigned': env['signature']={'mode':'unsigned'}
    elif change=='absent': del env['signature']
    elif change=='algorithm': env['signature']['alg']='none'
    elif change=='unknown_field': env['unbound_authority']='allow'
    elif change=='null': env['binding']=None
    elif change=='case_alias': env['Effect']='deny'
    body=json.dumps(env).encode()
    if change=='duplicate': body=body[:-1]+b',"effect":"deny"}'
    with pytest.raises(DecisionError): checked(body,credential,trust_file,verifier)

@pytest.mark.parametrize('field',['evaluator','mode','policy_package','policy_version','policy_package_digest'])
def test_independent_issuer_and_pin_refusal(credential, trust_file, verifier, field):
    trust=json.loads(trust_file.read_text()); trust[field]='wrong'; trust_file.write_text(json.dumps(trust)); admit(trust_file)
    with pytest.raises(DecisionError): checked(FIXTURE.read_bytes(),credential,trust_file,verifier)


def test_rotation_overlap_then_removal(credential, trust_file, verifier, signature_tools):
    trust=json.loads(trust_file.read_text()); trust['keys']['next-admitted']='Ivwpd5Lwtv_Av8_bftsMCqFOAlo2XsDjQuhuOCnLdLY'
    trust_file.write_text(json.dumps(trust)); admit(trust_file); checked(FIXTURE.read_bytes(),credential,trust_file,verifier)
    from tests.signature_support import sign_envelope
    next_body=json.dumps(sign_envelope(json.loads(FIXTURE.read_bytes()),signature_tools,next_key=True)).encode()
    checked(next_body,credential,trust_file,verifier)
    del trust['keys']['testdata-ed25519']; trust_file.write_text(json.dumps(trust)); admit(trust_file)
    with pytest.raises(DecisionError): checked(FIXTURE.read_bytes(),credential,trust_file,verifier)
    checked(next_body,credential,trust_file,verifier)


def test_missing_trust_precedes_credential_read(tmp_path):
    with pytest.raises(DecisionError,match='trust'):
        check_decision(base_url='http://127.0.0.1:1',token_file=tmp_path/'must-not-read',request={})


def test_production_authorization_has_no_unsigned_path(monkeypatch, credential, trust_file, verifier):
    observation=SimpleNamespace(request={})
    monkeypatch.setattr(approval_consume,'resolve_approval_observation',lambda *a,**kw:observation)
    with server(body=b'{"effect":"allow"}') as (origin,received):
        cfg=SimpleNamespace(pdp_url=origin,pdp_token_file=credential,pdp_trust_file=trust_file,pdp_verifier=verifier,
                            authorization_policy_package='package',authorization_policy_version='v1')
        with pytest.raises(DecisionError,match='authentication'):
            approval_consume.authorize_action(cfg,None,'rotate')
        assert received == [True]


def test_missing_trust_precedes_approval_observation(monkeypatch):
    def forbidden(*a,**kw): raise AssertionError('approval credential path reached')
    monkeypatch.setattr(approval_consume,'resolve_approval_observation',forbidden)
    with pytest.raises(DecisionError,match='trust'):
        approval_consume.authorize_action(SimpleNamespace(pdp_url='http://127.0.0.1:1',pdp_token_file=Path('/must-not-read')),None,'rotate')

@pytest.mark.parametrize('body', [b'{"effect":NaN}',b'{"effect":Infinity}',b'['*2000+b'0'+b']'*2000,b'{"x":1,"x":2}',b'\xff'])
def test_malformed_body_refuses_as_decision_error(credential,trust_file,verifier,body):
    with pytest.raises(DecisionError): checked(body,credential,trust_file,verifier)


def test_default_check_requires_authentication_before_credential_read(tmp_path):
    with pytest.raises(DecisionError,match='trust'):
        check_decision(base_url='http://127.0.0.1:1',token_file=tmp_path/'must-not-read',request={})