tegwick
4dd004a225
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786
80 lines
4.2 KiB
Python
80 lines
4.2 KiB
Python
"""Owner signature interoperability, independent synthetic admission only."""
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
|
|
import pytest
|
|
from secrets_engine.decision_signature import authenticate_response,load_trust
|
|
from secrets_engine.errors import DecisionError
|
|
from tests.test_decision_signature import FIXTURE,trust_file,verifier
|
|
from tests.signature_support import signature_tools
|
|
from tests.test_task_execution_handoff import lane,prepared,consume_task_execution
|
|
|
|
VECTORS=json.loads((Path(__file__).parent/'fixtures/pdp-signature/approval-native-vectors.json').read_text())
|
|
|
|
@pytest.mark.parametrize('vector',VECTORS)
|
|
def test_fourteen_owner_native_types_retain_signed_bytes(vector,trust_file,signature_tools):
|
|
env=json.loads(FIXTURE.read_text());env.pop('signature',None)
|
|
env['diagnostics']={'native_vector':'REPLACE_RAW_TARGET'}
|
|
raw=json.dumps(env).replace('"REPLACE_RAW_TARGET"',vector['input_target_json']).encode()
|
|
signed=subprocess.run([str(signature_tools/'sign')],input=raw,capture_output=True,check=True).stdout
|
|
result=authenticate_response(signed,trust=load_trust(trust_file),verifier=signature_tools/'verify')
|
|
again=authenticate_response(result.authenticated_json.encode(),trust=load_trust(trust_file),verifier=signature_tools/'verify')
|
|
assert again['diagnostics']['native_vector']==json.loads(vector['input_target_json'])
|
|
tampered=signed.replace(b'"action":"rotate"',b'"action":"destroy"')
|
|
assert tampered!=signed
|
|
with pytest.raises(DecisionError):authenticate_response(tampered,trust=load_trust(trust_file),verifier=signature_tools/'verify')
|
|
|
|
|
|
def test_unknown_provenance_refuses(trust_file,signature_tools):
|
|
env=json.loads(FIXTURE.read_text());env['provenance']['unknown_source_authority']=True
|
|
with pytest.raises(DecisionError):
|
|
authenticate_response(json.dumps(env).encode(),trust=load_trust(trust_file),verifier=signature_tools/'verify')
|
|
|
|
|
|
def test_exact_original_numeric_lexeme_survives_strict_cas(lane,monkeypatch,signature_tools):
|
|
import tests.test_task_execution_handoff as fixtures
|
|
cfg,e,contract,state=lane
|
|
def raw_sign(env,tools):
|
|
env['diagnostics']={'numeric':'LEXEME'}
|
|
raw=json.dumps(env).replace('"LEXEME"','1e-7').encode()
|
|
return subprocess.run([str(tools/'sign')],input=raw,capture_output=True,check=True).stdout
|
|
monkeypatch.setattr(fixtures,'sign_envelope',raw_sign)
|
|
authorization=prepared(lane)
|
|
assert '1e-7' in authorization._action.artifacts
|
|
assert consume_task_execution(cfg,authorization).idempotent is False
|
|
|
|
|
|
def source_evidence(envelope,claim):
|
|
from datetime import datetime,timezone
|
|
return {'source_id':'synthetic-source','source_revision':'synthetic-revision',
|
|
'configuration_digest':'sha256:'+'2'*64,'claim_digest':'sha256:'+'3'*64,
|
|
'native_binding_digest':claim['binding']['digest'],
|
|
'observed_at':datetime.now(timezone.utc).isoformat(),
|
|
'validated_at':datetime.now(timezone.utc).isoformat(),
|
|
'not_after':envelope['lifetime']['expires_at']}
|
|
|
|
|
|
def test_signed_source_evidence_keeps_original_context_and_cas(lane):
|
|
cfg,e,contract,state=lane
|
|
state['decision_change']=lambda env:env['provenance'].__setitem__('approval_observation',source_evidence(env,state['claim']))
|
|
authorization=prepared(lane)
|
|
snapshot=json.loads(authorization._action.artifacts)
|
|
assert snapshot['envelope']['binding']['context']==snapshot['request']['context']
|
|
assert snapshot['request']['context']['approval']==state['claim']
|
|
assert consume_task_execution(cfg,authorization).idempotent is False
|
|
|
|
|
|
@pytest.mark.parametrize('field,value',[
|
|
('source_id',''),('source_revision',''),('configuration_digest','wrong'),('claim_digest','wrong'),
|
|
('native_binding_digest','sha256:'+'0'*64),('observed_at','2099-01-01T00:00:00Z'),
|
|
('validated_at','2099-01-01T00:00:00Z'),('not_after','2000-01-01T00:00:00Z')])
|
|
def test_signed_invalid_source_evidence_refuses_before_cas(lane,field,value):
|
|
cfg,e,contract,state=lane
|
|
def change(env):
|
|
evidence=source_evidence(env,state['claim']);evidence[field]=value
|
|
env['provenance']['approval_observation']=evidence
|
|
state['decision_change']=change
|
|
with pytest.raises(DecisionError):prepared(lane)
|
|
assert 'consume' not in state['calls']
|