"""Installed-client current-authority boundaries, including actual local TLS. The offline transport records an emission only after the trusted setup callback. The TLS tests separately exercise that callback in the real transport. """ import copy from pathlib import Path import socket import tempfile import threading from types import SimpleNamespace import unittest from unittest import mock from odexa_ref import crypto from odexa_runtime.bootstrap import initialize from odexa_runtime.client import Client from odexa_runtime.runtime import Runtime from odexa_runtime.server import Server from profiles.authority_transport import AuthoritySnapshot, utc_now from profiles.free_service import later AID='12345678-1234-4234-8234-123456789abc' class ClientAuthorityTests(unittest.TestCase): def setUp(self): self.tmp=tempfile.TemporaryDirectory(); self.root=Path(self.tmp.name) initialize(self.root,origin='https://127.0.0.1:9443',service_origin='https://127.0.0.1:9444',paid=True) self.client=Client(self.root) self.metadata=crypto.strict_json((self.root/'public/odexa-service.json').read_bytes()) self.policy=(self.root/'public/odexa.json').read_bytes() self.instant=utc_now(); self.clock=mock.patch('odexa_runtime.client.utc_now',lambda:self.instant);self.clock.start() self.emitted=[];self.setups=[];self.setup_hook=lambda:None;self.response_hook=lambda:None self.client.transport=SimpleNamespace(fetch=self.fetch,request=self.request) def tearDown(self): self.clock.stop();self.tmp.cleanup() def fetch(self,origin): self.assertEqual(origin,self.client.value['origin']) raw=crypto.json_bytes(self.metadata) return AuthoritySnapshot(raw,self.policy,dict(source='https',origin=origin, authority_url=origin+'/odexa-service.json',policy_url=origin+'/odexa.json', authority_digest=crypto.digest(raw),policy_digest=crypto.digest(self.policy), checked_at=self.instant,available=True,tls_verified=True,redirected=False,revalidated=True)) def request(self,origin,path,**kwargs): self.setups.append((origin,path));self.setup_hook() self.assertTrue(callable(kwargs.get('before_send'))) self.assertIs(kwargs['before_send'](),True) self.emitted.append((origin,path,kwargs)) self.response_hook() return SimpleNamespace(body=b'{}') def operation(self,kind): request=dict(url=self.client.value['origin']+'/about',actions=['retrieve'], purposes=['public_retrieval'],supported_obligations=['attribution']) return self.client._operation(kind,request, access=600 if kind in ('offer','accept','token') else 0, use=1200 if kind in ('offer','accept','token') else 0, offer_seconds=60 if kind=='offer' else 1 if kind=='accept' else 0,agreement_id=AID) def send(self,kind): op=self.operation(kind);route=op['endpoint'][len(self.client.value['base_url']):] return self.client.call(route,{} if kind!='status' else None,operation=op, role='payer' if kind=='payment_mandate' else 'agent',method='GET' if kind=='status' else 'POST') def test_every_authenticated_operation_denies_withdrawn_delegation_before_setup(self): self.metadata['delegations']=[] for kind in ('offer','accept','token','status','revoke','payment_mandate'): with self.subTest(kind=kind),self.assertRaisesRegex(ValueError,'delegation_binding'): self.send(kind) self.assertEqual(self.setups,[]);self.assertEqual(self.emitted,[]) def test_narrowed_scope_denies_all_operations_before_setup(self): grant=next(g for g in self.metadata['delegations'] if g['id']==self.client.value['binding']['delegation_id']) # A valid but narrower purpose, still inside the service's advertised scope. self.metadata['services'][0]['scope']['purposes'].append('internal_knowledge') grant['purposes']=['internal_knowledge'] for kind in ('offer','accept','token','status','revoke','payment_mandate'): with self.subTest(kind=kind),self.assertRaisesRegex(ValueError,'scope_exceeded'):self.send(kind) self.assertEqual(self.setups,[]);self.assertEqual(self.emitted,[]) def test_delegation_capability_removed_for_every_operation(self): original=copy.deepcopy(self.metadata) for revision,kind in enumerate(('offer','accept','token','status','revoke','payment_mandate'),1): self.metadata=copy.deepcopy(original) self.metadata['revision']=revision grant=next(g for g in self.metadata['delegations'] if g['id']==self.client.value['binding']['delegation_id']) grant['capabilities'].remove('issue_tokens' if kind=='token' else 'issue_agreements') with self.subTest(kind=kind),self.assertRaisesRegex(ValueError,'delegation_capability'):self.send(kind) self.assertEqual(self.setups,[]);self.assertEqual(self.emitted,[]) def test_service_capability_removed_denies_token_before_setup(self): self.metadata['services'][0]['capabilities'].remove('issue_tokens') self.metadata['delegations'][0]['capabilities'].remove('issue_tokens') with self.assertRaisesRegex(ValueError,'capability_denied'):self.send('token') self.assertEqual(self.setups,[]);self.assertEqual(self.emitted,[]) def test_expired_observation_during_setup_prevents_emission_for_all_routes(self): self.setup_hook=lambda:setattr(self,'instant',later(self.instant,6)) for kind in ('offer','accept','token','status','revoke','payment_mandate'): with self.subTest(kind=kind),self.assertRaisesRegex(ValueError,'authority_stale'):self.send(kind) self.assertEqual(len(self.setups),6);self.assertEqual(self.emitted,[]) def test_locally_observed_withdrawal_during_setup_prevents_emission(self): def withdraw(): self.metadata['revision']+=1;self.metadata['delegations']=[] self.client.history.observe(self.fetch(self.client.value['origin']),now=lambda:self.instant) self.setup_hook=withdraw with self.assertRaises(ValueError):self.send('offer') self.assertEqual(len(self.setups),1);self.assertEqual(self.emitted,[]) def test_expired_assent_deadline_during_setup_prevents_emission(self): op=self.operation('accept');deadline=later(self.instant,1) self.setup_hook=lambda:setattr(self,'instant',later(self.instant,1)) with self.assertRaisesRegex(ValueError,'deadline expired'): self.client.call('agreements',{},operation=op,deadline=deadline) self.assertEqual(self.emitted,[]) def test_response_after_withdrawal_is_not_returned(self): def withdraw():self.metadata.update(revision=2,delegations=[]) self.response_hook=withdraw with self.assertRaisesRegex(ValueError,'delegation_binding'):self.send('status') self.assertEqual(len(self.emitted),1) def test_mismatched_route_cannot_reuse_operation(self): with self.assertRaisesRegex(ValueError,'request/operation binding'): self.client.call('tokens',{},operation=self.operation('offer')) self.assertEqual(self.setups,[]) def test_agree_withdrawal_stops_before_offer_and_writes_no_exchange(self): self.metadata['delegations']=[] with self.assertRaisesRegex(ValueError,'delegation_binding'):self.client.agree(self.root/'denied.json') self.assertEqual(self.setups,[]);self.assertFalse((self.root/'denied.json').exists()) def test_current_key_retirement_prevents_offer_and_current_signature_use(self): service=self.metadata['services'][0];key=service['signing_keys'][0] key.update(state='retired',retired_at=self.instant) with self.assertRaisesRegex(ValueError,'key_inactive'):self.send('offer') with self.assertRaisesRegex(ValueError,'key_inactive'): self.client.service_signature('not-a-jws','odexa-status+jws',operation=self.operation('status')) self.assertEqual(self.setups,[]) def test_authority_outage_has_no_stale_fallback(self): self.client._snapshot() self.client.transport.fetch=mock.Mock(side_effect=ValueError('origin unavailable')) with self.assertRaisesRegex(ValueError,'origin unavailable'):self.send('offer') self.assertEqual(self.setups,[]) def test_valid_scoped_operations_all_have_pre_and_post_checks(self): for kind in ('offer','accept','token','status','revoke','payment_mandate'): with self.subTest(kind=kind):self.assertEqual(self.send(kind),b'{}') self.assertEqual(len(self.emitted),6) for origin,path,kw in self.emitted: self.assertEqual(kw['credential'].origin,origin) self.assertTrue(path.startswith(kw['credential'].path_prefix)) class ClientAuthorityTLS(unittest.TestCase): def setUp(self): self.tmp=tempfile.TemporaryDirectory();self.root=Path(self.tmp.name);self.servers=[] def tearDown(self): for server,thread in reversed(self.servers): server.shutdown();server.server_close();thread.join(timeout=2) self.tmp.cleanup() def start(self,*,delegated=False,paid=False): holders=[] try: for _ in range(2 if delegated else 1): sock=socket.socket();holders.append(sock);sock.bind(('127.0.0.1',0)) origins=['https://127.0.0.1:'+str(s.getsockname()[1]) for s in holders] finally: for sock in holders:sock.close() initialize(self.root,origin=origins[0],service_origin=origins[-1],paid=paid) runtime=Runtime(self.root) for role in ('origin','service') if delegated else ('origin',): server=Server(runtime,role) thread=threading.Thread(target=server.serve_forever,kwargs={'poll_interval':0.01},daemon=True) thread.start();self.servers.append((server,thread)) return runtime,Client(self.root) def flow(self,*,delegated=False,paid=False): runtime,client=self.start(delegated=delegated,paid=paid) path=self.root/'exchange.json';result=client.agree(path) self.assertEqual(result['state'],'pending_payment' if paid else 'active') if paid: client.mandate(path);runtime.verifier_state('confirmed') result=client.retrieve(path,self.root/'body.txt');self.assertGreater(result['bytes_received'],0) self.assertEqual(client.status(path)['state'],'active') self.assertEqual(client.status(path,revoke=True)['state'],'revoked') with self.assertRaises(ValueError):client.retrieve(path,self.root/'after-revoke.txt') self.assertFalse((self.root/'after-revoke.txt').exists()) def test_direct_free_full_installed_client_flow(self):self.flow() def test_delegated_free_full_installed_client_flow(self):self.flow(delegated=True) def test_delegated_paid_full_installed_client_flow(self):self.flow(delegated=True,paid=True) def test_actual_tls_setup_guard_denies_without_provider_http(self): runtime,client=self.start(delegated=True) original=client.transport._perform def delayed(*args,**kwargs): guard=kwargs.get('before_send') if guard is not None: def stale(): with mock.patch('odexa_runtime.client.utc_now',return_value=later(utc_now(),6)): return guard() kwargs['before_send']=stale return original(*args,**kwargs) with mock.patch.object(client.transport,'_perform',side_effect=delayed),mock.patch.object(runtime,'route',wraps=runtime.route) as routed: with self.assertRaises(ValueError):client.agree(self.root/'denied.json') self.assertFalse(any(call.args[0]=='service' for call in routed.call_args_list)) self.assertFalse((self.root/'denied.json').exists()) if __name__=='__main__':unittest.main()