import copy import json import unittest from pathlib import Path from cryptography.hazmat.primitives.asymmetric import ec from odexa_ref import crypto from profiles import delegation as d NOW = '2026-09-17T00:00:00Z' BEFORE = '2026-09-16T00:00:00Z' AFTER = '2026-09-18T00:00:00Z' ORIGIN = 'https://publisher.example' PROVIDER = 'https://provider.example' def fixture(external=True): pol = {'protocol_version': '1.2.0-draft.1', 'policy_id': ORIGIN + '/policies/main', 'revision': 1, 'origin': ORIGIN, 'issued_at': BEFORE, 'expires_at': AFTER, 'default': 'no_grant', 'resources': [{'id': 'article', 'selectors': [{'type': 'exact', 'path': '/about', 'query': 'absent'}]}], 'rules': [{'id': 'read', 'resource_ids': ['article'], 'actions': ['retrieve', 'store'], 'purposes': ['public_retrieval', 'internal_knowledge'], 'effect': 'require_agreement', 'obligations': [{'type': 'attribution', 'name': 'Example originator', 'url': ORIGIN + '/about'}]}]} host = PROVIDER if external else ORIGIN key = {'kid': host + '/keys/one', 'public_jwk': crypto.public_jwk(ec.derive_private_key(1, ec.SECP256R1())), 'uses': ['odexa-receipt+jws', 'odexa-status+jws'], 'not_before': BEFORE, 'not_after': AFTER, 'state': 'active', 'retired_at': None, 'revoked_at': None} scope = {'resource_ids': ['article'], 'actions': ['retrieve', 'store'], 'purposes': ['public_retrieval', 'internal_knowledge'], 'asset_id_prefixes': [], 'version_id_prefixes': []} service = {'id': ORIGIN + '/services/one', 'base_url': host + '/tenant-a/api/', 'issuer': host + '/operator', 'capabilities': ['issue_agreements', 'issue_tokens', 'introspect'], 'signing_keys': [key], 'limits': {'max_offer_seconds': 600, 'max_access_seconds': 3600, 'max_use_seconds': 7200, 'payment_mode': 'none'}, 'scope': copy.deepcopy(scope)} grant = {'id': ORIGIN + '/delegations/one', 'origin': ORIGIN, 'delegate_service_id': service['id'], 'capabilities': list(service['capabilities']), **copy.deepcopy(scope), 'issued_at': BEFORE, 'expires_at': AFTER, 'max_access_seconds': 300, 'max_use_seconds': 600, 'payment_mode': 'none', 'allow_subdelegation': False} authority = {'protocol_version': d.VERSION, 'policy_protocol_version': '1.2.0-draft.1', 'origin': ORIGIN, 'policy_id': pol['policy_id'], 'revision': 1, 'issued_at': BEFORE, 'expires_at': AFTER, 'services': [service], 'delegations': [grant] if external else []} operation = {'operation_id': '12345678-1234-4234-8234-123456789abc', 'payment_request_digest': None, 'kind': 'offer', 'origin': ORIGIN, 'service_id': service['id'], 'issuer': service['issuer'], 'endpoint': service['base_url'] + 'offers', 'delegation_id': grant['id'] if external else None, 'key_id': key['kid'], 'key_use': 'odexa-receipt+jws', 'request': {'url': ORIGIN + '/about', 'actions': ['retrieve'], 'purposes': ['public_retrieval'], 'supported_obligations': ['attribution']}, 'offer_seconds': 60, 'access_seconds': 300, 'use_seconds': 600, 'payment_mode': 'none', 'asset_id': None, 'version_id': None} return authority, pol, operation def observation(authority, pol, source='https'): return {'source': source, 'origin': ORIGIN, 'authority_url': ORIGIN + '/odexa-service.json', 'policy_url': ORIGIN + '/odexa.json', 'authority_digest': d.digest(crypto.json_bytes(authority)), 'policy_digest': d.digest(crypto.json_bytes(pol)), 'checked_at': NOW, 'available': True, 'tls_verified': source == 'https', 'redirected': False, 'revalidated': True} class DelegationProfileTests(unittest.TestCase): def test_status_and_revoke_are_exact_agreement_bound_zero_duration_operations(self): self.setUp() aid='00000000-0000-4000-8000-000000000001' for kind in ('status','revoke'): self.op.update(kind=kind,agreement_id=aid,key_use='odexa-status+jws',offer_seconds=0, access_seconds=0,use_seconds=0,endpoint=self.authority['services'][0]['base_url']+'agreements/'+aid+'/'+kind) result=self.evaluate() self.assertEqual(result['decision'],'allow',result) self.assertEqual(result['authority']['agreement_id'],aid) self.assertFalse(result['resource_permission_granted']) original=self.op['endpoint'];self.op['endpoint']=original.replace(aid,'00000000-0000-4000-8000-000000000002') self.denied('endpoint_binding');self.op['endpoint']=original self.op['access_seconds']=1;self.denied('invalid_input');self.op['access_seconds']=0 del self.op['agreement_id'];self.denied('invalid_input') def test_legacy_operation_rejects_unexpected_agreement_id(self): self.setUp();self.op['agreement_id']='00000000-0000-4000-8000-000000000001' self.denied('invalid_input') def setUp(self): self.authority, self.pol, self.op = fixture() def evaluate(self, obs=None, **kwargs): return d.evaluate_authority(crypto.json_bytes(self.authority), crypto.json_bytes(self.pol), self.op, now=NOW, observation=obs or observation(self.authority, self.pol), **kwargs) def denied(self, reason=None, obs=None): result = self.evaluate(obs) self.assertEqual(result['decision'], 'deny', result) self.assertFalse(result['resource_permission_granted']) self.assertIsNone(result['authority']) if reason: self.assertEqual(result['reason'], reason, result) def test_external_zero_price_and_exact_context(self): result = self.evaluate() self.assertEqual(result['decision'], 'allow', result) self.assertFalse(result['resource_permission_granted']) self.assertEqual(result['authority']['endpoint'], PROVIDER + '/tenant-a/api/offers') self.assertEqual(result['authority']['operation_id'], self.op['operation_id']) self.assertEqual(result['authority']['operation_digest'], d.digest(crypto.json_bytes(self.op))) self.assertEqual(result['authority']['request'], self.op['request']) self.assertEqual(result['authority']['request_digest'], d.digest(crypto.json_bytes(self.op['request']))) self.assertEqual(result['policy_decision'], 'require_agreement') self.assertEqual(result['obligations'], self.pol['rules'][0]['obligations']) def test_same_origin_no_delegate_and_local_observation(self): self.authority, self.pol, self.op = fixture(False) self.assertEqual(self.evaluate(observation(self.authority, self.pol, 'colocated'))['decision'], 'allow') def test_external_requires_named_grant_and_remote_observation(self): self.denied('remote_observation_required', observation(self.authority, self.pol, 'colocated')) self.op['delegation_id'] = None self.denied('delegation_required') def test_no_union_of_partial_delegations(self): first = self.authority['delegations'][0] second = copy.deepcopy(first); second['id'] = ORIGIN + '/delegations/two' first['actions'] = ['retrieve']; second['actions'] = ['store'] self.authority['delegations'].append(second) self.op['request']['actions'] = ['retrieve', 'store'] self.denied('scope_exceeded') def test_provider_tenant_endpoint_issuer_key_substitution(self): for field, value, reason in [('endpoint', PROVIDER + '/tenant-b/api/offers', 'endpoint_binding'), ('issuer', PROVIDER + '/other-operator', 'service_binding'), ('service_id', ORIGIN + '/services/other', 'service_binding'), ('key_id', PROVIDER + '/keys/attacker', 'key_binding')]: with self.subTest(field=field): old = self.op[field]; self.op[field] = value; self.denied(reason); self.op[field] = old def test_unknown_capability_subdelegation_and_scope_rejected(self): for mutate in [lambda a: a['delegations'][0].update(allow_subdelegation=True), lambda a: a['delegations'][0]['capabilities'].append('admin'), lambda a: a['delegations'][0]['resource_ids'].append('unknown')]: with self.subTest(mutate=mutate): saved = copy.deepcopy(self.authority); mutate(self.authority); self.denied('invalid_input'); self.authority = saved def test_prohibit_and_every_action_purpose_pair(self): self.op['request']['actions'] = ['retrieve', 'store'] self.op['request']['purposes'] = ['public_retrieval', 'internal_knowledge'] self.pol['rules'].append({'id': 'no-internal-store', 'resource_ids': ['article'], 'actions': ['store'], 'purposes': ['internal_knowledge'], 'effect': 'prohibit', 'obligations': []}) self.denied('policy_denied') def test_duties_not_weakened(self): self.op['request']['supported_obligations'] = [] self.denied('policy_denied') def test_all_overlapping_policy_scopes_must_be_covered(self): self.pol['resources'].append({'id': 'whole-site', 'selectors': [{'type': 'directory_prefix', 'path': '/', 'query': 'any'}]}) self.denied('scope_exceeded') def test_duration_and_paid_mode_bounds(self): self.op['access_seconds'] = 301 self.denied('duration_exceeded') self.op['access_seconds'] = 300; self.op['payment_mode'] = 'external' self.denied('payment_not_authorized') def test_offer_lifetime_bound_by_current_authority_and_key(self): self.authority['services'][0]['signing_keys'][0]['not_after'] = '2026-09-17T00:00:30Z' self.denied('offer_lifetime_exceeded') def test_key_retirement_revocation_and_wrong_use(self): key = self.authority['services'][0]['signing_keys'][0] key.update(state='retired', retired_at=NOW) self.denied('key_inactive') key.update(state='revoked', revoked_at=NOW) self.denied('key_inactive') key.update(state='active', retired_at=None, revoked_at=None, uses=['odexa-status+jws']) self.denied('key_binding') def test_grant_removal_revokes_next_operation(self): prior = self.evaluate(); self.assertEqual(prior['decision'], 'allow') self.authority['delegations'] = []; self.authority['revision'] = 2 self.denied('delegation_binding') self.assertEqual(prior['authority']['authority_revision'], 1) def test_current_time_boundaries(self): for checked, allowed in [('2026-09-16T23:59:55Z', True), ('2026-09-16T23:59:54Z', False), ('2026-09-17T00:00:01Z', False)]: with self.subTest(checked=checked): obs = observation(self.authority, self.pol); obs['checked_at'] = checked self.assertEqual(self.evaluate(obs)['decision'], 'allow' if allowed else 'deny') def test_outage_redirect_tls_cached_and_wrong_bytes_fail_closed(self): for field, value, reason in [('available', False, 'authority_unavailable'), ('redirected', True, 'authority_not_current'), ('tls_verified', False, 'unverified_transport'), ('revalidated', False, 'authority_not_current'), ('authority_digest', 'sha256:' + '0' * 64, 'observation_digest'), ('policy_url', PROVIDER + '/odexa.json', 'observation_binding')]: with self.subTest(field=field): obs = observation(self.authority, self.pol); obs[field] = value; self.denied(reason, obs) def test_expiry_equality_denied(self): self.authority['delegations'][0]['expires_at'] = NOW self.denied('delegation_expired') def test_exact_bytes_duplicate_members_and_numeric_tokens(self): raw = crypto.json_bytes(self.authority) for changed in [raw.replace(b'"revision":1', b'"revision":1.0'), raw.replace(b'"revision":1', b'"revision":-0'), raw.replace(b'"revision":1', b'"revision":1,"revision":1')]: with self.subTest(raw=changed[:80]): result = d.evaluate_authority(changed, crypto.json_bytes(self.pol), self.op, now=NOW, observation=observation(self.authority, self.pol)) self.assertEqual(result['decision'], 'deny') result = d.evaluate_authority(raw + b' ', crypto.json_bytes(self.pol), self.op, now=NOW, observation=observation(self.authority, self.pol)) self.assertEqual(result['reason'], 'observation_digest') def test_report_only_service_no_agreement_or_payment_authority(self): service = self.authority['services'][0]; grant = self.authority['delegations'][0] service['capabilities'] = grant['capabilities'] = ['receive_events']; service['limits'] = None service['signing_keys'][0]['uses'] = ['odexa-event-record+jws'] grant['max_access_seconds'] = grant['max_use_seconds'] = 0 self.op.update(kind='report', endpoint=service['base_url'] + 'events', key_use='odexa-event-record+jws', offer_seconds=0, access_seconds=0, use_seconds=0) self.pol['rules'][0].update(effect='prohibit', obligations=[]) result = self.evaluate(); self.assertEqual(result['decision'], 'allow', result) self.assertEqual(result['policy_decision'], 'prohibit') self.assertFalse(result['resource_permission_granted']) self.op.update(kind='accept', endpoint=service['base_url'] + 'agreements', key_use='odexa-receipt+jws', offer_seconds=60, access_seconds=300, use_seconds=600) self.denied('capability_denied') def test_unsigned_token_operation_still_bound(self): self.op.update(kind='token', endpoint=self.authority['services'][0]['base_url'] + 'tokens', key_id=None, key_use=None, offer_seconds=0) result = self.evaluate(); self.assertEqual(result['decision'], 'allow', result) self.assertFalse(result['resource_permission_granted']) def test_asset_prefix_and_version_are_independently_bounded(self): service = self.authority['services'][0]; grant = self.authority['delegations'][0] service['capabilities'] = grant['capabilities'] = ['publish_assets']; service['limits'] = None service['signing_keys'][0]['uses'] = ['odexa-asset-manifest+jws'] grant['max_access_seconds'] = grant['max_use_seconds'] = 0 for scope in [service['scope'], grant]: scope['asset_id_prefixes'] = [ORIGIN + '/assets/']; scope['version_id_prefixes'] = [ORIGIN + '/versions/'] self.op.update(kind='publish_manifest', endpoint=ORIGIN + '/versions/one', key_use='odexa-asset-manifest+jws', offer_seconds=0, access_seconds=0, use_seconds=0, asset_id=ORIGIN + '/assets/article', version_id=ORIGIN + '/versions/one') self.assertEqual(self.evaluate()['decision'], 'allow') self.op['asset_id'] = ORIGIN + '/assets-evil/article'; self.denied('asset_scope') self.op['asset_id'] = ORIGIN + '/assets/article'; self.op['version_id'] = ORIGIN + '/other/one'; self.op['endpoint'] = self.op['version_id']; self.denied('asset_scope') def test_payment_decision_binds_operation_and_exact_check_digest(self): service = self.authority['services'][0]; grant = self.authority['delegations'][0] service['capabilities'] = grant['capabilities'] = ['verify_payments']; service['limits'] = None service['signing_keys'][0]['uses'] = ['odexa-payment-status+jws'] grant.update(max_access_seconds=0, max_use_seconds=0, payment_mode='external') self.op.update(kind='verify_payment', endpoint=service['base_url'] + 'verify', key_use='odexa-payment-status+jws', offer_seconds=0, access_seconds=0, use_seconds=0, payment_mode='external', payment_request_digest='sha256:' + 'a' * 64) prior = self.evaluate(); self.assertEqual(prior['decision'], 'allow', prior) self.assertEqual(prior['authority']['payment_request_digest'], self.op['payment_request_digest']) self.op['payment_request_digest'] = 'sha256:' + 'b' * 64 current = self.evaluate(); self.assertNotEqual(current['authority']['operation_digest'], prior['authority']['operation_digest']) self.op['payment_request_digest'] = None; self.denied('invalid_input') def test_migration_requires_new_descriptor_grant_and_current_observation(self): prior = self.evaluate() self.authority['services'][0]['base_url'] = 'https://replacement.example/tenant-a/api/' self.authority['revision'] = 2 self.denied('endpoint_binding') self.op['endpoint'] = self.authority['services'][0]['base_url'] + 'offers' result = self.evaluate(); self.assertEqual(result['decision'], 'allow', result) self.assertNotEqual(result['authority']['authority_digest'], prior['authority']['authority_digest']) self.assertNotEqual(result['authority']['operation_digest'], prior['authority']['operation_digest']) def test_does_not_mutate_input(self): old = copy.deepcopy((self.authority, self.pol, self.op)) self.evaluate(); self.assertEqual(old, (self.authority, self.pol, self.op)) def test_categorized_semantic_fixture_corpus(self): corpus = json.loads((Path(__file__).parents[1] / 'profiles/fixtures/delegation-cases.json').read_text()) for case in corpus['cases']: if 'semantic_valid' not in case: continue with self.subTest(case=case['name']): try: if case['definition'] == 'authority': d.validate_authority(case['value'], corpus['policy_document']) else: d.validate_operation(case['value']) valid = True except (ValueError, TypeError, KeyError): valid = False self.assertEqual(valid, case['semantic_valid']) if __name__ == '__main__': unittest.main()