From e759ab61e8a22235c45192750b536612a6c2cb8a Mon Sep 17 00:00:00 2001 From: Victor Shyba Date: Wed, 27 Jan 2016 21:35:43 -0300 Subject: [refactor] isolate requests Isolate requests lib related code and update docstrings. --- src/leap/keymanager/tests/test_keymanager.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) (limited to 'src/leap/keymanager/tests') diff --git a/src/leap/keymanager/tests/test_keymanager.py b/src/leap/keymanager/tests/test_keymanager.py index 856d6da..afcbd5f 100644 --- a/src/leap/keymanager/tests/test_keymanager.py +++ b/src/leap/keymanager/tests/test_keymanager.py @@ -342,7 +342,7 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): yield km.fetch_key(ADDRESS_OTHER, REMOTE_KEY_URL, OpenPGPKey) - get_mock.assert_called_once_with(REMOTE_KEY_URL, data=None, + get_mock.assert_called_once_with(REMOTE_KEY_URL, verify=ca_bundle.where()) @inlineCallbacks @@ -353,7 +353,7 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): yield km.fetch_key(ADDRESS_OTHER, REMOTE_KEY_URL, OpenPGPKey) - get_mock.assert_called_once_with(REMOTE_KEY_URL, data=None, + get_mock.assert_called_once_with(REMOTE_KEY_URL, verify=ca_bundle.where()) @inlineCallbacks @@ -364,7 +364,7 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): yield km.fetch_key(ADDRESS_OTHER, REMOTE_KEY_URL, OpenPGPKey) - get_mock.assert_called_once_with(REMOTE_KEY_URL, data=None, + get_mock.assert_called_once_with(REMOTE_KEY_URL, verify=ca_bundle.where()) @inlineCallbacks @@ -383,7 +383,7 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): yield km.fetch_key(ADDRESS_OTHER, REMOTE_KEY_URL, OpenPGPKey) # assert that combined bundle file is passed to get call - get_mock.assert_called_once_with(REMOTE_KEY_URL, data=None, + get_mock.assert_called_once_with(REMOTE_KEY_URL, verify=tmp_output.name) # assert that files got appended -- cgit v1.2.3 From cb32761d3dc97956085cd34a8f1e2e06432e8141 Mon Sep 17 00:00:00 2001 From: Victor Shyba Date: Wed, 27 Jan 2016 23:18:04 -0300 Subject: [feat] use HTTPClient instead of requests This commit adapts code to use HTTPClient instead of requests. requests library receives a certificate as parameter during requests while HTTPClient recelives a cert only on constructor. In order to have both types (leap cert and commercial certs) working together we introduced two clients on constructor. --- src/leap/keymanager/tests/test_keymanager.py | 109 +++++++++++---------------- 1 file changed, 45 insertions(+), 64 deletions(-) (limited to 'src/leap/keymanager/tests') diff --git a/src/leap/keymanager/tests/test_keymanager.py b/src/leap/keymanager/tests/test_keymanager.py index afcbd5f..c3426b6 100644 --- a/src/leap/keymanager/tests/test_keymanager.py +++ b/src/leap/keymanager/tests/test_keymanager.py @@ -21,11 +21,14 @@ Tests for the Key Manager. """ from os import path +import json +import urllib from datetime import datetime import tempfile +import pkg_resources from leap.common import ca_bundle from mock import Mock, MagicMock, patch -from twisted.internet.defer import inlineCallbacks +from twisted.internet import defer from twisted.trial import unittest from leap.keymanager import ( @@ -127,7 +130,7 @@ class KeyManagerUtilTestCase(unittest.TestCase): class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): - @inlineCallbacks + @defer.inlineCallbacks def test_get_all_keys_in_db(self): km = self._key_manager() yield km._wrapper_map[OpenPGPKey].put_ascii_key(PRIVATE_KEY, ADDRESS) @@ -142,7 +145,7 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): self.assertTrue(ADDRESS in keys[0].address) self.assertTrue(keys[0].private) - @inlineCallbacks + @defer.inlineCallbacks def test_get_public_key(self): km = self._key_manager() yield km._wrapper_map[OpenPGPKey].put_ascii_key(PRIVATE_KEY, ADDRESS) @@ -155,7 +158,7 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): key.fingerprint.lower(), KEY_FINGERPRINT.lower()) self.assertFalse(key.private) - @inlineCallbacks + @defer.inlineCallbacks def test_get_private_key(self): km = self._key_manager() yield km._wrapper_map[OpenPGPKey].put_ascii_key(PRIVATE_KEY, ADDRESS) @@ -173,7 +176,7 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): d = km.send_key(OpenPGPKey) return self.assertFailure(d, KeyNotFound) - @inlineCallbacks + @defer.inlineCallbacks def test_send_key(self): """ Test that request is well formed when sending keys to server. @@ -181,7 +184,7 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): token = "mytoken" km = self._key_manager(token=token) yield km._wrapper_map[OpenPGPKey].put_ascii_key(PUBLIC_KEY, ADDRESS) - km._fetcher.put = Mock() + km._async_client_pinned.request = Mock(return_value=defer.succeed('')) # the following data will be used on the send km.ca_cert_path = 'capath' km.session_id = 'sessionid' @@ -191,13 +194,15 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): yield km.send_key(OpenPGPKey) # setup expected args pubkey = yield km.get_key(km._address, OpenPGPKey) - data = { + data = urllib.urlencode({ km.PUBKEY_KEY: pubkey.key_data, - } + }) + headers = {'Authorization': [str('Token token=%s' % token)]} + headers['Content-Type'] = ['application/x-www-form-urlencoded'] url = '%s/%s/users/%s.json' % ('apiuri', 'apiver', 'myuid') - km._fetcher.put.assert_called_once_with( - url, data=data, verify='capath', - headers={'Authorization': 'Token token=%s' % token}, + km._async_client_pinned.request.assert_called_once_with( + str(url), 'PUT', body=str(data), + headers=headers ) def test_fetch_keys_from_server(self): @@ -205,19 +210,19 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): Test that the request is well formed when fetching keys from server. """ km = self._key_manager(url=NICKSERVER_URI) + expected_url = NICKSERVER_URI + '?address=' + ADDRESS_2 def verify_the_call(_): - km._fetcher.get.assert_called_once_with( - NICKSERVER_URI, - data={'address': ADDRESS_2}, - verify='cacertpath', + km._async_client_pinned.request.assert_called_once_with( + expected_url, + 'GET', ) d = self._fetch_key(km, ADDRESS_2, PUBLIC_KEY_2) d.addCallback(verify_the_call) return d - @inlineCallbacks + @defer.inlineCallbacks def test_get_key_fetches_from_server(self): """ Test that getting a key successfuly fetches from server. @@ -229,7 +234,7 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): self.assertTrue(ADDRESS in key.address) self.assertEqual(key.validation, ValidationLevels.Provider_Trust) - @inlineCallbacks + @defer.inlineCallbacks def test_get_key_fetches_other_domain(self): """ Test that getting a key successfuly fetches from server. @@ -245,18 +250,10 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): """ :returns: a Deferred that will fire with the OpenPGPKey """ - class Response(object): - status_code = 200 - headers = {'content-type': 'application/json'} - - def json(self): - return {'address': address, 'openpgp': key} - - def raise_for_status(self): - pass + data = json.dumps({'address': address, 'openpgp': key}) # mock the fetcher so it returns the key for ADDRESS_2 - km._fetcher.get = Mock(return_value=Response()) + km._async_client_pinned.request = Mock(return_value=defer.succeed(data)) km.ca_cert_path = 'cacertpath' # try to key get without fetching from server d_fail = km.get_key(address, OpenPGPKey, fetch_remote=False) @@ -265,7 +262,7 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): d.addCallback(lambda _: km.get_key(address, OpenPGPKey)) return d - @inlineCallbacks + @defer.inlineCallbacks def test_put_key_ascii(self): """ Test that putting ascii key works @@ -277,7 +274,7 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): self.assertIsInstance(key, OpenPGPKey) self.assertTrue(ADDRESS in key.address) - @inlineCallbacks + @defer.inlineCallbacks def test_fetch_uri_ascii_key(self): """ Test that fetch key downloads the ascii key and gets included in @@ -285,11 +282,7 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): """ km = self._key_manager() - class Response(object): - ok = True - content = PUBLIC_KEY - - km._fetcher.get = Mock(return_value=Response()) + km._async_client.request = Mock(return_value=defer.succeed(PUBLIC_KEY)) yield km.fetch_key(ADDRESS, "http://site.domain/key", OpenPGPKey) key = yield km.get_key(ADDRESS, OpenPGPKey) @@ -316,25 +309,16 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): """ km = self._key_manager() - class Response(object): - ok = True - content = PUBLIC_KEY - - km._fetcher.get = Mock(return_value=Response()) + km._async_client.request = Mock(return_value=defer.succeed(PUBLIC_KEY)) d = km.fetch_key(ADDRESS_2, "http://site.domain/key", OpenPGPKey) return self.assertFailure(d, KeyAddressMismatch) def _mock_get_response(self, km, body): - class Response(object): - ok = True - content = body - - mock = MagicMock(return_value=Response()) - km._fetcher.get = mock + km._async_client.request = MagicMock(return_value=defer.succeed(body)) - return mock + return km._async_client.request - @inlineCallbacks + @defer.inlineCallbacks def test_fetch_key_uses_ca_bundle_if_none_specified(self): ca_cert_path = None km = self._key_manager(ca_cert_path=ca_cert_path) @@ -342,10 +326,9 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): yield km.fetch_key(ADDRESS_OTHER, REMOTE_KEY_URL, OpenPGPKey) - get_mock.assert_called_once_with(REMOTE_KEY_URL, - verify=ca_bundle.where()) + get_mock.assert_called_once_with(REMOTE_KEY_URL, 'GET') - @inlineCallbacks + @defer.inlineCallbacks def test_fetch_key_uses_ca_bundle_if_empty_string_specified(self): ca_cert_path = '' km = self._key_manager(ca_cert_path=ca_cert_path) @@ -353,10 +336,9 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): yield km.fetch_key(ADDRESS_OTHER, REMOTE_KEY_URL, OpenPGPKey) - get_mock.assert_called_once_with(REMOTE_KEY_URL, - verify=ca_bundle.where()) + get_mock.assert_called_once_with(REMOTE_KEY_URL, 'GET') - @inlineCallbacks + @defer.inlineCallbacks def test_fetch_key_use_default_ca_bundle_if_set_as_ca_cert_path(self): ca_cert_path = ca_bundle.where() km = self._key_manager(ca_cert_path=ca_cert_path) @@ -364,14 +346,14 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): yield km.fetch_key(ADDRESS_OTHER, REMOTE_KEY_URL, OpenPGPKey) - get_mock.assert_called_once_with(REMOTE_KEY_URL, - verify=ca_bundle.where()) + get_mock.assert_called_once_with(REMOTE_KEY_URL, 'GET') - @inlineCallbacks + @defer.inlineCallbacks def test_fetch_uses_combined_ca_bundle_otherwise(self): with tempfile.NamedTemporaryFile() as tmp_input, \ tempfile.NamedTemporaryFile(delete=False) as tmp_output: - ca_content = 'some\ncontent\n' + ca_content = pkg_resources.resource_string('leap.common.testing', + 'cacert.pem') ca_cert_path = tmp_input.name self._dump_to_file(ca_cert_path, ca_content) @@ -383,8 +365,7 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): yield km.fetch_key(ADDRESS_OTHER, REMOTE_KEY_URL, OpenPGPKey) # assert that combined bundle file is passed to get call - get_mock.assert_called_once_with(REMOTE_KEY_URL, - verify=tmp_output.name) + get_mock.assert_called_once_with(REMOTE_KEY_URL, 'GET') # assert that files got appended expected = self._slurp_file(ca_bundle.where()) + ca_content @@ -402,7 +383,7 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): content = f.read() return content - @inlineCallbacks + @defer.inlineCallbacks def test_decrypt_updates_sign_used_for_signer(self): # given km = self._key_manager() @@ -420,7 +401,7 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): # then self.assertEqual(True, key.sign_used) - @inlineCallbacks + @defer.inlineCallbacks def test_decrypt_does_not_update_sign_used_for_recipient(self): # given km = self._key_manager() @@ -445,7 +426,7 @@ class KeyManagerCryptoTestCase(KeyManagerWithSoledadTestCase): RAW_DATA = 'data' - @inlineCallbacks + @defer.inlineCallbacks def test_keymanager_openpgp_encrypt_decrypt(self): km = self._key_manager() # put raw private key @@ -464,7 +445,7 @@ class KeyManagerCryptoTestCase(KeyManagerWithSoledadTestCase): fetch_remote=False) self.assertEqual(signingkey.fingerprint, key.fingerprint) - @inlineCallbacks + @defer.inlineCallbacks def test_keymanager_openpgp_encrypt_decrypt_wrong_sign(self): km = self._key_manager() # put raw keys @@ -481,7 +462,7 @@ class KeyManagerCryptoTestCase(KeyManagerWithSoledadTestCase): self.assertEqual(self.RAW_DATA, rawdata) self.assertTrue(isinstance(signingkey, errors.InvalidSignature)) - @inlineCallbacks + @defer.inlineCallbacks def test_keymanager_openpgp_sign_verify(self): km = self._key_manager() # put raw private keys -- cgit v1.2.3 From 4e13b7595fc5c8e245244eb535525ae8333ef9dc Mon Sep 17 00:00:00 2001 From: Ruben Pollan Date: Tue, 9 Feb 2016 16:32:53 +0100 Subject: [style] fix pep8 --- src/leap/keymanager/tests/test_keymanager.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) (limited to 'src/leap/keymanager/tests') diff --git a/src/leap/keymanager/tests/test_keymanager.py b/src/leap/keymanager/tests/test_keymanager.py index c3426b6..0f08326 100644 --- a/src/leap/keymanager/tests/test_keymanager.py +++ b/src/leap/keymanager/tests/test_keymanager.py @@ -253,7 +253,8 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): data = json.dumps({'address': address, 'openpgp': key}) # mock the fetcher so it returns the key for ADDRESS_2 - km._async_client_pinned.request = Mock(return_value=defer.succeed(data)) + km._async_client_pinned.request = Mock( + return_value=defer.succeed(data)) km.ca_cert_path = 'cacertpath' # try to key get without fetching from server d_fail = km.get_key(address, OpenPGPKey, fetch_remote=False) -- cgit v1.2.3 From c18b69441e24c57a3130e27356edc0b9395d78f8 Mon Sep 17 00:00:00 2001 From: Victor Shyba Date: Thu, 11 Feb 2016 19:22:34 -0300 Subject: [feat] defer decrypt, gen_key and encrypt This commit put those gnupg operations to be run on external threads limited by the amount of cores present on user machine. Some gnupg calls spawn processes and communicating to them is a synchronous operation, so running outside of a reactor should improve response time by avoiding reactor locking. --- src/leap/keymanager/tests/test_openpgp.py | 30 +++++++++++++++--------------- 1 file changed, 15 insertions(+), 15 deletions(-) (limited to 'src/leap/keymanager/tests') diff --git a/src/leap/keymanager/tests/test_openpgp.py b/src/leap/keymanager/tests/test_openpgp.py index bae83db..96b40a0 100644 --- a/src/leap/keymanager/tests/test_openpgp.py +++ b/src/leap/keymanager/tests/test_openpgp.py @@ -109,7 +109,7 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase): # encrypt yield pgp.put_ascii_key(PUBLIC_KEY, ADDRESS) pubkey = yield pgp.get_key(ADDRESS, private=False) - cyphertext = pgp.encrypt(data, pubkey) + cyphertext = yield pgp.encrypt(data, pubkey) self.assertTrue(cyphertext is not None) self.assertTrue(cyphertext != '') @@ -121,7 +121,7 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase): yield self._assert_key_not_found(pgp, ADDRESS, private=True) yield pgp.put_ascii_key(PRIVATE_KEY, ADDRESS) privkey = yield pgp.get_key(ADDRESS, private=True) - decrypted, _ = pgp.decrypt(cyphertext, privkey) + decrypted, _ = yield pgp.decrypt(cyphertext, privkey) self.assertEqual(decrypted, data) yield pgp.delete_key(pubkey) @@ -171,9 +171,9 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase): yield pgp.put_ascii_key(PRIVATE_KEY, ADDRESS) privkey = yield pgp.get_key(ADDRESS, private=True) pubkey = yield pgp.get_key(ADDRESS, private=False) - self.assertRaises( - AssertionError, - pgp.encrypt, data, privkey, sign=pubkey) + self.failureResultOf( + pgp.encrypt(data, privkey, sign=pubkey), + AssertionError) @inlineCallbacks def test_decrypt_verify_with_private_raises(self): @@ -183,12 +183,11 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase): yield pgp.put_ascii_key(PRIVATE_KEY, ADDRESS) privkey = yield pgp.get_key(ADDRESS, private=True) pubkey = yield pgp.get_key(ADDRESS, private=False) - encrypted_and_signed = pgp.encrypt( + encrypted_and_signed = yield pgp.encrypt( data, pubkey, sign=privkey) - self.assertRaises( - AssertionError, - pgp.decrypt, - encrypted_and_signed, privkey, verify=privkey) + self.failureResultOf( + pgp.decrypt(encrypted_and_signed, privkey, verify=privkey), + AssertionError) @inlineCallbacks def test_decrypt_verify_with_wrong_key(self): @@ -198,11 +197,12 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase): yield pgp.put_ascii_key(PRIVATE_KEY, ADDRESS) privkey = yield pgp.get_key(ADDRESS, private=True) pubkey = yield pgp.get_key(ADDRESS, private=False) - encrypted_and_signed = pgp.encrypt(data, pubkey, sign=privkey) + encrypted_and_signed = yield pgp.encrypt(data, pubkey, sign=privkey) yield pgp.put_ascii_key(PUBLIC_KEY_2, ADDRESS_2) wrongkey = yield pgp.get_key(ADDRESS_2) - decrypted, validsign = pgp.decrypt(encrypted_and_signed, privkey, - verify=wrongkey) + decrypted, validsign = yield pgp.decrypt(encrypted_and_signed, + privkey, + verify=wrongkey) self.assertEqual(decrypted, data) self.assertFalse(validsign) @@ -232,9 +232,9 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase): privkey2 = yield pgp.get_key(ADDRESS_2, private=True) data = 'data' - encrypted_and_signed = pgp.encrypt( + encrypted_and_signed = yield pgp.encrypt( data, pubkey2, sign=privkey) - res, validsign = pgp.decrypt( + res, validsign = yield pgp.decrypt( encrypted_and_signed, privkey2, verify=pubkey) self.assertEqual(data, res) self.assertTrue(validsign) -- cgit v1.2.3 From 64eaf2ee426623072bc2d9b1faf77ab831cb3be1 Mon Sep 17 00:00:00 2001 From: Ruben Pollan Date: Tue, 15 Dec 2015 13:29:44 +0100 Subject: [feat] move validation, usage and audited date to the active document - Resolves: #7485 --- src/leap/keymanager/tests/test_keymanager.py | 15 ++- src/leap/keymanager/tests/test_openpgp.py | 54 ++++++++- src/leap/keymanager/tests/test_validation.py | 164 ++++++++++++++++++++++++--- 3 files changed, 214 insertions(+), 19 deletions(-) (limited to 'src/leap/keymanager/tests') diff --git a/src/leap/keymanager/tests/test_keymanager.py b/src/leap/keymanager/tests/test_keymanager.py index 0f08326..e4e0d8b 100644 --- a/src/leap/keymanager/tests/test_keymanager.py +++ b/src/leap/keymanager/tests/test_keymanager.py @@ -83,13 +83,18 @@ class KeyManagerUtilTestCase(unittest.TestCase): 'private': False, 'length': 4096, 'expiry_date': 0, - 'last_audited_at': 0, 'refreshed_at': 1311239602, + } + adict = { + 'address': ADDRESS, + 'key_id': KEY_FINGERPRINT[-16:], + 'private': False, + 'last_audited_at': 0, 'validation': str(ValidationLevels.Weak_Chain), 'encr_used': False, 'sign_used': True, } - key = build_key_from_dict(OpenPGPKey, kdict) + key = build_key_from_dict(OpenPGPKey, kdict, adict) self.assertEqual( kdict['address'], key.address, 'Wrong data in key.') @@ -118,13 +123,13 @@ class KeyManagerUtilTestCase(unittest.TestCase): datetime.fromtimestamp(kdict['refreshed_at']), key.refreshed_at, 'Wrong data in key.') self.assertEqual( - ValidationLevels.get(kdict['validation']), key.validation, + ValidationLevels.get(adict['validation']), key.validation, 'Wrong data in key.') self.assertEqual( - kdict['encr_used'], key.encr_used, + adict['encr_used'], key.encr_used, 'Wrong data in key.') self.assertEqual( - kdict['sign_used'], key.sign_used, + adict['sign_used'], key.sign_used, 'Wrong data in key.') diff --git a/src/leap/keymanager/tests/test_openpgp.py b/src/leap/keymanager/tests/test_openpgp.py index 96b40a0..6641591 100644 --- a/src/leap/keymanager/tests/test_openpgp.py +++ b/src/leap/keymanager/tests/test_openpgp.py @@ -29,7 +29,10 @@ from leap.keymanager import ( KeyNotFound, openpgp, ) -from leap.keymanager.keys import TYPE_ID_PRIVATE_INDEX +from leap.keymanager.keys import ( + TYPE_ID_PRIVATE_INDEX, + TYPE_ADDRESS_PRIVATE_INDEX, +) from leap.keymanager.openpgp import OpenPGPKey from leap.keymanager.tests import ( KeyManagerWithSoledadTestCase, @@ -308,6 +311,7 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase): try: yield self.assertFailure(pgp.get_key(ADDRESS, private=False), KeyNotFound) + # it should have deleted the index self.assertEqual(self._soledad.delete_doc.call_count, 1) finally: self._soledad.get_from_index = get_from_index @@ -349,6 +353,54 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase): self._soledad.get_from_index = get_from_index self._soledad.delete_doc = delete_doc + @inlineCallbacks + def test_self_repair_five_active_docs(self): + pgp = openpgp.OpenPGPScheme( + self._soledad, gpgbinary=self.gpg_binary_path) + + get_from_index = self._soledad.get_from_index + delete_doc = self._soledad.delete_doc + + def my_get_from_index(*args): + if (args[0] == TYPE_ADDRESS_PRIVATE_INDEX and + args[2] == ADDRESS): + k1 = OpenPGPKey(ADDRESS, key_id="1", + last_audited_at=datetime(2005, 1, 1)) + k2 = OpenPGPKey(ADDRESS, key_id="2", + last_audited_at=datetime(2007, 1, 1)) + k3 = OpenPGPKey(ADDRESS, key_id="3", + last_audited_at=datetime(2007, 1, 1), + encr_used=True, sign_used=True) + k4 = OpenPGPKey(ADDRESS, key_id="4", + last_audited_at=datetime(2007, 1, 1), + sign_used=True) + k5 = OpenPGPKey(ADDRESS, key_id="5", + last_audited_at=datetime(2007, 1, 1), + encr_used=True) + deferreds = [] + for k in [k1, k2, k3, k4, k5]: + d = self._soledad.create_doc_from_json( + k.get_active_json(ADDRESS)) + deferreds.append(d) + return gatherResults(deferreds) + elif args[0] == TYPE_ID_PRIVATE_INDEX: + key_id = args[2] + self.assertEqual(key_id, "3") + k = OpenPGPKey(ADDRESS, key_id="3") + return succeed( + [self._soledad.create_doc_from_json(k.get_json())]) + return get_from_index(*args) + + self._soledad.get_from_index = my_get_from_index + self._soledad.delete_doc = Mock(return_value=succeed(None)) + + try: + yield pgp.get_key(ADDRESS, private=False) + self.assertEqual(self._soledad.delete_doc.call_count, 4) + finally: + self._soledad.get_from_index = get_from_index + self._soledad.delete_doc = delete_doc + def _assert_key_not_found(self, pgp, address, private=False): d = pgp.get_key(address, private=private) return self.assertFailure(d, KeyNotFound) diff --git a/src/leap/keymanager/tests/test_validation.py b/src/leap/keymanager/tests/test_validation.py index bcf41c4..44d6e39 100644 --- a/src/leap/keymanager/tests/test_validation.py +++ b/src/leap/keymanager/tests/test_validation.py @@ -108,14 +108,14 @@ class ValidationLevelsTestCase(KeyManagerWithSoledadTestCase): TEXT = "some text" km = self._key_manager() + yield km.put_raw_key(UNEXPIRED_PRIVATE, OpenPGPKey, ADDRESS) + signature = yield km.sign(TEXT, ADDRESS, OpenPGPKey) + yield self.delete_all_keys(km) + yield km.put_raw_key(UNEXPIRED_KEY, OpenPGPKey, ADDRESS) yield km.encrypt(TEXT, ADDRESS, OpenPGPKey) - - km2 = self._key_manager() - yield km2.put_raw_key(UNEXPIRED_PRIVATE, OpenPGPKey, ADDRESS) - signature = yield km2.sign(TEXT, ADDRESS, OpenPGPKey) - yield km.verify(TEXT, ADDRESS, OpenPGPKey, detached_sig=signature) + d = km.put_raw_key( UNRELATED_KEY, OpenPGPKey, ADDRESS, validation=ValidationLevels.Provider_Endorsement) @@ -126,17 +126,17 @@ class ValidationLevelsTestCase(KeyManagerWithSoledadTestCase): TEXT = "some text" km = self._key_manager() + yield km.put_raw_key(UNEXPIRED_PRIVATE, OpenPGPKey, ADDRESS) + yield km.put_raw_key(PUBLIC_KEY_2, OpenPGPKey, ADDRESS_2) + encrypted = yield km.encrypt(TEXT, ADDRESS_2, OpenPGPKey, + sign=ADDRESS) + yield self.delete_all_keys(km) + yield km.put_raw_key(UNEXPIRED_KEY, OpenPGPKey, ADDRESS) yield km.put_raw_key(PRIVATE_KEY_2, OpenPGPKey, ADDRESS_2) yield km.encrypt(TEXT, ADDRESS, OpenPGPKey) - - km2 = self._key_manager() - yield km2.put_raw_key(UNEXPIRED_PRIVATE, OpenPGPKey, ADDRESS) - yield km2.put_raw_key(PUBLIC_KEY_2, OpenPGPKey, ADDRESS_2) - encrypted = yield km2.encrypt(TEXT, ADDRESS_2, OpenPGPKey, - sign=ADDRESS) - yield km.decrypt(encrypted, ADDRESS_2, OpenPGPKey, verify=ADDRESS) + d = km.put_raw_key( UNRELATED_KEY, OpenPGPKey, ADDRESS, validation=ValidationLevels.Provider_Endorsement) @@ -150,6 +150,33 @@ class ValidationLevelsTestCase(KeyManagerWithSoledadTestCase): key = yield km.get_key(ADDRESS, OpenPGPKey, fetch_remote=False) self.assertEqual(key.fingerprint, SIGNED_FINGERPRINT) + @inlineCallbacks + def test_two_uuids(self): + TEXT = "some text" + + km = self._key_manager() + yield km.put_raw_key(UUIDS_PRIVATE, OpenPGPKey, ADDRESS_2) + signature = yield km.sign(TEXT, ADDRESS_2, OpenPGPKey) + yield self.delete_all_keys(km) + + yield km.put_raw_key(UUIDS_KEY, OpenPGPKey, ADDRESS_2) + yield km.put_raw_key(UUIDS_KEY, OpenPGPKey, ADDRESS) + yield km.encrypt(TEXT, ADDRESS_2, OpenPGPKey) + yield km.verify(TEXT, ADDRESS_2, OpenPGPKey, detached_sig=signature) + + d = km.put_raw_key( + PUBLIC_KEY_2, OpenPGPKey, ADDRESS_2, + validation=ValidationLevels.Provider_Endorsement) + yield self.assertFailure(d, KeyNotValidUpgrade) + key = yield km.get_key(ADDRESS_2, OpenPGPKey, fetch_remote=False) + self.assertEqual(key.fingerprint, UUIDS_FINGERPRINT) + + yield km.put_raw_key( + PUBLIC_KEY, OpenPGPKey, ADDRESS, + validation=ValidationLevels.Provider_Endorsement) + key = yield km.get_key(ADDRESS, OpenPGPKey, fetch_remote=False) + self.assertEqual(key.fingerprint, KEY_FINGERPRINT) + # Key material for testing @@ -364,6 +391,117 @@ X2+l7IOSt+31KQCBFN/VmhTySJOVQC1d2A56lSH2c/DWVClji+x3suzn -----END PGP PUBLIC KEY BLOCK----- """ +# key 0x1DDBAEB928D982F7: public key two uuids +# uid anotheruser +# uid Leap Test Key +UUIDS_FINGERPRINT = "21690ED054C1B2F3ACE963D38FCC7DEFB4EE5A9B" +UUIDS_KEY = """ +-----BEGIN PGP PUBLIC KEY BLOCK----- +Version: GnuPG v1 + +mQENBFZwjz0BCADHpVg7js8PK9gQXx3Z+Jtj6gswYZpeXRdIUfZBSebWNGKwXxC9 +ZZDjnQc3l6Kezh7ra/hB6xulDbj3vXi66V279QSOuFAKMIudlJehb86bUiVk9Ppy +kdrn44P40ZdVmw2m61WrKnvBelKW7pIF/EB/dY1uUonSfR56f/BxL5a5jGi2uI2y +2hTPnZEksoKQsjsp1FckPFwVGzRKVzYl3lsorL5oiHd450G2B3VRw8AZ8Eqq6bzf +jNrrA3TOMmEIYdADPqqnBj+xbnOCGBsvx+iAhGRxUckVJqW92SXlNNds8kEyoE0t +9B6eqe0MrrlcK3SLe03j85T9f1o3An53rV/3ABEBAAG0HExlYXAgVGVzdCBLZXkg +PGxlYXBAbGVhcC5zZT6JAT0EEwEIACcFAlZwjz0CGwMFCRLMAwAFCwkIBwMFFQoJ +CAsFFgMCAQACHgECF4AACgkQj8x977TuWpu4ZggAgk6rVtOCqYwM720Bs3k+wsWu +IVPUqnlPbSWph/PKBKWYE/5HoIGdvfN9jJxwpCM5x/ivPe1zeJ0qa9SnO66kolHU +7qC8HRWC67R4znO4Zrs2I+SgwRHAPPbqMVPsNs5rS0D6DCcr+LXtJF+LLAsIwDfw +mXEsKbX5H5aBmmDnfq0pGx05E3tKs5l09VVESvVZYOCM9b4FtdLVvgbKAD+KYDW1 +5A/PzOvyYjZu2FGtPKmNmqHD3KW8cmrcI/7mRR08FnNGbbpgXPZ2GPKgkUllY9N7 +E9lg4eaYH2fIWun293kbqp8ueELZvoU1jUQrP5B+eqBWTvIucqdQqJaiWn9pELQh +YW5vdGhlcnVzZXIgPGFub3RoZXJ1c2VyQGxlYXAuc2U+iQE9BBMBCAAnBQJWcI9a +AhsDBQkSzAMABQsJCAcDBRUKCQgLBRYDAgEAAh4BAheAAAoJEI/Mfe+07lqblRMH +/17vK2WOd0F7EegA5ELOrM+QJPKpLK4e6WdROUpS1hvRQcAOqadCCpYPSTTG/HnO +d9/Q9Q/G3xHHlk6nl1qHRkVlp0iVWyBZFn1s8lgGz/FFUEXXRj7I5RGmKSNgDlqA +un2OrwB2m+DH6pMjizu/RUfIJM2bSgViuvjCQoaLYLcFiULJlWHb/2WgpvesFyAc +0oc9AkXuaCEo50XQlrt8Bh++6rfbAMAS7ZrHluFPIY+y4eVl+MU/QkoGYAVgiLLV +5tnvbDZWNs8ixw4ubqKXYj5mK55sapokhOqObEfY6D3p7YpdQO/IhBneCw9gKOxa +wYAPhCOrJC8JmE69I1Nk8Bu5AQ0EVnCPPQEIANUivsrR2jwb8C9wPONn0HS3YYCI +/IVlLdw/Ia23ogBF1Uh8ORNg1G/t0/6S7IKDZ2gGgWw25u9TjWRRWsxO9tjOPi2d +YuhwlQRnq5Or+LzIEKRs9GnJMLFT0kR9Nrhw4UyaN6tWkR9p1Py7ux8RLmDEMAs3 +fBfVMLMmQRerJ5SyCUiq/lm9aYTLCC+vkjE01C+2oI0gcWGfLDjiJbaD4AazzibP +fBe41BIk7WaCJyEcBqfrgW/Seof43FhSKRGgc5nx3HH1DMz9AtYfKnVS5DgoBGpO +hxgzIJN3/hFHPicRlYoHxLUE48GcFfQFEJ78RXeBuieXAkDxNczHnLkEPx8AEQEA +AYkBJQQYAQgADwUCVnCPPQIbDAUJEswDAAAKCRCPzH3vtO5amyRIB/9IsWUWrvbH +njw2ZCiIV++lHgIntAcuQIbZIjyMXoM+imHsPrsDOUT65yi9Xp1pUyZEKtGkZvP4 +p7HRzJL+RWiWEN7sgQfNqqev8BF2/xmxkmiSuXHJ0PSaC5DmLfFDyvvSU6H5VPud +NszKIHtyoS6ph6TH8GXzFmNiHaTOZKdmVxlyLj1/DN+d63M+ARZIe7gB6jmP/gg4 +o52icfTcqLkkppYn8g1A9bdJ3k8qqExNPTf2llDscuD9VzpebFbPqfcYqR71GfG7 +Kmg7qGnZtNac1ERvknI/fmmCQydGk5pOh0KUTgeLG2qB07cqCUBbOXaweNWbiM9E +vtQLNMD9Gn7D +=MCXv +-----END PGP PUBLIC KEY BLOCK----- +""" +UUIDS_PRIVATE = """ +-----BEGIN PGP PRIVATE KEY BLOCK----- +Version: GnuPG v1 + +lQOYBFZwjz0BCADHpVg7js8PK9gQXx3Z+Jtj6gswYZpeXRdIUfZBSebWNGKwXxC9 +ZZDjnQc3l6Kezh7ra/hB6xulDbj3vXi66V279QSOuFAKMIudlJehb86bUiVk9Ppy +kdrn44P40ZdVmw2m61WrKnvBelKW7pIF/EB/dY1uUonSfR56f/BxL5a5jGi2uI2y +2hTPnZEksoKQsjsp1FckPFwVGzRKVzYl3lsorL5oiHd450G2B3VRw8AZ8Eqq6bzf +jNrrA3TOMmEIYdADPqqnBj+xbnOCGBsvx+iAhGRxUckVJqW92SXlNNds8kEyoE0t +9B6eqe0MrrlcK3SLe03j85T9f1o3An53rV/3ABEBAAEAB/9Lzeg2lP7hz8/2R2da +QB8gTNl6wVSPx+DzQMuz9o+DfdiLB02f3FSrWBBJd3XzvmfXE+Prg423mgJFbtfM +gJdqqpnUZv9dHxmj96urTHyyVPqF3s7JecAYlDaj31EK3BjO7ERW/YaH7B432NXx +F9qVitjsrsJN/dv4v2NYVq1wPcdDB05ge9WP+KRec7xvdTudH4Kov0iMZ+1Nksfn +lrowGuMYBGWDlTNoBoEYxDD2lqVaiOjyjx5Ss8btS59IlXxApOFZTezekl7hUI2B +1fDQ1GELL6BKVKxApGSD5XAgVlqkek4RhoHmg4gKSymfbFV5oRp1v1kC0JIGvnB1 +W5/BBADKzagL4JRnhWGubLec917B3se/3y1aHrEmO3T0IzxnUMD5yvg81uJWi5Co +M05Nu/Ny/Fw1VgoF8MjiGnumB2KKytylu8LKLarDxPpLxabOBCQLHrLQOMsmesjR +Cg3iYl/EeM/ooAufaN4IObcu6Pa8//rwNE7Fz1ZsIyJefN4fnwQA/AOpqA2BvHoH +VRYh4NVuMLhF1YdKqcd/T/dtSqszcadkmG4+vAL76r3jYxScRoNGQaIkpBnzP0ry +Adb0NDuBgSe/Cn44kqT7JJxTMfJNrw2rBMMUYZHdQrck2pf5R4fZ74yJyvCKg5pQ +QAl1gTSi6PJvPwpc7m7Kr4kHBVDlgKkEAJKkVrtq/v2+jSA+/osa4YC5brsDQ08X +pvZf0MBkc5/GDfusHyE8HGFnVY5ycrS85TBCrhc7suFu59pF4wEeXsqxqNf02gRe +B+btPwR7yki73iyXs4cbuszHMD03UnbvItFAybD5CC+oR9kG5noI0TzJNUNX9Vkq +xATf819dhwtgTha0HExlYXAgVGVzdCBLZXkgPGxlYXBAbGVhcC5zZT6JAT0EEwEI +ACcFAlZwjz0CGwMFCRLMAwAFCwkIBwMFFQoJCAsFFgMCAQACHgECF4AACgkQj8x9 +77TuWpu4ZggAgk6rVtOCqYwM720Bs3k+wsWuIVPUqnlPbSWph/PKBKWYE/5HoIGd +vfN9jJxwpCM5x/ivPe1zeJ0qa9SnO66kolHU7qC8HRWC67R4znO4Zrs2I+SgwRHA +PPbqMVPsNs5rS0D6DCcr+LXtJF+LLAsIwDfwmXEsKbX5H5aBmmDnfq0pGx05E3tK +s5l09VVESvVZYOCM9b4FtdLVvgbKAD+KYDW15A/PzOvyYjZu2FGtPKmNmqHD3KW8 +cmrcI/7mRR08FnNGbbpgXPZ2GPKgkUllY9N7E9lg4eaYH2fIWun293kbqp8ueELZ +voU1jUQrP5B+eqBWTvIucqdQqJaiWn9pELQhYW5vdGhlcnVzZXIgPGFub3RoZXJ1 +c2VyQGxlYXAuc2U+iQE9BBMBCAAnBQJWcI9aAhsDBQkSzAMABQsJCAcDBRUKCQgL +BRYDAgEAAh4BAheAAAoJEI/Mfe+07lqblRMH/17vK2WOd0F7EegA5ELOrM+QJPKp +LK4e6WdROUpS1hvRQcAOqadCCpYPSTTG/HnOd9/Q9Q/G3xHHlk6nl1qHRkVlp0iV +WyBZFn1s8lgGz/FFUEXXRj7I5RGmKSNgDlqAun2OrwB2m+DH6pMjizu/RUfIJM2b +SgViuvjCQoaLYLcFiULJlWHb/2WgpvesFyAc0oc9AkXuaCEo50XQlrt8Bh++6rfb +AMAS7ZrHluFPIY+y4eVl+MU/QkoGYAVgiLLV5tnvbDZWNs8ixw4ubqKXYj5mK55s +apokhOqObEfY6D3p7YpdQO/IhBneCw9gKOxawYAPhCOrJC8JmE69I1Nk8BudA5gE +VnCPPQEIANUivsrR2jwb8C9wPONn0HS3YYCI/IVlLdw/Ia23ogBF1Uh8ORNg1G/t +0/6S7IKDZ2gGgWw25u9TjWRRWsxO9tjOPi2dYuhwlQRnq5Or+LzIEKRs9GnJMLFT +0kR9Nrhw4UyaN6tWkR9p1Py7ux8RLmDEMAs3fBfVMLMmQRerJ5SyCUiq/lm9aYTL +CC+vkjE01C+2oI0gcWGfLDjiJbaD4AazzibPfBe41BIk7WaCJyEcBqfrgW/Seof4 +3FhSKRGgc5nx3HH1DMz9AtYfKnVS5DgoBGpOhxgzIJN3/hFHPicRlYoHxLUE48Gc +FfQFEJ78RXeBuieXAkDxNczHnLkEPx8AEQEAAQAH+wRSCn0RCPP7+v/zLgDMG3Eq +QHs7C6dmmCnlS7j6Rnnr8HliL0QBy/yi3Q/Fia7RnBiDPT9k04SZdH3KmmUW2rEl +aSRCkv00PwkSUuuQ6l9lTNUQclnsnqSRlusVgLT3cNG9NJCwFgwFeLBQ2+ey0PZc +M78edlEDXNPc3CfvK8O7WK74YiNJqIQCs7aDJSv0s7O/asRQyMCsl/UYtMV6W03d +eauS3bM41ll7GVfHMgkChFUQHb+19JHzSq4yeqQ/vS30ASugFxB3Omomp95sRL/w +60y51faLyTKD4AN3FhDfeIEfh1ggN2UT70qzC3+F8TvxQQHEBhNQKlhWVbWTp+0E +ANkcyokvn+09OIO/YDxF3aB37gA3J37d6NXos6pdvqUPOVSHvmF/hiM0FO2To6vu +ex/WcDQafPm4eiW6oNllwtZhWU2tr34bZD4PIuktSX2Ax2v5QtZ4d1CVdDEwbYn/ +fmR+nif1fTKTljZragaI9Rt6NWhfh7UGt62iIKh0lbhLBAD7T5wHY8V1yqlnyByG +K7nt+IHnND2I7Hk58yxKjv2KUNYxWZeOAQTQmfQXjJ+BOmw6oHMmDmGvdjSxIE+9 +j9nezEONxzVVjEDTKBeEnUeDY1QGDyDyW1/AhLJ52yWGTNrmKcGV4KmaYnhDzq7Z +aqJVRcFMF9TAfhrEGGhRdD83/QQA6xAqjWiu6tbaDurVuce64mA1R3T7HJ81gEaX +I+eJNDJb7PK3dhFETgyc3mcepWFNJkoXqx2ADhG8jLqK4o/N/QlV5PQgeHmhz09V +Z7MNhivGxDKZoxX6Bouh+qs5OkatcGFhTz//+FHSfusV2emxNiwd4QIsizxaltqh +W1ke0bA7eYkBJQQYAQgADwUCVnCPPQIbDAUJEswDAAAKCRCPzH3vtO5amyRIB/9I +sWUWrvbHnjw2ZCiIV++lHgIntAcuQIbZIjyMXoM+imHsPrsDOUT65yi9Xp1pUyZE +KtGkZvP4p7HRzJL+RWiWEN7sgQfNqqev8BF2/xmxkmiSuXHJ0PSaC5DmLfFDyvvS +U6H5VPudNszKIHtyoS6ph6TH8GXzFmNiHaTOZKdmVxlyLj1/DN+d63M+ARZIe7gB +6jmP/gg4o52icfTcqLkkppYn8g1A9bdJ3k8qqExNPTf2llDscuD9VzpebFbPqfcY +qR71GfG7Kmg7qGnZtNac1ERvknI/fmmCQydGk5pOh0KUTgeLG2qB07cqCUBbOXaw +eNWbiM9EvtQLNMD9Gn7D +=/3u/ +-----END PGP PRIVATE KEY BLOCK----- +""" + if __name__ == "__main__": - import unittest unittest.main() -- cgit v1.2.3 From 81232da09286f7f1812f6d3d182cd57665feaa1f Mon Sep 17 00:00:00 2001 From: Ruben Pollan Date: Fri, 18 Dec 2015 19:37:44 +0100 Subject: [feat] Migrate soledad documents by adding versioning field - Resolves: #7713 --- src/leap/keymanager/tests/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'src/leap/keymanager/tests') diff --git a/src/leap/keymanager/tests/__init__.py b/src/leap/keymanager/tests/__init__.py index cd612c4..d02f187 100644 --- a/src/leap/keymanager/tests/__init__.py +++ b/src/leap/keymanager/tests/__init__.py @@ -76,7 +76,7 @@ class KeyManagerWithSoledadTestCase(unittest.TestCase, BaseLeapTest): return d # wait for the indexes to be ready for the tear down - d = km._wrapper_map[OpenPGPKey].deferred_indexes + d = km._wrapper_map[OpenPGPKey].deferred_init d.addCallback(get_and_delete_keys) d.addCallback(lambda _: self.tearDownEnv()) d.addCallback(lambda _: self._soledad.close()) -- cgit v1.2.3 From 3d544f4a85930c5d1611d193500744fc97f0aee1 Mon Sep 17 00:00:00 2001 From: Ruben Pollan Date: Fri, 18 Dec 2015 20:31:18 +0100 Subject: [feat] Use fingerprints instead of key ids - Resolves: #7500 --- src/leap/keymanager/tests/__init__.py | 1 - src/leap/keymanager/tests/test_keymanager.py | 5 -- src/leap/keymanager/tests/test_openpgp.py | 123 +++++++++++---------------- 3 files changed, 48 insertions(+), 81 deletions(-) (limited to 'src/leap/keymanager/tests') diff --git a/src/leap/keymanager/tests/__init__.py b/src/leap/keymanager/tests/__init__.py index d02f187..4fbf63e 100644 --- a/src/leap/keymanager/tests/__init__.py +++ b/src/leap/keymanager/tests/__init__.py @@ -97,7 +97,6 @@ class KeyManagerWithSoledadTestCase(unittest.TestCase, BaseLeapTest): # key 24D18DDF: public key "Leap Test Key " -KEY_ID = "2F455E2824D18DDF" KEY_FINGERPRINT = "E36E738D69173C13D709E44F2F455E2824D18DDF" PUBLIC_KEY = """ -----BEGIN PGP PUBLIC KEY BLOCK----- diff --git a/src/leap/keymanager/tests/test_keymanager.py b/src/leap/keymanager/tests/test_keymanager.py index e4e0d8b..2fe9e4c 100644 --- a/src/leap/keymanager/tests/test_keymanager.py +++ b/src/leap/keymanager/tests/test_keymanager.py @@ -77,7 +77,6 @@ class KeyManagerUtilTestCase(unittest.TestCase): def test_build_key_from_dict(self): kdict = { 'address': [ADDRESS], - 'key_id': KEY_FINGERPRINT[-16:], 'fingerprint': KEY_FINGERPRINT, 'key_data': PUBLIC_KEY, 'private': False, @@ -87,7 +86,6 @@ class KeyManagerUtilTestCase(unittest.TestCase): } adict = { 'address': ADDRESS, - 'key_id': KEY_FINGERPRINT[-16:], 'private': False, 'last_audited_at': 0, 'validation': str(ValidationLevels.Weak_Chain), @@ -98,9 +96,6 @@ class KeyManagerUtilTestCase(unittest.TestCase): self.assertEqual( kdict['address'], key.address, 'Wrong data in key.') - self.assertEqual( - kdict['key_id'], key.key_id, - 'Wrong data in key.') self.assertEqual( kdict['fingerprint'], key.fingerprint, 'Wrong data in key.') diff --git a/src/leap/keymanager/tests/test_openpgp.py b/src/leap/keymanager/tests/test_openpgp.py index 6641591..8ed049f 100644 --- a/src/leap/keymanager/tests/test_openpgp.py +++ b/src/leap/keymanager/tests/test_openpgp.py @@ -30,7 +30,7 @@ from leap.keymanager import ( openpgp, ) from leap.keymanager.keys import ( - TYPE_ID_PRIVATE_INDEX, + TYPE_FINGERPRINT_PRIVATE_INDEX, TYPE_ADDRESS_PRIVATE_INDEX, ) from leap.keymanager.openpgp import OpenPGPKey @@ -40,7 +40,6 @@ from leap.keymanager.tests import ( ADDRESS_2, KEY_FINGERPRINT, PUBLIC_KEY, - KEY_ID, PUBLIC_KEY_2, PRIVATE_KEY, PRIVATE_KEY_2, @@ -256,39 +255,18 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase): @inlineCallbacks def test_self_repair_three_keys(self): + refreshed_keep = datetime(2007, 1, 1) + self._insert_key_docs([datetime(2005, 1, 1), + refreshed_keep, + datetime(2001, 1, 1)]) + delete_doc = self._mock_delete_doc() + pgp = openpgp.OpenPGPScheme( self._soledad, gpgbinary=self.gpg_binary_path) - yield pgp.put_ascii_key(PUBLIC_KEY, ADDRESS) - - get_from_index = self._soledad.get_from_index - delete_doc = self._soledad.delete_doc - - def my_get_from_index(*args): - if (args[0] == TYPE_ID_PRIVATE_INDEX and - args[2] == KEY_ID): - k1 = OpenPGPKey(ADDRESS, key_id="1", - refreshed_at=datetime(2005, 1, 1)) - k2 = OpenPGPKey(ADDRESS, key_id="2", - refreshed_at=datetime(2007, 1, 1)) - k3 = OpenPGPKey(ADDRESS, key_id="3", - refreshed_at=datetime(2001, 1, 1)) - d1 = self._soledad.create_doc_from_json(k1.get_json()) - d2 = self._soledad.create_doc_from_json(k2.get_json()) - d3 = self._soledad.create_doc_from_json(k3.get_json()) - return gatherResults([d1, d2, d3]) - return get_from_index(*args) - - self._soledad.get_from_index = my_get_from_index - self._soledad.delete_doc = Mock(return_value=succeed(None)) - key = yield pgp.get_key(ADDRESS, private=False) - - try: - self.assertEqual(key.key_id, "2") - self.assertEqual(self._soledad.delete_doc.call_count, 2) - finally: - self._soledad.get_from_index = get_from_index - self._soledad.delete_doc = delete_doc + self.assertEqual(key.refreshed_at, refreshed_keep) + self.assertEqual(self.count, 2) + self._soledad.delete_doc = delete_doc @inlineCallbacks def test_self_repair_no_keys(self): @@ -300,8 +278,8 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase): delete_doc = self._soledad.delete_doc def my_get_from_index(*args): - if (args[0] == TYPE_ID_PRIVATE_INDEX and - args[2] == KEY_ID): + if (args[0] == TYPE_FINGERPRINT_PRIVATE_INDEX and + args[2] == KEY_FINGERPRINT): return succeed([]) return get_from_index(*args) @@ -319,39 +297,16 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase): @inlineCallbacks def test_self_repair_put_keys(self): + self._insert_key_docs([datetime(2005, 1, 1), + datetime(2007, 1, 1), + datetime(2001, 1, 1)]) + delete_doc = self._mock_delete_doc() + pgp = openpgp.OpenPGPScheme( self._soledad, gpgbinary=self.gpg_binary_path) - - get_from_index = self._soledad.get_from_index - delete_doc = self._soledad.delete_doc - - def my_get_from_index(*args): - if (args[0] == TYPE_ID_PRIVATE_INDEX and - args[2] == KEY_ID): - k1 = OpenPGPKey(ADDRESS, key_id="1", - fingerprint=KEY_FINGERPRINT, - refreshed_at=datetime(2005, 1, 1)) - k2 = OpenPGPKey(ADDRESS, key_id="2", - fingerprint=KEY_FINGERPRINT, - refreshed_at=datetime(2007, 1, 1)) - k3 = OpenPGPKey(ADDRESS, key_id="3", - fingerprint=KEY_FINGERPRINT, - refreshed_at=datetime(2001, 1, 1)) - d1 = self._soledad.create_doc_from_json(k1.get_json()) - d2 = self._soledad.create_doc_from_json(k2.get_json()) - d3 = self._soledad.create_doc_from_json(k3.get_json()) - return gatherResults([d1, d2, d3]) - return get_from_index(*args) - - self._soledad.get_from_index = my_get_from_index - self._soledad.delete_doc = Mock(return_value=succeed(None)) - - try: - yield pgp.put_ascii_key(PUBLIC_KEY, ADDRESS) - self.assertEqual(self._soledad.delete_doc.call_count, 2) - finally: - self._soledad.get_from_index = get_from_index - self._soledad.delete_doc = delete_doc + yield pgp.put_ascii_key(PUBLIC_KEY, ADDRESS) + self.assertEqual(self.count, 2) + self._soledad.delete_doc = delete_doc @inlineCallbacks def test_self_repair_five_active_docs(self): @@ -364,29 +319,29 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase): def my_get_from_index(*args): if (args[0] == TYPE_ADDRESS_PRIVATE_INDEX and args[2] == ADDRESS): - k1 = OpenPGPKey(ADDRESS, key_id="1", + k1 = OpenPGPKey(ADDRESS, fingerprint="1", last_audited_at=datetime(2005, 1, 1)) - k2 = OpenPGPKey(ADDRESS, key_id="2", + k2 = OpenPGPKey(ADDRESS, fingerprint="2", last_audited_at=datetime(2007, 1, 1)) - k3 = OpenPGPKey(ADDRESS, key_id="3", + k3 = OpenPGPKey(ADDRESS, fingerprint="3", last_audited_at=datetime(2007, 1, 1), encr_used=True, sign_used=True) - k4 = OpenPGPKey(ADDRESS, key_id="4", + k4 = OpenPGPKey(ADDRESS, fingerprint="4", last_audited_at=datetime(2007, 1, 1), sign_used=True) - k5 = OpenPGPKey(ADDRESS, key_id="5", + k5 = OpenPGPKey(ADDRESS, fingerprint="5", last_audited_at=datetime(2007, 1, 1), encr_used=True) deferreds = [] - for k in [k1, k2, k3, k4, k5]: + for k in (k1, k2, k3, k4, k5): d = self._soledad.create_doc_from_json( k.get_active_json(ADDRESS)) deferreds.append(d) return gatherResults(deferreds) - elif args[0] == TYPE_ID_PRIVATE_INDEX: - key_id = args[2] - self.assertEqual(key_id, "3") - k = OpenPGPKey(ADDRESS, key_id="3") + elif args[0] == TYPE_FINGERPRINT_PRIVATE_INDEX: + fingerprint = args[2] + self.assertEqual(fingerprint, "3") + k = OpenPGPKey(ADDRESS, fingerprint="3") return succeed( [self._soledad.create_doc_from_json(k.get_json())]) return get_from_index(*args) @@ -404,3 +359,21 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase): def _assert_key_not_found(self, pgp, address, private=False): d = pgp.get_key(address, private=private) return self.assertFailure(d, KeyNotFound) + + @inlineCallbacks + def _insert_key_docs(self, refreshed_at): + for date in refreshed_at: + key = OpenPGPKey(ADDRESS, fingerprint=KEY_FINGERPRINT, + refreshed_at=date) + yield self._soledad.create_doc_from_json(key.get_json()) + yield self._soledad.create_doc_from_json(key.get_active_json()) + + def _mock_delete_doc(self): + delete_doc = self._soledad.delete_doc + self.count = 0 + + def my_delete_doc(*args): + self.count += 1 + return delete_doc(*args) + self._soledad.delete_doc = my_delete_doc + return delete_doc -- cgit v1.2.3 From 6222996d0805dfec1fab949b536adb0af08df0be Mon Sep 17 00:00:00 2001 From: Ruben Pollan Date: Mon, 21 Dec 2015 20:24:52 +0100 Subject: [test] add updater tests --- src/leap/keymanager/tests/test_migrator.py | 175 +++++++++++++++++++++++++++++ 1 file changed, 175 insertions(+) create mode 100644 src/leap/keymanager/tests/test_migrator.py (limited to 'src/leap/keymanager/tests') diff --git a/src/leap/keymanager/tests/test_migrator.py b/src/leap/keymanager/tests/test_migrator.py new file mode 100644 index 0000000..2d9782b --- /dev/null +++ b/src/leap/keymanager/tests/test_migrator.py @@ -0,0 +1,175 @@ +# -*- coding: utf-8 -*- +# test_migrator.py +# Copyright (C) 2015 LEAP +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + + +""" +Tests for the migrator. +""" + + +from collections import namedtuple +from mock import Mock +from twisted.internet.defer import succeed, inlineCallbacks + +from leap.keymanager.migrator import KeyDocumentsMigrator, KEY_ID_KEY +from leap.keymanager.keys import ( + TAGS_PRIVATE_INDEX, + KEYMANAGER_ACTIVE_TAG, + KEYMANAGER_KEY_TAG, + KEYMANAGER_DOC_VERSION, + + KEY_ADDRESS_KEY, + KEY_UIDS_KEY, + KEY_VERSION_KEY, + KEY_FINGERPRINT_KEY, + KEY_VALIDATION_KEY, + KEY_LAST_AUDITED_AT_KEY, + KEY_ENCR_USED_KEY, + KEY_SIGN_USED_KEY, +) +from leap.keymanager.validation import ValidationLevels +from leap.keymanager.tests import ( + KeyManagerWithSoledadTestCase, + ADDRESS, + ADDRESS_2, + KEY_FINGERPRINT, +) + + +class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase): + @inlineCallbacks + def test_simple_migration(self): + get_from_index = self._soledad.get_from_index + delete_doc = self._soledad.delete_doc + put_doc = self._soledad.put_doc + + def my_get_from_index(*args): + docs = [] + if (args[0] == TAGS_PRIVATE_INDEX and + args[2] == '0'): + SoledadDocument = namedtuple("SoledadDocument", ["content"]) + if args[1] == KEYMANAGER_KEY_TAG: + docs = [SoledadDocument({ + KEY_ADDRESS_KEY: [ADDRESS], + KEY_ID_KEY: KEY_FINGERPRINT[-16:], + KEY_FINGERPRINT_KEY: KEY_FINGERPRINT, + KEY_VALIDATION_KEY: str(ValidationLevels.Weak_Chain), + KEY_LAST_AUDITED_AT_KEY: 0, + KEY_ENCR_USED_KEY: True, + KEY_SIGN_USED_KEY: False, + })] + if args[1] == KEYMANAGER_ACTIVE_TAG: + docs = [SoledadDocument({ + KEY_ID_KEY: KEY_FINGERPRINT[-16:], + })] + return succeed(docs) + + self._soledad.get_from_index = my_get_from_index + self._soledad.delete_doc = Mock(return_value=succeed(None)) + self._soledad.put_doc = Mock(return_value=succeed(None)) + + try: + migrator = KeyDocumentsMigrator(self._soledad) + yield migrator.migrate() + call_list = self._soledad.put_doc.call_args_list + finally: + self._soledad.get_from_index = get_from_index + self._soledad.delete_doc = delete_doc + self._soledad.put_doc = put_doc + + self.assertEqual(len(call_list), 2) + active = call_list[0][0][0] + key = call_list[1][0][0] + + self.assertTrue(KEY_ID_KEY not in active.content) + self.assertEqual(active.content[KEY_VERSION_KEY], + KEYMANAGER_DOC_VERSION) + self.assertEqual(active.content[KEY_FINGERPRINT_KEY], KEY_FINGERPRINT) + self.assertEqual(active.content[KEY_VALIDATION_KEY], + str(ValidationLevels.Weak_Chain)) + self.assertEqual(active.content[KEY_LAST_AUDITED_AT_KEY], 0) + self.assertEqual(active.content[KEY_ENCR_USED_KEY], True) + self.assertEqual(active.content[KEY_SIGN_USED_KEY], False) + + self.assertTrue(KEY_ID_KEY not in key.content) + self.assertTrue(KEY_ADDRESS_KEY not in key.content) + self.assertTrue(KEY_VALIDATION_KEY not in key.content) + self.assertTrue(KEY_LAST_AUDITED_AT_KEY not in key.content) + self.assertTrue(KEY_ENCR_USED_KEY not in key.content) + self.assertTrue(KEY_SIGN_USED_KEY not in key.content) + self.assertEqual(key.content[KEY_UIDS_KEY], [ADDRESS]) + + @inlineCallbacks + def test_two_active_docs(self): + get_from_index = self._soledad.get_from_index + delete_doc = self._soledad.delete_doc + put_doc = self._soledad.put_doc + + def my_get_from_index(*args): + docs = [] + if (args[0] == TAGS_PRIVATE_INDEX and + args[2] == '0'): + SoledadDocument = namedtuple("SoledadDocument", ["content"]) + if args[1] == KEYMANAGER_KEY_TAG: + validation = str(ValidationLevels.Provider_Trust) + docs = [SoledadDocument({ + KEY_ADDRESS_KEY: [ADDRESS, ADDRESS_2], + KEY_ID_KEY: KEY_FINGERPRINT[-16:], + KEY_FINGERPRINT_KEY: KEY_FINGERPRINT, + KEY_VALIDATION_KEY: validation, + KEY_LAST_AUDITED_AT_KEY: 1984, + KEY_ENCR_USED_KEY: True, + KEY_SIGN_USED_KEY: False, + })] + if args[1] == KEYMANAGER_ACTIVE_TAG: + docs = [ + SoledadDocument({ + KEY_ADDRESS_KEY: ADDRESS, + KEY_ID_KEY: KEY_FINGERPRINT[-16:], + }), + SoledadDocument({ + KEY_ADDRESS_KEY: ADDRESS_2, + KEY_ID_KEY: KEY_FINGERPRINT[-16:], + }), + ] + return succeed(docs) + + self._soledad.get_from_index = my_get_from_index + self._soledad.delete_doc = Mock(return_value=succeed(None)) + self._soledad.put_doc = Mock(return_value=succeed(None)) + + try: + migrator = KeyDocumentsMigrator(self._soledad) + yield migrator.migrate() + call_list = self._soledad.put_doc.call_args_list + finally: + self._soledad.get_from_index = get_from_index + self._soledad.delete_doc = delete_doc + self._soledad.put_doc = put_doc + + self.assertEqual(len(call_list), 3) + for active in [call[0][0] for call in call_list][:2]: + self.assertTrue(KEY_ID_KEY not in active.content) + self.assertEqual(active.content[KEY_VERSION_KEY], + KEYMANAGER_DOC_VERSION) + self.assertEqual(active.content[KEY_FINGERPRINT_KEY], + KEY_FINGERPRINT) + self.assertEqual(active.content[KEY_VALIDATION_KEY], + str(ValidationLevels.Weak_Chain)) + self.assertEqual(active.content[KEY_LAST_AUDITED_AT_KEY], 0) + self.assertEqual(active.content[KEY_ENCR_USED_KEY], False) + self.assertEqual(active.content[KEY_SIGN_USED_KEY], False) -- cgit v1.2.3 From fdb6e285a97d5af21c7b3bdc02cba6fc21382f74 Mon Sep 17 00:00:00 2001 From: Ruben Pollan Date: Mon, 21 Dec 2015 19:26:55 +0100 Subject: [feat] Make EncryptionKey aware of the active address --- src/leap/keymanager/tests/test_keymanager.py | 21 ++++++++++++--------- src/leap/keymanager/tests/test_openpgp.py | 2 +- 2 files changed, 13 insertions(+), 10 deletions(-) (limited to 'src/leap/keymanager/tests') diff --git a/src/leap/keymanager/tests/test_keymanager.py b/src/leap/keymanager/tests/test_keymanager.py index 2fe9e4c..6347d56 100644 --- a/src/leap/keymanager/tests/test_keymanager.py +++ b/src/leap/keymanager/tests/test_keymanager.py @@ -76,7 +76,7 @@ class KeyManagerUtilTestCase(unittest.TestCase): def test_build_key_from_dict(self): kdict = { - 'address': [ADDRESS], + 'uids': [ADDRESS], 'fingerprint': KEY_FINGERPRINT, 'key_data': PUBLIC_KEY, 'private': False, @@ -94,7 +94,7 @@ class KeyManagerUtilTestCase(unittest.TestCase): } key = build_key_from_dict(OpenPGPKey, kdict, adict) self.assertEqual( - kdict['address'], key.address, + kdict['uids'], key.uids, 'Wrong data in key.') self.assertEqual( kdict['fingerprint'], key.fingerprint, @@ -117,6 +117,9 @@ class KeyManagerUtilTestCase(unittest.TestCase): self.assertEqual( datetime.fromtimestamp(kdict['refreshed_at']), key.refreshed_at, 'Wrong data in key.') + self.assertEqual( + adict['address'], key.address, + 'Wrong data in key.') self.assertEqual( ValidationLevels.get(adict['validation']), key.validation, 'Wrong data in key.') @@ -137,12 +140,12 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): # get public keys keys = yield km.get_all_keys(False) self.assertEqual(len(keys), 1, 'Wrong number of keys') - self.assertTrue(ADDRESS in keys[0].address) + self.assertTrue(ADDRESS in keys[0].uids) self.assertFalse(keys[0].private) # get private keys keys = yield km.get_all_keys(True) self.assertEqual(len(keys), 1, 'Wrong number of keys') - self.assertTrue(ADDRESS in keys[0].address) + self.assertTrue(ADDRESS in keys[0].uids) self.assertTrue(keys[0].private) @defer.inlineCallbacks @@ -153,7 +156,7 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): key = yield km.get_key(ADDRESS, OpenPGPKey, private=False, fetch_remote=False) self.assertTrue(key is not None) - self.assertTrue(ADDRESS in key.address) + self.assertTrue(ADDRESS in key.uids) self.assertEqual( key.fingerprint.lower(), KEY_FINGERPRINT.lower()) self.assertFalse(key.private) @@ -166,7 +169,7 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): key = yield km.get_key(ADDRESS, OpenPGPKey, private=True, fetch_remote=False) self.assertTrue(key is not None) - self.assertTrue(ADDRESS in key.address) + self.assertTrue(ADDRESS in key.uids) self.assertEqual( key.fingerprint.lower(), KEY_FINGERPRINT.lower()) self.assertTrue(key.private) @@ -231,7 +234,7 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): key = yield self._fetch_key(km, ADDRESS, PUBLIC_KEY) self.assertIsInstance(key, OpenPGPKey) - self.assertTrue(ADDRESS in key.address) + self.assertTrue(ADDRESS in key.uids) self.assertEqual(key.validation, ValidationLevels.Provider_Trust) @defer.inlineCallbacks @@ -243,7 +246,7 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): key = yield self._fetch_key(km, ADDRESS_OTHER, PUBLIC_KEY_OTHER) self.assertIsInstance(key, OpenPGPKey) - self.assertTrue(ADDRESS_OTHER in key.address) + self.assertTrue(ADDRESS_OTHER in key.uids) self.assertEqual(key.validation, ValidationLevels.Weak_Chain) def _fetch_key(self, km, address, key): @@ -273,7 +276,7 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase): yield km.put_raw_key(PUBLIC_KEY, OpenPGPKey, ADDRESS) key = yield km.get_key(ADDRESS, OpenPGPKey) self.assertIsInstance(key, OpenPGPKey) - self.assertTrue(ADDRESS in key.address) + self.assertTrue(ADDRESS in key.uids) @defer.inlineCallbacks def test_fetch_uri_ascii_key(self): diff --git a/src/leap/keymanager/tests/test_openpgp.py b/src/leap/keymanager/tests/test_openpgp.py index 8ed049f..0e5f6be 100644 --- a/src/leap/keymanager/tests/test_openpgp.py +++ b/src/leap/keymanager/tests/test_openpgp.py @@ -335,7 +335,7 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase): deferreds = [] for k in (k1, k2, k3, k4, k5): d = self._soledad.create_doc_from_json( - k.get_active_json(ADDRESS)) + k.get_active_json()) deferreds.append(d) return gatherResults(deferreds) elif args[0] == TYPE_FINGERPRINT_PRIVATE_INDEX: -- cgit v1.2.3 From d44d72c816ebc668931071cf5bd8e37e0be32e05 Mon Sep 17 00:00:00 2001 From: Ruben Pollan Date: Thu, 25 Feb 2016 11:34:45 -0600 Subject: [test] refactor key deletion tests --- src/leap/keymanager/tests/__init__.py | 30 +++++++++++++++--------------- 1 file changed, 15 insertions(+), 15 deletions(-) (limited to 'src/leap/keymanager/tests') diff --git a/src/leap/keymanager/tests/__init__.py b/src/leap/keymanager/tests/__init__.py index 4fbf63e..20d05e8 100644 --- a/src/leap/keymanager/tests/__init__.py +++ b/src/leap/keymanager/tests/__init__.py @@ -54,6 +54,14 @@ class KeyManagerWithSoledadTestCase(unittest.TestCase, BaseLeapTest): def tearDown(self): km = self._key_manager() + # wait for the indexes to be ready for the tear down + d = km._wrapper_map[OpenPGPKey].deferred_init + d.addCallback(lambda _: self.delete_all_keys(km)) + d.addCallback(lambda _: self.tearDownEnv()) + d.addCallback(lambda _: self._soledad.close()) + return d + + def delete_all_keys(self, km): def delete_keys(keys): deferreds = [] for key in keys: @@ -61,26 +69,18 @@ class KeyManagerWithSoledadTestCase(unittest.TestCase, BaseLeapTest): deferreds.append(d) return gatherResults(deferreds) - def get_and_delete_keys(_): - deferreds = [] - for private in [True, False]: - d = km.get_all_keys(private=private) - d.addCallback(delete_keys) - d.addCallback(check_deleted, private) - deferreds.append(d) - return gatherResults(deferreds) - def check_deleted(_, private): d = km.get_all_keys(private=private) d.addCallback(lambda keys: self.assertEqual(keys, [])) return d - # wait for the indexes to be ready for the tear down - d = km._wrapper_map[OpenPGPKey].deferred_init - d.addCallback(get_and_delete_keys) - d.addCallback(lambda _: self.tearDownEnv()) - d.addCallback(lambda _: self._soledad.close()) - return d + deferreds = [] + for private in [True, False]: + d = km.get_all_keys(private=private) + d.addCallback(delete_keys) + d.addCallback(check_deleted, private) + deferreds.append(d) + return gatherResults(deferreds) def _key_manager(self, user=ADDRESS, url='', token=None, ca_cert_path=None): -- cgit v1.2.3