"""Paid-aware cross-message validation; no free-envelope rewriting.""" import copy import json import sys import unittest from odexa_ref import crypto from profiles import paid_contracts as p, free_contracts as f from profiles import delegation as d from test_free_contracts import fixtures as free_fixtures, later, NOW, uid from test_payment_profile import fixtures as payment_fixtures from test_delegation_profile import fixture as delegation_fixture, observation def fixtures(): values,_=free_fixtures(); quote,_,_,_=payment_fixtures() offer=values['offer']; quote.update(origin=offer['origin'],service_id=offer['service_id'],offer_id=offer['offer_id'], verification_service_id=offer['origin']+'/services/payments',issued_at=offer['issued_at'],expires_at=offer['expires_at']) raw=crypto.json_bytes(quote) values['offer_request']['payer_id']=quote['payer_id'] offer['payment']=dict(required=True,quote_digest=crypto.digest(raw),quote_url='https://provider.example/tenant-a/api/contexts/'+crypto.digest(raw)[7:]) values['acceptance']['offer_digest']=crypto.digest(crypto.json_bytes(offer)) values['receipt'].update(state_at_issue='pending_payment',payment_required=True, offer_digest=values['acceptance']['offer_digest'],acceptance_digest=crypto.digest(crypto.json_bytes(values['acceptance']))) values['status']['state']='pending_payment' return values,raw def schema_cases(): values,_=fixtures() cases=[dict(name='paid_valid_'+name,definition=name,value=copy.deepcopy(value),structural_valid=True,semantic_valid=True) for name,value in values.items()] def bad(name,definition,change,structural=False): value=copy.deepcopy(values[definition]);change(value) cases.append(dict(name=name,definition=definition,value=value,structural_valid=structural,semantic_valid=False)) bad('paid_request_missing_payer','offer_request',lambda v:v.pop('payer_id')) bad('paid_offer_without_quote','offer',lambda v:v.update(payment={'required':True})) bad('paid_offer_free_flag','offer',lambda v:v['payment'].update(required=False)) bad('paid_receipt_free_state','receipt',lambda v:v.update(state_at_issue='active')) bad('paid_receipt_free_flag','receipt',lambda v:v.update(payment_required=False)) bad('paid_receipt_bearer','receipt',lambda v:v.update(access_credential=True)) bad('paid_acceptance_unknown_field','acceptance',lambda v:v.update(payer_approved=True)) bad('paid_status_unknown_state','status',lambda v:v.update(state='settled')) bad('paid_wrong_version','receipt',lambda v:v.update(protocol_version='1.2.0-draft.2')) # These remain structurally valid: their failure requires original contexts. bad('paid_receipt_extends_offer','receipt',lambda v:v.update(access_expires_at=later(301)),True) bad('paid_receipt_commit_at_expiry','receipt',lambda v:v.update(recorded_at=later(60),access_expires_at=later(360),use_expires_at=later(660)),True) bad('paid_acceptance_wrong_nonce','acceptance',lambda v:v.update(nonce=crypto.b64u(b'x'*32)),True) bad('paid_token_other_agreement','token_response',lambda v:v.update(agreement_id=uid(99)),True) bad('paid_request_noncanonical_path','offer_request',lambda v:v['request'].update(url=v['origin']+'/a%2Fb'),True) bad('paid_status_active_after_expiry','status',lambda v:v.update(state='active',issued_at=later(300),authority_checked_at=later(300)),True) return dict(schema='paid-agreement',cases=cases) class PaidContractsTests(unittest.TestCase): def setUp(self): self.v,self.quote=fixtures();self.offer=crypto.json_bytes(self.v['offer']);self.acceptance=crypto.json_bytes(self.v['acceptance']) def test_full_paid_context_validates_without_changing_free_or_crypto(self): before=set(crypto.TYPES); version=f.VERSION p.validate_offer(self.v['offer'],self.v['offer_request']);p.bind_quote(self.v['offer'],self.quote) p.validate_acceptance(self.v['acceptance'],self.offer,NOW) p.validate_receipt(self.v['receipt'],offer_bytes=self.offer,acceptance_bytes=self.acceptance) p.validate_token_response(self.v['token_response'],self.v['receipt'],NOW) p.validate_introspection_response(self.v['introspection_response'],self.v['introspection_request'],self.v['receipt'],self.v['token_response']) p.validate_status(self.v['status'],self.v['receipt']) with self.assertRaises(ValueError):f.validate_offer(self.v['offer']) with self.assertRaises(ValueError):f.validate_receipt(self.v['receipt']) self.assertEqual(set(crypto.TYPES),before);self.assertEqual(f.VERSION,version) def test_acceptance_exact_nonce_principal_scope_and_offer_bytes(self): for field,value in [('nonce',crypto.b64u(b'x'*32)),('principal_id','https://other.example/principal'), ('client_id','other'),('service_id',self.v['binding']['origin']+'/services/other'), ('offer_digest','sha256:'+'0'*64),('offer_id',uid(99))]: with self.subTest(field=field),self.assertRaises(ValueError): p.validate_acceptance(dict(self.v['acceptance'],**{field:value}),self.offer,NOW) with self.assertRaises(ValueError):p.validate_acceptance(self.v['acceptance'],self.offer+b' ',NOW) def test_acceptance_interval_and_clock_are_explicit(self): for when in [later(-1),later(60),later(61)]: with self.subTest(when=when),self.assertRaises(ValueError): p.validate_acceptance(dict(self.v['acceptance'],accepted_at=when),self.offer,later(62)) with self.assertRaises(ValueError):p.validate_acceptance(dict(self.v['acceptance'],accepted_at=later(1)),self.offer,NOW) def test_receipt_must_preserve_exact_access_and_use_durations(self): for field,value in [('access_expires_at',later(301)),('use_expires_at',later(601)), ('access_expires_at',later(299)),('use_expires_at',later(599))]: with self.subTest(field=field,value=value),self.assertRaises(ValueError): p.validate_receipt(dict(self.v['receipt'],**{field:value}),offer_bytes=self.offer,acceptance_bytes=self.acceptance) with self.assertRaises(ValueError): p.validate_receipt(dict(self.v['receipt'],use_expires_at=later(f.MAX_DURATION+1))) def test_receipt_commit_must_fit_offer_and_follow_assent(self): for when in [-1,60,61]: receipt=dict(self.v['receipt'],recorded_at=later(when),access_expires_at=later(when+300),use_expires_at=later(when+600)) with self.subTest(when=when),self.assertRaises(ValueError): p.validate_receipt(receipt,offer_bytes=self.offer,acceptance_bytes=self.acceptance) acceptance=dict(self.v['acceptance'],accepted_at=later(1));raw=crypto.json_bytes(acceptance) with self.assertRaises(ValueError): p.validate_receipt(dict(self.v['receipt'],acceptance_digest=crypto.digest(raw)),offer_bytes=self.offer,acceptance_bytes=raw) def test_receipt_does_not_skip_offer_acceptance_nonce_chain(self): acceptance=dict(self.v['acceptance'],nonce=crypto.b64u(b'x'*32));raw=crypto.json_bytes(acceptance) receipt=dict(self.v['receipt'],acceptance_digest=crypto.digest(raw)) with self.assertRaises(ValueError):p.validate_receipt(receipt,offer_bytes=self.offer,acceptance_bytes=raw) with self.assertRaises(ValueError):p.validate_receipt(self.v['receipt'],offer_bytes=self.offer,acceptance_bytes=self.acceptance+b' ') def test_offer_request_context_cannot_change_scope_or_principal(self): for field,value in [('principal_id','https://other.example/principal'),('access_seconds',301),('use_seconds',601), ('request',dict(self.v['offer']['request'],url=self.v['offer']['origin']+'/other'))]: with self.subTest(field=field),self.assertRaises(ValueError): p.validate_offer(dict(self.v['offer'],**{field:value}),self.v['offer_request']) def test_token_bound_to_paid_receipt_and_lifetime(self): for field,value in [('agreement_id',uid(99)),('expires_at',later(301)),('issuer','https://other.example/operator')]: with self.subTest(field=field),self.assertRaises(ValueError): p.validate_token_response(dict(self.v['token_response'],**{field:value}),self.v['receipt'],NOW) with self.assertRaises(ValueError):p.validate_token_response(dict(self.v['token_response'],expires_at=NOW),self.v['receipt'],NOW) def test_introspection_binds_paid_principal_agreement_token_and_request(self): for field,value in [('agreement_id',uid(99)),('client_id','other'),('principal_id','https://other.example/principal'), ('resource_url',self.v['binding']['origin']+'/other'),('expires_at',later(121))]: response=copy.deepcopy(self.v['introspection_response']);response['admission'][field]=value with self.subTest(field=field),self.assertRaises(ValueError): p.validate_introspection_response(response,self.v['introspection_request'],self.v['receipt'],self.v['token_response']) def test_paid_denial_has_no_admission_or_false_activation(self): for active,reason in [(False,'inactive_agreement'),(False,'invalid_token'),(True,'scope_mismatch')]: denied=dict(self.v['binding'],request_id=self.v['introspection_request']['request_id'],active=active,permitted=False,reason=reason,admission=None) p.validate_introspection_response(denied,self.v['introspection_request'],self.v['receipt'],self.v['token_response']) with self.assertRaises(ValueError):p.validate_introspection_response(dict(denied,admission=self.v['introspection_response']['admission'])) def test_status_paid_context_and_expiry_are_correlated(self): for changes in [dict(agreement_id=uid(99)),dict(effective_at=later(-1)), dict(issued_at=later(300),authority_checked_at=later(300)), dict(state='expired',issued_at=later(300),authority_checked_at=later(300),effective_at=later(299))]: with self.subTest(changes=changes),self.assertRaises(ValueError): p.validate_status(dict(self.v['status'],**changes),self.v['receipt']) p.validate_status(dict(self.v['status'],state='expired',issued_at=later(300),authority_checked_at=later(300),effective_at=later(300)),self.v['receipt']) def test_paid_and_free_receipts_cannot_be_interchanged(self): values,_=free_fixtures() with self.assertRaises(ValueError):p.validate_receipt(values['receipt']) for changed in [dict(self.v['receipt'],payment_required=False),dict(self.v['receipt'],state_at_issue='active'), dict(self.v['receipt'],access_credential=True)]: with self.assertRaises(ValueError):p.validate_receipt(changed) class PaidOperationalAuthorityTests(unittest.TestCase): def context(self,kind): authority,policy,op=delegation_fixture();service=authority['services'][0];grant=authority['delegations'][0] op.update(kind=kind,agreement_id=uid(90),offer_seconds=0,access_seconds=0,use_seconds=0, key_id=None,key_use=None,payment_mode='external') service['limits']['payment_mode']=grant['payment_mode']='external' op['endpoint']=service['base_url']+'agreements/'+op['agreement_id']+'/'+kind.replace('_','-') if kind=='export_evidence': service['capabilities'].append('export_evidence');grant['capabilities'].append('export_evidence') key=service['signing_keys'][0];key['uses'].append('odexa-evidence-export+jws') op.update(endpoint=service['base_url']+'exports',key_id=key['kid'],key_use='odexa-evidence-export+jws',payment_mode='none') return authority,policy,op def evaluate(self,authority,policy,op): return d.evaluate_authority(crypto.json_bytes(authority),crypto.json_bytes(policy),op,now=NOW, observation=observation(authority,policy)) def test_new_operations_have_exact_positive_agreement_context_without_access_grant(self): for kind in ('payment_mandate','payment_check','export_evidence'): authority,policy,op=self.context(kind);result=self.evaluate(authority,policy,op) self.assertEqual(result['decision'],'allow',result) self.assertEqual(result['authority']['agreement_id'],op['agreement_id']) self.assertFalse(result['resource_permission_granted']) def test_new_operations_reject_outside_scope_wrong_endpoint_and_new_duration(self): for kind in ('payment_mandate','payment_check','export_evidence'): authority,policy,op=self.context(kind) for change in [dict(endpoint=op['endpoint']+'/other'),dict(access_seconds=1), dict(request=dict(op['request'],url=policy['origin']+'/unlisted')), dict(request=dict(op['request'],actions=['redistribute'])), dict(request=dict(op['request'],purposes=['model_training']))]: with self.subTest(kind=kind,change=change): self.assertEqual(self.evaluate(authority,policy,dict(op,**change))['decision'],'deny') missing=dict(op);missing.pop('agreement_id') self.assertEqual(self.evaluate(authority,policy,missing)['decision'],'deny') if kind!='export_evidence': self.assertEqual(self.evaluate(authority,policy,dict(op,agreement_id=uid(91)))['decision'],'deny') def test_paid_operations_cannot_downgrade_mode_or_exceed_service_payment_permission(self): for kind in ('payment_mandate','payment_check'): authority,policy,op=self.context(kind) self.assertEqual(self.evaluate(authority,policy,dict(op,payment_mode='none'))['decision'],'deny') authority['delegations'][0]['payment_mode']='none' self.assertEqual(self.evaluate(authority,policy,op)['decision'],'deny') authority['services'][0]['limits']['payment_mode']='none' self.assertEqual(self.evaluate(authority,policy,op)['decision'],'deny') def test_export_requires_distinct_capability_key_use_and_no_payment_authority(self): authority,policy,op=self.context('export_evidence') self.assertEqual(self.evaluate(authority,policy,dict(op,payment_mode='external'))['decision'],'deny') self.assertEqual(self.evaluate(authority,policy,dict(op,key_use='odexa-receipt+jws'))['decision'],'deny') authority['delegations'][0]['capabilities'].remove('export_evidence') self.assertEqual(self.evaluate(authority,policy,op)['decision'],'deny') authority,policy,op=self.context('export_evidence') key=authority['services'][0]['signing_keys'][0];key.update(state='revoked',revoked_at=NOW) self.assertEqual(self.evaluate(authority,policy,op)['decision'],'deny') if __name__=='__main__': if '--cases' in sys.argv: print(json.dumps(schema_cases(),indent=2)) else: unittest.main()