"""Draft-3 delegated operational authority; pure, no network or access issuance.""" import copy import hashlib import json import re from odexa_ref import policy, crypto VERSION = '1.2.0-draft.3' CAPABILITIES = {'issue_agreements', 'issue_tokens', 'introspect', 'receive_events', 'verify_payments', 'publish_assets', 'export_evidence'} USES = set(crypto.TYPES) - {'odexa-acceptance+jws'} | {'odexa-payment-status+jws', 'odexa-evidence-export+jws'} GRANT_CAPS = {'issue_agreements', 'issue_tokens', 'introspect'} ACCESS_KINDS = {'offer', 'accept', 'token', 'introspect'} KINDS = { 'offer': ('issue_agreements', 'offers', 'odexa-receipt+jws'), 'accept': ('issue_agreements', 'agreements', 'odexa-receipt+jws'), 'token': ('issue_tokens', 'tokens', None), 'introspect': ('introspect', 'introspect', None), 'report': ('receive_events', 'events', 'odexa-event-record+jws'), 'verify_payment': ('verify_payments', 'verify', 'odexa-payment-status+jws'), 'publish_manifest': ('publish_assets', None, 'odexa-asset-manifest+jws'), 'export_evidence': ('export_evidence', 'exports', 'odexa-evidence-export+jws'), 'payment_mandate': ('issue_agreements', 'payment-mandate', None), 'payment_check': ('issue_agreements', 'payment-check', None), 'status': ('issue_agreements', None, 'odexa-status+jws'), 'revoke': ('issue_agreements', None, 'odexa-status+jws'), } KEY_CAPS = {'odexa-receipt+jws': 'issue_agreements', 'odexa-status+jws': 'issue_agreements', 'odexa-event+jws': 'receive_events', 'odexa-event-record+jws': 'receive_events', 'odexa-asset-manifest+jws': 'publish_assets', 'odexa-payment-status+jws': 'verify_payments', 'odexa-evidence-export+jws': 'export_evidence'} AGREEMENT_KINDS = {'status', 'revoke', 'export_evidence', 'payment_mandate', 'payment_check'} MAX_DURATION = 31536000 need, keys = policy.need, policy.keys def digest(raw): return 'sha256:' + hashlib.sha256(raw).hexdigest() def _json(raw): need(isinstance(raw, bytes), 'json', 'exact UTF-8 bytes required') return crypto.strict_json(raw, 131072) def _time(value): return policy.timestamp(value, 'time') def _url(value, *, origin=None, prefix=False): p = policy.parse_url(value) need(not p['has_query'] and not p['has_fragment'], 'url', 'query and fragment forbidden') need(value == p['origin'] + p['path'], 'url', 'canonical absolute HTTPS URL required') if origin is not None: need(p['origin'] == origin, 'url', 'wrong origin') if prefix: need(p['path'].endswith('/'), 'url', 'directory prefix must end in slash') return p def _interval(start, end): need(_time(start) < _time(end), 'interval', 'empty or reversed validity') def _duration(value, minimum=0, maximum=MAX_DURATION): policy.integer(value, 'duration', minimum) need(value <= maximum, 'duration', 'duration exceeds profile maximum') def _ids(values, allowed, path, minimum=1): policy.array(values, path, minimum, 256) need(all(isinstance(v, str) and v in allowed for v in values), path, 'unknown scope value') need(len(set(values)) == len(values), path, 'duplicate scope value') def _prefixes(values, origin): policy.array(values, 'prefixes', 0, 64) need(len(set(values)) == len(values), 'prefixes', 'duplicate prefix') for value in values: _url(value, origin=origin, prefix=True) def _scope(scope, pol): keys(scope, {'resource_ids', 'actions', 'purposes', 'asset_id_prefixes', 'version_id_prefixes'}, 'scope') _ids(scope['resource_ids'], {r['id'] for r in pol['resources']}, 'resource_ids') policy.enum_array(scope['actions'], policy.ACTIONS, 'actions') policy.enum_array(scope['purposes'], policy.PURPOSES, 'purposes') for name in ('asset_id_prefixes', 'version_id_prefixes'): _prefixes(scope[name], pol['origin']) def _keys(service): policy.array(service['signing_keys'], 'signing_keys', 1, 16) seen = set() for key in service['signing_keys']: keys(key, {'kid', 'public_jwk', 'uses', 'not_before', 'not_after', 'state', 'retired_at', 'revoked_at'}, 'key') _url(key['kid']) need(key['kid'] not in seen, 'kid', 'duplicate key identifier') seen.add(key['kid']) crypto.load_public_jwk(key['public_jwk']) policy.enum_array(key['uses'], USES, 'uses') need(all(KEY_CAPS[u] in service['capabilities'] for u in key['uses']), 'uses', 'key use exceeds service capability') _interval(key['not_before'], key['not_after']) need(key['state'] in {'active', 'retired', 'revoked'}, 'state', 'unknown key state') if key['state'] == 'active': need(key['retired_at'] is None and key['revoked_at'] is None, 'state', 'active key has terminal time') elif key['state'] == 'retired': need(key['revoked_at'] is None and _time(key['not_before']) <= _time(key['retired_at']), 'state', 'invalid retirement') else: need(_time(key['not_before']) <= _time(key['revoked_at']), 'state', 'invalid revocation') if key['retired_at'] is not None: need(_time(key['not_before']) <= _time(key['retired_at']) <= _time(key['revoked_at']), 'state', 'invalid retirement/revocation order') def validate_authority(value, policy_document): """Validate the full authority document structurally/semantically, not freshness.""" pol = policy.validate_policy(policy_document) keys(value, {'protocol_version', 'policy_protocol_version', 'origin', 'policy_id', 'revision', 'issued_at', 'expires_at', 'services', 'delegations'}, 'authority') need(value['protocol_version'] == VERSION and value['policy_protocol_version'] == policy.VERSION, 'version', 'unsupported version') need(value['origin'] == pol['origin'] and value['policy_id'] == pol['policy_id'], 'authority', 'policy identity mismatch') policy.origin(value['origin']); policy.integer(value['revision'], 'revision', 1) _interval(value['issued_at'], value['expires_at']) policy.array(value['services'], 'services', 1, 16) services = {} key_material = {} for service in value['services']: keys(service, {'id', 'base_url', 'issuer', 'capabilities', 'signing_keys', 'limits', 'scope'}, 'service') _url(service['id'], origin=pol['origin']); _url(service['base_url'], prefix=True); _url(service['issuer']) need(service['id'] not in services, 'service.id', 'duplicate service') services[service['id']] = service policy.enum_array(service['capabilities'], CAPABILITIES, 'capabilities') _scope(service['scope'], pol); _keys(service) has_grants = bool(set(service['capabilities']) & GRANT_CAPS) if has_grants: keys(service['limits'], {'max_offer_seconds', 'max_access_seconds', 'max_use_seconds', 'payment_mode'}, 'limits') _duration(service['limits']['max_offer_seconds'], 1, 600) _duration(service['limits']['max_access_seconds'], 1) _duration(service['limits']['max_use_seconds'], 1) need(service['limits']['payment_mode'] in {'none', 'external'}, 'payment_mode', 'unsupported mode') else: need(service['limits'] is None, 'limits', 'non-grant service limits must be null') if 'publish_assets' in service['capabilities']: need(all(service['scope'][p] for p in ('asset_id_prefixes', 'version_id_prefixes')), 'prefixes', 'manifest scope prefixes required') else: need(not service['scope']['asset_id_prefixes'] and not service['scope']['version_id_prefixes'], 'prefixes', 'asset prefixes without capability') for key in service['signing_keys']: material = json.dumps(key['public_jwk'], sort_keys=True) need(key['kid'] not in key_material or key_material[key['kid']] == material, 'kid', 'key material differs under one identifier') key_material[key['kid']] = material policy.array(value['delegations'], 'delegations', 0, 64) seen = set() for grant in value['delegations']: keys(grant, {'id', 'origin', 'delegate_service_id', 'capabilities', 'resource_ids', 'actions', 'purposes', 'issued_at', 'expires_at', 'max_access_seconds', 'max_use_seconds', 'payment_mode', 'allow_subdelegation', 'asset_id_prefixes', 'version_id_prefixes'}, 'delegation') _url(grant['id'], origin=pol['origin']) need(grant['id'] not in seen, 'delegation.id', 'duplicate delegation') seen.add(grant['id']) need(grant['origin'] == pol['origin'] and grant['delegate_service_id'] in services, 'delegation', 'origin/service mismatch') need(grant['allow_subdelegation'] is False, 'allow_subdelegation', 'subdelegation forbidden') policy.enum_array(grant['capabilities'], CAPABILITIES, 'capabilities') service = services[grant['delegate_service_id']] need(set(grant['capabilities']) <= set(service['capabilities']), 'capabilities', 'delegation exceeds service') scope = {k: grant[k] for k in service['scope']} _scope(scope, pol) for field in ('resource_ids', 'actions', 'purposes'): need(set(grant[field]) <= set(service['scope'][field]), field, 'delegation exceeds service scope') _interval(grant['issued_at'], grant['expires_at']) need(_time(value['issued_at']) <= _time(grant['issued_at']) < _time(grant['expires_at']) <= _time(value['expires_at']), 'delegation', 'validity outside metadata') grant_caps = bool(set(grant['capabilities']) & GRANT_CAPS) for field in ('max_access_seconds', 'max_use_seconds'): _duration(grant[field], 1 if grant_caps else 0) need(grant[field] <= service['limits'][field] if grant_caps else grant[field] == 0, field, 'duration exceeds service or non-grant mode') need(grant['payment_mode'] in {'none', 'external'}, 'payment_mode', 'unsupported mode') if grant['payment_mode'] == 'external': need('verify_payments' in grant['capabilities'] or (grant_caps and service['limits']['payment_mode'] == 'external'), 'payment_mode', 'external payment authority absent') for field in ('asset_id_prefixes', 'version_id_prefixes'): if 'publish_assets' in grant['capabilities']: need(bool(grant[field]), field, 'manifest prefixes required') need(all(any(prefix.startswith(parent) for parent in service['scope'][field]) for prefix in grant[field]), field, 'delegation exceeds service prefix') else: need(not grant[field], field, 'asset prefixes without capability') return copy.deepcopy(value) def validate_operation(value): expected = {'operation_id', 'payment_request_digest', 'kind', 'origin', 'service_id', 'issuer', 'endpoint', 'delegation_id', 'key_id', 'key_use', 'request', 'offer_seconds', 'access_seconds', 'use_seconds', 'payment_mode', 'asset_id', 'version_id'} if isinstance(value, dict) and value.get('kind') in AGREEMENT_KINDS: expected.add('agreement_id') keys(value, expected, 'operation') need(isinstance(value['operation_id'], str) and bool(re.fullmatch(r'[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}', value['operation_id'])), 'operation_id', 'canonical UUIDv4 required') need(value['kind'] in KINDS, 'kind', 'unknown operation') if value['kind'] in AGREEMENT_KINDS: need(isinstance(value['agreement_id'], str) and bool(re.fullmatch(r'[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}', value['agreement_id'])), 'agreement_id', 'canonical UUIDv4 required') policy.origin(value['origin']); _url(value['service_id'], origin=value['origin']); _url(value['issuer']); _url(value['endpoint']) if value['delegation_id'] is not None: _url(value['delegation_id'], origin=value['origin']) request = value['request'] keys(request, {'url', 'actions', 'purposes', 'supported_obligations'}, 'request') parsed = policy.parse_url(request['url']) canonical = parsed['origin'] + parsed['path'] + (('?' + parsed['parsed'].query) if parsed['has_query'] else '') need(not parsed['has_fragment'] and request['url'] == canonical and parsed['origin'] == value['origin'], 'request.url', 'noncanonical or wrong origin') policy.enum_array(request['actions'], policy.ACTIONS, 'actions'); policy.enum_array(request['purposes'], policy.PURPOSES, 'purposes') policy.enum_array(request['supported_obligations'], policy.DUTIES, 'supported_obligations', 0) capability, _, required_use = KINDS[value['kind']] if required_use is None: need(value['key_id'] is None and value['key_use'] is None, 'key', 'unsigned operation has no signing key') else: _url(value['key_id']); need(value['key_use'] == required_use, 'key_use', 'wrong message signing type') grants = value['kind'] in ACCESS_KINDS _duration(value['offer_seconds'], 1 if value['kind'] in {'offer', 'accept'} else 0, 600) if value['kind'] not in {'offer', 'accept'}: need(value['offer_seconds'] == 0, 'offer_seconds', 'non-offer lifetime must be zero') for field in ('access_seconds', 'use_seconds'): _duration(value[field], 1 if grants else 0) if not grants: need(value[field] == 0, field, 'non-grant duration must be zero') if grants: need(value['access_seconds'] <= value['use_seconds'], 'duration', 'access exceeds use window') need(value['payment_mode'] in {'none', 'external'}, 'payment_mode', 'unsupported mode') if value['kind'] == 'verify_payment': need(value['payment_mode'] == 'external', 'payment_mode', 'verification requires external mode') need(isinstance(value['payment_request_digest'], str) and bool(re.fullmatch(r'sha256:[0-9a-f]{64}', value['payment_request_digest'])), 'payment_request_digest', 'payment check digest required') else: need(value['payment_request_digest'] is None, 'payment_request_digest', 'unexpected payment check digest') if value['kind'] in {'report', 'publish_manifest', 'status', 'revoke', 'export_evidence'}: need(value['payment_mode'] == 'none', 'payment_mode', 'operation grants no payment authority') if value['kind'] in {'payment_mandate', 'payment_check'}: need(value['payment_mode'] == 'external', 'payment_mode', 'paid operation requires external mode') if value['kind'] == 'publish_manifest': _url(value['asset_id'], origin=value['origin']); _url(value['version_id'], origin=value['origin']) need(value['asset_id'] != value['version_id'], 'asset', 'asset and version identifiers must differ') else: need(value['asset_id'] is None and value['version_id'] is None, 'asset', 'unexpected asset publication identifiers') return value def evaluate_authority(authority_bytes, policy_bytes, operation, *, now, observation): """Allow/deny operational authority, never an access token or spending mandate. `observation` must come from the trusted caller's transport/transaction adapter, never the requesting agent. This pure function does not verify that testimony. """ deny = lambda reason: {'protocol_version': VERSION, 'decision': 'deny', 'reason': reason, 'resource_permission_granted': False, 'authority': None} try: keys(observation, {'source', 'origin', 'authority_url', 'policy_url', 'authority_digest', 'policy_digest', 'checked_at', 'available', 'tls_verified', 'redirected', 'revalidated'}, 'observation') for field in ('available', 'tls_verified', 'redirected', 'revalidated'): need(type(observation[field]) is bool, field, 'boolean required') if not observation['available']: return deny('authority_unavailable') op = validate_operation(operation); instant = _time(now); checked = _time(observation['checked_at']) pol = policy.validate_policy(_json(policy_bytes)); metadata = validate_authority(_json(authority_bytes), pol) if op['origin'] != metadata['origin']: return deny('origin_mismatch') if observation['origin'] != op['origin'] or observation['authority_url'] != op['origin'] + '/odexa-service.json' or observation['policy_url'] != op['origin'] + '/odexa.json': return deny('observation_binding') if observation['authority_digest'] != digest(authority_bytes) or observation['policy_digest'] != digest(policy_bytes): return deny('observation_digest') if observation['source'] not in {'colocated', 'https'}: return deny('observation_source') if not 0 <= (instant - checked).total_seconds() <= 5: return deny('authority_stale') if not observation['revalidated'] or observation['redirected']: return deny('authority_not_current') if observation['source'] == 'https' and not observation['tls_verified']: return deny('unverified_transport') if not _time(metadata['issued_at']) <= instant < _time(metadata['expires_at']): return deny('authority_expired') if not _time(pol['issued_at']) <= instant < _time(pol['expires_at']): return deny('policy_expired') service = next((s for s in metadata['services'] if s['id'] == op['service_id']), None) if service is None or service['issuer'] != op['issuer']: return deny('service_binding') capability, suffix, required_use = KINDS[op['kind']] if op['kind'] == 'publish_manifest': endpoint = op['version_id'] elif op['kind'] in {'status', 'revoke', 'payment_mandate', 'payment_check'}: endpoint = service['base_url'] + 'agreements/' + op['agreement_id'] + '/' + op['kind'].replace('_', '-') else: endpoint = service['base_url'] + suffix if op['endpoint'] != endpoint: return deny('endpoint_binding') if capability not in service['capabilities']: return deny('capability_denied') external = _url(service['base_url'])['origin'] != op['origin'] if external and observation['source'] != 'https': return deny('remote_observation_required') grant = None if op['delegation_id'] is not None: grant = next((d for d in metadata['delegations'] if d['id'] == op['delegation_id']), None) if grant is None or grant['delegate_service_id'] != service['id']: return deny('delegation_binding') if not _time(grant['issued_at']) <= instant < _time(grant['expires_at']): return deny('delegation_expired') if capability not in grant['capabilities']: return deny('delegation_capability') elif external: return deny('delegation_required') if required_use: key = next((k for k in service['signing_keys'] if k['kid'] == op['key_id']), None) if key is None or required_use not in key['uses']: return deny('key_binding') if key['state'] != 'active' or not _time(key['not_before']) <= instant < _time(key['not_after']): return deny('key_inactive') if op['kind'] in {'offer', 'accept'}: ceilings = [service['limits']['max_offer_seconds'], (_time(metadata['expires_at']) - instant).total_seconds(), (_time(pol['expires_at']) - instant).total_seconds(), (_time(key['not_after']) - instant).total_seconds()] if grant: ceilings.append((_time(grant['expires_at']) - instant).total_seconds()) if op['offer_seconds'] > min(ceilings): return deny('offer_lifetime_exceeded') decision = policy.evaluate(pol, op['request'], now) if decision['decision'] in {'invalid_policy', 'invalid_request', 'expired', 'not_yet_valid'}: return deny('invalid_policy_scope') selected = set(decision['matched_resources']) if not selected: return deny('resource_scope') for scope in [service['scope']] + ([grant] if grant else []): if not selected <= set(scope['resource_ids']) or not set(op['request']['actions']) <= set(scope['actions']) or not set(op['request']['purposes']) <= set(scope['purposes']): return deny('scope_exceeded') if op['kind'] == 'publish_manifest': if not all(any(op[field].startswith(prefix) for prefix in scope[field + '_prefixes']) for field in ('asset_id', 'version_id')): return deny('asset_scope') if capability in GRANT_CAPS: for limits in [service['limits']] + ([grant] if grant else []): if op['access_seconds'] > limits['max_access_seconds'] or op['use_seconds'] > limits['max_use_seconds']: return deny('duration_exceeded') if op['payment_mode'] == 'external' and limits['payment_mode'] != 'external': return deny('payment_not_authorized') if op['kind'] in {'payment_mandate', 'payment_check'}: if service['limits']['payment_mode'] != 'external' or (grant and grant['payment_mode'] != 'external'): return deny('payment_not_authorized') if op['kind'] == 'verify_payment' and grant and grant['payment_mode'] != 'external': return deny('payment_not_authorized') if op['kind'] in {'offer', 'accept'} and decision['decision'] not in {'permit', 'require_agreement'}: return deny('policy_denied') context = {'operation_id': op['operation_id'], 'operation_digest': digest(crypto.json_bytes(op)), 'payment_request_digest': op['payment_request_digest'], 'request': copy.deepcopy(op['request']), 'request_digest': digest(crypto.json_bytes(op['request'])), 'origin': op['origin'], 'service_id': service['id'], 'issuer': service['issuer'], 'capability': capability, 'endpoint': op['endpoint'], 'key_id': op['key_id'], 'key_use': op['key_use'], 'delegation_id': op['delegation_id'], 'authority_revision': metadata['revision'], 'authority_digest': digest(authority_bytes), 'policy_digest': digest(policy_bytes), 'checked_at': observation['checked_at']} if op['kind'] in AGREEMENT_KINDS: context['agreement_id'] = op['agreement_id'] return {'protocol_version': VERSION, 'decision': 'allow', 'reason': 'within_operational_authority', 'resource_permission_granted': False, 'authority': context, 'policy_decision': decision['decision'], 'obligations': decision['obligations'], 'obligation_status': 'not_verified'} except (ValueError, TypeError, KeyError, OverflowError) as error: return {**deny('invalid_input'), 'detail': str(error)[:256]}