"""Test sqlcipher backend internals.""" import os import time import unittest try: import simplejson as json except ImportError: import json # noqa import threading from pysqlcipher import dbapi2 from StringIO import StringIO # u1db stuff. from u1db import ( errors, query_parser, sync, vectorclock, ) from u1db.backends.sqlite_backend import SQLitePartialExpandDatabase # soledad stuff. from leap.soledad.backends.sqlcipher import ( SQLCipherDatabase, DatabaseIsNotEncrypted, ) from leap.soledad.backends.sqlcipher import open as u1db_open from leap.soledad.backends.leap_backend import ( LeapDocument, EncryptionSchemes, decrypt_doc_json, encrypt_doc_json, ) # u1db tests stuff. from leap.soledad.tests import u1db_tests as tests, BaseSoledadTest from leap.soledad.tests.u1db_tests import test_sqlite_backend from leap.soledad.tests.u1db_tests import test_backends from leap.soledad.tests.u1db_tests import test_open from leap.soledad.tests.u1db_tests import test_sync from leap.soledad.backends.leap_backend import LeapSyncTarget from leap.common.testing.basetest import BaseLeapTest PASSWORD = '123456' #----------------------------------------------------------------------------- # The following tests come from `u1db.tests.test_common_backend`. #----------------------------------------------------------------------------- class TestSQLCipherBackendImpl(tests.TestCase): def test__allocate_doc_id(self): db = SQLCipherDatabase(':memory:', PASSWORD) doc_id1 = db._allocate_doc_id() self.assertTrue(doc_id1.startswith('D-')) self.assertEqual(34, len(doc_id1)) int(doc_id1[len('D-'):], 16) self.assertNotEqual(doc_id1, db._allocate_doc_id()) #----------------------------------------------------------------------------- # The following tests come from `u1db.tests.test_backends`. #----------------------------------------------------------------------------- def make_sqlcipher_database_for_test(test, replica_uid): db = SQLCipherDatabase(':memory:', PASSWORD) db._set_replica_uid(replica_uid) return db def copy_sqlcipher_database_for_test(test, db): # DO NOT COPY OR REUSE THIS CODE OUTSIDE TESTS: COPYING U1DB DATABASES IS # THE WRONG THING TO DO, THE ONLY REASON WE DO SO HERE IS TO TEST THAT WE # CORRECTLY DETECT IT HAPPENING SO THAT WE CAN RAISE ERRORS RATHER THAN # CORRUPT USER DATA. USE SYNC INSTEAD, OR WE WILL SEND NINJA TO YOUR # HOUSE. new_db = SQLCipherDatabase(':memory:', PASSWORD) tmpfile = StringIO() for line in db._db_handle.iterdump(): if not 'sqlite_sequence' in line: # work around bug in iterdump tmpfile.write('%s\n' % line) tmpfile.seek(0) new_db._db_handle = dbapi2.connect(':memory:') new_db._db_handle.cursor().executescript(tmpfile.read()) new_db._db_handle.commit() new_db._set_replica_uid(db._replica_uid) new_db._factory = db._factory return new_db def make_document_for_test(test, doc_id, rev, content, has_conflicts=False): return LeapDocument(doc_id, rev, content, has_conflicts=has_conflicts) SQLCIPHER_SCENARIOS = [ ('sqlcipher', {'make_database_for_test': make_sqlcipher_database_for_test, 'copy_database_for_test': copy_sqlcipher_database_for_test, 'make_document_for_test': make_document_for_test, }), ] class SQLCipherTests(test_backends.AllDatabaseTests): scenarios = SQLCIPHER_SCENARIOS class SQLCipherDatabaseTests(test_backends.LocalDatabaseTests): scenarios = SQLCIPHER_SCENARIOS class SQLCipherValidateGenNTransIdTests( test_backends.LocalDatabaseValidateGenNTransIdTests): scenarios = SQLCIPHER_SCENARIOS class SQLCipherValidateSourceGenTests( test_backends.LocalDatabaseValidateSourceGenTests): scenarios = SQLCIPHER_SCENARIOS class SQLCipherWithConflictsTests( test_backends.LocalDatabaseWithConflictsTests): scenarios = SQLCIPHER_SCENARIOS class SQLCipherIndexTests(test_backends.DatabaseIndexTests): scenarios = SQLCIPHER_SCENARIOS load_tests = tests.load_with_scenarios #----------------------------------------------------------------------------- # The following tests come from `u1db.tests.test_sqlite_backend`. #----------------------------------------------------------------------------- class TestSQLCipherDatabase(test_sqlite_backend.TestSQLiteDatabase): def test_atomic_initialize(self): tmpdir = self.createTempDir() dbname = os.path.join(tmpdir, 'atomic.db') t2 = None # will be a thread class SQLCipherDatabaseTesting(SQLCipherDatabase): _index_storage_value = "testing" def __init__(self, dbname, ntry): self._try = ntry self._is_initialized_invocations = 0 SQLCipherDatabase.__init__(self, dbname, PASSWORD) def _is_initialized(self, c): res = \ SQLCipherDatabase._is_initialized(self, c) if self._try == 1: self._is_initialized_invocations += 1 if self._is_initialized_invocations == 2: t2.start() # hard to do better and have a generic test time.sleep(0.05) return res outcome2 = [] def second_try(): try: db2 = SQLCipherDatabaseTesting(dbname, 2) except Exception, e: outcome2.append(e) else: outcome2.append(db2) t2 = threading.Thread(target=second_try) db1 = SQLCipherDatabaseTesting(dbname, 1) t2.join() self.assertIsInstance(outcome2[0], SQLCipherDatabaseTesting) db2 = outcome2[0] self.assertTrue(db2._is_initialized(db1._get_sqlite_handle().cursor())) class TestAlternativeDocument(LeapDocument): """A (not very) alternative implementation of Document.""" class TestSQLCipherPartialExpandDatabase( test_sqlite_backend.TestSQLitePartialExpandDatabase): # The following tests had to be cloned from u1db because they all # instantiate the backend directly, so we need to change that in order to # our backend be instantiated in place. def setUp(self): test_sqlite_backend.TestSQLitePartialExpandDatabase.setUp(self) self.db = SQLCipherDatabase(':memory:', PASSWORD) self.db._set_replica_uid('test') def test_default_replica_uid(self): self.db = SQLCipherDatabase(':memory:', PASSWORD) self.assertIsNot(None, self.db._replica_uid) self.assertEqual(32, len(self.db._replica_uid)) int(self.db._replica_uid, 16) def test__parse_index(self): self.db = SQLCipherDatabase(':memory:', PASSWORD) g = self.db._parse_index_definition('fieldname') self.assertIsInstance(g, query_parser.ExtractField) self.assertEqual(['fieldname'], g.field) def test__update_indexes(self): self.db = SQLCipherDatabase(':memory:', PASSWORD) g = self.db._parse_index_definition('fieldname') c = self.db._get_sqlite_handle().cursor() self.db._update_indexes('doc-id', {'fieldname': 'val'}, [('fieldname', g)], c) c.execute('SELECT doc_id, field_name, value FROM document_fields') self.assertEqual([('doc-id', 'fieldname', 'val')], c.fetchall()) def test__set_replica_uid(self): # Start from scratch, so that replica_uid isn't set. self.db = SQLCipherDatabase(':memory:', PASSWORD) self.assertIsNot(None, self.db._real_replica_uid) self.assertIsNot(None, self.db._replica_uid) self.db._set_replica_uid('foo') c = self.db._get_sqlite_handle().cursor() c.execute("SELECT value FROM u1db_config WHERE name='replica_uid'") self.assertEqual(('foo',), c.fetchone()) self.assertEqual('foo', self.db._real_replica_uid) self.assertEqual('foo', self.db._replica_uid) self.db._close_sqlite_handle() self.assertEqual('foo', self.db._replica_uid) def test__open_database(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/test.sqlite' SQLCipherDatabase(path, PASSWORD) db2 = SQLCipherDatabase._open_database(path, PASSWORD) self.assertIsInstance(db2, SQLCipherDatabase) def test__open_database_with_factory(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/test.sqlite' SQLCipherDatabase(path, PASSWORD) db2 = SQLCipherDatabase._open_database( path, PASSWORD, document_factory=TestAlternativeDocument) doc = db2.create_doc({}) self.assertTrue(isinstance(doc, LeapDocument)) def test__open_database_non_existent(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/non-existent.sqlite' self.assertRaises(errors.DatabaseDoesNotExist, SQLCipherDatabase._open_database, path, PASSWORD) def test__open_database_during_init(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/initialised.db' db = SQLCipherDatabase.__new__( SQLCipherDatabase) db._db_handle = dbapi2.connect(path) # db is there but not yet init-ed c = db._db_handle.cursor() c.execute('PRAGMA key="%s"' % PASSWORD) self.addCleanup(db.close) observed = [] class SQLiteDatabaseTesting(SQLCipherDatabase): WAIT_FOR_PARALLEL_INIT_HALF_INTERVAL = 0.1 @classmethod def _which_index_storage(cls, c): res = SQLCipherDatabase._which_index_storage(c) db._ensure_schema() # init db observed.append(res[0]) return res db2 = SQLiteDatabaseTesting._open_database(path, PASSWORD) self.addCleanup(db2.close) self.assertIsInstance(db2, SQLCipherDatabase) self.assertEqual( [None, SQLCipherDatabase._index_storage_value], observed) def test__open_database_invalid(self): class SQLiteDatabaseTesting(SQLCipherDatabase): WAIT_FOR_PARALLEL_INIT_HALF_INTERVAL = 0.1 temp_dir = self.createTempDir(prefix='u1db-test-') path1 = temp_dir + '/invalid1.db' with open(path1, 'wb') as f: f.write("") self.assertRaises(dbapi2.OperationalError, SQLiteDatabaseTesting._open_database, path1, PASSWORD) with open(path1, 'wb') as f: f.write("invalid") self.assertRaises(dbapi2.DatabaseError, SQLiteDatabaseTesting._open_database, path1, PASSWORD) def test_open_database_existing(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/existing.sqlite' SQLCipherDatabase(path, PASSWORD) db2 = SQLCipherDatabase.open_database(path, PASSWORD, create=False) self.assertIsInstance(db2, SQLCipherDatabase) def test_open_database_with_factory(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/existing.sqlite' SQLCipherDatabase(path, PASSWORD) db2 = SQLCipherDatabase.open_database( path, PASSWORD, create=False, document_factory=TestAlternativeDocument) doc = db2.create_doc({}) self.assertTrue(isinstance(doc, LeapDocument)) def test_open_database_create(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/new.sqlite' SQLCipherDatabase.open_database(path, PASSWORD, create=True) db2 = SQLCipherDatabase.open_database(path, PASSWORD, create=False) self.assertIsInstance(db2, SQLCipherDatabase) def test_create_database_initializes_schema(self): # This test had to be cloned because our implementation of SQLCipher # backend is referenced with an index_storage_value that includes the # word "encrypted". See u1db's sqlite_backend and our # sqlcipher_backend for reference. raw_db = self.db._get_sqlite_handle() c = raw_db.cursor() c.execute("SELECT * FROM u1db_config") config = dict([(r[0], r[1]) for r in c.fetchall()]) self.assertEqual({'sql_schema': '0', 'replica_uid': 'test', 'index_storage': 'expand referenced encrypted'}, config) def test_store_syncable(self): doc = self.db.create_doc_from_json(tests.simple_doc) # assert that docs are syncable by default self.assertEqual(True, doc.syncable) # assert that we can store syncable = False doc.syncable = False self.db.put_doc(doc) self.assertEqual(False, self.db.get_doc(doc.doc_id).syncable) # assert that we can store syncable = True doc.syncable = True self.db.put_doc(doc) self.assertEqual(True, self.db.get_doc(doc.doc_id).syncable) #----------------------------------------------------------------------------- # The following tests come from `u1db.tests.test_open`. #----------------------------------------------------------------------------- class SQLCipherOpen(test_open.TestU1DBOpen): def test_open_no_create(self): self.assertRaises(errors.DatabaseDoesNotExist, u1db_open, self.db_path, password=PASSWORD, create=False) self.assertFalse(os.path.exists(self.db_path)) def test_open_create(self): db = u1db_open(self.db_path, password=PASSWORD, create=True) self.addCleanup(db.close) self.assertTrue(os.path.exists(self.db_path)) self.assertIsInstance(db, SQLCipherDatabase) def test_open_with_factory(self): db = u1db_open(self.db_path, password=PASSWORD, create=True, document_factory=TestAlternativeDocument) self.addCleanup(db.close) doc = db.create_doc({}) self.assertTrue(isinstance(doc, LeapDocument)) def test_open_existing(self): db = SQLCipherDatabase(self.db_path, PASSWORD) self.addCleanup(db.close) doc = db.create_doc_from_json(tests.simple_doc) # Even though create=True, we shouldn't wipe the db db2 = u1db_open(self.db_path, password=PASSWORD, create=True) self.addCleanup(db2.close) doc2 = db2.get_doc(doc.doc_id) self.assertEqual(doc, doc2) def test_open_existing_no_create(self): db = SQLCipherDatabase(self.db_path, PASSWORD) self.addCleanup(db.close) db2 = u1db_open(self.db_path, password=PASSWORD, create=False) self.addCleanup(db2.close) self.assertIsInstance(db2, SQLCipherDatabase) #----------------------------------------------------------------------------- # The following tests come from `u1db.tests.test_sync`. #----------------------------------------------------------------------------- sync_scenarios = [] for name, scenario in SQLCIPHER_SCENARIOS: scenario = dict(scenario) scenario['do_sync'] = test_sync.sync_via_synchronizer sync_scenarios.append((name, scenario)) scenario = dict(scenario) def sync_via_synchronizer_and_leap(test, db_source, db_target, trace_hook=None, trace_hook_shallow=None): if trace_hook: test.skipTest("full trace hook unsupported over http") path = test._http_at[db_target] target = LeapSyncTarget.connect(test.getURL(path), test._soledad._crypto) if trace_hook_shallow: target._set_trace_hook_shallow(trace_hook_shallow) return sync.Synchronizer(db_source, target).sync() sync_scenarios.append(('pyleap', { 'make_database_for_test': test_sync.make_database_for_http_test, 'copy_database_for_test': test_sync.copy_database_for_http_test, 'make_document_for_test': make_document_for_test, 'make_app_with_state': tests.test_remote_sync_target.make_http_app, 'do_sync': sync_via_synchronizer_and_leap, })) class SQLCipherDatabaseSyncTests( test_sync.DatabaseSyncTests, BaseSoledadTest): scenarios = sync_scenarios def setUp(self): test_sync.DatabaseSyncTests.setUp(self) BaseSoledadTest.setUp(self) def tearDown(self): test_sync.DatabaseSyncTests.tearDown(self) BaseSoledadTest.tearDown(self) def test_sync_autoresolves(self): # The remote database can't autoresolve conflicts based on magic # content convergence, so we modify this test to leave the possibility # of the remode document ending up in conflicted state. self.db1 = self.create_database('test1', 'source') self.db2 = self.create_database('test2', 'target') doc1 = self.db1.create_doc_from_json(tests.simple_doc, doc_id='doc') rev1 = doc1.rev doc2 = self.db2.create_doc_from_json(tests.simple_doc, doc_id='doc') rev2 = doc2.rev self.sync(self.db1, self.db2) doc = self.db1.get_doc('doc') self.assertFalse(doc.has_conflicts) # if remote content is in conflicted state, then document revisions # will be different. #self.assertEqual(doc.rev, self.db2.get_doc('doc').rev) v = vectorclock.VectorClockRev(doc.rev) self.assertTrue(v.is_newer(vectorclock.VectorClockRev(rev1))) self.assertTrue(v.is_newer(vectorclock.VectorClockRev(rev2))) def test_sync_autoresolves_moar(self): # here we test that when a database that has a conflicted document is # the source of a sync, and the target database has a revision of the # conflicted document that is newer than the source database's, and # that target's database's document's content is the same as the # source's document's conflict's, the source's document's conflict gets # autoresolved, and the source's document's revision bumped. # # idea is as follows: # A B # a1 - # `-------> # a1 a1 # v v # a2 a1b1 # `-------> # a1b1+a2 a1b1 # v # a1b1+a2 a1b2 (a1b2 has same content as a2) # `-------> # a3b2 a1b2 (autoresolved) # `-------> # a3b2 a3b2 self.db1 = self.create_database('test1', 'source') self.db2 = self.create_database('test2', 'target') self.db1.create_doc_from_json(tests.simple_doc, doc_id='doc') self.sync(self.db1, self.db2) for db, content in [(self.db1, '{}'), (self.db2, '{"hi": 42}')]: doc = db.get_doc('doc') doc.set_json(content) db.put_doc(doc) self.sync(self.db1, self.db2) # db1 and db2 now both have a doc of {hi:42}, but db1 has a conflict doc = self.db1.get_doc('doc') rev1 = doc.rev self.assertTrue(doc.has_conflicts) # set db2 to have a doc of {} (same as db1 before the conflict) doc = self.db2.get_doc('doc') doc.set_json('{}') self.db2.put_doc(doc) rev2 = doc.rev # sync it across self.sync(self.db1, self.db2) # tadaa! doc = self.db1.get_doc('doc') self.assertFalse(doc.has_conflicts) vec1 = vectorclock.VectorClockRev(rev1) vec2 = vectorclock.VectorClockRev(rev2) vec3 = vectorclock.VectorClockRev(doc.rev) self.assertTrue(vec3.is_newer(vec1)) self.assertTrue(vec3.is_newer(vec2)) # because the conflict is on the source, sync it another time self.sync(self.db1, self.db2) # make sure db2 now has the exact same thing doc1 = self.db1.get_doc('doc') doc2 = self.db1.get_doc('doc') if '_encryption_scheme' in doc2.content: doc2.set_json( decrypt_doc_json( self._soledad._crypto, doc2, doc2.get_json())) self.assertEqual(doc1, doc2) def test_sync_autoresolves_moar_backwards(self): # here we would test that when a database that has a conflicted # document is the target of a sync, and the source database has a # revision of the conflicted document that is newer than the target # database's, and that source's database's document's content is the # same as the target's document's conflict's, the target's document's # conflict gets autoresolved, and the document's revision bumped. # # Despite that, in Soledad we suppose that the server never syncs, so # it never has conflicted documents. Also, if it had, convergence # would not be possible in terms of document's contents. # # Therefore we suppress this test. pass def test_sync_autoresolves_moar_backwards_three(self): # We use the same reasoning from the last test to suppress this one. pass def test_sync_propagates_resolution(self): self.db1 = self.create_database('test1', 'both') self.db2 = self.create_database('test2', 'both') doc1 = self.db1.create_doc_from_json('{"a": 1}', doc_id='the-doc') db3 = self.create_database('test3', 'both') self.sync(self.db2, self.db1) self.assertEqual( self.db1._get_generation_info(), self.db2._get_replica_gen_and_trans_id(self.db1._replica_uid)) self.assertEqual( self.db2._get_generation_info(), self.db1._get_replica_gen_and_trans_id(self.db2._replica_uid)) self.sync(db3, self.db1) # update on 2 doc2 = self.make_document('the-doc', doc1.rev, '{"a": 2}') self.db2.put_doc(doc2) self.sync(self.db2, db3) self.assertEqual(db3.get_doc('the-doc').rev, doc2.rev) # update on 1 doc1.set_json('{"a": 3}') self.db1.put_doc(doc1) # conflicts self.sync(self.db2, self.db1) self.sync(db3, self.db1) self.assertTrue(self.db2.get_doc('the-doc').has_conflicts) self.assertTrue(db3.get_doc('the-doc').has_conflicts) # resolve conflicts = self.db2.get_doc_conflicts('the-doc') doc4 = self.make_document('the-doc', None, '{"a": 4}') revs = [doc.rev for doc in conflicts] self.db2.resolve_doc(doc4, revs) doc2 = self.db2.get_doc('the-doc') self.assertEqual(doc4.get_json(), doc2.get_json()) self.assertFalse(doc2.has_conflicts) self.sync(self.db2, db3) doc3 = db3.get_doc('the-doc') if '_encryption_scheme' in doc3.content: doc3.set_json( decrypt_doc_json( self._soledad._crypto, doc3.doc_id, doc3.get_json())) self.assertEqual(doc4.get_json(), doc3.get_json()) self.assertFalse(doc3.has_conflicts) def test_sync_puts_changes(self): self.db1 = self.create_database('test1', 'source') self.db2 = self.create_database('test2', 'target') doc = self.db1.create_doc_from_json(tests.simple_doc) self.assertEqual(1, self.sync(self.db1, self.db2)) exp_doc = self.make_document( doc.doc_id, doc.rev, tests.simple_doc, False) doc2 = self.db2.get_doc(doc.doc_id) # decrypt to compare it it is the case if '_encryption_scheme' in doc2.content: doc2 = self.db2.get_doc(doc.doc_id) doc2.set_json( decrypt_doc_json( self._soledad._crypto, doc.doc_id, doc2.get_json())) self.assertEqual(exp_doc, doc2) self.assertEqual(1, self.db1._get_replica_gen_and_trans_id('test2')[0]) self.assertEqual(1, self.db2._get_replica_gen_and_trans_id('test1')[0]) self.assertLastExchangeLog( self.db2, {'receive': {'docs': [(doc.doc_id, doc.rev)], 'source_uid': 'test1', 'source_gen': 1, 'last_known_gen': 0}, 'return': {'docs': [], 'last_gen': 1}}) def _make_local_db_and_leap_target(test, path='test'): test.startServer() db = test.request_state._create_database(os.path.basename(path)) st = LeapSyncTarget.connect(test.getURL(path), test._soledad._crypto) return db, st target_scenarios = [ ('leap', { 'create_db_and_target': _make_local_db_and_leap_target, 'make_app_with_state': tests.test_remote_sync_target.make_http_app}), ] class SQLCipherSyncTargetTests( test_sync.DatabaseSyncTargetTests, BaseSoledadTest): scenarios = (tests.multiply_scenarios(SQLCIPHER_SCENARIOS, target_scenarios)) def setUp(self): test_sync.DatabaseSyncTargetTests.setUp(self) BaseSoledadTest.setUp(self) def tearDown(self): test_sync.DatabaseSyncTargetTests.tearDown(self) BaseSoledadTest.tearDown(self) def test_sync_exchange(self): """ Modified to account for possibly receiving encrypted documents from sever-side. """ docs_by_gen = [ (self.make_document('doc-id', 'replica:1', tests.simple_doc), 10, 'T-sid')] new_gen, trans_id = self.st.sync_exchange( docs_by_gen, 'replica', last_known_generation=0, last_known_trans_id=None, return_doc_cb=self.receive_doc) # decrypt doc1 for comparison if needed tmpdoc = self.db.get_doc('doc-id') if '_encryption_scheme' in tmpdoc.content: tmpdoc.set_json( decrypt_doc_json( self._soledad._crypto, tmpdoc.doc_id, tmpdoc.get_json())) self.assertEqual(tmpdoc.rev, 'replica:1') self.assertEqual(tmpdoc.content, json.loads(tests.simple_doc)) self.assertFalse(tmpdoc.has_conflicts) self.assertTransactionLog(['doc-id'], self.db) last_trans_id = self.getLastTransId(self.db) self.assertEqual(([], 1, last_trans_id), (self.other_changes, new_gen, last_trans_id)) self.assertEqual(10, self.st.get_sync_info('replica')[3]) def test_sync_exchange_push_many(self): """ Modified to account for possibly receiving encrypted documents from sever-side. """ docs_by_gen = [ (self.make_document('doc-id', 'replica:1', tests.simple_doc), 10, 'T-1'), (self.make_document('doc-id2', 'replica:1', tests.nested_doc), 11, 'T-2')] new_gen, trans_id = self.st.sync_exchange( docs_by_gen, 'replica', last_known_generation=0, last_known_trans_id=None, return_doc_cb=self.receive_doc) # decrypt doc1 for comparison if needed tmpdoc1 = self.db.get_doc('doc-id') if '_encryption_scheme' in tmpdoc1.content: tmpdoc1.set_json( decrypt_doc_json( self._soledad._crypto, tmpdoc1.doc_id, tmpdoc1.get_json())) self.assertEqual(tmpdoc1.rev, 'replica:1') self.assertEqual(tmpdoc1.content, json.loads(tests.simple_doc)) self.assertFalse(tmpdoc1.has_conflicts) # decrypt doc2 for comparison if needed tmpdoc2 = self.db.get_doc('doc-id2') if '_encryption_scheme' in tmpdoc2.content: tmpdoc2.set_json( decrypt_doc_json( self._soledad._crypto, tmpdoc2.doc_id, tmpdoc2.get_json())) self.assertEqual(tmpdoc2.rev, 'replica:1') self.assertEqual(tmpdoc2.content, json.loads(tests.nested_doc)) self.assertFalse(tmpdoc2.has_conflicts) self.assertTransactionLog(['doc-id', 'doc-id2'], self.db) last_trans_id = self.getLastTransId(self.db) self.assertEqual(([], 2, last_trans_id), (self.other_changes, new_gen, trans_id)) self.assertEqual(11, self.st.get_sync_info('replica')[3]) def test_sync_exchange_returns_many_new_docs(self): """ Modified to account for JSON serialization differences. """ doc = self.db.create_doc_from_json(tests.simple_doc) doc2 = self.db.create_doc_from_json(tests.nested_doc) self.assertTransactionLog([doc.doc_id, doc2.doc_id], self.db) new_gen, _ = self.st.sync_exchange( [], 'other-replica', last_known_generation=0, last_known_trans_id=None, return_doc_cb=self.receive_doc) self.assertTransactionLog([doc.doc_id, doc2.doc_id], self.db) self.assertEqual(2, new_gen) self.assertEqual( [(doc.doc_id, doc.rev, 1), (doc2.doc_id, doc2.rev, 2)], [c[:2]+c[3:4] for c in self.other_changes]) self.assertEqual( json.dumps(tests.simple_doc), json.dumps(self.other_changes[0][2])) self.assertEqual( json.loads(tests.nested_doc), json.loads(self.other_changes[1][2])) if self.whitebox: self.assertEqual( self.db._last_exchange_log['return'], {'last_gen': 2, 'docs': [(doc.doc_id, doc.rev), (doc2.doc_id, doc2.rev)]}) #----------------------------------------------------------------------------- # Tests for actual encryption of the database #----------------------------------------------------------------------------- class SQLCipherEncryptionTest(BaseLeapTest): """ Tests to guarantee SQLCipher is indeed encrypting data when storing. """ def _delete_dbfiles(self): for dbfile in [self.DB_FILE]: if os.path.exists(dbfile): os.unlink(dbfile) def setUp(self): self.DB_FILE = self.tempdir + '/test.db' self._delete_dbfiles() def tearDown(self): self._delete_dbfiles() def test_try_to_open_encrypted_db_with_sqlite_backend(self): """ SQLite backend should not succeed to open SQLCipher databases. """ db = SQLCipherDatabase(self.DB_FILE, PASSWORD) doc = db.create_doc_from_json(tests.simple_doc) db.close() try: # trying to open an encrypted database with the regular u1db # backend should raise a DatabaseError exception. SQLitePartialExpandDatabase(self.DB_FILE, document_factory=LeapDocument) raise DatabaseIsNotEncrypted() except dbapi2.DatabaseError: # at this point we know that the regular U1DB sqlcipher backend # did not succeed on opening the database, so it was indeed # encrypted. db = SQLCipherDatabase(self.DB_FILE, PASSWORD) doc = db.get_doc(doc.doc_id) self.assertEqual(tests.simple_doc, doc.get_json(), 'decrypted content mismatch') def test_try_to_open_raw_db_with_sqlcipher_backend(self): """ SQLCipher backend should not succeed to open unencrypted databases. """ db = SQLitePartialExpandDatabase(self.DB_FILE, document_factory=LeapDocument) db.create_doc_from_json(tests.simple_doc) db.close() try: # trying to open the a non-encrypted database with sqlcipher # backend should raise a DatabaseIsNotEncrypted exception. SQLCipherDatabase(self.DB_FILE, PASSWORD) raise db1pi2.DatabaseError( "SQLCipher backend should not be able to open non-encrypted " "dbs.") except DatabaseIsNotEncrypted: pass load_tests = tests.load_with_scenarios