"""Selected storage archive preserves exact signed parent closure and source limits.""" import copy import json import sys import unittest from odexa_ref import crypto from profiles import portable_evidence as p, storage_evidence as s from profiles.free_service import uid, later import test_free_service as fixture from test_storage_service import configure, agree, action, session, signed def enable_export(f): service = f.authority['services'][0] service['capabilities'].append('export_evidence') service['signing_keys'][0]['uses'].append(p.TYPE) if f.binding['delegation_id'] is not None: f.authority['delegations'][0]['capabilities'].append('export_evidence') def export_material(f, agreement_id): snapshot = f.transport.fetch(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=copy.deepcopy(f.request), offer_seconds=0, access_seconds=0, use_seconds=0, payment_mode='none', asset_id=None, version_id=None, agreement_id=agreement_id) # Trust is taken from a separately acquired publisher snapshot and the local # provisioning registry. It is never populated from the candidate bundle. 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')} trust = s.ExternalTrust(snapshot, {crypto.digest(x): x for x in (snapshot.authority_bytes,snapshot.policy_bytes)}, keys) kwargs = dict(agreement_id=agreement_id, snapshot=snapshot, history=f.history, export_operation=operation, private_key=f.signing_key, key_id=f.service.key_id, now=f.now) return s.build_bundle(f.store, **kwargs), trust, kwargs class StorageEvidenceTests(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(); enable_export(self.f); configure(self.f); agree(self.f) def tearDown(self): self.f.tearDown() def send(self, event): self.f.call('events', signed(self.f,event), wire=True, status=201) def chain(self): w = action(self.f); start = session(w) child = action(self.f) child['operation']['copy_context']['acquired_at'] = w['operation']['copy_context']['acquired_at'] child['operation']['copy_context']['retain_until'] = w['operation']['copy_context']['retain_until'] child['operation']['copy_context']['parent_write'] = dict(reporter_id=w['reporter_id'],event_id=w['event_id'],payload_digest=crypto.digest(crypto.json_bytes(w))) child_start = session(child) for event in (w,start,child,child_start): self.send(event) return w,start,child,child_start def test_signed_selected_chain_imports_inertly_and_exact_retries_survive_restart(self): self.chain(); body,trust,_ = export_material(self.f,self.f.agreement_id) result = s.verify_bundle(body,trust=trust,now=self.f.now) self.assertEqual(result['storage_metrics']['sessions_known'],2) self.assertEqual(result['storage_metrics']['unresolved_sessions'],0) self.assertFalse(result['downstream_use_verified']) destination=self.f.path/'archive'/'storage.sqlite3' first=s.StorageEvidenceArchive(destination).import_bundle(body,trust=trust,now=self.f.now) archive=s.StorageEvidenceArchive(destination) self.assertEqual(archive.import_bundle(body,trust=trust,now=self.f.now),first) saved=archive.get(result['origin'],result['bundle_id']) self.assertEqual(saved['bundle_bytes'],body);self.assertFalse(saved['reactivates_access']) with self.assertRaises(p.PortabilityError):p.verify_bundle(body,trust=trust,now=self.f.now) def test_ordinary_exporter_never_infers_selected_storage(self): _,_,kwargs=export_material(self.f,self.f.agreement_id) with self.assertRaises(p.PortabilityError):p.build_bundle(self.f.store,**kwargs) def test_copy_parent_outside_snapshot_cannot_be_silently_dropped(self): child=action(self.f) child['operation']['copy_context']['parent_write']=dict(reporter_id=child['reporter_id'],event_id=uid(),payload_digest='sha256:'+'0'*64) self.assertEqual(child['related_events'],[]) self.send(child) with self.assertRaisesRegex(p.PortabilityError,'outside selected closure'): export_material(self.f,self.f.agreement_id) def test_authorized_exporter_omission_still_fails_copy_parent_closure(self): w=action(self.f);child=action(self.f) child['operation']['copy_context']['acquired_at']=w['operation']['copy_context']['acquired_at'] child['operation']['copy_context']['retain_until']=w['operation']['copy_context']['retain_until'] child['operation']['copy_context']['parent_write']=dict(reporter_id=w['reporter_id'],event_id=w['event_id'],payload_digest=crypto.digest(crypto.json_bytes(w))) self.send(w);self.send(child) body,trust,_=export_material(self.f,self.f.agreement_id) envelope=p.load(body);manifest=p.load(crypto.split_jws(envelope['manifest_jws'],allowed_types={p.TYPE})[3]) for entry in list(manifest['inventory']): if entry['kind']!='intake_jws':continue blob=next(x for x in envelope['blobs'] if x['digest']==entry['digest']) intake=p.load(crypto.split_jws(crypto.unb64u(blob['body_b64url']))[3]) event=p.load(crypto.unb64u(intake['payload_b64url'])) if event['event_id']==w['event_id']: manifest['inventory'].remove(entry);envelope['blobs'].remove(blob);manifest['cut']['records_count']-=1 envelope['manifest_jws']=crypto.sign_jws(p.raw(manifest),self.f.signing_key,self.f.service.key_id,p.TYPE,allowed_types={p.TYPE}) with self.assertRaisesRegex(p.PortabilityError,'related event missing'): s.verify_bundle(p.raw(envelope),trust=trust,now=self.f.now) def test_independent_client_pins_and_fresh_export_authority_still_required(self): self.chain();body,trust,_=export_material(self.f,self.f.agreement_id) with self.assertRaises(p.PortabilityError): s.verify_bundle(body,trust=s.ExternalTrust(trust.current_snapshot,trust.pinned_documents,{}),now=self.f.now) with self.assertRaises(p.PortabilityError): s.verify_bundle(body,trust=trust,now=lambda:later(trust.current_snapshot.observation['checked_at'],6)) def test_conflicting_sessions_remain_qualified_unresolved_claims(self): w=action(self.f);a=session(w);b=session(w) for event in (w,a,b):self.send(event) body,trust,_=export_material(self.f,self.f.agreement_id) result=s.verify_bundle(body,trust=trust,now=self.f.now) self.assertGreater(result['storage_metrics']['unresolved_sessions'],0) self.assertFalse(result['downstream_use_verified']) def schema_cases(): StorageEvidenceTests.setUpClass() t=StorageEvidenceTests();t.setUp() try: t.chain();body,_,_=export_material(t.f,t.f.agreement_id) bundle=p.load(body);manifest=p.load(crypto.split_jws(bundle['manifest_jws'],allowed_types={p.TYPE})[3]) cases=[] for name,value in [('bundle',bundle),('manifest',manifest)]: cases.append(dict(name='storage_'+name,definition=name,value=value,structural_valid=True)) for field,replacement in [('profile',p.PROFILE),('unexpected',True)]: changed=copy.deepcopy(value);changed[field]=replacement cases.append(dict(name='storage_'+name+'_'+field,definition=name,value=changed,structural_valid=False)) changed=copy.deepcopy(manifest);changed['source_scope']='all_retained_free_material_for_one_agreement_at_sqlite_snapshot' cases.append(dict(name='wrong_selected_source_scope',definition='manifest',value=changed,structural_valid=False)) return dict(schema='storage-portable',cases=cases) finally:t.tearDown();StorageEvidenceTests.tearDownClass() if __name__=='__main__': if '--cases' in sys.argv:print(json.dumps(schema_cases(),indent=2)) else:unittest.main()