import copy import concurrent.futures import hashlib import json from pathlib import Path import sqlite3 import tempfile import threading import time import sys import unittest from unittest import mock from cryptography.hazmat.primitives.asymmetric import ec from odexa_ref import crypto from profiles import delegation as d, free_contracts as f from profiles.authority_history import AuthorityHistory from profiles.authority_transport import AuthoritySnapshot from profiles.free_store import FreeStore from profiles import portable_evidence as p from test_free_contracts import fixtures, NOW, later, uid class PortableEvidenceTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(); self.path = Path(self.temp.name) self.values, context = fixtures() self.pol, self.meta, self.terms = crypto.strict_json(context[0]), crypto.strict_json(context[1]), context[2] service = self.meta['services'][0] service['capabilities'] += ['receive_events', 'export_evidence'] service['signing_keys'][0]['uses'] += ['odexa-event-record+jws', p.TYPE] self.meta['delegations'][0]['capabilities'] += ['receive_events', 'export_evidence'] self.signer = ec.derive_private_key(1, ec.SECP256R1()); self.agent = ec.derive_private_key(3, ec.SECP256R1()) self.kid = service['signing_keys'][0]['kid'] self.clock = lambda: later(3) self.snapshot = self.snap(self.meta) offer = self.values['offer']; offer['authority_digest'] = crypto.digest(self.snapshot.authority_bytes) for entry in offer['context']: if entry['role'] == 'origin_authority': entry['digest'] = offer['authority_digest'] self.offer_raw = p.raw(offer) acceptance = self.values['acceptance']; acceptance['offer_digest'] = p.digest(self.offer_raw) self.acceptance_raw = p.raw(acceptance) self.acceptance_jws = crypto.sign_jws(self.acceptance_raw, self.agent, 'https://agent.example/keys/one', 'odexa-acceptance+jws') receipt = self.values['receipt']; receipt.update(offer_digest=p.digest(self.offer_raw), acceptance_digest=p.digest(self.acceptance_raw)) self.receipt_jws = crypto.sign_jws(p.raw(receipt), self.signer, self.kid, 'odexa-receipt+jws') self.values['status']['authority_digest'] = offer['authority_digest'] self.status_jws = crypto.sign_jws(p.raw(self.values['status']), self.signer, self.kid, 'odexa-status+jws') self.values['intake']['authority']['digest'] = offer['authority_digest'] self.intake_jws = crypto.sign_jws(p.raw(self.values['intake']), self.signer, self.kid, 'odexa-event-record+jws') self.agreement_id = receipt['agreement_id'] self.store = FreeStore(self.path / 'source' / 'free.sqlite3', {k: offer[k] for k in ('origin','service_id','issuer')}) self.descriptor = dict(client_id=offer['client_id'], role='agent', principals=[offer['principal_id']], reporter_id=self.values['report']['reporter_id'], key_id='https://agent.example/keys/one', public_jwk=crypto.public_jwk(self.agent)) with self.store.transaction() as db: db.execute('INSERT INTO clients VALUES(?,?,?,?,?,?,?,1)', (offer['client_id'], hashlib.sha256(b'synthetic-secret').hexdigest(), 'agent', p.raw(self.descriptor['principals']), p.raw(self.descriptor['public_jwk']), self.descriptor['key_id'], self.descriptor['reporter_id'])) db.execute('INSERT INTO offers VALUES(?,?,?,?,?,?)', (offer['offer_id'], offer['service_id'], offer['client_id'], offer['principal_id'], offer['nonce'], self.offer_raw)) db.execute('INSERT INTO agreements VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)', (self.agreement_id, offer['service_id'], offer['offer_id'], offer['client_id'], offer['principal_id'], acceptance['idempotency_key'], self.acceptance_raw, self.acceptance_jws, self.receipt_jws, 'active', 1, NOW, 'accepted', receipt['access_expires_at'], receipt['use_expires_at'])) db.execute('INSERT INTO status_history VALUES(?,?,?)', (self.agreement_id, 1, self.status_jws)) db.execute('INSERT INTO status_attestations VALUES(?,?,?,?)', (p.digest(self.status_jws.encode()), self.agreement_id, 1, self.status_jws)) self.store.record_record(db, self.values['intake']['record_id'], self.descriptor['reporter_id'], self.values['report']['event_id'], p.raw(self.values['report']), self.intake_jws, offer['client_id'], self.agreement_id) for body, media in [(self.snapshot.authority_bytes,'application/json'), (self.snapshot.policy_bytes,'application/json'), (self.terms,'text/plain')]: self.store.put_document(db, body, media) self.history = AuthorityHistory(self.path / 'history' / 'authority.sqlite3', clock=self.clock) self.operation = dict(operation_id=uid(90), payment_request_digest=None, kind='export_evidence', origin=offer['origin'], service_id=offer['service_id'], issuer=offer['issuer'], endpoint=service['base_url']+'exports', delegation_id=offer['delegation_id'], key_id=self.kid, key_use=p.TYPE, request=copy.deepcopy(offer['request']), offer_seconds=0, access_seconds=0, use_seconds=0, payment_mode='none', asset_id=None, version_id=None, agreement_id=self.agreement_id) self.trust = p.ExternalTrust(self.snapshot, {p.digest(self.snapshot.authority_bytes): self.snapshot.authority_bytes, p.digest(self.snapshot.policy_bytes): self.snapshot.policy_bytes}, {self.descriptor['key_id']: dict(self.descriptor, revoked=False)}) def tearDown(self): self.temp.cleanup() def snap(self, meta, *, checked=None): ar, pr = p.raw(meta), p.raw(self.pol) return AuthoritySnapshot(ar, pr, dict(source='https', origin=meta['origin'], authority_url=meta['origin']+'/odexa-service.json', policy_url=meta['origin']+'/odexa.json', authority_digest=p.digest(ar), policy_digest=p.digest(pr), checked_at=checked or later(3), available=True, tls_verified=True, redirected=False, revalidated=True)) def bundle(self): return p.build_bundle(self.store, agreement_id=self.agreement_id, snapshot=self.snapshot, history=self.history, export_operation=self.operation, private_key=self.signer, key_id=self.kid, now=self.clock) def verify(self, body=None, trust=None): return p.verify_bundle(body or self.bundle(), trust=trust or self.trust, now=self.clock) def mutate(self, body, callback, resign=False): envelope = p.load(body); manifest = p.load(crypto.split_jws(envelope['manifest_jws'], allowed_types={p.TYPE})[3]) callback(envelope, manifest) if resign: envelope['manifest_jws'] = crypto.sign_jws(p.raw(manifest), self.signer, self.kid, p.TYPE, allowed_types={p.TYPE}) return p.raw(envelope) def artifact(self, body, kind, replacement): def alter(envelope, manifest): entry = next(x for x in manifest['inventory'] if x['kind'] == kind) old, new = entry['digest'], p.digest(replacement) blob = next(x for x in envelope['blobs'] if x['digest'] == old) blob.update(digest=new,body_b64url=crypto.b64u(replacement)) entry.update(digest=new,size=len(replacement)) if kind == 'status_jws': entry['id'] = new return self.mutate(body,alter,resign=True) def extra_report(self, change): event = copy.deepcopy(self.values['report']); event['event_id'] = uid(51) change(event) event_raw = p.raw(event) signed = crypto.sign_jws(event_raw,self.agent,self.descriptor['key_id'],'odexa-event+jws') intake = copy.deepcopy(self.values['intake']) intake.update(record_id=uid(52),payload_b64url=crypto.b64u(event_raw),payload_digest=p.digest(event_raw),reporter_jws=signed) compact = crypto.sign_jws(p.raw(intake),self.signer,self.kid,'odexa-event-record+jws') with self.store.transaction() as db: self.store.record_record(db,intake['record_id'],self.descriptor['reporter_id'],event['event_id'], event_raw,compact,self.descriptor['client_id'],self.agreement_id) def unresolved_admission(self, change=lambda a: None): gateway=ec.derive_private_key(4,ec.SECP256R1()) descriptor=dict(self.descriptor,client_id='gateway-one',role='gateway', reporter_id=self.values['binding']['origin']+'/gateway',key_id='https://gateway.example/key',public_jwk=crypto.public_jwk(gateway)) response=copy.deepcopy(self.values['introspection_response']) response['admission']['authority_digest']=p.digest(self.snapshot.authority_bytes) change(response['admission']) with self.store.transaction() as db: db.execute('INSERT INTO clients VALUES(?,?,?,?,?,?,?,1)',(descriptor['client_id'],'0'*64,'gateway', p.raw(descriptor['principals']),p.raw(descriptor['public_jwk']),descriptor['key_id'],descriptor['reporter_id'])) db.execute('INSERT INTO admissions VALUES(?,?,?,?,?,NULL)',(response['admission']['decision_id'],response['request_id'], descriptor['client_id'],self.agreement_id,p.raw(response))) pins=dict(self.trust.client_keys);pins[descriptor['key_id']]=dict(descriptor,revoked=False) self.trust=p.ExternalTrust(self.snapshot,self.trust.pinned_documents,pins) return response def test_indexed_lifecycle_cannot_invent_unattested_revocation_or_versions(self): body=self.bundle() with self.store.connect() as db: state=dict(db.execute('SELECT * FROM agreements').fetchone()) names='id offer_id client_id principal_id state status_version effective_at reason access_expires_at use_expires_at'.split() original={key:state[key] for key in names} mutations=[dict(state='revoked',status_version=99,effective_at='2040-01-01T00:00:00Z',reason={'unexpected':'object'}), dict(state='revoked',status_version=2,effective_at=later(2),reason='withdrawn'), dict(reason={'unexpected':'object'}),dict(reason='x'*257),dict(effective_at=later(4)), dict(effective_at=later(-1)),dict(status_version=2),dict(reason='different')] for change in mutations: with self.subTest(change=change),self.assertRaises(p.PortabilityError): self.verify(self.artifact(body,'agreement',p.raw(dict(original,**change)))) def test_signed_revocation_and_indexed_state_correlate(self): status=dict(self.values['status'],state='revoked',status_version=2,effective_at=later(2), reason='withdrawn',issued_at=later(2),authority_checked_at=later(2)) signed=crypto.sign_jws(p.raw(status),self.signer,self.kid,'odexa-status+jws') with self.store.transaction() as db: db.execute("UPDATE agreements SET state='revoked',status_version=2,effective_at=?,reason='withdrawn'",(later(2),)) db.execute('INSERT INTO status_history VALUES(?,?,?)',(self.agreement_id,2,signed)) db.execute('INSERT INTO status_attestations VALUES(?,?,?,?)',(p.digest(signed.encode()),self.agreement_id,2,signed)) result=self.verify()['lifecycle'] self.assertEqual(result['indexed_state_assurance'],'signed_status_correlated') self.assertEqual(result['signed_status_versions'],[1,2]) def test_lazy_expiry_is_qualified_indexed_statement_without_inventing_status(self): self.clock=lambda:later(303);self.snapshot=self.snap(self.meta,checked=self.clock()) self.trust=p.ExternalTrust(self.snapshot,self.trust.pinned_documents,self.trust.client_keys) with self.store.transaction() as db: db.execute("UPDATE agreements SET state='expired',status_version=2,effective_at=access_expires_at,reason='access_expired'") result=self.verify()['lifecycle'] self.assertEqual(result['indexed_state_assurance'],'exporter_asserted_lazy_expiry_at_receipt_deadline') self.assertEqual(result['signed_status_versions'],[1]);self.assertEqual(result['access_at_snapshot'],'expired') def test_stale_initial_index_does_not_claim_access_beyond_receipt_deadline(self): self.clock=lambda:later(303);self.snapshot=self.snap(self.meta,checked=self.clock()) self.trust=p.ExternalTrust(self.snapshot,self.trust.pinned_documents,self.trust.client_keys) result=self.verify()['lifecycle'] self.assertEqual(result['indexed_state'],'active');self.assertEqual(result['access_at_snapshot'],'expired') def test_unresolved_admission_cannot_exceed_snapshot_or_predate_receipt(self): self.unresolved_admission() body=self.bundle();self.assertEqual(self.verify(body)['manifest']['cut']['admissions_count'],1) envelope=p.load(body);manifest=p.load(crypto.split_jws(envelope['manifest_jws'],allowed_types={p.TYPE})[3]) entry=next(x for x in manifest['inventory'] if x['kind']=='admission') blob=next(x for x in envelope['blobs'] if x['digest']==entry['digest']) original=p.load(crypto.unb64u(blob['body_b64url'])) for change in [dict(checked_at=later(30),authority_checked_at=later(30)), dict(checked_at=later(-1),authority_checked_at=later(-1)),dict(status_version=2), dict(authority_checked_at=later(1))]: wrapper=copy.deepcopy(original);response=p.load(crypto.unb64u(wrapper['response_b64url'])) response['admission'].update(change);wrapper['response_b64url']=crypto.b64u(p.raw(response)) with self.subTest(change=change),self.assertRaises(p.PortabilityError): self.verify(self.artifact(body,'admission',p.raw(wrapper))) def test_unresolved_admission_after_signed_revocation_is_rejected(self): self.unresolved_admission(lambda a:a.update(checked_at=later(3),authority_checked_at=later(3))) status=dict(self.values['status'],state='revoked',status_version=2,effective_at=later(2), reason='withdrawn',issued_at=later(2),authority_checked_at=later(2)) signed=crypto.sign_jws(p.raw(status),self.signer,self.kid,'odexa-status+jws') with self.store.transaction() as db: db.execute("UPDATE agreements SET state='revoked',status_version=2,effective_at=?,reason='withdrawn'",(later(2),)) db.execute('INSERT INTO status_history VALUES(?,?,?)',(self.agreement_id,2,signed)) with self.assertRaises(p.PortabilityError):self.verify() def test_real_resolver_export_capability_and_closed_free_chain(self): decision = d.evaluate_authority(self.snapshot.authority_bytes, self.snapshot.policy_bytes, self.operation, now=self.clock(), observation=self.snapshot.observation) self.assertEqual(decision['decision'], 'allow', decision) self.assertEqual(decision['authority']['agreement_id'], self.agreement_id) result = self.verify() self.assertTrue(result['archive_only']); self.assertFalse(result['reactivates_access']) self.assertFalse(result['global_completeness_verified']); self.assertFalse(result['downstream_use_verified']) self.assertEqual(result['manifest']['cut']['records_count'], 1) def test_archive_import_restart_and_exact_retry_are_inert(self): body = self.bundle() archive = p.EvidenceArchive(self.path / 'archive' / 'evidence.sqlite3') report = archive.import_bundle(body, trust=self.trust, now=self.clock) self.assertEqual(archive.import_bundle(body, trust=self.trust, now=self.clock), report) reopened = p.EvidenceArchive(archive.path) saved = reopened.get(report['origin'], report['bundle_id']) self.assertEqual(saved['bundle_bytes'], body) self.assertEqual(saved['authority_status'], 'historical_only') with sqlite3.connect(archive.path) as db: tables = [x[0] for x in db.execute("SELECT name FROM sqlite_master WHERE type='table'")] self.assertEqual(tables, ['archives']) def test_no_credential_tables_or_token_plaintext_exported(self): with self.store.transaction() as db: db.execute('INSERT INTO tokens VALUES(?,?,?)', (hashlib.sha256(b'private-token').hexdigest(), self.agreement_id, later(100))) self.store.record_command(db, self.descriptor['client_id'], '/test', 'opaque-command', b'command', 200, b'{}', b'not-exportable-command') body = self.bundle() decoded = b''.join(crypto.unb64u(x['body_b64url']) for x in p.load(body)['blobs']) for secret in (b'private-token', b'synthetic-secret', b'secret_hash', b'not-exportable-command'): self.assertNotIn(secret, decoded) self.verify(body) def test_missing_extra_duplicate_and_changed_blobs_rejected(self): body = self.bundle() mutations = [lambda b,m: b['blobs'].pop(), lambda b,m: b['blobs'].append(copy.deepcopy(b['blobs'][0])), lambda b,m: b['blobs'].append({'digest':p.digest(b'extra'),'body_b64url':crypto.b64u(b'extra')}), lambda b,m: b['blobs'][0].update(body_b64url=crypto.b64u(b'changed'))] for callback in mutations: with self.subTest(callback=callback), self.assertRaises(ValueError): self.verify(self.mutate(body, callback)) def test_manifest_signature_tamper_and_false_count_rejected(self): body = self.bundle() with self.assertRaises(ValueError): self.verify(self.mutate(body, lambda b,m: m['cut'].update(records_count=2), resign=True)) value = p.load(body); compact = value['manifest_jws']; value['manifest_jws'] = compact[:-2] + ('AA' if not compact.endswith('AA') else 'BB') with self.assertRaises(ValueError): self.verify(p.raw(value)) def test_exporter_keys_cannot_bootstrap_external_client_trust(self): trust = p.ExternalTrust(self.snapshot, self.trust.pinned_documents, {}) with self.assertRaises(ValueError): self.verify(trust=trust) changed = copy.deepcopy(self.trust.client_keys); changed[self.descriptor['key_id']]['principals'] = ['https://other.example/person'] with self.assertRaises(ValueError): self.verify(trust=p.ExternalTrust(self.snapshot,self.trust.pinned_documents,changed)) def test_known_compromise_rejects_historical_agent_and_service_signatures(self): body = self.bundle() changed = copy.deepcopy(self.trust.client_keys); changed[self.descriptor['key_id']]['revoked'] = True with self.assertRaises(ValueError): self.verify(body, p.ExternalTrust(self.snapshot,self.trust.pinned_documents,changed)) current = copy.deepcopy(self.meta); current['revision'] = 2 current['services'][0]['signing_keys'][0].update(state='revoked', revoked_at=later(3)) with self.assertRaises(ValueError): self.verify(body, p.ExternalTrust(self.snap(current),self.trust.pinned_documents,self.trust.client_keys)) def test_removed_export_delegation_or_wrong_origin_denied(self): body = self.bundle() current = copy.deepcopy(self.meta); current.update(revision=2,delegations=[]) with self.assertRaises(ValueError): self.verify(body,p.ExternalTrust(self.snap(current),self.trust.pinned_documents,self.trust.client_keys)) broken = dict(self.snapshot.observation,tls_verified=False) snapshot = AuthoritySnapshot(self.snapshot.authority_bytes,self.snapshot.policy_bytes,broken) with self.assertRaises(ValueError): self.verify(body,p.ExternalTrust(snapshot,self.trust.pinned_documents,self.trust.client_keys)) def test_retired_historical_signer_survives_real_new_export_key_rotation(self): old_ar = self.snapshot.authority_bytes current = copy.deepcopy(self.meta); current['revision'] = 2 old = current['services'][0]['signing_keys'][0]; old.update(state='retired',retired_at=later(2)) new = copy.deepcopy(old); self.signer = ec.derive_private_key(5, ec.SECP256R1()); self.kid = 'https://provider.example/keys/new' new.update(kid=self.kid,public_jwk=crypto.public_jwk(self.signer),state='active',retired_at=None) current['services'][0]['signing_keys'].append(new) self.snapshot = self.snap(current); self.operation['key_id'] = self.kid trust = p.ExternalTrust(self.snapshot,self.trust.pinned_documents,self.trust.client_keys) body = self.bundle() self.assertTrue(self.verify(body,trust)['archive_only']) without_old = {p.digest(self.snapshot.policy_bytes):self.snapshot.policy_bytes} with self.assertRaises(ValueError): self.verify(body,p.ExternalTrust(self.snapshot,without_old,self.trust.client_keys)) def test_paid_storage_and_wrong_export_scope_fail_closed(self): original = self.store.storage_version self.store.storage_version = b'paid-1' with self.assertRaises(ValueError): self.bundle() self.store.storage_version = original self.operation['agreement_id'] = uid(999) with self.assertRaises(ValueError): self.bundle() def test_archive_same_id_different_exact_bytes_conflicts(self): body = self.bundle(); archive = p.EvidenceArchive(self.path/'archive'/'evidence.sqlite3') archive.import_bundle(body,trust=self.trust,now=self.clock) changed = json.dumps(p.load(body),indent=2).encode() self.verify(changed) with self.assertRaises(ValueError): archive.import_bundle(changed,trust=self.trust,now=self.clock) def test_archive_failed_verification_leaves_no_rows(self): body = self.bundle(); archive = p.EvidenceArchive(self.path/'archive'/'evidence.sqlite3') changed = self.mutate(body,lambda b,m:b['blobs'].pop()) with self.assertRaises(ValueError): archive.import_bundle(changed,trust=self.trust,now=self.clock) with sqlite3.connect(archive.path) as db: self.assertEqual(db.execute('SELECT COUNT(*) FROM archives').fetchone()[0],0) def test_closed_inventory_rejects_unknown_wire_fields_and_version_shim(self): body = self.bundle() for change in [lambda b,m:b.update(protocol_version='1.2.0-draft.2'), lambda b,m:b.update(token='private'),lambda b,m:m.update(reactivates_access=True)]: with self.subTest(change=change),self.assertRaises(ValueError): self.verify(self.mutate(body,change,resign=True)) def test_manifest_future_time_and_stale_current_snapshot_rejected(self): body = self.bundle() with self.assertRaises(ValueError): self.verify(self.mutate(body,lambda b,m:m.update(created_at=later(4)),resign=True)) with self.assertRaises(ValueError): p.verify_bundle(body,trust=self.trust,now=lambda:later(9)) def test_re_signed_paid_receipt_or_wrong_exact_assent_rejected(self): body = self.bundle() paid = dict(self.values['receipt'],payment_required=True) wire = crypto.sign_jws(p.raw(paid),self.signer,self.kid,'odexa-receipt+jws').encode() with self.assertRaises(ValueError):self.verify(self.artifact(body,'receipt_jws',wire)) changed = self.acceptance_raw+b' ' with self.assertRaises(ValueError):self.verify(self.artifact(body,'acceptance',changed)) def test_incorrect_authority_revision_is_not_rescued_by_valid_signature(self): body=self.bundle();status=dict(self.values['status'],authority_revision=99) wire=crypto.sign_jws(p.raw(status),self.signer,self.kid,'odexa-status+jws').encode() with self.assertRaises(ValueError):self.verify(self.artifact(body,'status_jws',wire)) def test_missing_external_asset_or_related_event_closure_fails(self): self.extra_report(lambda e:e['related_events'].append(dict(reporter_id=self.descriptor['reporter_id'], event_id=uid(999),payload_digest='sha256:'+'0'*64))) with self.assertRaises(ValueError):self.verify() def test_asset_manifest_cannot_be_silently_dropped(self): origin=self.values['binding']['origin'] self.extra_report(lambda e:e.update(asset_ref=dict(asset_id=origin+'/assets/a',version_id=origin+'/assets/a/v1', manifest_digest='sha256:'+'1'*64,representation_id=uid(61)))) with self.assertRaises(ValueError):self.verify() def test_conflicting_signed_status_at_one_lifecycle_version_rejected(self): status=dict(self.values['status'],reason='different meaning',issued_at=later(2),authority_checked_at=later(2)) signed=crypto.sign_jws(p.raw(status),self.signer,self.kid,'odexa-status+jws') with self.store.transaction() as db: db.execute('INSERT INTO status_attestations VALUES(?,?,?,?)',(p.digest(signed.encode()),self.agreement_id,1,signed)) with self.assertRaises(ValueError):self.verify() def test_fresh_distinct_status_attestations_all_survive_import(self): status=dict(self.values['status'],issued_at=later(2),authority_checked_at=later(2)) signed=crypto.sign_jws(p.raw(status),self.signer,self.kid,'odexa-status+jws') with self.store.transaction() as db: db.execute('INSERT INTO status_attestations VALUES(?,?,?,?)',(p.digest(signed.encode()),self.agreement_id,1,signed)) result=self.verify() self.assertEqual(result['manifest']['cut']['status_attestations_count'],2) self.assertEqual(len([x for x in result['manifest']['inventory'] if x['kind']=='status_jws']),2) def test_authority_expiry_during_verification_is_not_hidden_by_initial_clock(self): body=self.bundle();samples=iter([later(3),later(9)]) with self.assertRaises(ValueError):p.verify_bundle(body,trust=self.trust,now=lambda:next(samples)) def test_export_signing_delay_cannot_outlive_current_authority_observation(self): value=[later(3)];self.clock=lambda:value[0] original=crypto.sign_jws def delayed(*args,**kwargs): result=original(*args,**kwargs);value[0]=later(9);return result with mock.patch.object(crypto,'sign_jws',side_effect=delayed): with self.assertRaises(ValueError):self.bundle() def test_archive_lock_wait_resamples_current_authority(self): body=self.bundle();archive=p.EvidenceArchive(self.path/'archive'/'evidence.sqlite3') lock=sqlite3.connect(archive.path,isolation_level=None);lock.execute('BEGIN IMMEDIATE') value=[later(3)];ready=threading.Event() def clock():ready.set();return value[0] try: with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: result=pool.submit(archive.import_bundle,body,trust=self.trust,now=clock) self.assertTrue(ready.wait(2));time.sleep(.05) self.assertFalse(result.done());value[0]=later(9);lock.rollback() with self.assertRaises(ValueError):result.result(timeout=5) finally:lock.close() with sqlite3.connect(archive.path) as db:self.assertEqual(db.execute('SELECT COUNT(*) FROM archives').fetchone()[0],0) def test_concurrent_exact_imports_create_one_inert_record(self): body=self.bundle();archive=p.EvidenceArchive(self.path/'archive'/'evidence.sqlite3') def enter(_):return archive.import_bundle(body,trust=self.trust,now=self.clock)['bundle_digest'] with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:results=list(pool.map(enter,range(2))) self.assertEqual(results[0],results[1]) with sqlite3.connect(archive.path) as db:self.assertEqual(db.execute('SELECT COUNT(*) FROM archives').fetchone()[0],1) def test_gateway_delivery_admission_and_external_reporter_key_close_together(self): gateway=ec.derive_private_key(4,ec.SECP256R1()) descriptor=dict(self.descriptor,client_id='gateway-one',role='gateway', reporter_id=self.values['binding']['origin']+'/gateway',key_id='https://gateway.example/key',public_jwk=crypto.public_jwk(gateway)) admission=copy.deepcopy(self.values['introspection_response']) admission['admission']['authority_digest']=p.digest(self.snapshot.authority_bytes) event=copy.deepcopy(self.values['report']);event.update(event_id=uid(71),source='origin_observed', event_type='delivery.completed',reporter_id=descriptor['reporter_id'],actions=['retrieve'],operation=None,quantity=None,unit=None) origin=event['origin'] event['http']=dict(method='GET',status=200,kind='full',delivery_id=admission['request_id'],hop_id=uid(72), ingress_id=origin+'/edge',hop_role='end_client',boundary_id=origin+'/end-client',cache_status='bypass', content_codings=[],content_bytes=3,content_digest=p.digest(b'abc'),decoded_bytes=3,decoded_digest=p.digest(b'abc'), range=None,representation_metadata=None) event_raw=p.raw(event);report_jws=crypto.sign_jws(event_raw,gateway,descriptor['key_id'],'odexa-event+jws') intake=copy.deepcopy(self.values['intake']);intake.update(record_id=uid(73),assurance='origin_key_verified', authenticated_reporter_id=descriptor['reporter_id'],payload_digest=p.digest(event_raw), payload_b64url=crypto.b64u(event_raw),reporter_jws=report_jws) intake_jws=crypto.sign_jws(p.raw(intake),self.signer,self.kid,'odexa-event-record+jws') with self.store.transaction() as db: db.execute('INSERT INTO clients VALUES(?,?,?,?,?,?,?,1)',(descriptor['client_id'],'0'*64,'gateway', p.raw(descriptor['principals']),p.raw(descriptor['public_jwk']),descriptor['key_id'],descriptor['reporter_id'])) db.execute('INSERT INTO admissions VALUES(?,?,?,?,?,?)',(admission['admission']['decision_id'],admission['request_id'], descriptor['client_id'],self.agreement_id,p.raw(admission),event['event_id'])) self.store.record_record(db,intake['record_id'],descriptor['reporter_id'],event['event_id'],event_raw, intake_jws,descriptor['client_id'],self.agreement_id) pins=dict(self.trust.client_keys);pins[descriptor['key_id']]=dict(descriptor,revoked=False) self.trust=p.ExternalTrust(self.snapshot,self.trust.pinned_documents,pins) result=self.verify();self.assertEqual(result['manifest']['cut']['admissions_count'],1) self.assertEqual(result['manifest']['cut']['records_count'],2) body=self.bundle() self.assertFalse(result['downstream_use_verified']) value=p.load(body) def alter(envelope,manifest): entry=next(x for x in manifest['inventory'] if x['kind']=='admission') encoded=next(x for x in envelope['blobs'] if x['digest']==entry['digest']) wrapper=p.load(crypto.unb64u(encoded['body_b64url'])) wrong=p.load(crypto.unb64u(wrapper['response_b64url']));wrong['admission']['method']='HEAD' wrapper['response_b64url']=crypto.b64u(p.raw(wrong));replacement=p.raw(wrapper) encoded.update(digest=p.digest(replacement),body_b64url=crypto.b64u(replacement)) entry.update(digest=p.digest(replacement),size=len(replacement)) with self.assertRaises(ValueError):self.verify(self.mutate(body,alter,resign=True)) def test_historical_operation_decision_is_recomputed_not_blindly_trusted(self): outcome=d.evaluate_authority(self.snapshot.authority_bytes,self.snapshot.policy_bytes,self.operation, now=self.clock(),observation=self.snapshot.observation) with self.store.transaction() as db: db.execute('INSERT INTO decisions VALUES(?,?,?,?,?,?,?)',(self.operation['operation_id'],self.agreement_id, p.raw(self.operation),p.raw(outcome),self.snapshot.authority_bytes,self.snapshot.policy_bytes,p.raw(self.snapshot.observation))) body=self.bundle();self.assertEqual(self.verify(body)['manifest']['cut']['decisions_count'],1) envelope=p.load(body);manifest=p.load(crypto.split_jws(envelope['manifest_jws'],allowed_types={p.TYPE})[3]) entry=next(x for x in manifest['inventory'] if x['kind']=='decision') encoded=next(x for x in envelope['blobs'] if x['digest']==entry['digest']) wrapper=p.load(crypto.unb64u(encoded['body_b64url'])) outcome['authority']['issuer']='https://attacker.example/operator' wrapper['decision_b64url']=crypto.b64u(p.raw(outcome)) with self.assertRaises(ValueError):self.verify(self.artifact(body,'decision',p.raw(wrapper))) def test_archive_private_path_and_symlink_boundaries(self): archive=p.EvidenceArchive(self.path/'archive'/'evidence.sqlite3') self.assertEqual(archive.path.stat().st_mode&0o777,0o600) alias=archive.path.parent/'alias';alias.symlink_to(archive.path) with self.assertRaises(OSError):p.EvidenceArchive(alias) def test_archive_cannot_be_constructed_over_live_service_database(self): with self.assertRaises(p.PortabilityError):p.EvidenceArchive(self.store.path) with self.store.connect() as db: self.assertEqual(db.execute("SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='archives'").fetchone()[0],0) def schema_cases(): case=PortableEvidenceTests();case.setUp() try: body=case.bundle();bundle=p.load(body) manifest=p.load(crypto.split_jws(bundle['manifest_jws'],allowed_types={p.TYPE})[3]) cases=[dict(name='valid_portable_bundle',definition='bundle',value=bundle,structural_valid=True,semantic_valid=True), dict(name='valid_portable_manifest',definition='manifest',value=manifest,structural_valid=True,semantic_valid=True)] for kind in ('agreement','client_key'): entry=next(x for x in manifest['inventory'] if x['kind']==kind) blob=next(x for x in bundle['blobs'] if x['digest']==entry['digest']) cases.append(dict(name='valid_'+kind,definition=kind,value=p.load(crypto.unb64u(blob['body_b64url'])),structural_valid=True,semantic_valid=True)) for name,definition,value in [('wrong_profile','bundle',dict(bundle,profile='paid')), ('unknown_token','bundle',dict(bundle,access_token='forbidden')), ('reactivation','manifest',dict(manifest,reactivates_access=True)), ('global_completeness','manifest',dict(manifest,global_completeness_claimed=True))]: cases.append(dict(name=name,definition=definition,value=value,structural_valid=False,semantic_valid=False)) missing=copy.deepcopy(bundle);missing['blobs'].pop() cases.append(dict(name='missing_blob_requires_semantic_closure',definition='bundle',value=missing,structural_valid=True,semantic_valid=False)) return {'schema':'portable','cases':cases} finally:case.tearDown() if __name__=='__main__': if '--cases' in sys.argv:print(json.dumps(schema_cases(),sort_keys=True)) else:unittest.main()