import concurrent.futures import hashlib from pathlib import Path import sqlite3 import subprocess import sys import tempfile import threading import unittest from profiles.free_store import FreeStore, FreeStoreError, FreeStoreConflict from odexa_ref import crypto IDENTITY = {'origin': 'https://origin.example', 'service_id': 'https://origin.example/services/free', 'issuer': 'https://provider.example/operator'} NOW = '2026-09-17T00:00:00Z' ACCESS_END = '2026-09-17T01:00:00Z' USE_END = '2026-09-18T00:00:00Z' def client(db, name='agent', role='agent'): db.execute('INSERT INTO clients VALUES (?,?,?,?,?,?,?,1)', (name, hashlib.sha256((name + '-secret').encode()).hexdigest(), role, b'["https://principal.example/one"]', b'{"public":"only"}' if role == 'agent' else None, 'https://agent.example/keys/one' if role == 'agent' else None, 'https://agent.example/reporters/' + name)) def offer(db, ident='offer-1', owner='agent', nonce='nonce-1'): db.execute('INSERT INTO offers VALUES (?,?,?,?,?,?)', (ident, IDENTITY['service_id'], owner, 'https://principal.example/one', nonce, b'{ "offer": "original exact bytes" }')) def agreement(db, ident='agreement-1', offer_id='offer-1', key='accept-1', owner='agent'): db.execute('INSERT INTO agreements VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)', (ident, IDENTITY['service_id'], offer_id, owner, 'https://principal.example/one', key, b'{ "acceptance": true }', 'signed.original.acceptance', 'signed.original.receipt', 'active', 1, NOW, 'accepted', ACCESS_END, USE_END)) db.execute('INSERT INTO status_history VALUES (?,?,?)', (ident, 1, 'signed.original.status')) class FreeStoreTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.path = Path(self.temp.name) / 'store' / 'free.sqlite3' self.store = FreeStore(self.path, IDENTITY) with self.store.transaction() as db: client(db); client(db, 'gateway', 'gateway'); client(db, 'admin', 'admin') offer(db) def tearDown(self): self.temp.cleanup() def accept(self): with self.store.transaction() as db: agreement(db) def command(self, db, *, body=b'{ "command": true }', route='/agreements', key='key-1', owner='agent', result=b'original.response'): return self.store.record_command(db, owner, route, key, body, 201, b'{"Content-Type":"application/json"}', result) def count(self, table): with self.store.connect() as db: return db.execute('SELECT COUNT(*) FROM ' + table).fetchone()[0] def test_restart_preserves_exact_offer_acceptance_receipt_and_status(self): self.accept() reopened = FreeStore(self.path, dict(IDENTITY)) value = reopened.raw_materials('agreement-1') self.assertEqual(value['offer'], b'{ "offer": "original exact bytes" }') self.assertEqual(value['acceptance'], b'{ "acceptance": true }') self.assertEqual(value['acceptance_jws'], 'signed.original.acceptance') self.assertEqual(value['receipt'], 'signed.original.receipt') self.assertEqual(value['statuses'], [{'version': 1, 'jws': 'signed.original.status'}]) self.assertEqual(value['kind'], 'local_raw_materials') self.assertFalse(value['portable_verification_established']) def test_identity_and_version_are_immutable_on_restart(self): for field, replacement in [('origin', 'https://elsewhere.example'), ('service_id', 'https://origin.example/services/other'), ('issuer', 'https://other-provider.example/operator')]: identity = dict(IDENTITY, **{field: replacement}) if field == 'origin': identity['service_id'] = replacement + '/services/free' with self.subTest(field=field), self.assertRaises(FreeStoreError): FreeStore(self.path, identity) with self.store.transaction() as db: with self.assertRaises(sqlite3.IntegrityError): db.execute("UPDATE settings SET value=? WHERE name='identity'", (b'{}',)) with self.assertRaises(sqlite3.IntegrityError): db.execute("DELETE FROM settings WHERE name='storage_version'") with self.assertRaises(sqlite3.IntegrityError): db.execute("INSERT OR REPLACE INTO settings VALUES ('identity',?)", (b'{}',)) def test_private_directory_file_and_symlinks(self): self.assertEqual(self.path.parent.stat().st_mode & 0o777, 0o700) self.assertEqual(self.path.stat().st_mode & 0o777, 0o600) public = Path(self.temp.name) / 'public'; public.mkdir(mode=0o755) with self.assertRaises(FreeStoreError): FreeStore(public / 'db', IDENTITY) link = self.path.parent / 'link'; link.symlink_to(self.path) with self.assertRaises(OSError): FreeStore(link, IDENTITY) def test_hash_fields_refuse_plaintext_and_no_private_fields_in_materials(self): self.accept() with self.store.transaction() as db: with self.assertRaises(sqlite3.IntegrityError): db.execute('UPDATE clients SET secret_hash=? WHERE client_id=?', ('plaintext-secret', 'agent')) with self.assertRaises(sqlite3.IntegrityError): db.execute('INSERT INTO tokens VALUES (?,?,?)', ('plaintext-token', 'agreement-1', ACCESS_END)) sha = hashlib.sha256(b'secret-access-token').hexdigest() db.execute('INSERT INTO tokens VALUES (?,?,?)', (sha, 'agreement-1', ACCESS_END)) self.command(db, result=b'{"access_token":"secret-access-token"}') value = repr(self.store.raw_materials('agreement-1')) for secret in ('secret-access-token', 'agent-secret', 'secret_hash', 'token_hash', sha): self.assertNotIn(secret, value) def test_duplicate_offer_nonce_and_one_agreement_per_offer(self): with self.store.transaction() as db: with self.assertRaises(sqlite3.IntegrityError): offer(db, 'offer-2') agreement(db) with self.assertRaises(sqlite3.IntegrityError): agreement(db, 'agreement-2', key='accept-2') self.assertEqual(self.count('agreements'), 1) def test_agreement_must_match_offer_owner_and_service(self): with self.store.transaction() as db: with self.assertRaises(sqlite3.IntegrityError): agreement(db, owner='admin') self.assertEqual(self.count('agreements'), 0) def test_exact_command_retry_returns_original_response_and_is_route_scoped(self): with self.store.transaction() as db: first = dict(self.command(db)) retry = dict(self.command(db, result=b'new-signature-must-not-replace-old')) self.assertEqual(first, retry) self.command(db, route='/revoke') self.command(db, owner='admin') self.assertEqual(self.count('commands'), 3) reopened = FreeStore(self.path, IDENTITY) with reopened.transaction() as db: row = reopened.check_command(db, 'agent', '/agreements', 'key-1', b'{ "command": true }') self.assertEqual(dict(row), first) def test_idempotency_key_conflict_compares_exact_request_bytes(self): with self.store.transaction() as db: self.command(db) with self.assertRaises(FreeStoreConflict): self.command(db, body=b'{"command":true}') with self.assertRaises(FreeStoreConflict): self.command(db, body=b'{ "command": false }') self.assertEqual(self.count('commands'), 1) def test_opaque_documents_preserve_bytes_and_media_type_binding(self): body = b'Non-JSON exact\x00bytes\xff' with self.store.transaction() as db: sha = self.store.put_document(db, body, 'application/octet-stream') self.assertEqual(self.store.put_document(db, body, 'application/octet-stream'), sha) with self.assertRaises(FreeStoreConflict): self.store.put_document(db, body, 'text/plain') self.assertEqual(db.execute('SELECT body FROM documents WHERE digest=?', (sha,)).fetchone()[0], body) self.assertEqual(self.count('documents'), 1) def test_records_are_idempotent_by_reporter_and_exact_event_bytes(self): self.accept() with self.store.transaction() as db: one = self.store.record_record(db, 'record-1', 'verified:agent', 'event-1', b'original.payload', 'original.intake.jws', 'agent', 'agreement-1') retry = self.store.record_record(db, 'new-record-id', 'verified:agent', 'event-1', b'original.payload', 'new.signature.jws', 'agent', 'agreement-1') self.assertEqual(dict(one), dict(retry)) with self.assertRaises(FreeStoreConflict): self.store.record_record(db, 'record-2', 'verified:agent', 'event-1', b'changed.payload', 'some.jws', 'agent', 'agreement-1') with self.assertRaises(FreeStoreConflict): self.store.record_record(db, 'record-2', 'verified:agent', 'event-1', b'original.payload', 'some.jws', 'admin', 'agreement-1') self.store.record_record(db, 'record-2', 'verified:admin', 'event-1', b'original.payload', 'different.intake.jws', 'admin', 'agreement-1') self.assertEqual(self.count('records'), 2) def test_receipt_acceptance_offer_status_and_decision_history_immutable(self): self.accept() with self.store.transaction() as db: db.execute('INSERT INTO decisions VALUES (?,?,?,?,?,?,?)', ('operation-1', 'agreement-1', b'operation', b'decision', b'authority', b'policy', b'observation')) checks = ["UPDATE offers SET body=X'00'", "UPDATE agreements SET receipt='changed'", "UPDATE agreements SET acceptance=X'00'", "UPDATE agreements SET use_expires_at='changed'", "UPDATE status_history SET jws='changed'", "UPDATE decisions SET decision=X'00'", 'DELETE FROM agreements', 'DELETE FROM decisions'] for sql in checks: with self.subTest(sql=sql), self.assertRaises(sqlite3.IntegrityError): db.execute(sql) reopened = FreeStore(self.path, IDENTITY) with reopened.connect() as db: row = db.execute('SELECT * FROM decisions').fetchone() self.assertEqual(row['authority_bytes'], b'authority') self.assertEqual(row['observation'], b'observation') def test_status_update_needs_new_version_and_terminal_state_never_reactivates(self): self.accept() with self.store.transaction() as db: with self.assertRaises(sqlite3.IntegrityError): db.execute("UPDATE agreements SET state='revoked'") db.execute("UPDATE agreements SET state='revoked',status_version=2,reason='originator revocation'") db.execute('INSERT INTO status_history VALUES (?,?,?)', ('agreement-1', 2, 'signed.revoked.status')) with self.assertRaises(sqlite3.IntegrityError): db.execute("UPDATE agreements SET state='active',status_version=3") with self.assertRaises(sqlite3.IntegrityError): db.execute('UPDATE agreements SET status_version=1') reopened = FreeStore(self.path, IDENTITY) with reopened.connect() as db: row = db.execute('SELECT state,use_expires_at FROM agreements').fetchone() self.assertEqual(tuple(row), ('revoked', USE_END)) self.assertEqual(len(reopened.raw_materials('agreement-1')['statuses']), 2) def test_multiple_current_status_attestations_retained_at_one_lifecycle_version(self): self.accept() signatures = ['first.current.status', 'second.current.status'] with self.store.transaction() as db: for signed in signatures: db.execute('INSERT INTO status_attestations VALUES(?,?,?,?)', (crypto.digest(signed.encode('ascii')), 'agreement-1', 1, signed)) with self.assertRaises(sqlite3.IntegrityError): db.execute("UPDATE status_attestations SET jws='replacement'") with self.assertRaises(sqlite3.IntegrityError): db.execute('DELETE FROM status_attestations') reopened = FreeStore(self.path, IDENTITY) materials = reopened.raw_materials('agreement-1') self.assertEqual([row['jws'] for row in materials['status_attestations']], signatures) self.assertEqual([row['version'] for row in materials['status_attestations']], [1, 1]) self.assertEqual(len(materials['statuses']), 1) def test_status_attestation_rolls_back_with_lifecycle_and_command(self): self.accept() with self.assertRaises(RuntimeError): with self.store.transaction() as db: db.execute("UPDATE agreements SET state='revoked',status_version=2,reason='revoked'") db.execute('INSERT INTO status_attestations VALUES(?,?,?,?)', (crypto.digest(b'revoked.status'), 'agreement-1', 2, 'revoked.status')) self.command(db, route='/revoke') raise RuntimeError('late guard denied') self.assertEqual(self.count('status_attestations'), 0) self.assertEqual(self.count('commands'), 0) with self.store.connect() as db: self.assertEqual(db.execute('SELECT state FROM agreements').fetchone()[0], 'active') def test_admission_gateway_request_unique_and_terminal_event_single_assignment(self): self.accept() with self.store.transaction() as db: db.execute('INSERT INTO admissions VALUES (?,?,?,?,?,?)', ('decision-1', 'request-1', 'gateway', 'agreement-1', b'bound.context', None)) with self.assertRaises(sqlite3.IntegrityError): db.execute('INSERT INTO admissions VALUES (?,?,?,?,?,?)', ('decision-2', 'request-1', 'gateway', 'agreement-1', b'changed.context', None)) db.execute("UPDATE admissions SET terminal_event_id='event-1'") db.execute("UPDATE admissions SET terminal_event_id='event-1'") with self.assertRaises(sqlite3.IntegrityError): db.execute("UPDATE admissions SET terminal_event_id='event-2'") with self.assertRaises(sqlite3.IntegrityError): db.execute("UPDATE admissions SET payload=X'00'") def test_foreign_keys_and_role_constraints(self): with self.store.transaction() as db: self.assertEqual(db.execute('PRAGMA foreign_keys').fetchone()[0], 1) self.assertEqual(db.execute('PRAGMA synchronous').fetchone()[0], 2) with self.assertRaises(sqlite3.IntegrityError): client(db, 'bad', 'unknown') with self.assertRaises(sqlite3.IntegrityError): offer(db, 'offer-2', owner='unknown', nonce='nonce-2') with self.assertRaises(sqlite3.IntegrityError): db.execute('INSERT INTO status_history VALUES (?,?,?)', ('unknown', 1, 'status')) def test_before_commit_exception_rolls_back_all_acceptance_records(self): def fail(stage): if stage == 'before_commit': raise RuntimeError('injected failure') self.store.fault_hook = fail with self.assertRaises(RuntimeError): with self.store.transaction() as db: agreement(db); self.command(db) self.store.fault_hook = None self.assertEqual(self.count('agreements'), 0) self.assertEqual(self.count('status_history'), 0) self.assertEqual(self.count('commands'), 0) self.accept() def test_after_commit_failure_preserves_response_for_retry(self): def fail(stage): if stage == 'after_commit': raise RuntimeError('lost response') self.store.fault_hook = fail with self.assertRaises(RuntimeError): with self.store.transaction() as db: agreement(db); self.command(db) reopened = FreeStore(self.path, IDENTITY) with reopened.transaction() as db: row = reopened.check_command(db, 'agent', '/agreements', 'key-1', b'{ "command": true }') self.assertEqual(row['result'], b'original.response') self.assertEqual(self.count('agreements'), 1) def test_actual_process_exit_before_and_after_commit(self): script = ''' import os,sys sys.path.insert(0,'tests') from test_free_store import IDENTITY,agreement from profiles.free_store import FreeStore def fault(stage): if stage==sys.argv[2]: os._exit(84) store=FreeStore(sys.argv[1],IDENTITY,fault) with store.transaction() as db: agreement(db) store.record_command(db,'agent','/agreements','key-1',b'request',201,b'{}',b'original-receipt') ''' for stage, expected in [('before_commit', 0), ('after_commit', 1)]: with self.subTest(stage=stage): result = subprocess.run([sys.executable, '-c', script, str(self.path), stage], text=True, capture_output=True, timeout=10) self.assertEqual(result.returncode, 84, result.stderr) reopened = FreeStore(self.path, IDENTITY) with reopened.connect() as db: self.assertEqual(db.execute('SELECT COUNT(*) FROM agreements').fetchone()[0], expected) self.assertEqual(db.execute('SELECT COUNT(*) FROM status_history').fetchone()[0], expected) self.assertEqual(db.execute('SELECT COUNT(*) FROM commands').fetchone()[0], expected) def test_concurrent_exact_acceptance_retries_have_single_receipt(self): barrier = threading.Barrier(2) def writer(index): store = FreeStore(self.path, IDENTITY); barrier.wait() with store.transaction() as db: prior = store.check_command(db, 'agent', '/agreements', 'key-1', b'acceptance') if prior: return prior['result'] agreement(db) result = ('receipt-' + str(index)).encode() return store.record_command(db, 'agent', '/agreements', 'key-1', b'acceptance', 201, b'{}', result)['result'] with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool: values = list(pool.map(writer, (1, 2))) self.assertEqual(values[0], values[1]) self.assertEqual(self.count('agreements'), 1) self.assertEqual(self.count('commands'), 1) self.assertEqual(self.count('status_history'), 1) def test_concurrent_conflicting_commands_have_single_winner(self): barrier = threading.Barrier(2) def writer(index): store = FreeStore(self.path, IDENTITY); barrier.wait() try: with store.transaction() as db: store.record_command(db, 'agent', '/revoke', 'shared-key', str(index).encode(), 200, b'{}', b'response') return True except FreeStoreConflict: return False with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool: values = list(pool.map(writer, (1, 2))) self.assertEqual(sorted(values), [False, True]) self.assertEqual(self.count('commands'), 1) def test_connection_context_closes_handle(self): with self.store.connect() as db: db.execute('SELECT 1') with self.assertRaises(sqlite3.ProgrammingError): db.execute('SELECT 1') def test_invalid_helper_inputs_rollback_without_records(self): for operation in [lambda db: self.store.put_document(db, 'text-not-bytes', 'text/plain'), lambda db: self.store.put_document(db, b'x' * 1048577, 'text/plain'), lambda db: self.store.record_command(db, 'agent', '/path', 'k', b'body', True, b'{}', b'response'), lambda db: self.store.record_record(db, 'r', 'agent', 'e', 'not-bytes', 'jws')]: with self.subTest(operation=operation), self.assertRaises(FreeStoreError): with self.store.transaction() as db: operation(db) for table in ('documents', 'commands', 'records'): self.assertEqual(self.count(table), 0) if __name__ == '__main__': unittest.main()