"""Independent native asset verification over actual Python HTTPS publication. Only this harness imports Python fixtures. The client uses native Node crypto, HTTP and filesystem APIs and retains its own authority/version history. """ import copy import gzip import json import os from pathlib import Path import subprocess import time import unittest from cryptography.hazmat.primitives.asymmetric import ec from odexa_ref import crypto from profiles import asset_evidence as a from profiles.authority_transport import AuthoritySnapshot import test_asset_runtime as runtime from test_asset_evidence import manifest,ref class IndependentAssetNodeTests(unittest.TestCase): @classmethod def setUpClass(cls):runtime.AssetRuntimeTests.setUpClass() @classmethod def tearDownClass(cls):runtime.AssetRuntimeTests.tearDownClass() def setUp(self): self.t=runtime.AssetRuntimeTests();self.t.setUp();self.results=[] def tearDown(self):self.t.tearDown() def configure(self,external=True): self.t.configure(external);self.f=self.t.f;f=self.f self.first=b'First immutable originator asset.\n';self.second=b'Updated immutable originator asset.\n' self.m1=runtime.publish(f,self.t.catalog,1,self.first) self.m2=runtime.publish(f,self.t.catalog,2,self.second,parent=ref(self.m1),coded=True) self.bundle=a.load(self.t.catalog.bundle_for([ref(self.m2)]),a.MAX_BUNDLE) self.original=f.origin_server.route;self.override={};self.after_bundle=None;self.after_bytes=None def route(method,path,headers,body): if path=='/asset-evidence.json': raw=a.raw(self.bundle) if self.after_bundle:self.after_bundle() return 200,{'Content-Type':'application/json'},raw if path in self.override:return self.override[path] if path in {'/asset-bytes/first','/asset-bytes/second'}: data=self.first if path.endswith('first') else gzip.compress(self.second,mtime=0) if self.after_bytes:self.after_bytes() return 200,{'Content-Type':'text/plain',**({'Content-Encoding':'gzip'} if path.endswith('second') else {})},data return self.original(method,path,headers,body) f.origin_server.route=route self.cfg=dict(origins=[f.origin],bundle_url=f.origin+'/asset-evidence.json',expected_roots=[ref(self.m2)],ca_file=str(f.tls_path/'ca.pem'), document_pins={},manifest_pins={},byte_probes=[dict(url=f.origin+'/asset-bytes/'+name,reference=ref(m), metadata={k:m['representations'][0][k] for k in ('media_type','media_parameters','languages')}) for name,m in [('first',self.m1),('second',self.m2)]]) self.config=f.path/'asset-node-config.json';self.state=f.path/'asset-node-state.json' def run_client(self,*,fail=False): self.config.write_bytes(a.raw(self.cfg));self.config.chmod(0o600) node=os.environ.get('ODEXA_NODE','node') result=subprocess.run([node,'verification/asset-client.mjs','--config',str(self.config),'--state',str(self.state)],capture_output=True,text=True,timeout=30) self.assertTrue(result.stdout,result.stderr);value=json.loads(result.stdout) if fail:self.assertNotEqual(result.returncode,0,value);self.assertTrue(value.get('failed'),value) else:self.assertEqual(result.returncode,0,value) self.results.append(dict(expected_failure=fail,result=value));return value def save(self,name): for server,_ in self.f.servers: for method,path,headers,body in server.requests: self.assertNotIn('Authorization',headers);self.assertNotIn('Cookie',headers) folder=os.environ.get('ODEXA_ASSET_NODE_REPORT_DIR') if folder: out=Path(folder);out.mkdir(parents=True,exist_ok=True) (out/(name+'.json')).write_text(json.dumps(dict(scenario=name,runs=self.results,count=len(self.results), actual_tls=True,separate_node_process_per_run=True,python_imports_in_client=False,credentials_sent=False),indent=2)+'\n') def lifecycle(self,external): self.configure(external) for _ in range(2): r=self.run_client();self.assertEqual(r['verified_manifests'],2);self.assertEqual(r['retained_versions'],2) self.assertTrue(all(v['representation_bytes_match'] for v in r['byte_comparisons'])) self.assertEqual(r['byte_comparisons'][1]['encoded_variant_match'],True) self.assertFalse(r['resource_permission_granted']);self.assertFalse(r['downstream_use_verified']) state=json.loads(self.state.read_text());self.assertEqual(len(state['versions']),2) self.assertEqual(self.state.stat().st_mode&0o077,0) if not external:self.assertFalse(self.f.provider_server.requests) self.save('delegated-lifecycle' if external else 'direct-lifecycle') def test_direct_graph_original_publications_identity_gzip_and_restart(self):self.lifecycle(False) def test_delegated_graph_original_publications_identity_gzip_and_restart(self):self.lifecycle(True) def test_missing_ancestor_and_wrong_original_publication_are_rejected(self): self.configure();original=copy.deepcopy(self.bundle) self.bundle['manifests']=self.bundle['manifests'][1:] self.assertIn('Missing',self.run_client(fail=True)['error']) self.bundle=original self.override['/versions/1']=(200,{'Content-Type':'application/jose'},self.t.catalog.get_manifest(self.m2['version_id'])['compact_jws'].encode()) self.run_client(fail=True);self.assertEqual(json.loads(self.state.read_text())['versions'],{}) self.save('missing-or-substituted-publication') def test_observed_retirement_and_policy_revision_survive_restart_and_reject_rollback(self): self.configure();self.run_client();old=copy.deepcopy(self.f.authority) self.f.authority['services'][0]['issuer']+='/changed' self.assertIn('same revision',self.run_client(fail=True)['error']);self.f.authority=copy.deepcopy(old) self.f.authority['revision']=2 self.f.authority['services'][0]['signing_keys'][0].update(state='retired',retired_at=self.f.now()) r=self.run_client();self.assertEqual(r['publication_authorized_now'],[False,False]) self.f.authority=old;self.assertIn('rollback',self.run_client(fail=True)['error']) self.assertEqual(json.loads(self.state.read_text())['history'][self.f.origin]['authority_revision'],2) self.save('retirement-and-rollback') def test_redefined_immutable_version_is_rejected_after_process_restart(self): self.configure();self.run_client();changed=copy.deepcopy(self.m2) changed['representations'][0]['languages']=['fr'] signed=a.sign_manifest(a.raw(changed),self.f.signing_key,self.f.service.key_id) self.bundle['manifests'][-1]=signed;self.bundle['roots']=[ref(changed)];self.cfg['expected_roots']=[ref(changed)] self.assertIn('Pinned version',self.run_client(fail=True)['error']) self.assertEqual(json.loads(self.state.read_text())['versions'][self.m2['version_id']],ref(self.m2)['manifest_digest']) self.save('immutable-version-restart') def test_current_compromise_after_bytes_and_removed_key_alias_remain_rejected(self): self.configure();self.run_client();f=self.f def compromise(): f.authority['revision']=2;f.authority['services'][0]['signing_keys'][0].update(state='revoked',revoked_at=f.now()) self.after_bytes=compromise self.assertIn('revoked',self.run_client(fail=True)['error']);self.after_bytes=None f.authority['revision']=3;key=f.authority['services'][0]['signing_keys'][0] key.update(kid=key['kid']+'/alias',state='active',revoked_at=None) self.assertIn('revoked',self.run_client(fail=True)['error']) self.save('compromise-and-alias') def test_untrusted_history_cannot_bootstrap_itself_from_bundle(self): self.configure();self.f.authority['revision']=2 self.assertIn('independent origin pin',self.run_client(fail=True)['error']) self.cfg['document_pins']={d['digest']:d['body_b64url'] for d in self.bundle['documents']} self.run_client();self.save('separately-provisioned-history') def test_cross_origin_parent_requires_both_independent_current_origins(self): self.configure();f=self.f def replace(value):return json.loads(json.dumps(value).replace(f.origin,f.provider)) meta,pol=map(replace,(f.authority,f.policy));svc=meta['services'][0];meta['delegations']=[] key=ec.derive_private_key(29,ec.SECP256R1());svc['signing_keys'][0]['public_jwk']=crypto.public_jwk(key) operation=replace(self.m2['publication']);operation['delegation_id']=None child=manifest(meta,pol,operation,self.second,3,parents=[ref(self.m2)]);child['published_at']=f.now();child['derived_from'][0]['relationship']='transform' signed=a.sign_manifest(a.raw(child),key,svc['signing_keys'][0]['kid']) self.bundle['manifests'].append(signed);self.bundle['roots']=[ref(child)];self.cfg['expected_roots']=[ref(child)] for v in (meta,pol):self.bundle['documents'].append(dict(digest=a.digest(a.raw(v)),body_b64url=crypto.b64u(a.raw(v)))) f.provider_server.route=lambda method,path,headers,body: (200,{'Content-Type':'application/jose' if path.startswith('/versions/') else 'application/json','Cache-Control':'no-store'}, a.raw(meta) if path=='/odexa-service.json' else a.raw(pol) if path=='/odexa.json' else signed.encode()) # Python independently agrees that the same closed graph is valid. s1=f.transport.fetch(f.origin);s2=f.transport.fetch(f.provider) pins={a.digest(b):b for b in (s1.authority_bytes,s1.policy_bytes,s2.authority_bytes,s2.policy_bytes)} checked=a.verify_closure(a.raw(self.bundle),trust=a.AssetTrust({f.origin:s1,f.provider:s2},pins),now=f.now) self.assertEqual(len(checked.manifests),3) self.run_client(fail=True) self.cfg['origins'].append(f.provider) # Changing the configured trust domain uses a separate journal deliberately. self.state=f.path/'cross-origin-state.json' r=self.run_client();self.assertEqual(r['independent_origins'],2);self.assertEqual(r['verified_manifests'],3) self.save('cross-origin-closure') def test_redirect_outage_and_wrong_coded_bytes_do_not_grant_verified_result(self): self.configure();self.override['/versions/1']=(302,{'Content-Type':'application/jose','Location':'https://unconfigured.example/'},b'') self.assertIn('302',self.run_client(fail=True)['error']);self.override.clear() self.f.origin_status=503;self.assertIn('503',self.run_client(fail=True)['error']);self.f.origin_status=200 self.override['/asset-bytes/second']=(200,{'Content-Type':'text/plain','Content-Encoding':'gzip'},gzip.compress(b'wrong',mtime=0)) self.assertIn('Decoded bytes',self.run_client(fail=True)['error']);self.save('transport-and-bytes-rejections') def test_slow_or_cached_origin_observation_cannot_be_minted_as_fresh(self): self.configure();route=self.f.origin_server.route def slow(method,path,headers,body): if path=='/odexa-service.json':time.sleep(5.1) return route(method,path,headers,body) self.f.origin_server.route=slow self.assertIn('freshness budget',self.run_client(fail=True)['error']);self.assertFalse(self.state.exists()) def cached(method,path,headers,body): result=route(method,path,headers,body) if path=='/odexa-service.json':return result[0],dict(result[1],Age='1'),result[2] return result self.f.origin_server.route=cached self.assertIn('Stale/cache-warning',self.run_client(fail=True)['error']);self.assertFalse(self.state.exists()) for server,_ in self.f.servers: for method,path,headers,body in server.requests: if path in {'/odexa-service.json','/odexa.json'}: self.assertEqual(headers.get('Cache-Control'),'no-cache, no-store') self.save('slow-and-cached-origin-rejections') if __name__=='__main__':unittest.main()