summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--src/leap/bitmask/keymanager/__init__.py59
-rw-r--r--src/leap/bitmask/keymanager/keys.py22
-rw-r--r--src/leap/bitmask/keymanager/nicknym.py181
-rw-r--r--src/leap/bitmask/keymanager/openpgp.py15
-rw-r--r--src/leap/bitmask/keymanager/refresher.py132
-rw-r--r--tests/integration/keymanager/test_keymanager.py61
-rw-r--r--tests/test_refresher.py149
7 files changed, 587 insertions, 32 deletions
diff --git a/src/leap/bitmask/keymanager/__init__.py b/src/leap/bitmask/keymanager/__init__.py
index 27c9853d..637391c6 100644
--- a/src/leap/bitmask/keymanager/__init__.py
+++ b/src/leap/bitmask/keymanager/__init__.py
@@ -22,8 +22,6 @@ import fileinput
import os
import sys
import tempfile
-import json
-import urllib
from urlparse import urlparse
@@ -34,9 +32,10 @@ from twisted.web._responses import NOT_FOUND
from leap.common import ca_bundle
from leap.common.check import leap_assert
+from leap.common.decorators import memoized_method
from leap.common.http import HTTPClient
from leap.common.events import emit_async, catalog
-from leap.common.decorators import memoized_method
+from leap.bitmask.keymanager.nicknym import Nicknym
from leap.bitmask.keymanager.errors import (
KeyNotFound,
@@ -105,12 +104,20 @@ class KeyManager(object):
self._combined_ca_bundle = ''
self._async_client = HTTPClient(self._combined_ca_bundle)
- self._async_client_pinned = HTTPClient(self._ca_cert_path)
+ self._nicknym = Nicknym(self._nickserver_uri, self._ca_cert_path, self._token)
+ self.refresher = None
#
# utilities
#
+ def start_refresher(self):
+ self.refresher = RandomRefreshPublicKey(self._openpgp, self)
+ self.refresher.start()
+
+ def stop_refresher(self):
+ self.refresher.stop()
+
def _create_combined_bundle_file(self):
leap_ca_bundle = ca_bundle.where()
@@ -303,14 +310,7 @@ class KeyManager(object):
:raise UnsupportedKeyTypeError: if invalid key type
"""
def send(pubkey):
- data = {
- self.PUBKEY_KEY: pubkey.key_data
- }
- uri = "%s/%s/users/%s.json" % (
- self._api_uri,
- self._api_version,
- self._uid)
- d = self._put(uri, data)
+ d = self._nicknym.put_key(self.uid, pubkey.key_data, self._api_uri, self._api_version)
d.addCallback(lambda _:
emit_async(catalog.KEYMANAGER_DONE_UPLOADING_KEYS,
self._address))
@@ -321,6 +321,36 @@ class KeyManager(object):
d.addCallback(send)
return d
+ @defer.inlineCallbacks
+ def _fetch_keys_from_server_and_store_local(self, address):
+ """
+ Fetch keys from nickserver and insert them in locale database.
+
+ :param address: The address bound to the keys.
+ :type address: str
+
+ :return: A Deferred which fires when the key is in the storage,
+ or which fails with KeyNotFound if the key was not found on
+ nickserver.
+ :rtype: Deferred
+
+ """
+ server_keys = yield self._nicknym.fetch_key_with_address(address)
+
+ # insert keys in local database
+ if self.OPENPGP_KEY in server_keys:
+ # nicknym server is authoritative for its own domain,
+ # for other domains the key might come from key servers.
+ validation_level = ValidationLevels.Weak_Chain
+ _, domain = _split_email(address)
+ if (domain == _get_domain(self._nickserver_uri)):
+ validation_level = ValidationLevels.Provider_Trust
+
+ yield self.put_raw_key(
+ server_keys['openpgp'],
+ address=address,
+ validation=validation_level)
+
def get_key(self, address, private=False, fetch_remote=True):
"""
Return a key bound to address.
@@ -364,7 +394,7 @@ class KeyManager(object):
return failure
emit_async(catalog.KEYMANAGER_LOOKING_FOR_KEY, address)
- d = self._fetch_keys_from_server(address)
+ d = self._fetch_keys_from_server_and_store_local(address)
d.addCallback(
lambda _: self._openpgp.get_key(address, private=False))
d.addCallback(key_found)
@@ -396,7 +426,6 @@ class KeyManager(object):
:raise UnsupportedKeyTypeError: if invalid key type
"""
-
def signal_finished(key):
emit_async(
catalog.KEYMANAGER_FINISHED_KEY_GENERATION, self._address)
@@ -636,7 +665,6 @@ class KeyManager(object):
:raise UnsupportedKeyTypeError: if invalid key type
"""
-
def verify(pubkey):
signed = self._openpgp.verify(
data, pubkey, detached_sig=detached_sig)
@@ -686,7 +714,6 @@ class KeyManager(object):
:raise UnsupportedKeyTypeError: if invalid key type
"""
-
def old_key_not_found(failure):
if failure.check(KeyNotFound):
return None
diff --git a/src/leap/bitmask/keymanager/keys.py b/src/leap/bitmask/keymanager/keys.py
index 45a7d720..1b236c3d 100644
--- a/src/leap/bitmask/keymanager/keys.py
+++ b/src/leap/bitmask/keymanager/keys.py
@@ -272,6 +272,28 @@ class OpenPGPKey(object):
self.address,
"priv" if self.private else "publ")
+ def is_active(self):
+ """
+ Indicates if a key is active.
+ :return: True if key is active.
+ :rtype: bool
+ """
+ return True if self.address is not None else False
+
+ def set_unactive(self):
+ """
+ Sets a key as unactive.
+ """
+ self.address = None
+
+ def is_expired(self):
+ """
+ Indicates if a key is expired.
+ :return: True if key expired.
+ :rtype: bool
+ """
+ return False if self.expiry_date is None else self.expiry_date < datetime.now()
+
def parse_address(address):
"""
diff --git a/src/leap/bitmask/keymanager/nicknym.py b/src/leap/bitmask/keymanager/nicknym.py
new file mode 100644
index 00000000..7af49d6d
--- /dev/null
+++ b/src/leap/bitmask/keymanager/nicknym.py
@@ -0,0 +1,181 @@
+# -*- coding: utf-8 -*-
+# nicknym.py
+# Copyright (C) 2016 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 <http://www.gnu.org/licenses/>.
+
+import json
+import sys
+import logging
+import urllib
+
+from twisted.internet import defer
+from twisted.web import client
+from twisted.web._responses import NOT_FOUND
+
+from leap.bitmask.keymanager.errors import KeyNotFound
+from leap.common.check import leap_assert
+from leap.common.http import HTTPClient
+from leap.common.decorators import memoized_method
+
+
+logger = logging.getLogger(__name__)
+
+
+class Nicknym(object):
+ """
+ Responsible for communication to the nicknym server.
+ """
+
+ PUBKEY_KEY = "user[public_key]"
+
+ def __init__(self, nickserver_uri, ca_cert_path, token):
+ self._nickserver_uri = nickserver_uri
+ self._async_client_pinned = HTTPClient(ca_cert_path)
+ self._token = token
+
+ @defer.inlineCallbacks
+ def put_key(self, uid, key_data, api_uri, api_version):
+ """
+ Send a PUT request to C{uri} containing C{data}.
+
+ The request will be sent using the configured CA certificate path to
+ verify the server certificate and the configured session id for
+ authentication.
+
+ :param uid: The URI of the request.
+ :type uid: str
+ :param key_data: The body of the request.
+ :type key_data: dict, str or file
+
+ :return: A deferred that will be fired when PUT request finishes
+ :rtype: Deferred
+ """
+ data = {
+ self.PUBKEY_KEY: key_data
+ }
+
+ uri = "%s/%s/users/%s.json" % (
+ api_uri,
+ api_version,
+ uid)
+
+ leap_assert(
+ self._token is not None,
+ 'We need a token to interact with webapp!')
+ if type(data) == dict:
+ data = urllib.urlencode(data)
+ headers = {'Authorization': [str('Token token=%s' % self._token)]}
+ headers['Content-Type'] = ['application/x-www-form-urlencoded']
+ try:
+ res = yield self._async_client_pinned.request(str(uri), 'PUT',
+ body=str(data),
+ headers=headers)
+ except Exception as e:
+ logger.warning("Error uploading key: %r" % (e,))
+ raise e
+ if 'error' in res:
+ # FIXME: That's a workaround for 500,
+ # we need to implement a readBody to assert response code
+ logger.warning("Error uploading key: %r" % (res,))
+ raise Exception(res)
+
+ @defer.inlineCallbacks
+ def _get_key_from_nicknym(self, uri):
+ """
+ Send a GET request to C{uri} containing C{data}.
+
+ :param uri: The URI of the request.
+ :type uri: str
+
+ :return: A deferred that will be fired with GET content as json (dict)
+ :rtype: Deferred
+ """
+ try:
+ content = yield self._fetch_and_handle_404_from_nicknym(uri)
+ json_content = json.loads(content)
+
+ except KeyNotFound:
+ raise
+ except IOError as e:
+ logger.warning("HTTP error retrieving key: %r" % (e,))
+ logger.warning("%s" % (content,))
+ raise KeyNotFound(e.message), None, sys.exc_info()[2]
+ except ValueError as v:
+ logger.warning("Invalid JSON data from key: %s" % (uri,))
+ raise KeyNotFound(v.message + ' - ' + uri), None, sys.exc_info()[2]
+
+ except Exception as e:
+ logger.warning("Error retrieving key: %r" % (e,))
+ raise KeyNotFound(e.message), None, sys.exc_info()[2]
+ # Responses are now text/plain, although it's json anyway, but
+ # this will fail when it shouldn't
+ # leap_assert(
+ # res.headers['content-type'].startswith('application/json'),
+ # 'Content-type is not JSON.')
+ defer.returnValue(json_content)
+
+ def _fetch_and_handle_404_from_nicknym(self, uri):
+ """
+ Send a GET request to C{uri} containing C{data}.
+
+ :param uri: The URI of the request.
+ :type uri: str
+
+ :return: A deferred that will be fired with GET content as json (dict)
+ :rtype: Deferred
+ """
+
+ def check_404(response):
+ if response.code == NOT_FOUND:
+ message = ' %s: Key not found. Request: %s' % (response.code, uri)
+ logger.warning(message)
+ raise KeyNotFound(message), None, sys.exc_info()[2]
+ return response
+
+ d = self._async_client_pinned.request(str(uri), 'GET', callback=check_404)
+ d.addCallback(client.readBody)
+ return d
+
+ @memoized_method(invalidation=300)
+ def fetch_key_with_address(self, address):
+ """
+ Fetch keys bound to address from nickserver.
+
+ :param address: The address bound to the keys.
+ :type address: str
+
+ :return: A Deferred which fires when the key is in the storage,
+ or which fails with KeyNotFound if the key was not found on
+ nickserver.
+ :rtype: Deferred
+
+ """
+ return self._get_key_from_nicknym(self._nickserver_uri + '?address=' + address)
+
+ @memoized_method(invalidation=300)
+ def fetch_key_with_fingerprint(self, fingerprint):
+ """
+ Fetch keys bound to fingerprint from nickserver.
+
+ :param fingerprint: The fingerprint bound to the keys.
+ :type fingerprint: str
+
+ :return: A Deferred which fires when the key is in the storage,
+ or which fails with KeyNotFound if the key was not found on
+ nickserver.
+ :rtype: Deferred
+
+ """
+ return self._get_key_from_nicknym(self._nickserver_uri + '?fingerprint=' + fingerprint)
diff --git a/src/leap/bitmask/keymanager/openpgp.py b/src/leap/bitmask/keymanager/openpgp.py
index 02b54563..b256ff00 100644
--- a/src/leap/bitmask/keymanager/openpgp.py
+++ b/src/leap/bitmask/keymanager/openpgp.py
@@ -410,10 +410,9 @@ class OpenPGPScheme(object):
active_json = key.get_active_json()
if activedoc:
activedoc.set_json(active_json)
- d = self._soledad.put_doc(activedoc)
- else:
- d = self._soledad.create_doc_from_json(active_json)
- return d
+ return self._soledad.put_doc(activedoc)
+ elif key.is_active():
+ return self._soledad.create_doc_from_json(active_json)
def get_active_doc(keydoc):
d = self._get_active_doc_from_address(key.address, key.private)
@@ -534,6 +533,14 @@ class OpenPGPScheme(object):
d.addCallback(delete_key)
return d
+ def unactivate_key(self, address):
+ """
+ Mark a active doc as deleted.
+ :param address: The unique address for the active content.
+ """
+ active_doc = self._get_active_doc_from_address(address, False)
+ yield self._soledad.delete_doc(active_doc)
+
#
# Data encryption, decryption, signing and verifying
#
diff --git a/src/leap/bitmask/keymanager/refresher.py b/src/leap/bitmask/keymanager/refresher.py
new file mode 100644
index 00000000..cd9db281
--- /dev/null
+++ b/src/leap/bitmask/keymanager/refresher.py
@@ -0,0 +1,132 @@
+# -*- coding: utf-8 -*-
+# refresher.py
+# Copyright (C) 2016 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 <http://www.gnu.org/licenses/>.
+
+
+"""
+A service which continuous refreshes the (public) key directories randomly in a random time interval.
+"""
+
+import logging
+
+from twisted.internet.task import LoopingCall
+from twisted.internet import defer
+from random import choice, randrange
+
+DEBUG_STOP_REFRESH = "Stop to refresh the key directory ..."
+DEBUG_START_REFRESH = "Start to refresh the key directory ..."
+ERROR_UNEQUAL_FINGERPRINTS = "[WARNING] Your provider might be cheat on you, " \
+ "and gave a wrong key back. Fingerprints are unequal, old %s new %s "
+
+MIN_RANDOM_INTERVAL_RANGE = 4 * 60 # four minutes
+MAX_RANDOM_INTERVAL_RANGE = 6 * 60 # six minutes
+
+logger = logging.getLogger(__name__)
+
+
+class RandomRefreshPublicKey(object):
+
+ def __init__(self, openpgp, keymanager):
+ """
+ Initialize the RandomRefreshPublicKey.
+ :param openpgp: Openpgp object.
+ :param keymanager: The key manager.
+ """
+ self._openpgp = openpgp
+ self._keymanger = keymanager
+ self._loop = LoopingCall(self._refresh_continuous)
+ self._loop.interval = self._random_interval_to_refersh()
+
+ def start(self):
+ """
+ Start the looping call with random interval
+ [MIN_RANDOM_INTERVAL_RANGE, MAX_RANDOM_INTERVAL_RANGE]
+ :return: LoopingCall to start the service.
+ :rtype: A deferred.
+ """
+ self._loop.start(self._random_interval_to_refersh(), False)
+ logger.debug(DEBUG_START_REFRESH)
+
+ def stop(self):
+ """
+ Stop the looping call with random interval.
+ """
+ self._loop.stop()
+ logger.debug(DEBUG_STOP_REFRESH)
+
+ @defer.inlineCallbacks
+ def _get_random_key(self):
+ """
+ Get a random key of all the keys in a users key doc.
+ :return: A random key.
+ :rtype: A deferred.
+ """
+ # TODO maybe make a check first if key is active and get another one then.
+ keys = yield self._openpgp.get_all_keys()
+ defer.returnValue(None if keys is None or keys == [] else choice(keys))
+
+ @defer.inlineCallbacks
+ def _refresh_continuous(self):
+ """
+ The LoopingCall to refresh the key doc continuously.
+ """
+ self._loop.interval = self._random_interval_to_refersh()
+ yield self.maybe_refresh_key()
+
+ @defer.inlineCallbacks
+ def _maybe_unactivate_key(self, key):
+ """
+ Unactivate a given key.
+ :param key: The key to be unactivated.
+ """
+ if key.is_expired() and key.is_active(): # TODO or is_revoked
+ yield self._openpgp.unactivate_key(key.address)
+ key.set_unactive()
+
+ @defer.inlineCallbacks
+ def maybe_refresh_key(self):
+ """
+ Get key from nicknym and try to refresh.
+ """
+ old_key = yield self._get_random_key()
+
+ if old_key is None:
+ defer.returnValue(None)
+
+ old_updated_key = yield self._keymanger._nicknym.\
+ fetch_key_with_fingerprint(old_key.fingerprint)
+
+ if old_updated_key.fingerprint != old_key.fingerprint:
+ logger.error(ERROR_UNEQUAL_FINGERPRINTS %
+ (old_key.fingerprint, old_updated_key.fingerprint))
+ defer.returnValue(None)
+
+ yield self._maybe_unactivate_key(old_updated_key)
+ yield self._openpgp.put_key(old_updated_key)
+
+ # No new fetch by address needed, bc that will happen before sending an email
+ # could be discussed since fetching before sending an email leaks information.
+
+ def _random_interval_to_refersh(self):
+ """
+ Random interval.
+ :return: A number in this interval.
+ """
+ return randrange(MIN_RANDOM_INTERVAL_RANGE, MAX_RANDOM_INTERVAL_RANGE)
+
+
+
+
diff --git a/tests/integration/keymanager/test_keymanager.py b/tests/integration/keymanager/test_keymanager.py
index 443902ab..9b5de831 100644
--- a/tests/integration/keymanager/test_keymanager.py
+++ b/tests/integration/keymanager/test_keymanager.py
@@ -27,12 +27,16 @@ from os import path
from twisted.internet import defer
from twisted.trial import unittest
from twisted.web._responses import NOT_FOUND
+from twisted.web import client
import mock
from leap.common import ca_bundle
from leap.bitmask.keymanager import client
from leap.bitmask.keymanager import errors
from leap.bitmask.keymanager.keys import (
+
+from leap.keymanager import errors
+from leap.keymanager.keys import (
OpenPGPKey,
is_address,
build_key_from_dict,
@@ -222,24 +226,42 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase):
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._async_client_pinned.request.assert_called_once_with(
+ km._nicknym._async_client_pinned.request.assert_called_once_with(
str(url), 'PUT', body=str(data),
headers=headers
)
def test_fetch_keys_from_server(self):
"""
- Test that the request is well formed when fetching keys from server.
+ Test that the request is well formed when fetching keys from server
+ with address.
"""
km = self._key_manager(url=NICKSERVER_URI)
expected_url = NICKSERVER_URI + '?address=' + ADDRESS_2
def verify_the_call(_):
- used_kwargs = km._async_client_pinned.request.call_args[1]
- km._async_client_pinned.request.assert_called_once_with(
+ used_kwargs = km._nicknym._async_client_pinned.request.call_args[1]
+ km._nicknym._async_client_pinned.request.assert_called_once_with(
expected_url, 'GET', **used_kwargs)
- d = self._fetch_key(km, ADDRESS_2, PUBLIC_KEY_2)
+ d = self._fetch_key_with_address(km, ADDRESS_2, PUBLIC_KEY_2)
+ d.addCallback(verify_the_call)
+ return d
+
+ def test_fetch_keys_from_server_with_fingerprint(self):
+ """
+ Test that the request is well formed when fetching keys from server
+ with fingerprint.
+ """
+ km = self._key_manager(url=NICKSERVER_URI)
+ expected_url = NICKSERVER_URI + '?fingerprint=' + KEY_FINGERPRINT
+
+ def verify_the_call(_):
+ used_kwargs = km._nicknym._async_client_pinned.request.call_args[1]
+ km._nicknym._async_client_pinned.request.assert_called_once_with(
+ expected_url, 'GET', **used_kwargs)
+
+ d = self._fetch_key_with_fingerprint(km, KEY_FINGERPRINT, PUBLIC_KEY)
d.addCallback(verify_the_call)
return d
@@ -254,16 +276,16 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase):
return_value=defer.succeed(None))
url = NICKSERVER_URI + '?address=' + INVALID_MAIL_ADDRESS
- d = km._fetch_and_handle_404_from_nicknym(url, INVALID_MAIL_ADDRESS)
+ d = km._nicknym._fetch_and_handle_404_from_nicknym(url)
def check_key_not_found_is_raised_if_404(_):
- used_kwargs = km._async_client_pinned.request.call_args[1]
+ used_kwargs = km._nicknym._async_client_pinned.request.call_args[1]
check_404_callback = used_kwargs['callback']
fake_response = mock.Mock()
fake_response.code = NOT_FOUND
with self.assertRaisesRegexp(
errors.KeyNotFound,
- '404: %s key not found.' % INVALID_MAIL_ADDRESS):
+ '404: Key not found. Request: %s' % url.replace('?', '\?')):
check_404_callback(fake_response)
d.addCallback(check_key_not_found_is_raised_if_404)
@@ -282,7 +304,7 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase):
def assert_key_not_found_raised(error):
self.assertEqual(error.value, key_not_found_exception)
- d = km._get_key_from_nicknym(INVALID_MAIL_ADDRESS)
+ d = km._nicknym.fetch_key_with_address(INVALID_MAIL_ADDRESS)
d.addErrback(assert_key_not_found_raised)
@defer.inlineCallbacks
@@ -292,7 +314,7 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase):
"""
km = self._key_manager(url=NICKSERVER_URI)
- key = yield self._fetch_key(km, ADDRESS, PUBLIC_KEY)
+ key = yield self._fetch_key_with_address(km, ADDRESS, PUBLIC_KEY)
self.assertIsInstance(key, OpenPGPKey)
self.assertTrue(ADDRESS in key.uids)
self.assertEqual(key.validation, ValidationLevels.Provider_Trust)
@@ -304,12 +326,12 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase):
"""
km = self._key_manager(url=NICKSERVER_URI)
- key = yield self._fetch_key(km, ADDRESS_OTHER, PUBLIC_KEY_OTHER)
+ key = yield self._fetch_key_with_address(km, ADDRESS_OTHER, PUBLIC_KEY_OTHER)
self.assertIsInstance(key, OpenPGPKey)
self.assertTrue(ADDRESS_OTHER in key.uids)
self.assertEqual(key.validation, ValidationLevels.Weak_Chain)
- def _fetch_key(self, km, address, key):
+ def _fetch_key_with_address(self, km, address, key):
"""
:returns: a Deferred that will fire with the OpenPGPKey
"""
@@ -328,6 +350,21 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase):
d.addCallback(lambda _: km.get_key(address))
return d
+ def _fetch_key_with_fingerprint(self, km, fingerprint, key):
+ """
+ :returns: a Deferred that will fire with the OpenPGPKey
+ """
+ data = json.dumps({'fingerprint': fingerprint, 'openpgp': key})
+
+ client.readBody = Mock(return_value=defer.succeed(data))
+
+ # mock the fetcher so it returns the key for KEY_FINGERPRINT
+ km._nicknym._async_client_pinned.request = Mock(
+ return_value=defer.succeed(None))
+ km.ca_cert_path = 'cacertpath'
+ key = km._nicknym.fetch_key_with_fingerprint(fingerprint)
+ return key
+
@defer.inlineCallbacks
def test_put_key_ascii(self):
"""
diff --git a/tests/test_refresher.py b/tests/test_refresher.py
new file mode 100644
index 00000000..13a46d47
--- /dev/null
+++ b/tests/test_refresher.py
@@ -0,0 +1,149 @@
+# -*- coding: utf-8 -*-
+# test_refresher.py
+# Copyright (C) 2016 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 <http://www.gnu.org/licenses/>.
+
+"""
+Tests for refreshing the key directory.
+"""
+
+import logging
+from datetime import datetime
+
+from mock import Mock, patch
+from twisted.internet import defer
+
+from common import KeyManagerWithSoledadTestCase, KEY_FINGERPRINT
+from leap.keymanager import openpgp
+from leap.keymanager.keys import OpenPGPKey
+from leap.keymanager.refresher import RandomRefreshPublicKey, MIN_RANDOM_INTERVAL_RANGE, DEBUG_START_REFRESH, \
+ DEBUG_STOP_REFRESH, ERROR_UNEQUAL_FINGERPRINTS
+
+ANOTHER_FP = 'ANOTHERFINGERPRINT'
+
+logger = logging.getLogger(__name__)
+
+
+class RandomRefreshPublicKeyTestCase(KeyManagerWithSoledadTestCase):
+
+ @defer.inlineCallbacks
+ def test_get_random_address(self):
+ pgp = openpgp.OpenPGPScheme(
+ self._soledad, gpgbinary=self.gpg_binary_path)
+ rf = RandomRefreshPublicKey(pgp, self._key_manager())
+ key = OpenPGPKey(address='user@leap.se')
+ key_another = OpenPGPKey(address='zara@leap.se')
+
+ pgp.get_all_keys = Mock(return_value=defer.succeed([key, key_another]))
+
+ random_key = yield rf._get_random_key()
+ self.assertTrue(random_key.address == key.address or random_key.address == key_another.address)
+
+ @defer.inlineCallbacks
+ def test_do_not_throw_error_for_empty_key_dict(self):
+ pgp = openpgp.OpenPGPScheme(
+ self._soledad, gpgbinary=self.gpg_binary_path)
+ rf = RandomRefreshPublicKey(pgp, self._key_manager())
+
+ pgp.get_all_keys = Mock(return_value=defer.succeed([]))
+ random_address = yield rf._get_random_key()
+ self.assertTrue(random_address is None)
+
+ @defer.inlineCallbacks
+ def test_log_error_if_fetch_by_fingerprint_returns_wrong_key(self):
+ pgp = openpgp.OpenPGPScheme(
+ self._soledad, gpgbinary=self.gpg_binary_path)
+ km = self._key_manager()
+
+ with patch.object(logging.Logger, 'error') as mock_logger_error:
+ rf = RandomRefreshPublicKey(pgp, km)
+ rf._get_random_key = \
+ Mock(return_value=defer.succeed(OpenPGPKey(fingerprint=KEY_FINGERPRINT)))
+
+ km._nicknym.fetch_key_with_fingerprint = \
+ Mock(return_value=defer.succeed(OpenPGPKey(fingerprint=ANOTHER_FP)))
+
+ yield rf.maybe_refresh_key()
+
+ mock_logger_error.assert_called_with(ERROR_UNEQUAL_FINGERPRINTS %
+ (KEY_FINGERPRINT, ANOTHER_FP))
+
+ @defer.inlineCallbacks
+ def test_put_new_key_in_local_storage(self):
+ pgp = openpgp.OpenPGPScheme(
+ self._soledad, gpgbinary=self.gpg_binary_path)
+ km = self._key_manager()
+
+ rf = RandomRefreshPublicKey(pgp, km)
+ rf._get_random_key = Mock(return_value=defer.succeed(OpenPGPKey(fingerprint=KEY_FINGERPRINT)))
+
+ km._nicknym.fetch_key_with_fingerprint = \
+ Mock(return_value=defer.succeed(OpenPGPKey(fingerprint=ANOTHER_FP)))
+
+ yield rf.maybe_refresh_key()
+
+ @defer.inlineCallbacks
+ def test_key_expired_will_be_deactivatet(self):
+ pgp = openpgp.OpenPGPScheme(
+ self._soledad, gpgbinary=self.gpg_binary_path)
+ km = self._key_manager()
+ rf = RandomRefreshPublicKey(pgp, km)
+ key = OpenPGPKey(address='zara@leap.se', expiry_date=datetime.now())
+
+ self.assertTrue(key.address is 'zara@leap.se')
+
+ km._openpgp.unactivate_key = Mock(return_value=defer.succeed(None))
+
+ yield rf._maybe_unactivate_key(key)
+
+ self.assertTrue(key.address is None)
+ self.assertFalse(key.is_active())
+
+ def test_start_refreshing(self):
+ pgp = openpgp.OpenPGPScheme(
+ self._soledad, gpgbinary=self.gpg_binary_path)
+
+ with patch.object(logging.Logger, 'debug') as mock_logger_start:
+ rf = RandomRefreshPublicKey(pgp, self._key_manager())
+ rf.start()
+ mock_logger_start.assert_called_with(DEBUG_START_REFRESH)
+ rf.stop()
+ mock_logger_start.assert_called_with(DEBUG_STOP_REFRESH)
+
+ def test_random_interval_is_set_properly(self):
+ pgp = openpgp.OpenPGPScheme(
+ self._soledad, gpgbinary=self.gpg_binary_path)
+ rf = RandomRefreshPublicKey(pgp, self._key_manager())
+ self.assertTrue(rf._loop.interval >= MIN_RANDOM_INTERVAL_RANGE)
+
+ def test_is_random_really_random(self):
+ pgp = openpgp.OpenPGPScheme(
+ self._soledad, gpgbinary=self.gpg_binary_path)
+ rf = RandomRefreshPublicKey(pgp, self._key_manager())
+
+ for x in range(0, 5):
+ random_numbers = []
+
+ for y in range(0, 5):
+ random_numbers.append(rf._random_interval_to_refersh())
+
+ # there are different numbers in the list
+ if len(random_numbers) == len(set(random_numbers)) \
+ or len(random_numbers) == len(set(random_numbers)) + 1:
+ self.assertTrue(True)
+ else:
+ self.assertTrue(False)
+
+