"""Real TLS manifest publication, protected versions, signed reports and archives.""" import copy import gzip import http.client import json from pathlib import Path import tempfile import threading import sys import unittest from odexa_ref import crypto from profiles import asset_evidence as a, asset_portability as ap, portable_evidence as p from profiles.asset_catalog import AssetCatalog from profiles.authority_history import AuthorityHistory from profiles.authority_transport import ScopedCredential from profiles.free_gateway import FreeGateway from profiles.gateway_reports import GatewayReporter from profiles.free_service import uid import test_free_service as fixture from test_asset_evidence import manifest,ref from test_storage_evidence import enable_export from test_storage_service import configure as storage_configure,agree as storage_agree,action,session,signed def enable_assets(f): enable_export(f);service=f.authority['services'][0] service['capabilities'].append('publish_assets');service['signing_keys'][0]['uses'].append(a.TYPE) service['scope'].update(asset_id_prefixes=[f.origin+'/assets/'],version_id_prefixes=[f.origin+'/versions/']) for grant in f.authority['delegations']: grant['capabilities'].append('publish_assets') grant.update(asset_id_prefixes=[f.origin+'/assets/'],version_id_prefixes=[f.origin+'/versions/']) f.policy['resources'][0]['selectors'].append(dict(type='exact',path='/moved',query='absent')) def asset_trust(f): snap=f.transport.fetch(f.origin) return a.AssetTrust({f.origin:snap},{a.digest(x):x for x in (snap.authority_bytes,snap.policy_bytes)}) def publish(f,catalog,number,body,*,parent=None,coded=False): operation=dict(operation_id=uid(),payment_request_digest=None,kind='publish_manifest', **{k:f.binding[k] for k in ('origin','service_id','issuer','delegation_id')}, endpoint=f.origin+'/versions/'+str(number),key_id=f.service.key_id,key_use=a.TYPE, request=copy.deepcopy(f.request),offer_seconds=0,access_seconds=0,use_seconds=0,payment_mode='none', asset_id=f.origin+'/assets/article',version_id=f.origin+'/versions/'+str(number)) m=manifest(f.authority,f.policy,operation,body,number,parents=[parent] if parent else []) m['published_at']=f.now() if parent:m['derived_from'][0]['relationship']='transform' if coded: encoded=gzip.compress(body,mtime=0) m['content_variants']=[dict(representation_id=m['representations'][0]['representation_id'],content_codings=['gzip'],content_length=len(encoded),content_digest=a.digest(encoded))] signed=a.sign_manifest(a.raw(m),f.signing_key,f.service.key_id) parents=a.load(catalog.bundle_for([parent]),a.MAX_BUNDLE) if parent else None docs={a.digest(x):x for x in (a.raw(f.authority),a.raw(f.policy))} if parents: docs.update({e['digest']:crypto.unb64u(e['body_b64url']) for e in parents['documents']}) bundle=a.raw(dict(protocol_version=a.VERSION,profile=a.PROFILE,roots=[ref(m)], manifests=(parents['manifests'] if parents else [])+[signed], documents=[dict(digest=d,body_b64url=crypto.b64u(b)) for d,b in docs.items()])) catalog.retain(bundle,trust=asset_trust(f),now=f.now) return m def archive_material(f,catalog,agreement_id,request,*,storage=False): trust=asset_trust(f);snap=trust.current_snapshots[f.origin] operation=dict(operation_id=uid(),payment_request_digest=None,kind='export_evidence', **{k:f.binding[k] for k in ('origin','service_id','issuer','delegation_id')}, endpoint=f.base+'exports',key_id=f.service.key_id,key_use=p.TYPE,request=request, offer_seconds=0,access_seconds=0,use_seconds=0,payment_mode='none',asset_id=None,version_id=None,agreement_id=agreement_id) with f.store.connect() as db: keys={row['key_id']:dict(p._key_descriptor(row),revoked=False) for row in db.execute('SELECT * FROM clients WHERE jwk IS NOT NULL')} events=[a.load(bytes(row[0])) for row in db.execute('SELECT payload FROM records WHERE agreement_id=?',(agreement_id,))] roots={a.ref_key(r):r for e in events for r in p._asset_references(e)} closure=catalog.bundle_for(list(roots.values())) args=dict(agreement_id=agreement_id,snapshot=snap,history=f.history,export_operation=operation, private_key=f.signing_key,key_id=f.service.key_id,now=f.now) body=ap.build_bundle(f.store,asset_bundle_bytes=closure,asset_trust=trust,storage=storage,**args) external=ap.AssetExternalTrust(snap,trust.pinned_documents,keys,trust) return body,external,args class AssetRuntimeTests(unittest.TestCase): @classmethod def setUpClass(cls):fixture.FreeServiceTests.setUpClass() @classmethod def tearDownClass(cls):fixture.FreeServiceTests.tearDownClass() def setUp(self):self.f=fixture.FreeServiceTests();self.f.setUp();self.finishes=[] def tearDown(self): for done,errors in self.finishes: self.assertTrue(done.wait(2),'actual delivery callback incomplete');self.assertEqual(errors,[]) self.f.tearDown() def configure(self,external=True,storage=False): f=self.f if not external: f.temp.cleanup();f.temp=tempfile.TemporaryDirectory();f.path=Path(f.temp.name);f.make_service(False) enable_assets(f) if storage:storage_configure(f) self.catalog=AssetCatalog(f.path/'catalog.sqlite3') credential=ScopedCredential(f.host,f.base[len(f.host):],f.clients['gateway']['authorization']) reporter=GatewayReporter(f.clients['gateway']['reporter_id'],f.clients['gateway']['key_id'],f.gateway_key,credential) self.gateway=FreeGateway(f.path/'gateway.sqlite3',binding=f.binding,base_url=f.base,transport=f.transport, history=AuthorityHistory(f.path/'gateway-history.sqlite3'),credential=credential,clock=f.now,reporting=reporter) self.resources={};self.original=f.origin_route;f.origin_server.route=self.route def route(self,method,path,headers,body): f=self.f if method=='GET' and path.startswith('/versions/'): try:saved=self.catalog.get_manifest(f.origin+path) except ValueError:return 404,{'Content-Type':'application/json'},b'{}' return 200,{'Content-Type':'application/jose'},saved['compact_jws'].encode() if method!='GET' or path not in self.resources:return self.original(method,path,headers,body) m,decoded,coded=self.resources[path];wire=gzip.compress(decoded,mtime=0) if coded else decoded try: auth=headers.get('Authorization','') if not auth.startswith('Bearer '):raise ValueError('no token') rid=self.gateway.admit(auth[7:],resource_url=f.origin+path,actions=['retrieve'],purposes=['public_retrieval']) except ValueError:return 403,{'Content-Type':'application/json'},b'{}' done,errors=threading.Event(),[];self.finishes.append((done,errors));rep=m['representations'][0] def finish(written,complete): try: self.gateway.finish_delivery(rid,bytes_written=written,representation_digest=a.digest(wire), complete=complete,representation_metadata={k:rep[k] for k in ('media_type','media_parameters','languages')}, content_codings=['gzip'] if coded else [],decoded_bytes=len(decoded) if complete else None, decoded_digest=a.digest(decoded) if complete else None,asset_ref=ref(m)) except BaseException as e:errors.append(type(e).__name__) finally:done.set() return 200,{'Content-Type':'text/plain',**({'Content-Encoding':'gzip'} if coded else {})},wire,{ 'begin':lambda:self.gateway.begin_delivery(rid),'finish':finish} def get(self,path,token=None): from urllib.parse import urlsplit parsed=urlsplit(self.f.origin) connection=http.client.HTTPSConnection(parsed.hostname,parsed.port,context=self.f.transport.context,timeout=5) before=len(self.finishes) try: connection.request('GET',path,headers={'Authorization':'Bearer '+token} if token else {}) response=connection.getresponse();result=(response.status,dict(response.getheaders()),response.read()) finally:connection.close() if len(self.finishes)>before: done,errors=self.finishes[-1];self.assertTrue(done.wait(2));self.assertEqual(errors,[]) return result def exercise(self,external): self.configure(external);f=self.f first=b'Originator article version one.\n';second=b'Originator article version two with an update.\n' m1=publish(f,self.catalog,1,first);m2=publish(f,self.catalog,2,second,parent=ref(m1),coded=True) status,_,wire=self.get('/versions/1');self.assertEqual(status,200) self.assertEqual(crypto.split_jws(wire,allowed_types={a.TYPE})[3],a.raw(m1)) f.agree(wire=True);old_id=f.agreement_id;request=copy.deepcopy(f.request);token=f.token(wire=True) self.resources['/about']=(m1,first,False) self.assertEqual(self.get('/about',token)[2],first) self.resources['/about']=(m2,second,True) self.assertEqual(gzip.decompress(self.get('/about',token)[2]),second) self.assertEqual([r['state'] for r in self.gateway.publish_reports()],['acknowledged']*2) body,trust,args=archive_material(f,self.catalog,old_id,request) with self.assertRaises(p.PortabilityError):p.build_bundle(f.store,**args) report=ap.AssetEvidenceArchive(f.path/'archive'/'assets.sqlite3').import_bundle(body,trust=trust,now=f.now) self.assertEqual(len(report['asset_correlations']),2) self.assertTrue(all(r['binding']=='full_representation_consistent' for r in report['asset_correlations'])) self.assertTrue(all(not r['full_version_delivered'] and not r['downstream_use_verified'] for r in report['asset_correlations'])) self.resources['/moved']=(m2,second,True) self.assertEqual(self.get('/moved',token)[0],403) f.request['url']=f.origin+'/moved' original_acceptance=f.acceptance def distinct_acceptance(offer=None): original_acceptance(offer);f.accept_doc['idempotency_key']=uid() f.accept_jws=crypto.sign_jws(crypto.json_bytes(f.accept_doc),f.agent_key,f.clients['agent']['key_id'],'odexa-acceptance+jws').encode() return f.accept_jws f.acceptance=distinct_acceptance f.agree(wire=True);new_token=f.token(wire=True) self.assertEqual(gzip.decompress(self.get('/moved',new_token)[2]),second) self.assertEqual(self.gateway.publish_reports()[0]['state'],'acknowledged') self.assertEqual(self.catalog.get_manifest(m2['version_id'])['manifest_digest'],ref(m2)['manifest_digest']) if not external:self.assertFalse(f.provider_server.requests) def test_direct_versions_coding_relocation_and_closed_archive(self):self.exercise(False) def test_delegated_versions_coding_relocation_and_closed_archive(self):self.exercise(True) def test_wrong_manifest_claim_survives_as_report_but_fails_asset_verification(self): self.configure();f=self.f;m=publish(f,self.catalog,1,b'original') f.agree();token=f.token();self.resources['/about']=(m,b'substitution',False) self.assertEqual(self.get('/about',token)[0],200) self.assertEqual(self.gateway.publish_reports()[0]['state'],'acknowledged') with self.assertRaisesRegex(ValueError,'decoded representation conflicts'): archive_material(f,self.catalog,f.agreement_id,f.request) def test_selected_storage_archive_keeps_asset_and_copy_write_closure(self): self.configure(storage=True);f=self.f;m=publish(f,self.catalog,1,b'asset') storage_agree(f);w=action(f);w['asset_ref']=ref(m);s=session(w) for event in (w,s):f.call('events',signed(f,event),wire=True,status=201) body,trust,_=archive_material(f,self.catalog,f.agreement_id,f.request,storage=True) report=ap.AssetEvidenceArchive(f.path/'archive'/'assets.sqlite3',storage=True).import_bundle(body,trust=trust,now=f.now) self.assertEqual(report['storage_metrics']['sessions_known'],1) self.assertEqual([r['binding'] for r in report['asset_correlations']],['referenced_only']*2) self.assertFalse(report['downstream_use_verified']) with self.assertRaises(p.PortabilityError):ap.verify_bundle(body,trust=trust,now=f.now) def test_importer_rejects_authorized_exporter_asset_omission_and_untrusted_origin(self): self.configure();f=self.f;m1=publish(f,self.catalog,1,b'one');m2=publish(f,self.catalog,2,b'two',parent=ref(m1)) f.agree();token=f.token();self.resources['/about']=(m2,b'two',False);self.get('/about',token) self.gateway.publish_reports();body,trust,_=archive_material(f,self.catalog,f.agreement_id,f.request) untrusted=ap.AssetExternalTrust(trust.current_snapshot,trust.pinned_documents,trust.client_keys,a.AssetTrust({},{})) with self.assertRaises(p.PortabilityError):ap.verify_bundle(body,trust=untrusted,now=f.now) envelope=p.load(body);outer=p.load(crypto.split_jws(envelope['manifest_jws'],allowed_types={p.TYPE})[3]) item=next(x for x in outer['inventory'] if x['kind']=='asset_bundle') blob=next(x for x in envelope['blobs'] if x['digest']==item['digest']) nested=a.load(crypto.unb64u(blob['body_b64url']),a.MAX_BUNDLE) nested['manifests']=[s for s in nested['manifests'] if a.digest(crypto.split_jws(s,allowed_types={a.TYPE})[3])!=ref(m1)['manifest_digest']] altered=a.raw(nested);sha=a.digest(altered) item.update(digest=sha,size=len(altered));blob.update(digest=sha,body_b64url=crypto.b64u(altered)) envelope['manifest_jws']=crypto.sign_jws(p.raw(outer),f.signing_key,f.service.key_id,p.TYPE,allowed_types={p.TYPE}) with self.assertRaisesRegex(p.PortabilityError,'missing referenced manifest'): ap.verify_bundle(p.raw(envelope),trust=trust,now=f.now) def schema_cases(): AssetRuntimeTests.setUpClass();cases=[] try: for storage in (False,True): t=AssetRuntimeTests();t.setUp() try: t.configure(storage=storage);f=t.f;m=publish(f,t.catalog,1,b'asset') if storage:storage_agree(f) else:f.agree() event=action(f);event['asset_ref']=ref(m) if not storage: event.pop('event_profile');event['actions']=['retrieve'];event['operation']={k:event['operation'][k] for k in ('id','started_at','ended_at','state')} f.call('events',signed(f,event),wire=True,status=201) body,_,_=archive_material(f,t.catalog,f.agreement_id,f.request,storage=storage) b=p.load(body);m=p.load(crypto.split_jws(b['manifest_jws'],allowed_types={p.TYPE})[3]) schema='assets-storage-portable' if storage else 'assets-portable' for definition,value in [('bundle',b),('manifest',m)]: cases.append(dict(schema=schema,name=schema+'_'+definition,definition=definition,value=value,structural_valid=True)) for field,replacement in [('profile',p.PROFILE),('unknown',True)]: changed=copy.deepcopy(value);changed[field]=replacement cases.append(dict(schema=schema,name=schema+'_'+definition+'_'+field,definition=definition,value=changed,structural_valid=False)) finally:t.tearDown() finally:AssetRuntimeTests.tearDownClass() return dict(schema='assets-portable',cases=cases) if __name__=='__main__': if '--cases' in sys.argv:print(json.dumps(schema_cases(),indent=2)) else:unittest.main()