"""Native Node paid client across real Python HTTPS; only the harness imports fixtures.""" import copy import json import os from pathlib import Path import subprocess import unittest from odexa_ref import crypto from profiles.authority_history import AuthorityHistory from profiles.authority_transport import ScopedCredential from profiles.free_gateway import FreeGateway from profiles.paid_store import PaidStore from profiles.paid_service import PaidAgreementService from profiles.free_service import uid import test_paid_service as fixture class IndependentPaidNodeTests(unittest.TestCase): @classmethod def setUpClass(cls): fixture.PaidServiceTests.setUpClass() @classmethod def tearDownClass(cls): fixture.PaidServiceTests.tearDownClass() def setUp(self): self.t=fixture.PaidServiceTests();self.t.setUp();self.f=self.t.f self.steps=[] def tearDown(self): self.t.tearDown() def direct(self): f,t=self.f,self.t service=f.authority['services'][0] main_grant=f.binding['delegation_id'] service.update(base_url=f.origin+'/tenant-a/api/',issuer=f.origin+'/operator') f.authority['delegations']=[g for g in f.authority['delegations'] if g['id']!=main_grant] f.binding.update(issuer=service['issuer'],delegation_id=None) f.base=service['base_url'];f.host=f.origin f.service_args.update(binding=f.binding,base_url=f.base) f.store=PaidStore(f.path/'direct-paid.sqlite3',{k:f.binding[k] for k in ('origin','service_id','issuer')}) f.service=PaidAgreementService(f.store,**f.service_args,quote_factory=t.quote, verifier_delegation_id=t.verifier_grant['id'],verifier_credential=t.verify_credential) for role,key in [('agent',f.agent_key),('gateway',f.gateway_key),('admin',None)]: c=f.clients[role] f.service.provision_client(role,role[0]*43,[f.principal],role=role,public_jwk=crypto.public_jwk(key) if key else None, key_id=c['key_id'],reporter_id=c['reporter_id']) f.service.provision_payer('payer','p'*43,t.quote_payer,key_id=t.payer_kid,public_jwk=crypto.public_jwk(t.payer_key)) t.gateway=FreeGateway(f.path/'direct-gateway.sqlite3',binding=f.binding,base_url=f.base,transport=f.transport, history=AuthorityHistory(f.path/'direct-gateway-history.sqlite3'),clock=f.now, credential=ScopedCredential(f.origin,'/tenant-a/api/',f.clients['gateway']['authorization'])) def configure(self): f,t=self.f,self.t def jwk(key): return dict(crypto.public_jwk(key),d=crypto.b64u(key.private_numbers().private_value.to_bytes(32,'big'))) self.cfg=dict(publisherOrigin=f.origin,serviceId=f.binding['service_id'],baseUrl=f.base,delegationId=f.binding['delegation_id'], caFile=str(f.tls_path/'ca.pem'),request=f.request,accessSeconds=120,useSeconds=240, expectedTermsDigest=crypto.digest(f.service.terms_bytes),expectedResourceDigest=crypto.digest(t.asset), agent=dict(clientId='agent',principalId=f.principal,password='a'*43,keyId=f.clients['agent']['key_id'], reporterId=f.clients['agent']['reporter_id'],privateJwk=jwk(f.agent_key)), payer=dict(clientId='payer',payerId=t.quote_payer,password='p'*43,keyId=t.payer_kid, privateJwk=jwk(t.payer_key),publicJwk=crypto.public_jwk(t.payer_key)), verifier=dict(serviceId=t.verifier_id,providerId=t.verifier['issuer'],endpoint=t.verifier['base_url']+'verify', delegationId=t.verifier_grant['id'],authorization=t.verify_credential.authorization), approvedQuote=dict(currency='USD',minor_unit_scale=2,total_minor='125',items=[ dict(item_id='resource',kind='resource_license',payee_id=f.origin+'/payee',amount_minor='100',description='Resource licence'), dict(item_id='service',kind='provider_service',payee_id=f.provider+'/payee',amount_minor='25',description='Payment management')])) self.config=f.path/'paid-node-config.json';self.config.write_bytes(crypto.json_bytes(self.cfg));self.config.chmod(0o600) self.state=f.path/'paid-node-state.json' def phase(self,name,*,fail=False,check_id=None): node=os.environ.get('ODEXA_NODE','node') args=[node,'verification/paid-client.mjs','--config',str(self.config),'--state',str(self.state),'--phase',name] if check_id: args+=['--check-id',check_id] response=subprocess.run(args,capture_output=True,text=True,timeout=20) self.assertTrue(response.stdout,response.stderr) result=json.loads(response.stdout) if fail: self.assertNotEqual(response.returncode,0,result);self.assertTrue(result.get('failed'),result) else:self.assertEqual(response.returncode,0,result) self.steps.append(dict(phase=name,expected_failure=fail,result=result)) return result def save_report(self,name): configured=os.environ.get('ODEXA_PAID_NODE_REPORT_DIR') if configured: out=Path(configured);out.mkdir(parents=True,exist_ok=True) (out/(name+'.json')).write_text(json.dumps(dict(scenario=name,phases=self.steps,count=len(self.steps), separate_node_process_per_phase=True,python_imports_in_client=False, actual_tls=any(server.requests for server,_ in self.f.servers)),indent=2)+'\n') def lifecycle(self,direct): if direct:self.direct() self.configure() self.phase('observe');self.phase('agree');self.phase('acceptance-retry') self.phase('retrieve',fail=True) self.phase('mandate-agent-credential');self.phase('mandate-agent-signature') self.phase('mandate');self.phase('mandate') self.phase('retrieve',fail=True) checks=[] for state in ('pending','failed','confirmed'): self.t.provider_state=state;check=uid();checks.append(check) result=self.phase('verify',check_id=check);self.assertEqual(result['payment_state'],state) self.phase('service-check',check_id=check) if state!='confirmed':self.phase('retrieve',fail=True) result=self.phase('retrieve');self.assertEqual(result['resource_digest'],crypto.digest(self.t.asset)) self.assertTrue(self.t.delivery_finished.wait(timeout=3));self.assertIsNone(self.t.delivery_error) one=self.phase('report');two=self.phase('report');self.assertEqual(one['record_id'],two['record_id']) self.assertEqual(self.phase('status')['state'],'active') self.t.provider_state='reversed';reversal=uid() self.assertEqual(self.phase('verify',check_id=reversal)['access_state'],'revoked') self.phase('service-check',check_id=reversal) self.phase('resource',fail=True);self.phase('retrieve',fail=True) replay=self.phase('verify',check_id=checks[-1]);self.assertTrue(replay['historical_retry']);self.assertEqual(replay['access_state'],'revoked') self.phase('service-check',check_id=checks[-1]) self.assertEqual(self.phase('revoke')['state'],'revoked');self.assertEqual(self.phase('status')['state'],'revoked') for method,path,headers,body in self.f.origin_server.requests: if path in {'/odexa.json','/odexa-service.json'}:self.assertNotIn('Authorization',headers) if direct: self.assertTrue(all(path=='/payment/verify' for _,path,_,_ in self.f.provider_server.requests)) for _,path,headers,_ in self.f.provider_server.requests: if path=='/payment/verify':self.assertEqual(headers.get('Authorization'),self.t.verify_credential.authorization) with self.f.store.connect() as db: self.assertEqual(db.execute('SELECT count(*) FROM records').fetchone()[0],1) self.assertEqual(db.execute('SELECT state FROM agreements').fetchone()[0],'revoked') self.save_report('direct-lifecycle' if direct else 'delegated-lifecycle') def test_delegated_paid_lifecycle_across_native_process_restarts(self):self.lifecycle(False) def test_direct_paid_lifecycle_with_only_read_only_verifier_delegated(self):self.lifecycle(True) def test_provider_wrong_amount_and_signer_do_not_commit_client_state(self): self.configure();self.phase('agree');self.phase('mandate');self.t.provider_state='confirmed' for field in ('wrong_amount','wrong_signature'): setattr(self.t,field,True) self.phase('verify',check_id=uid(),fail=True) state=json.loads(self.state.read_text());self.assertEqual(state['payment']['payment_state'],'pending');self.assertEqual(state['payment']['checks'],{}) setattr(self.t,field,False) self.save_report('provider-rejections') def test_live_origin_withdrawal_after_provider_reply_blocks_application_and_rollback(self): self.configure();self.phase('agree');self.phase('mandate');self.t.provider_state='confirmed' old=copy.deepcopy(self.f.authority) def withdraw(): self.f.authority['revision']=2 self.f.authority['delegations']=[g for g in self.f.authority['delegations'] if g['id']!=self.t.verifier_grant['id']] self.t.after_verify=withdraw self.phase('verify',check_id=uid(),fail=True) state=json.loads(self.state.read_text());self.assertEqual(state['payment']['payment_state'],'pending');self.assertEqual(state['history']['authorityRevision'],2) self.t.after_verify=None;self.f.authority=old calls=self.t.verify_calls;self.phase('verify',check_id=uid(),fail=True);self.assertEqual(self.t.verify_calls,calls) self.save_report('withdrawal-and-rollback') def test_valid_but_unapproved_price_and_payee_are_rejected_before_assent(self): self.configure();original=self.f.service.quote_factory for change in ('payee','price'): def altered(offer,payer): value=crypto.strict_json(original(offer,payer)) if change=='payee':value['items'][0]['payee_id']=self.f.origin+'/unapproved-payee' else:value['items'][0]['amount_minor']='101';value['total_minor']='126' return crypto.json_bytes(value) self.f.service.quote_factory=altered self.phase('agree',fail=True) with self.f.store.connect() as db:self.assertEqual(db.execute('SELECT count(*) FROM agreements').fetchone()[0],0) self.assertFalse(any(path.endswith('/agreements') for _,path,_,_ in self.f.provider_server.requests)) self.save_report('unapproved-quote') def test_authorised_signature_cannot_turn_paid_receipt_into_access_grant(self): self.configure();original=self.t.provider_route def route(method,path,headers,body): result=original(method,path,headers,body) if method=='POST' and path==self.f.base[len(self.f.provider):]+'agreements' and result[0]==201: raw,_=crypto.verify_jws(result[2],crypto.public_jwk(self.f.signing_key),'odexa-receipt+jws') value=crypto.strict_json(raw);value['state_at_issue']='active' wire=crypto.sign_jws(crypto.json_bytes(value),self.f.signing_key,self.f.service.key_id,'odexa-receipt+jws').encode() return result[0],result[1],wire return result self.f.provider_server.route=route self.phase('agree',fail=True) self.assertIsNone(json.loads(self.state.read_text())['receipt']) with self.f.store.connect() as db:self.assertEqual(db.execute('SELECT state FROM agreements').fetchone()[0],'pending_payment') self.save_report('hostile-signed-receipt') def test_unsupported_client_duty_declined_before_network_or_assent(self): self.configure();self.cfg['request']['supported_obligations'].append('retention') self.config.write_bytes(crypto.json_bytes(self.cfg)) before=[len(s.requests) for s,_ in self.f.servers] self.phase('observe',fail=True) self.assertEqual([len(s.requests) for s,_ in self.f.servers],before) self.save_report('unsupported-duty') def test_known_compromised_key_cannot_reappear_under_alias_after_process_restart(self): self.configure();self.phase('observe') key=self.f.authority['services'][1]['signing_keys'][0] key.update(state='revoked',revoked_at=self.f.now());self.f.authority['revision']=2 self.phase('observe');known=copy.deepcopy(self.f.authority) self.f.authority['revision']=3;key['retired_at']=key['revoked_at'] self.phase('observe',fail=True) self.f.authority=copy.deepcopy(known);self.f.authority['revision']=3 key=self.f.authority['services'][1]['signing_keys'][0] key.update(kid=key['kid']+'/alias',state='active',revoked_at=None,retired_at=None) self.phase('observe',fail=True) self.assertEqual(self.t.verify_calls,0) self.assertEqual(json.loads(self.state.read_text())['history']['authorityRevision'],2) self.save_report('known-compromise') if __name__=='__main__':unittest.main()