summaryrefslogtreecommitdiff
path: root/keymanager/src
diff options
context:
space:
mode:
Diffstat (limited to 'keymanager/src')
-rw-r--r--keymanager/src/leap/keymanager/__init__.py438
-rw-r--r--keymanager/src/leap/keymanager/errors.py6
-rw-r--r--keymanager/src/leap/keymanager/keys.py86
-rw-r--r--keymanager/src/leap/keymanager/openpgp.py366
-rw-r--r--keymanager/src/leap/keymanager/tests/__init__.py109
-rw-r--r--keymanager/src/leap/keymanager/tests/test_keymanager.py394
-rw-r--r--keymanager/src/leap/keymanager/tests/test_openpgp.py250
-rw-r--r--keymanager/src/leap/keymanager/tests/test_validation.py49
8 files changed, 1024 insertions, 674 deletions
diff --git a/keymanager/src/leap/keymanager/__init__.py b/keymanager/src/leap/keymanager/__init__.py
index b2b05f4a..7ff437e5 100644
--- a/keymanager/src/leap/keymanager/__init__.py
+++ b/keymanager/src/leap/keymanager/__init__.py
@@ -43,6 +43,8 @@ except (ImportError, AssertionError):
import logging
import requests
+from twisted.internet import defer
+
from leap.common.check import leap_assert
from leap.common.events import signal
from leap.common.events import events_pb2 as proto
@@ -51,7 +53,8 @@ from leap.common.decorators import memoized_method
from leap.keymanager.errors import (
KeyNotFound,
KeyAddressMismatch,
- KeyNotValidUpgrade
+ KeyNotValidUpgrade,
+ UnsupportedKeyTypeError
)
from leap.keymanager.validation import ValidationLevel, can_upgrade
@@ -196,15 +199,20 @@ class KeyManager(object):
@memoized_method(invalidation=300)
def _fetch_keys_from_server(self, address):
"""
- Fetch keys bound to C{address} from nickserver and insert them in
+ Fetch keys bound to address from nickserver and insert them in
local database.
:param address: The address bound to the keys.
:type address: str
- :raise KeyNotFound: If the key was not found on nickserver.
+ :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
+
"""
# request keys from the nickserver
+ d = defer.succeed(None)
res = None
try:
res = self._get(self._nickserver_uri, {'address': address})
@@ -212,18 +220,22 @@ class KeyManager(object):
server_keys = res.json()
# insert keys in local database
if self.OPENPGP_KEY in server_keys:
- self.put_raw_key(
+ d = self.put_raw_key(
server_keys['openpgp'],
OpenPGPKey,
address=address,
validation=ValidationLevel.Provider_Trust)
except requests.exceptions.HTTPError as e:
if e.response.status_code == 404:
- raise KeyNotFound(address)
+ d = defer.fail(KeyNotFound(address))
+ else:
+ d = defer.fail(e)
logger.warning("HTTP error retrieving key: %r" % (e,))
logger.warning("%s" % (res.content,))
except Exception as e:
+ d = defer.fail(e)
logger.warning("Error retrieving key: %r" % (e,))
+ return d
#
# key management
@@ -231,7 +243,7 @@ class KeyManager(object):
def send_key(self, ktype):
"""
- Send user's key of type C{ktype} to provider.
+ Send user's key of type ktype to provider.
Public key bound to user's is sent to provider, which will sign it and
replace any prior keys for the same address in its database.
@@ -239,27 +251,33 @@ class KeyManager(object):
:param ktype: The type of the key.
:type ktype: subclass of EncryptionKey
- :raise KeyNotFound: If the key was not found in local database.
+ :return: A Deferred which fires when the key is sent, or which fails
+ with KeyNotFound if the key was not found in local database.
+ :rtype: Deferred
+
+ :raise UnsupportedKeyTypeError: if invalid key type
"""
- leap_assert(
- ktype is OpenPGPKey,
- 'For now we only know how to send OpenPGP public keys.')
- # prepare the public key bound to address
- pubkey = self.get_key(
+ self._assert_supported_key_type(ktype)
+
+ def send(pubkey):
+ data = {
+ self.PUBKEY_KEY: pubkey.key_data
+ }
+ uri = "%s/%s/users/%s.json" % (
+ self._api_uri,
+ self._api_version,
+ self._uid)
+ self._put(uri, data)
+ signal(proto.KEYMANAGER_DONE_UPLOADING_KEYS, self._address)
+
+ d = self.get_key(
self._address, ktype, private=False, fetch_remote=False)
- data = {
- self.PUBKEY_KEY: pubkey.key_data
- }
- uri = "%s/%s/users/%s.json" % (
- self._api_uri,
- self._api_version,
- self._uid)
- self._put(uri, data)
- signal(proto.KEYMANAGER_DONE_UPLOADING_KEYS, self._address)
+ d.addCallback(send)
+ return d
def get_key(self, address, ktype, private=False, fetch_remote=True):
"""
- Return a key of type C{ktype} bound to C{address}.
+ Return a key of type ktype bound to address.
First, search for the key in local storage. If it is not available,
then try to fetch from nickserver.
@@ -274,36 +292,47 @@ class KeyManager(object):
from nickserver
:type fetch_remote: bool
- :return: A key of type C{ktype} bound to C{address}.
- :rtype: EncryptionKey
- :raise KeyNotFound: If the key was not found both locally and in
- keyserver.
+ :return: A Deferred which fires with an EncryptionKey of type ktype
+ bound to address, or which fails with KeyNotFound if no key
+ was found neither locally or in keyserver.
+ :rtype: Deferred
+
+ :raise UnsupportedKeyTypeError: if invalid key type
"""
+ self._assert_supported_key_type(ktype)
logger.debug("getting key for %s" % (address,))
leap_assert(
ktype in self._wrapper_map,
'Unkown key type: %s.' % str(ktype))
- try:
- signal(proto.KEYMANAGER_LOOKING_FOR_KEY, address)
- # return key if it exists in local database
- key = self._wrapper_map[ktype].get_key(address, private=private)
- signal(proto.KEYMANAGER_KEY_FOUND, address)
+ signal(proto.KEYMANAGER_LOOKING_FOR_KEY, address)
+ def key_found(key):
+ signal(proto.KEYMANAGER_KEY_FOUND, address)
return key
- except KeyNotFound:
+
+ def key_not_found(failure):
+ if not failure.check(KeyNotFound):
+ return failure
+
signal(proto.KEYMANAGER_KEY_NOT_FOUND, address)
# we will only try to fetch a key from nickserver if fetch_remote
# is True and the key is not private.
if fetch_remote is False or private is True:
- raise
+ return failure
signal(proto.KEYMANAGER_LOOKING_FOR_KEY, address)
- self._fetch_keys_from_server(address) # might raise KeyNotFound
- key = self._wrapper_map[ktype].get_key(address, private=False)
- signal(proto.KEYMANAGER_KEY_FOUND, address)
-
- return key
+ d = self._fetch_keys_from_server(address)
+ d.addCallback(
+ lambda _:
+ self._wrapper_map[ktype].get_key(address, private=False))
+ d.addCallback(key_found)
+ return d
+
+ # return key if it exists in local database
+ d = self._wrapper_map[ktype].get_key(address, private=private)
+ d.addCallbacks(key_found, key_not_found)
+ return d
def get_all_keys(self, private=False):
"""
@@ -312,33 +341,50 @@ class KeyManager(object):
:param private: Include private keys
:type private: bool
- :return: A list with all keys in local db.
- :rtype: list
- """
- return map(
- lambda doc: build_key_from_dict(
- self._key_class_from_type(doc.content['type']),
- doc.content),
- self._soledad.get_from_index(
- TAGS_PRIVATE_INDEX,
- KEYMANAGER_KEY_TAG,
- '1' if private else '0'))
+ :return: A Deferred which fires with a list of all keys in local db.
+ :rtype: Deferred
+ """
+ def build_keys(docs):
+ return map(
+ lambda doc: build_key_from_dict(
+ self._key_class_from_type(doc.content['type']),
+ doc.content),
+ docs)
+
+ # XXX: there is no check that the soledad indexes are ready, as it
+ # happens with EncryptionScheme.
+ # The usecases right now are not problematic. This could be solve
+ # adding a keytype to this funciont and moving the soledad request
+ # to the EncryptionScheme.
+ d = self._soledad.get_from_index(
+ TAGS_PRIVATE_INDEX,
+ KEYMANAGER_KEY_TAG,
+ '1' if private else '0')
+ d.addCallback(build_keys)
+ return d
def gen_key(self, ktype):
"""
- Generate a key of type C{ktype} bound to the user's address.
+ Generate a key of type ktype bound to the user's address.
:param ktype: The type of the key.
:type ktype: subclass of EncryptionKey
- :return: The generated key.
- :rtype: EncryptionKey
+ :return: A Deferred which fires with the generated EncryptionKey.
+ :rtype: Deferred
+
+ :raise UnsupportedKeyTypeError: if invalid key type
"""
- signal(proto.KEYMANAGER_STARTED_KEY_GENERATION, self._address)
- key = self._wrapper_map[ktype].gen_key(self._address)
- signal(proto.KEYMANAGER_FINISHED_KEY_GENERATION, self._address)
+ self._assert_supported_key_type(ktype)
- return key
+ def signal_finished(key):
+ signal(proto.KEYMANAGER_FINISHED_KEY_GENERATION, self._address)
+ return key
+
+ signal(proto.KEYMANAGER_STARTED_KEY_GENERATION, self._address)
+ d = self._wrapper_map[ktype].gen_key(self._address)
+ d.addCallback(signal_finished)
+ return d
#
# Setters/getters
@@ -397,7 +443,8 @@ class KeyManager(object):
def encrypt(self, data, address, ktype, passphrase=None, sign=None,
cipher_algo='AES256', fetch_remote=True):
"""
- Encrypt C{data} for C{address} and sign with C{sign} address.
+ Encrypt data with the public key bound to address and sign with with
+ the private key bound to sign address.
:param data: The data to be encrypted.
:type data: str
@@ -412,34 +459,39 @@ class KeyManager(object):
:type sign: str
:param cipher_algo: The cipher algorithm to use.
:type cipher_algo: str
- :param fetch_remote: If key not found in local storage try to fetch
+ :param fetch_remote: If key is not found in local storage try to fetch
from nickserver
:type fetch_remote: bool
- :return: The encrypted data.
- :rtype: str
+ :return: A Deferred which fires with the encrypted data as str, or
+ which fails with KeyNotFound if no keys were found neither
+ locally or in keyserver or fails with EncryptError if failed
+ encrypting for some reason.
+ :rtype: Deferred
- :raise KeyNotFound: If any of the keys was not found both locally and
- in keyserver.
- :raise EncryptError: Raised if failed encrypting for some reason.
+ :raise UnsupportedKeyTypeError: if invalid key type
"""
- pubkey = self.get_key(address, ktype, private=False,
- fetch_remote=fetch_remote)
- privkey = None
- if sign is not None:
- privkey = self.get_key(sign, ktype, private=True)
- return self._wrapper_map[ktype].encrypt(
- data, pubkey, passphrase, privkey, cipher_algo=cipher_algo)
+ self._assert_supported_key_type(ktype)
+
+ def encrypt(keys):
+ pubkey, signkey = keys
+ return self._wrapper_map[ktype].encrypt(
+ data, pubkey, passphrase, sign=signkey,
+ cipher_algo=cipher_algo)
+
+ d = self._get_keys(ktype, address, sign, fetch_remote=fetch_remote)
+ d.addCallback(encrypt)
+ return d
def decrypt(self, data, address, ktype, passphrase=None, verify=None,
fetch_remote=True):
"""
- Decrypt C{data} using private key from @{address} and verify with
- C{verify} address.
+ Decrypt data using private key from address and verify with public key
+ bound to verify address.
:param data: The data to be decrypted.
:type data: str
- :param address: The address to who was encrypted.
+ :param address: The address to whom data was encrypted.
:type address: str
:param ktype: The type of the key.
:type ktype: subclass of EncryptionKey
@@ -452,26 +504,59 @@ class KeyManager(object):
to fetch from nickserver
:type fetch_remote: bool
- :return: The decrypted data and the signing key if signature verifies
- :rtype: (unicode, EncryptionKey)
+ :return: A Deferred which fires with the decrypted data as str and the
+ signing EncryptionKey if signature verifies, or which fails
+ with KeyNotFound if no keys were found neither locally or in
+ keyserver or fails with DecryptError if failed decrypting for
+ some reason.
+ :rtype: Deferred
- :raise KeyNotFound: If any of the keys was not found both locally and
- in keyserver.
- :raise DecryptError: Raised if failed decrypting for some reason.
+ :raise UnsupportedKeyTypeError: if invalid key type
"""
- privkey = self.get_key(address, ktype, private=True)
- pubkey = None
- if verify is not None:
- pubkey = self.get_key(verify, ktype, private=False,
- fetch_remote=fetch_remote)
- decrypted, signed = self._wrapper_map[ktype].decrypt(
- data, privkey, passphrase, pubkey)
- return (decrypted, pubkey if signed else None)
+ self._assert_supported_key_type(ktype)
+
+ def decrypt(keys):
+ pubkey, privkey = keys
+ decrypted, signed = self._wrapper_map[ktype].decrypt(
+ data, privkey, passphrase=passphrase, verify=pubkey)
+ return (decrypted, pubkey if signed else None)
+
+ d = self._get_keys(ktype, verify, address, fetch_remote)
+ d.addCallback(decrypt)
+ return d
+
+ def _get_keys(self, ktype, public, private, fetch_remote=True):
+ """
+ Get public and private keys of ktype.
+
+ :param ktype: The type of the key.
+ :type ktype: subclass of EncryptionKey
+ :param public: The address of the public key.
+ :type public: str
+ :param private: The address of the private key.
+ :type private: str
+ :param fetch_remote: If key for verify not found in local storage try
+ to fetch from nickserver
+ :type fetch_remote: bool
+
+ :return: A Deferred which fires with a tuple with public and private
+ EncryptionKeys, or which fails with KeyNotFound if no keys
+ were found neither locally or in keyserver.
+ :rtype: Deferred
+ """
+ dpub = defer.succeed(None)
+ if public is not None:
+ dpub = self.get_key(public, ktype, private=False,
+ fetch_remote=fetch_remote)
+ dpriv = defer.succeed(None)
+ if private is not None:
+ dpriv = self.get_key(private, ktype, private=True)
+ return defer.gatherResults([dpub, dpriv])
def sign(self, data, address, ktype, digest_algo='SHA512', clearsign=False,
detach=True, binary=False):
"""
- Sign C{data} with C{address}.
+ Sign data with private key bound to address.
:param data: The data to be signed.
:type data: str
@@ -488,23 +573,30 @@ class KeyManager(object):
:param binary: If True, do not ascii armour the output.
:type binary: bool
- :return: The signed data.
- :rtype: str
+ :return: A Deferred which fires with the signed data as str or fails
+ with KeyNotFound if no key was found neither locally or in
+ keyserver or fails with SignFailed if there was any error
+ signing.
+ :rtype: Deferred
- :raise KeyNotFound: If the key was not found both locally and
- in keyserver.
- :raise SignFailed: If there was any error signing.
+ :raise UnsupportedKeyTypeError: if invalid key type
"""
- privkey = self.get_key(address, ktype, private=True)
- return self._wrapper_map[ktype].sign(
- data, privkey, digest_algo=digest_algo, clearsign=clearsign,
- detach=detach, binary=binary)
+ self._assert_supported_key_type(ktype)
+
+ def sign(privkey):
+ return self._wrapper_map[ktype].sign(
+ data, privkey, digest_algo=digest_algo, clearsign=clearsign,
+ detach=detach, binary=binary)
+
+ d = self.get_key(address, ktype, private=True)
+ d.addCallback(sign)
+ return d
def verify(self, data, address, ktype, detached_sig=None,
fetch_remote=True):
"""
- Verify signed C{data} with C{address}, eventually using
- C{detached_sig}.
+ Verify signed data with private key bound to address, eventually using
+ detached_sig.
:param data: The data to be verified.
:type data: str
@@ -519,71 +611,90 @@ class KeyManager(object):
to fetch from nickserver
:type fetch_remote: bool
- :return: The signing key if signature verifies else None
- :rtype: EncryptionKey
+ :return: A Deferred which fires with the signing EncryptionKey if
+ signature verifies else None, or which fails with KeyNotFound
+ if no key was found neither locally or in keyserver.
+ :rtype: Deferred
- :raise KeyNotFound: If the key was not found both locally and
- in keyserver.
+ :raise UnsupportedKeyTypeError: if invalid key type
"""
- pubkey = self.get_key(address, ktype, private=False,
- fetch_remote=fetch_remote)
- signed = self._wrapper_map[ktype].verify(
- data, pubkey, detached_sig=detached_sig)
- return pubkey if signed else None
+ self._assert_supported_key_type(ktype)
+
+ def verify(pubkey):
+ signed = self._wrapper_map[ktype].verify(
+ data, pubkey, detached_sig=detached_sig)
+ return pubkey if signed else None
+
+ d = self.get_key(address, ktype, private=False,
+ fetch_remote=fetch_remote)
+ d.addCallback(verify)
+ return d
def delete_key(self, key):
"""
- Remove C{key} from storage.
-
- May raise:
- openpgp.errors.KeyNotFound
- openpgp.errors.KeyAttributesDiffer
+ Remove key from storage.
:param key: The key to be removed.
:type key: EncryptionKey
+
+ :return: A Deferred which fires when the key is deleted, or which fails
+ KeyNotFound if the key was not found on local storage.
+ :rtype: Deferred
+
+ :raise UnsupportedKeyTypeError: if invalid key type
"""
- try:
- self._wrapper_map[type(key)].delete_key(key)
- except IndexError as e:
- leap_assert(False, "Unsupported key type. Error {0!r}".format(e))
+ self._assert_supported_key_type(type(key))
+ return self._wrapper_map[type(key)].delete_key(key)
def put_key(self, key, address):
"""
- Put C{key} in local storage.
+ Put key bound to address in local storage.
:param key: The key to be stored
:type key: EncryptionKey
:param address: address for which this key will be active
:type address: str
- :raises KeyAddressMismatch: if address doesn't match any uid on the key
- :raises KeyNotValidUpdate: if a key with the same uid exists and the
- new one is not a valid update for it
+ :return: A Deferred which fires when the key is in the storage, or
+ which fails with KeyAddressMismatch if address doesn't match
+ any uid on the key or fails with KeyNotValidUpdate if a key
+ with the same uid exists and the new one is not a valid update
+ for it.
+ :rtype: Deferred
+
+ :raise UnsupportedKeyTypeError: if invalid key type
"""
- if address not in key.address:
- raise KeyAddressMismatch("UID %s found, but expected %s"
- % (str(key.address), address))
+ self._assert_supported_key_type(type(key))
- try:
- old_key = self._wrapper_map[type(key)].get_key(address,
- private=key.private)
- except KeyNotFound:
- old_key = None
-
- if key.private or can_upgrade(key, old_key):
- try:
- self._wrapper_map[type(key)].put_key(key, address)
- except IndexError as e:
- leap_assert(
- False, "Unsupported key type. Error {0!r}".format(e))
- else:
- raise KeyNotValidUpgrade("Key %s can not be upgraded by new key %s"
- % (old_key.key_id, key.key_id))
+ if address not in key.address:
+ return defer.fail(
+ KeyAddressMismatch("UID %s found, but expected %s"
+ % (str(key.address), address)))
+
+ def old_key_not_found(failure):
+ if failure.check(KeyNotFound):
+ return None
+ else:
+ return failure
+
+ def check_upgrade(old_key):
+ if key.private or can_upgrade(key, old_key):
+ return self._wrapper_map[type(key)].put_key(key, address)
+ else:
+ raise KeyNotValidUpgrade(
+ "Key %s can not be upgraded by new key %s"
+ % (old_key.key_id, key.key_id))
+
+ d = self._wrapper_map[type(key)].get_key(address,
+ private=key.private)
+ d.addErrback(old_key_not_found)
+ d.addCallback(check_upgrade)
+ return d
def put_raw_key(self, key, ktype, address,
validation=ValidationLevel.Weak_Chain):
"""
- Put C{key} in local storage.
+ Put raw key bound to address in local storage.
:param key: The ascii key to be stored
:type key: str
@@ -595,19 +706,24 @@ class KeyManager(object):
(default: 'Weak_Chain')
:type validation: ValidationLevel
- :raises KeyAddressMismatch: if address doesn't match any uid on the key
- :raises KeyNotValidUpdate: if a key with the same uid exists and the
- new one is not a valid update for it
+ :return: A Deferred which fires when the key is in the storage, or
+ which fails with KeyAddressMismatch if address doesn't match
+ any uid on the key or fails with KeyNotValidUpdate if a key
+ with the same uid exists and the new one is not a valid update
+ for it.
+ :rtype: Deferred
+
+ :raise UnsupportedKeyTypeError: if invalid key type
"""
+ self._assert_supported_key_type(ktype)
pubkey, _ = self._wrapper_map[ktype].parse_ascii_key(key)
-
pubkey.validation = validation
- self.put_key(pubkey, address)
+ return self.put_key(pubkey, address)
def fetch_key(self, address, uri, ktype,
validation=ValidationLevel.Weak_Chain):
"""
- Fetch a public key for C{address} from the network and put it in
+ Fetch a public key bound to address from the network and put it in
local storage.
:param address: The email address of the key.
@@ -620,22 +736,40 @@ class KeyManager(object):
(default: 'Weak_Chain')
:type validation: ValidationLevel
- :raises KeyNotFound: if not valid key on C{uri}
- :raises KeyAddressMismatch: if address doesn't match any uid on the key
- :raises KeyNotValidUpdate: if a key with the same uid exists and the
- new one is not a valid update for it
+ :return: A Deferred which fires when the key is in the storage, or
+ which fails with KeyNotFound: if not valid key on uri or fails
+ with KeyAddressMismatch if address doesn't match any uid on
+ the key or fails with KeyNotValidUpdate if a key with the same
+ uid exists and the new one is not a valid update for it.
+ :rtype: Deferred
+
+ :raise UnsupportedKeyTypeError: if invalid key type
"""
+ self._assert_supported_key_type(ktype)
+
res = self._get(uri)
if not res.ok:
- raise KeyNotFound(uri)
+ return defer.fail(KeyNotFound(uri))
# XXX parse binary keys
pubkey, _ = self._wrapper_map[ktype].parse_ascii_key(res.content)
if pubkey is None:
- raise KeyNotFound(uri)
+ return defer.fail(KeyNotFound(uri))
pubkey.validation = validation
- self.put_key(pubkey, address)
+ return self.put_key(pubkey, address)
+
+ def _assert_supported_key_type(self, ktype):
+ """
+ Check if ktype is one of the supported key types
+
+ :param ktype: the type of the key.
+ :type ktype: subclass of EncryptionKey
+
+ :raise UnsupportedKeyTypeError: if invalid key type
+ """
+ if ktype not in self._wrapper_map:
+ raise UnsupportedKeyTypeError(str(ktype))
from ._version import get_versions
__version__ = get_versions()['version']
diff --git a/keymanager/src/leap/keymanager/errors.py b/keymanager/src/leap/keymanager/errors.py
index c068b27f..4041837e 100644
--- a/keymanager/src/leap/keymanager/errors.py
+++ b/keymanager/src/leap/keymanager/errors.py
@@ -95,3 +95,9 @@ class KeyNotValidUpgrade(Exception):
"""
Already existing key can not be upgraded with the new key
"""
+
+
+class UnsupportedKeyTypeError(Exception):
+ """
+ Invalid key type
+ """
diff --git a/keymanager/src/leap/keymanager/keys.py b/keymanager/src/leap/keymanager/keys.py
index 0e243ba0..4e98de6c 100644
--- a/keymanager/src/leap/keymanager/keys.py
+++ b/keymanager/src/leap/keymanager/keys.py
@@ -33,6 +33,7 @@ import time
from abc import ABCMeta, abstractmethod
from datetime import datetime
from leap.common.check import leap_assert
+from twisted.internet import defer
from leap.keymanager.validation import ValidationLevel, toValidationLevel
@@ -277,21 +278,61 @@ class EncryptionScheme(object):
"""
leap_assert(self._soledad is not None,
"Cannot init indexes with null soledad")
- # Ask the database for currently existing indexes.
- db_indexes = dict(self._soledad.list_indexes())
- # Loop through the indexes we expect to find.
- for name, expression in INDEXES.items():
- if name not in db_indexes:
- # The index does not yet exist.
- self._soledad.create_index(name, *expression)
- continue
- if expression == db_indexes[name]:
- # The index exists and is up to date.
- continue
- # The index exists but the definition is not what expected, so we
- # delete it and add the proper index expression.
- self._soledad.delete_index(name)
- self._soledad.create_index(name, *expression)
+
+ def init_idexes(indexes):
+ deferreds = []
+ db_indexes = dict(indexes)
+ # Loop through the indexes we expect to find.
+ for name, expression in INDEXES.items():
+ if name not in db_indexes:
+ # The index does not yet exist.
+ d = self._soledad.create_index(name, *expression)
+ deferreds.append(d)
+ elif expression != db_indexes[name]:
+ # The index exists but the definition is not what expected,
+ # so we delete it and add the proper index expression.
+ d = self._soledad.delete_index(name)
+ d.addCallback(
+ lambda _:
+ self._soledad.create_index(name, *expression))
+ deferreds.append(d)
+ return defer.gatherResults(deferreds, consumeErrors=True)
+
+ self.deferred_indexes = self._soledad.list_indexes()
+ self.deferred_indexes.addCallback(init_idexes)
+
+ def _wait_indexes(self, *methods):
+ """
+ Methods that need to wait for the indexes to be ready.
+
+ Heavily based on
+ http://blogs.fluidinfo.com/terry/2009/05/11/a-mixin-class-allowing-python-__init__-methods-to-work-with-twisted-deferreds/
+
+ :param methods: methods that need to wait for the indexes to be ready
+ :type methods: tuple(str)
+ """
+ self.waiting = []
+ self.stored = {}
+
+ def restore(_):
+ for method in self.stored:
+ setattr(self, method, self.stored[method])
+ for d in self.waiting:
+ d.callback(None)
+
+ def makeWrapper(method):
+ def wrapper(*args, **kw):
+ d = defer.Deferred()
+ d.addCallback(lambda _: self.stored[method](*args, **kw))
+ self.waiting.append(d)
+ return d
+ return wrapper
+
+ for method in methods:
+ self.stored[method] = getattr(self, method)
+ setattr(self, method, makeWrapper(method))
+
+ self.deferred_indexes.addCallback(restore)
@abstractmethod
def get_key(self, address, private=False):
@@ -303,9 +344,10 @@ class EncryptionScheme(object):
:param private: Look for a private key instead of a public one?
:type private: bool
- :return: The key bound to C{address}.
- :rtype: EncryptionKey
- @raise KeyNotFound: If the key was not found on local storage.
+ :return: A Deferred which fires with the EncryptionKey bound to
+ address, or which fails with KeyNotFound if the key was not
+ found on local storage.
+ :rtype: Deferred
"""
pass
@@ -318,6 +360,9 @@ class EncryptionScheme(object):
:type key: EncryptionKey
:param address: address for which this key will be active.
:type address: str
+
+ :return: A Deferred which fires when the key is in the storage.
+ :rtype: Deferred
"""
pass
@@ -341,6 +386,11 @@ class EncryptionScheme(object):
:param key: The key to be removed.
:type key: EncryptionKey
+
+ :return: A Deferred which fires when the key is deleted, or which
+ fails with KeyNotFound if the key was not found on local
+ storage.
+ :rtype: Deferred
"""
pass
diff --git a/keymanager/src/leap/keymanager/openpgp.py b/keymanager/src/leap/keymanager/openpgp.py
index 1d1de989..f81fb0a0 100644
--- a/keymanager/src/leap/keymanager/openpgp.py
+++ b/keymanager/src/leap/keymanager/openpgp.py
@@ -28,6 +28,7 @@ import io
from datetime import datetime
from gnupg import GPG
from gnupg.gnupg import GPGUtilities
+from twisted.internet import defer
from leap.common.check import leap_assert, leap_assert_type, leap_check
from leap.keymanager import errors
@@ -243,6 +244,7 @@ class OpenPGPScheme(EncryptionScheme):
:type gpgbinary: C{str}
"""
EncryptionScheme.__init__(self, soledad)
+ self._wait_indexes("get_key", "put_key")
self._gpgbinary = gpgbinary
#
@@ -255,59 +257,67 @@ class OpenPGPScheme(EncryptionScheme):
:param address: The address bound to the key.
:type address: str
- :return: The key bound to C{address}.
- :rtype: OpenPGPKey
- @raise KeyAlreadyExists: If key already exists in local database.
+
+ :return: A Deferred which fires with the key bound to address, or fails
+ with KeyAlreadyExists if key already exists in local database.
+ :rtype: Deferred
"""
# make sure the key does not already exist
leap_assert(is_address(address), 'Not an user address: %s' % address)
- try:
- self.get_key(address)
- raise errors.KeyAlreadyExists(address)
- except errors.KeyNotFound:
- logger.debug('Key for %s not found' % (address,))
-
- with self._temporary_gpgwrapper() as gpg:
- # TODO: inspect result, or use decorator
- params = gpg.gen_key_input(
- key_type='RSA',
- key_length=4096,
- name_real=address,
- name_email=address,
- name_comment='')
- logger.info("About to generate keys... This might take SOME time.")
- gpg.gen_key(params)
- logger.info("Keys for %s have been successfully "
- "generated." % (address,))
- pubkeys = gpg.list_keys()
-
- # assert for new key characteristics
-
- # XXX This exception is not properly catched by the soledad
- # bootstrapping, so if we do not finish generating the keys
- # we end with a blocked thread -- kali
- leap_assert(
- len(pubkeys) is 1, # a unitary keyring!
- 'Keyring has wrong number of keys: %d.' % len(pubkeys))
- key = gpg.list_keys(secret=True).pop()
- leap_assert(
- len(key['uids']) is 1, # with just one uid!
- 'Wrong number of uids for key: %d.' % len(key['uids']))
- uid_match = False
- for uid in key['uids']:
- if re.match('.*<%s>$' % address, uid) is not None:
- uid_match = True
- return
- leap_assert(uid_match, 'Key not correctly bound to address.')
- # insert both public and private keys in storage
- for secret in [True, False]:
- key = gpg.list_keys(secret=secret).pop()
- openpgp_key = _build_key_from_gpg(
- key, gpg.export_keys(key['fingerprint'], secret=secret))
- self.put_key(openpgp_key, address)
+ def _gen_key(_):
+ with self._temporary_gpgwrapper() as gpg:
+ # TODO: inspect result, or use decorator
+ params = gpg.gen_key_input(
+ key_type='RSA',
+ key_length=4096,
+ name_real=address,
+ name_email=address,
+ name_comment='')
+ logger.info("About to generate keys... "
+ "This might take SOME time.")
+ gpg.gen_key(params)
+ logger.info("Keys for %s have been successfully "
+ "generated." % (address,))
+ pubkeys = gpg.list_keys()
+
+ # assert for new key characteristics
+
+ # XXX This exception is not properly catched by the soledad
+ # bootstrapping, so if we do not finish generating the keys
+ # we end with a blocked thread -- kali
+
+ leap_assert(
+ len(pubkeys) is 1, # a unitary keyring!
+ 'Keyring has wrong number of keys: %d.' % len(pubkeys))
+ key = gpg.list_keys(secret=True).pop()
+ leap_assert(
+ len(key['uids']) is 1, # with just one uid!
+ 'Wrong number of uids for key: %d.' % len(key['uids']))
+ uid_match = False
+ for uid in key['uids']:
+ if re.match('.*<%s>$' % address, uid) is not None:
+ uid_match = True
+ return
+ leap_assert(uid_match, 'Key not correctly bound to address.')
+ # insert both public and private keys in storage
+ deferreds = []
+ for secret in [True, False]:
+ key = gpg.list_keys(secret=secret).pop()
+ openpgp_key = _build_key_from_gpg(
+ key,
+ gpg.export_keys(key['fingerprint'], secret=secret))
+ d = self.put_key(openpgp_key, address)
+ deferreds.append(d)
+ return defer.gatherResults(deferreds)
+
+ def key_already_exists(_):
+ raise errors.KeyAlreadyExists(address)
- return self.get_key(address, private=True)
+ d = self.get_key(address)
+ d.addCallbacks(key_already_exists, _gen_key)
+ d.addCallback(lambda _: self.get_key(address, private=True))
+ return d
def get_key(self, address, private=False):
"""
@@ -318,19 +328,24 @@ class OpenPGPScheme(EncryptionScheme):
:param private: Look for a private key instead of a public one?
:type private: bool
- :return: The key bound to C{address}.
- :rtype: OpenPGPKey
- @raise KeyNotFound: If the key was not found on local storage.
+ :return: A Deferred which fires with the OpenPGPKey bound to address,
+ or which fails with KeyNotFound if the key was not found on
+ local storage.
+ :rtype: Deferred
"""
address = _parse_address(address)
- doc = self._get_key_doc(address, private)
- if doc is None:
- raise errors.KeyNotFound(address)
- leap_assert(
- address in doc.content[KEY_ADDRESS_KEY],
- 'Wrong address in key data.')
- return build_key_from_dict(OpenPGPKey, doc.content)
+ def build_key(doc):
+ if doc is None:
+ raise errors.KeyNotFound(address)
+ leap_assert(
+ address in doc.content[KEY_ADDRESS_KEY],
+ 'Wrong address in key data.')
+ return build_key_from_dict(OpenPGPKey, doc.content)
+
+ d = self._get_key_doc(address, private)
+ d.addCallback(build_key)
+ return d
def parse_ascii_key(self, key_data):
"""
@@ -388,6 +403,9 @@ class OpenPGPScheme(EncryptionScheme):
:type key_data: str or unicode
:param address: address for which this key will be active
:type address: str
+
+ :return: A Deferred which fires when the OpenPGPKey is in the storage.
+ :rtype: Deferred
"""
leap_assert_type(key_data, (str, unicode))
@@ -395,12 +413,17 @@ class OpenPGPScheme(EncryptionScheme):
try:
openpgp_pubkey, openpgp_privkey = self.parse_ascii_key(key_data)
except (errors.KeyAddressMismatch, errors.KeyFingerprintMismatch) as e:
- leap_assert(False, repr(e))
+ return defer.fail(e)
+ def put_key(_, key):
+ return self.put_key(key, address)
+
+ d = defer.succeed(None)
if openpgp_pubkey is not None:
- self.put_key(openpgp_pubkey, address)
+ d.addCallback(put_key, openpgp_pubkey)
if openpgp_privkey is not None:
- self.put_key(openpgp_privkey, address)
+ d.addCallback(put_key, openpgp_privkey)
+ return d
def put_key(self, key, address):
"""
@@ -410,42 +433,59 @@ class OpenPGPScheme(EncryptionScheme):
:type key: OpenPGPKey
:param address: address for which this key will be active.
:type address: str
+
+ :return: A Deferred which fires when the key is in the storage.
+ :rtype: Deferred
"""
- self._put_key_doc(key)
- self._put_active_doc(key, address)
+ d = self._put_key_doc(key)
+ d.addCallback(lambda _: self._put_active_doc(key, address))
+ return d
def _put_key_doc(self, key):
"""
Put key document in soledad
:type key: OpenPGPKey
- """
- docs = self._soledad.get_from_index(
+ :rtype: Deferred
+ """
+ def check_and_put(docs, key):
+ if len(docs) == 1:
+ doc = docs.pop()
+ if key.fingerprint == doc.content[KEY_FINGERPRINT_KEY]:
+ # in case of an update of the key merge them with gnupg
+ with self._temporary_gpgwrapper() as gpg:
+ gpg.import_keys(doc.content[KEY_DATA_KEY])
+ gpg.import_keys(key.key_data)
+ gpgkey = gpg.list_keys(secret=key.private).pop()
+ key = _build_key_from_gpg(
+ gpgkey,
+ gpg.export_keys(gpgkey['fingerprint'],
+ secret=key.private))
+ doc.set_json(key.get_json())
+ d = self._soledad.put_doc(doc)
+ else:
+ logger.critical(
+ "Can't put a key whith the same key_id and different "
+ "fingerprint: %s, %s"
+ % (key.fingerprint, doc.content[KEY_FINGERPRINT_KEY]))
+ d = defer.fail(
+ errors.KeyFingerprintMismatch(key.fingerprint))
+ elif len(docs) > 1:
+ logger.critical(
+ "There is more than one key with the same key_id %s"
+ % (key.key_id,))
+ d = defer.fail(errors.KeyAttributesDiffer(key.key_id))
+ else:
+ d = self._soledad.create_doc_from_json(key.get_json())
+ return d
+
+ d = self._soledad.get_from_index(
TYPE_ID_PRIVATE_INDEX,
self.KEY_TYPE,
key.key_id,
'1' if key.private else '0')
- if len(docs) != 0:
- doc = docs.pop()
- if key.fingerprint == doc.content[KEY_FINGERPRINT_KEY]:
- # in case of an update of the key merge them with gnupg
- with self._temporary_gpgwrapper() as gpg:
- gpg.import_keys(doc.content[KEY_DATA_KEY])
- gpg.import_keys(key.key_data)
- gpgkey = gpg.list_keys(secret=key.private).pop()
- key = _build_key_from_gpg(
- gpgkey,
- gpg.export_keys(gpgkey['fingerprint'],
- secret=key.private))
- doc.set_json(key.get_json())
- self._soledad.put_doc(doc)
- else:
- logger.critical(
- "Can't put a key whith the same key_id and different "
- "fingerprint: %s, %s"
- % (key.fingerprint, doc.content[KEY_FINGERPRINT_KEY]))
- else:
- self._soledad.create_doc_from_json(key.get_json())
+ d.addCallback(check_and_put, key)
+ return d
def _put_active_doc(self, key, address):
"""
@@ -453,24 +493,37 @@ class OpenPGPScheme(EncryptionScheme):
:type key: OpenPGPKey
:type addresses: str
+ :rtype: Deferred
"""
- docs = self._soledad.get_from_index(
+ def check_and_put(docs):
+ if len(docs) == 1:
+ doc = docs.pop()
+ doc.set_json(key.get_active_json(address))
+ d = self._soledad.put_doc(doc)
+ else:
+ if len(docs) > 1:
+ logger.error("There is more than one active key document "
+ "for the address %s" % (address,))
+ deferreds = []
+ for doc in docs:
+ delete = self._soledad.delete_doc(doc)
+ deferreds.append(delete)
+ d = defer.gatherResults(deferreds, consumeErrors=True)
+ else:
+ d = defer.succeed(None)
+
+ d.addCallback(
+ lambda _: self._soledad.create_doc_from_json(
+ key.get_active_json(address)))
+ return d
+
+ d = self._soledad.get_from_index(
TYPE_ADDRESS_PRIVATE_INDEX,
self.ACTIVE_TYPE,
address,
'1' if key.private else '0')
- if len(docs) == 1:
- doc = docs.pop()
- doc.set_json(key.get_active_json(address))
- self._soledad.put_doc(doc)
- else:
- if len(docs) > 1:
- logger.error("There is more than one active key document "
- "for the address %s" % (address,))
- for doc in docs:
- self._soledad.delete_doc(doc)
- self._soledad.create_doc_from_json(
- key.get_active_json(address))
+ d.addCallback(check_and_put)
+ return d
def _get_key_doc(self, address, private=False):
"""
@@ -482,69 +535,94 @@ class OpenPGPScheme(EncryptionScheme):
:type address: str
:param private: Whether to look for a private key.
:type private: bool
- :return: The document with the key or None if it does not exist.
- :rtype: leap.soledad.document.SoledadDocument
+
+ :return: A Deferred which fires with the SoledadDocument with the key
+ or None if it does not exist.
+ :rtype: Deferred
"""
- activedoc = self._soledad.get_from_index(
+ def get_key_from_active_doc(activedoc):
+ if len(activedoc) is 0:
+ return None
+ leap_assert(
+ len(activedoc) is 1,
+ 'Found more than one key for address %s!' % (address,))
+
+ key_id = activedoc[0].content[KEY_ID_KEY]
+ d = self._soledad.get_from_index(
+ TYPE_ID_PRIVATE_INDEX,
+ self.KEY_TYPE,
+ key_id,
+ '1' if private else '0')
+ d.addCallback(get_doc, key_id)
+ return d
+
+ def get_doc(doclist, key_id):
+ leap_assert(
+ len(doclist) is 1,
+ 'There is %d keys for id %s!' % (len(doclist), key_id))
+ return doclist.pop()
+
+ d = self._soledad.get_from_index(
TYPE_ADDRESS_PRIVATE_INDEX,
self.ACTIVE_TYPE,
address,
'1' if private else '0')
- if len(activedoc) is 0:
- return None
- leap_assert(
- len(activedoc) is 1,
- 'Found more than one key for address %s!' % (address,))
-
- key_id = activedoc[0].content[KEY_ID_KEY]
- doclist = self._soledad.get_from_index(
- TYPE_ID_PRIVATE_INDEX,
- self.KEY_TYPE,
- key_id,
- '1' if private else '0')
- leap_assert(
- len(doclist) is 1,
- 'There is %d keys for id %s!' % (len(doclist), key_id))
- return doclist.pop()
+ d.addCallback(get_key_from_active_doc)
+ return d
def delete_key(self, key):
"""
Remove C{key} from storage.
- May raise:
- errors.KeyNotFound
-
:param key: The key to be removed.
:type key: EncryptionKey
+
+ :return: A Deferred which fires when the key is deleted, or which
+ fails with KeyNotFound if the key was not found on local
+ storage.
+ :rtype: Deferred
"""
leap_assert_type(key, OpenPGPKey)
- activedocs = self._soledad.get_from_index(
- TYPE_ID_PRIVATE_INDEX,
- self.ACTIVE_TYPE,
- key.key_id,
- '1' if key.private else '0')
- for doc in activedocs:
- self._soledad.delete_doc(doc)
- docs = self._soledad.get_from_index(
+ def delete_docs(activedocs):
+ deferreds = []
+ for doc in activedocs:
+ d = self._soledad.delete_doc(doc)
+ deferreds.append(d)
+ return defer.gatherResults(deferreds)
+
+ def get_key_docs(_):
+ return self._soledad.get_from_index(
+ TYPE_ID_PRIVATE_INDEX,
+ self.KEY_TYPE,
+ key.key_id,
+ '1' if key.private else '0')
+
+ def delete_key(docs):
+ if len(docs) == 0:
+ raise errors.KeyNotFound(key)
+ if len(docs) > 1:
+ logger.critical("There is more than one key for key_id %s"
+ % key.key_id)
+
+ doc = None
+ for d in docs:
+ if d.content['fingerprint'] == key.fingerprint:
+ doc = d
+ break
+ if doc is None:
+ raise errors.KeyNotFound(key)
+ return self._soledad.delete_doc(doc)
+
+ d = self._soledad.get_from_index(
TYPE_ID_PRIVATE_INDEX,
- self.KEY_TYPE,
+ self.ACTIVE_TYPE,
key.key_id,
'1' if key.private else '0')
- if len(docs) == 0:
- raise errors.KeyNotFound(key)
- if len(docs) > 1:
- logger.critical("There is more than one key for key_id %s"
- % key.key_id)
-
- doc = None
- for d in docs:
- if d.content['fingerprint'] == key.fingerprint:
- doc = d
- break
- if doc is None:
- raise errors.KeyNotFound(key)
- self._soledad.delete_doc(doc)
+ d.addCallback(delete_docs)
+ d.addCallback(get_key_docs)
+ d.addCallback(delete_key)
+ return d
#
# Data encryption, decryption, signing and verifying
diff --git a/keymanager/src/leap/keymanager/tests/__init__.py b/keymanager/src/leap/keymanager/tests/__init__.py
index 1ea33b57..05b44871 100644
--- a/keymanager/src/leap/keymanager/tests/__init__.py
+++ b/keymanager/src/leap/keymanager/tests/__init__.py
@@ -18,37 +18,25 @@
Base classes for the Key Manager tests.
"""
-from mock import Mock
+from twisted.internet.defer import gatherResults
+from twisted.trial import unittest
from leap.common.testing.basetest import BaseLeapTest
from leap.soledad.client import Soledad
from leap.keymanager import KeyManager
+from leap.keymanager.openpgp import OpenPGPKey
ADDRESS = 'leap@leap.se'
+ADDRESS_2 = 'anotheruser@leap.se'
# XXX discover the gpg binary path
GPG_BINARY_PATH = '/usr/bin/gpg'
-class KeyManagerWithSoledadTestCase(BaseLeapTest):
+class KeyManagerWithSoledadTestCase(unittest.TestCase, BaseLeapTest):
def setUp(self):
- # mock key fetching and storing so Soledad doesn't fail when trying to
- # reach the server.
- Soledad._get_secrets_from_shared_db = Mock(return_value=None)
- Soledad._put_secrets_in_shared_db = Mock(return_value=None)
-
- class MockSharedDB(object):
-
- get_doc = Mock(return_value=None)
- put_doc = Mock()
- lock = Mock(return_value=('atoken', 300))
- unlock = Mock(return_value=True)
-
- def __call__(self):
- return self
-
- Soledad._shared_db = MockSharedDB()
+ self.setUpEnv()
self._soledad = Soledad(
u"leap@leap.se",
@@ -58,14 +46,32 @@ class KeyManagerWithSoledadTestCase(BaseLeapTest):
server_url='',
cert_file=None,
auth_token=None,
+ syncable=False
)
def tearDown(self):
km = self._key_manager()
- for key in km.get_all_keys():
- km._wrapper_map[key.__class__].delete_key(key)
- for key in km.get_all_keys(private=True):
- km._wrapper_map[key.__class__].delete_key(key)
+
+ def delete_keys(keys):
+ deferreds = []
+ for key in keys:
+ d = km._wrapper_map[key.__class__].delete_key(key)
+ 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)
+ deferreds.append(d)
+ return gatherResults(deferreds)
+
+ # wait for the indexes to be ready for the tear down
+ d = km._wrapper_map[OpenPGPKey].deferred_indexes
+ d.addCallback(get_and_delete_keys)
+ d.addCallback(lambda _: self.tearDownEnv())
+ return d
def _key_manager(self, user=ADDRESS, url='', token=None):
return KeyManager(user, url, self._soledad, token=token,
@@ -234,3 +240,62 @@ RZXoH+FTg9UAW87eqU610npOkT6cRaBxaMK/mDtGNdc=
=JTFu
-----END PGP PRIVATE KEY BLOCK-----
"""
+
+# key 7FEE575A: public key "anotheruser <anotheruser@leap.se>"
+PUBLIC_KEY_2 = """
+-----BEGIN PGP PUBLIC KEY BLOCK-----
+Version: GnuPG v1.4.10 (GNU/Linux)
+
+mI0EUYwJXgEEAMbTKHuPJ5/Gk34l9Z06f+0WCXTDXdte1UBoDtZ1erAbudgC4MOR
+gquKqoj3Hhw0/ILqJ88GcOJmKK/bEoIAuKaqlzDF7UAYpOsPZZYmtRfPC2pTCnXq
+Z1vdeqLwTbUspqXflkCkFtfhGKMq5rH8GV5a3tXZkRWZhdNwhVXZagC3ABEBAAG0
+IWFub3RoZXJ1c2VyIDxhbm90aGVydXNlckBsZWFwLnNlPoi4BBMBAgAiBQJRjAle
+AhsDBgsJCAcDAgYVCAIJCgsEFgIDAQIeAQIXgAAKCRB/nfpof+5XWotuA/4tLN4E
+gUr7IfLy2HkHAxzw7A4rqfMN92DIM9mZrDGaWRrOn3aVF7VU1UG7MDkHfPvp/cFw
+ezoCw4s4IoHVc/pVlOkcHSyt4/Rfh248tYEJmFCJXGHpkK83VIKYJAithNccJ6Q4
+JE/o06Mtf4uh/cA1HUL4a4ceqUhtpLJULLeKo7iNBFGMCV4BBADsyQI7GR0wSAxz
+VayLjuPzgT+bjbFeymIhjuxKIEwnIKwYkovztW+4bbOcQs785k3Lp6RzvigTpQQt
+Z/hwcLOqZbZw8t/24+D+Pq9mMP2uUvCFFqLlVvA6D3vKSQ/XNN+YB919WQ04jh63
+yuRe94WenT1RJd6xU1aaUff4rKizuQARAQABiJ8EGAECAAkFAlGMCV4CGwwACgkQ
+f536aH/uV1rPZQQAqCzRysOlu8ez7PuiBD4SebgRqWlxa1TF1ujzfLmuPivROZ2X
+Kw5aQstxgGSjoB7tac49s0huh4X8XK+BtJBfU84JS8Jc2satlfwoyZ35LH6sDZck
+I+RS/3we6zpMfHs3vvp9xgca6ZupQxivGtxlJs294TpJorx+mFFqbV17AzQ=
+=Thdu
+-----END PGP PUBLIC KEY BLOCK-----
+"""
+
+PRIVATE_KEY_2 = """
+-----BEGIN PGP PRIVATE KEY BLOCK-----
+Version: GnuPG v1.4.10 (GNU/Linux)
+
+lQHYBFGMCV4BBADG0yh7jyefxpN+JfWdOn/tFgl0w13bXtVAaA7WdXqwG7nYAuDD
+kYKriqqI9x4cNPyC6ifPBnDiZiiv2xKCALimqpcwxe1AGKTrD2WWJrUXzwtqUwp1
+6mdb3Xqi8E21LKal35ZApBbX4RijKuax/BleWt7V2ZEVmYXTcIVV2WoAtwARAQAB
+AAP7BLuSAx7tOohnimEs74ks8l/L6dOcsFQZj2bqs4AoY3jFe7bV0tHr4llypb/8
+H3/DYvpf6DWnCjyUS1tTnXSW8JXtx01BUKaAufSmMNg9blKV6GGHlT/Whe9uVyks
+7XHk/+9mebVMNJ/kNlqq2k+uWqJohzC8WWLRK+d1tBeqDsECANZmzltPaqUsGV5X
+C3zszE3tUBgptV/mKnBtopKi+VH+t7K6fudGcG+bAcZDUoH/QVde52mIIjjIdLje
+uajJuHUCAO1mqh+vPoGv4eBLV7iBo3XrunyGXiys4a39eomhxTy3YktQanjjx+ty
+GltAGCs5PbWGO6/IRjjvd46wh53kzvsCAO0J97gsWhzLuFnkxFAJSPk7RRlyl7lI
+1XS/x0Og6j9XHCyY1OYkfBm0to3UlCfkgirzCYlTYObCofzdKFIPDmSqHbQhYW5v
+dGhlcnVzZXIgPGFub3RoZXJ1c2VyQGxlYXAuc2U+iLgEEwECACIFAlGMCV4CGwMG
+CwkIBwMCBhUIAgkKCwQWAgMBAh4BAheAAAoJEH+d+mh/7ldai24D/i0s3gSBSvsh
+8vLYeQcDHPDsDiup8w33YMgz2ZmsMZpZGs6fdpUXtVTVQbswOQd8++n9wXB7OgLD
+izgigdVz+lWU6RwdLK3j9F+Hbjy1gQmYUIlcYemQrzdUgpgkCK2E1xwnpDgkT+jT
+oy1/i6H9wDUdQvhrhx6pSG2kslQst4qjnQHYBFGMCV4BBADsyQI7GR0wSAxzVayL
+juPzgT+bjbFeymIhjuxKIEwnIKwYkovztW+4bbOcQs785k3Lp6RzvigTpQQtZ/hw
+cLOqZbZw8t/24+D+Pq9mMP2uUvCFFqLlVvA6D3vKSQ/XNN+YB919WQ04jh63yuRe
+94WenT1RJd6xU1aaUff4rKizuQARAQABAAP9EyElqJ3dq3EErXwwT4mMnbd1SrVC
+rUJrNWQZL59mm5oigS00uIyR0SvusOr+UzTtd8ysRuwHy5d/LAZsbjQStaOMBILx
+77TJveOel0a1QK0YSMF2ywZMCKvquvjli4hAtWYz/EwfuzQN3t23jc5ny+GqmqD2
+3FUxLJosFUfLNmECAO9KhVmJi+L9dswIs+2Dkjd1eiRQzNOEVffvYkGYZyKxNiXF
+UA5kvyZcB4iAN9sWCybE4WHZ9jd4myGB0MPDGxkCAP1RsXJbbuD6zS7BXe5gwunO
+2q4q7ptdSl/sJYQuTe1KNP5d/uGsvlcFfsYjpsopasPjFBIncc/2QThMKlhoEaEB
+/0mVAxpT6SrEvUbJ18z7kna24SgMPr3OnPMxPGfvNLJY/Xv/A17YfoqjmByCvsKE
+JCDjopXtmbcrZyoEZbEht9mko4ifBBgBAgAJBQJRjAleAhsMAAoJEH+d+mh/7lda
+z2UEAKgs0crDpbvHs+z7ogQ+Enm4EalpcWtUxdbo83y5rj4r0TmdlysOWkLLcYBk
+o6Ae7WnOPbNIboeF/FyvgbSQX1POCUvCXNrGrZX8KMmd+Sx+rA2XJCPkUv98Hus6
+THx7N776fcYHGumbqUMYrxrcZSbNveE6SaK8fphRam1dewM0
+=a5gs
+-----END PGP PRIVATE KEY BLOCK-----
+"""
diff --git a/keymanager/src/leap/keymanager/tests/test_keymanager.py b/keymanager/src/leap/keymanager/tests/test_keymanager.py
index ee4462a1..b8ef88ae 100644
--- a/keymanager/src/leap/keymanager/tests/test_keymanager.py
+++ b/keymanager/src/leap/keymanager/tests/test_keymanager.py
@@ -23,12 +23,12 @@ Tests for the Key Manager.
from datetime import datetime
from mock import Mock
-from leap.common.testing.basetest import BaseLeapTest
+from twisted.internet.defer import inlineCallbacks
+from twisted.trial import unittest
+
from leap.keymanager import (
- openpgp,
KeyNotFound,
KeyAddressMismatch,
- errors,
)
from leap.keymanager.openpgp import OpenPGPKey
from leap.keymanager.keys import (
@@ -42,23 +42,16 @@ from leap.keymanager.validation import (
from leap.keymanager.tests import (
KeyManagerWithSoledadTestCase,
ADDRESS,
+ ADDRESS_2,
KEY_FINGERPRINT,
PUBLIC_KEY,
+ PUBLIC_KEY_2,
PRIVATE_KEY,
- GPG_BINARY_PATH
+ PRIVATE_KEY_2,
)
-ADDRESS_2 = 'anotheruser@leap.se'
-
-
-class KeyManagerUtilTestCase(BaseLeapTest):
-
- def setUp(self):
- pass
-
- def tearDown(self):
- pass
+class KeyManagerUtilTestCase(unittest.TestCase):
def test_is_address(self):
self.assertTrue(
@@ -128,227 +121,43 @@ class KeyManagerUtilTestCase(BaseLeapTest):
'Wrong data in key.')
-class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase):
-
- def _test_openpgp_gen_key(self):
- pgp = openpgp.OpenPGPScheme(self._soledad)
- self.assertRaises(KeyNotFound, pgp.get_key, 'user@leap.se')
- key = pgp.gen_key('user@leap.se')
- self.assertIsInstance(key, openpgp.OpenPGPKey)
- self.assertEqual(
- ['user@leap.se'], key.address, 'Wrong address bound to key.')
- self.assertEqual(
- 4096, key.length, 'Wrong key length.')
-
- def test_openpgp_put_delete_key(self):
- pgp = openpgp.OpenPGPScheme(
- self._soledad, gpgbinary=GPG_BINARY_PATH)
- self.assertRaises(KeyNotFound, pgp.get_key, ADDRESS)
- pgp.put_ascii_key(PUBLIC_KEY, ADDRESS)
- key = pgp.get_key(ADDRESS, private=False)
- pgp.delete_key(key)
- self.assertRaises(KeyNotFound, pgp.get_key, ADDRESS)
-
- def test_openpgp_put_ascii_key(self):
- pgp = openpgp.OpenPGPScheme(
- self._soledad, gpgbinary=GPG_BINARY_PATH)
- self.assertRaises(KeyNotFound, pgp.get_key, ADDRESS)
- pgp.put_ascii_key(PUBLIC_KEY, ADDRESS)
- key = pgp.get_key(ADDRESS, private=False)
- self.assertIsInstance(key, openpgp.OpenPGPKey)
- self.assertTrue(
- ADDRESS in key.address, 'Wrong address bound to key.')
- self.assertEqual(
- 4096, key.length, 'Wrong key length.')
- pgp.delete_key(key)
- self.assertRaises(KeyNotFound, pgp.get_key, ADDRESS)
-
- def test_get_public_key(self):
- pgp = openpgp.OpenPGPScheme(
- self._soledad, gpgbinary=GPG_BINARY_PATH)
- self.assertRaises(KeyNotFound, pgp.get_key, ADDRESS)
- pgp.put_ascii_key(PUBLIC_KEY, ADDRESS)
- self.assertRaises(
- KeyNotFound, pgp.get_key, ADDRESS, private=True)
- key = pgp.get_key(ADDRESS, private=False)
- self.assertTrue(ADDRESS in key.address)
- self.assertFalse(key.private)
- self.assertEqual(KEY_FINGERPRINT, key.fingerprint)
- pgp.delete_key(key)
- self.assertRaises(KeyNotFound, pgp.get_key, ADDRESS)
-
- def test_openpgp_encrypt_decrypt(self):
- # encrypt
- pgp = openpgp.OpenPGPScheme(
- self._soledad, gpgbinary=GPG_BINARY_PATH)
- pgp.put_ascii_key(PUBLIC_KEY, ADDRESS)
- pubkey = pgp.get_key(ADDRESS, private=False)
- data = 'data'
- cyphertext = pgp.encrypt(data, pubkey)
- # assert
- self.assertTrue(cyphertext is not None)
- self.assertTrue(cyphertext != '')
- self.assertTrue(cyphertext != data)
- self.assertTrue(pgp.is_encrypted(cyphertext))
- self.assertTrue(pgp.is_encrypted(cyphertext))
- # decrypt
- self.assertRaises(
- KeyNotFound, pgp.get_key, ADDRESS, private=True)
- pgp.put_ascii_key(PRIVATE_KEY, ADDRESS)
- privkey = pgp.get_key(ADDRESS, private=True)
- decrypted, _ = pgp.decrypt(cyphertext, privkey)
- self.assertEqual(decrypted, data)
- pgp.delete_key(pubkey)
- pgp.delete_key(privkey)
- self.assertRaises(
- KeyNotFound, pgp.get_key, ADDRESS, private=False)
- self.assertRaises(
- KeyNotFound, pgp.get_key, ADDRESS, private=True)
-
- def test_verify_with_private_raises(self):
- pgp = openpgp.OpenPGPScheme(
- self._soledad, gpgbinary=GPG_BINARY_PATH)
- pgp.put_ascii_key(PRIVATE_KEY, ADDRESS)
- data = 'data'
- privkey = pgp.get_key(ADDRESS, private=True)
- signed = pgp.sign(data, privkey)
- self.assertRaises(
- AssertionError,
- pgp.verify, signed, privkey)
-
- def test_sign_with_public_raises(self):
- pgp = openpgp.OpenPGPScheme(
- self._soledad, gpgbinary=GPG_BINARY_PATH)
- pgp.put_ascii_key(PUBLIC_KEY, ADDRESS)
- data = 'data'
- self.assertRaises(
- AssertionError,
- pgp.sign, data, ADDRESS, OpenPGPKey)
-
- def test_verify_with_wrong_key_raises(self):
- pgp = openpgp.OpenPGPScheme(
- self._soledad, gpgbinary=GPG_BINARY_PATH)
- pgp.put_ascii_key(PRIVATE_KEY, ADDRESS)
- data = 'data'
- privkey = pgp.get_key(ADDRESS, private=True)
- signed = pgp.sign(data, privkey)
- pgp.put_ascii_key(PUBLIC_KEY_2, ADDRESS_2)
- wrongkey = pgp.get_key(ADDRESS_2)
- self.assertFalse(pgp.verify(signed, wrongkey))
-
- def test_encrypt_sign_with_public_raises(self):
- pgp = openpgp.OpenPGPScheme(
- self._soledad, gpgbinary=GPG_BINARY_PATH)
- pgp.put_ascii_key(PRIVATE_KEY, ADDRESS)
- data = 'data'
- privkey = pgp.get_key(ADDRESS, private=True)
- pubkey = pgp.get_key(ADDRESS, private=False)
- self.assertRaises(
- AssertionError,
- pgp.encrypt, data, privkey, sign=pubkey)
-
- def test_decrypt_verify_with_private_raises(self):
- pgp = openpgp.OpenPGPScheme(
- self._soledad, gpgbinary=GPG_BINARY_PATH)
- pgp.put_ascii_key(PRIVATE_KEY, ADDRESS)
- data = 'data'
- privkey = pgp.get_key(ADDRESS, private=True)
- pubkey = pgp.get_key(ADDRESS, private=False)
- encrypted_and_signed = pgp.encrypt(
- data, pubkey, sign=privkey)
- self.assertRaises(
- AssertionError,
- pgp.decrypt,
- encrypted_and_signed, privkey, verify=privkey)
-
- def test_decrypt_verify_with_wrong_key(self):
- pgp = openpgp.OpenPGPScheme(
- self._soledad, gpgbinary=GPG_BINARY_PATH)
- pgp.put_ascii_key(PRIVATE_KEY, ADDRESS)
- data = 'data'
- privkey = pgp.get_key(ADDRESS, private=True)
- pubkey = pgp.get_key(ADDRESS, private=False)
- encrypted_and_signed = pgp.encrypt(data, pubkey, sign=privkey)
- pgp.put_ascii_key(PUBLIC_KEY_2, ADDRESS_2)
- wrongkey = pgp.get_key(ADDRESS_2)
- decrypted, validsign = pgp.decrypt(encrypted_and_signed, privkey,
- verify=wrongkey)
- self.assertEqual(decrypted, data)
- self.assertFalse(validsign)
-
- def test_sign_verify(self):
- pgp = openpgp.OpenPGPScheme(
- self._soledad, gpgbinary=GPG_BINARY_PATH)
- pgp.put_ascii_key(PRIVATE_KEY, ADDRESS)
- data = 'data'
- privkey = pgp.get_key(ADDRESS, private=True)
- signed = pgp.sign(data, privkey, detach=False)
- pubkey = pgp.get_key(ADDRESS, private=False)
- self.assertTrue(pgp.verify(signed, pubkey))
-
- def test_encrypt_sign_decrypt_verify(self):
- pgp = openpgp.OpenPGPScheme(
- self._soledad, gpgbinary=GPG_BINARY_PATH)
- pgp.put_ascii_key(PRIVATE_KEY, ADDRESS)
- pubkey = pgp.get_key(ADDRESS, private=False)
- privkey = pgp.get_key(ADDRESS, private=True)
- pgp.put_ascii_key(PRIVATE_KEY_2, ADDRESS_2)
- pubkey2 = pgp.get_key(ADDRESS_2, private=False)
- privkey2 = pgp.get_key(ADDRESS_2, private=True)
- data = 'data'
- encrypted_and_signed = pgp.encrypt(
- data, pubkey2, sign=privkey)
- res, validsign = pgp.decrypt(
- encrypted_and_signed, privkey2, verify=pubkey)
- self.assertEqual(data, res)
- self.assertTrue(validsign)
-
- def test_sign_verify_detached_sig(self):
- pgp = openpgp.OpenPGPScheme(
- self._soledad, gpgbinary=GPG_BINARY_PATH)
- pgp.put_ascii_key(PRIVATE_KEY, ADDRESS)
- data = 'data'
- privkey = pgp.get_key(ADDRESS, private=True)
- signature = pgp.sign(data, privkey, detach=True)
- pubkey = pgp.get_key(ADDRESS, private=False)
- validsign = pgp.verify(data, pubkey, detached_sig=signature)
- self.assertTrue(validsign)
-
-
class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase):
+ @inlineCallbacks
def test_get_all_keys_in_db(self):
km = self._key_manager()
- km._wrapper_map[OpenPGPKey].put_ascii_key(PRIVATE_KEY, ADDRESS)
+ yield km._wrapper_map[OpenPGPKey].put_ascii_key(PRIVATE_KEY, ADDRESS)
# get public keys
- keys = km.get_all_keys(False)
+ keys = yield km.get_all_keys(False)
self.assertEqual(len(keys), 1, 'Wrong number of keys')
self.assertTrue(ADDRESS in keys[0].address)
self.assertFalse(keys[0].private)
# get private keys
- keys = km.get_all_keys(True)
+ 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(keys[0].private)
+ @inlineCallbacks
def test_get_public_key(self):
km = self._key_manager()
- km._wrapper_map[OpenPGPKey].put_ascii_key(PRIVATE_KEY, ADDRESS)
+ yield km._wrapper_map[OpenPGPKey].put_ascii_key(PRIVATE_KEY, ADDRESS)
# get the key
- key = km.get_key(ADDRESS, OpenPGPKey, private=False,
- fetch_remote=False)
+ 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.assertEqual(
key.fingerprint.lower(), KEY_FINGERPRINT.lower())
self.assertFalse(key.private)
+ @inlineCallbacks
def test_get_private_key(self):
km = self._key_manager()
- km._wrapper_map[OpenPGPKey].put_ascii_key(PRIVATE_KEY, ADDRESS)
+ yield km._wrapper_map[OpenPGPKey].put_ascii_key(PRIVATE_KEY, ADDRESS)
# get the key
- key = km.get_key(ADDRESS, OpenPGPKey, private=True,
- fetch_remote=False)
+ 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.assertEqual(
@@ -357,17 +166,17 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase):
def test_send_key_raises_key_not_found(self):
km = self._key_manager()
- self.assertRaises(
- KeyNotFound,
- km.send_key, OpenPGPKey)
+ d = km.send_key(OpenPGPKey)
+ return self.assertFailure(d, KeyNotFound)
+ @inlineCallbacks
def test_send_key(self):
"""
Test that request is well formed when sending keys to server.
"""
token = "mytoken"
km = self._key_manager(token=token)
- km._wrapper_map[OpenPGPKey].put_ascii_key(PUBLIC_KEY, ADDRESS)
+ yield km._wrapper_map[OpenPGPKey].put_ascii_key(PUBLIC_KEY, ADDRESS)
km._fetcher.put = Mock()
# the following data will be used on the send
km.ca_cert_path = 'capath'
@@ -375,10 +184,11 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase):
km.uid = 'myuid'
km.api_uri = 'apiuri'
km.api_version = 'apiver'
- km.send_key(OpenPGPKey)
+ yield km.send_key(OpenPGPKey)
# setup expected args
+ pubkey = yield km.get_key(km._address, OpenPGPKey)
data = {
- km.PUBKEY_KEY: km.get_key(km._address, OpenPGPKey).key_data,
+ km.PUBKEY_KEY: pubkey.key_data,
}
url = '%s/%s/users/%s.json' % ('apiuri', 'apiver', 'myuid')
km._fetcher.put.assert_called_once_with(
@@ -386,7 +196,7 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase):
headers={'Authorization': 'Token token=%s' % token},
)
- def test__fetch_keys_from_server(self):
+ def test_fetch_keys_from_server(self):
"""
Test that the request is well formed when fetching keys from server.
"""
@@ -406,15 +216,19 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase):
km._fetcher.get = Mock(
return_value=Response())
km.ca_cert_path = 'cacertpath'
- # do the fetch
- km._fetch_keys_from_server(ADDRESS_2)
- # and verify the call
- km._fetcher.get.assert_called_once_with(
- 'http://nickserver.domain',
- data={'address': ADDRESS_2},
- verify='cacertpath',
- )
+ def verify_the_call(_):
+ km._fetcher.get.assert_called_once_with(
+ 'http://nickserver.domain',
+ data={'address': ADDRESS_2},
+ verify='cacertpath',
+ )
+
+ d = km._fetch_keys_from_server(ADDRESS_2)
+ d.addCallback(verify_the_call)
+ return d
+
+ @inlineCallbacks
def test_get_key_fetches_from_server(self):
"""
Test that getting a key successfuly fetches from server.
@@ -435,26 +249,26 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase):
km._fetcher.get = Mock(return_value=Response())
km.ca_cert_path = 'cacertpath'
# try to key get without fetching from server
- self.assertRaises(
- KeyNotFound, km.get_key, ADDRESS, OpenPGPKey,
- fetch_remote=False
- )
+ d = km.get_key(ADDRESS, OpenPGPKey, fetch_remote=False)
+ yield self.assertFailure(d, KeyNotFound)
# try to get key fetching from server.
- key = km.get_key(ADDRESS, OpenPGPKey)
+ key = yield km.get_key(ADDRESS, OpenPGPKey)
self.assertIsInstance(key, OpenPGPKey)
self.assertTrue(ADDRESS in key.address)
+ @inlineCallbacks
def test_put_key_ascii(self):
"""
Test that putting ascii key works
"""
km = self._key_manager(url='http://nickserver.domain')
- km.put_raw_key(PUBLIC_KEY, OpenPGPKey, ADDRESS)
- key = km.get_key(ADDRESS, OpenPGPKey)
+ 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)
+ @inlineCallbacks
def test_fetch_uri_ascii_key(self):
"""
Test that fetch key downloads the ascii key and gets included in
@@ -469,8 +283,8 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase):
km._fetcher.get = Mock(return_value=Response())
km.ca_cert_path = 'cacertpath'
- km.fetch_key(ADDRESS, "http://site.domain/key", OpenPGPKey)
- key = km.get_key(ADDRESS, OpenPGPKey)
+ yield km.fetch_key(ADDRESS, "http://site.domain/key", OpenPGPKey)
+ key = yield km.get_key(ADDRESS, OpenPGPKey)
self.assertEqual(KEY_FINGERPRINT, key.fingerprint)
def test_fetch_uri_empty_key(self):
@@ -485,8 +299,8 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase):
km._fetcher.get = Mock(return_value=Response())
km.ca_cert_path = 'cacertpath'
- self.assertRaises(KeyNotFound, km.fetch_key,
- ADDRESS, "http://site.domain/key", OpenPGPKey)
+ d = km.fetch_key(ADDRESS, "http://site.domain/key", OpenPGPKey)
+ return self.assertFailure(d, KeyNotFound)
def test_fetch_uri_address_differ(self):
"""
@@ -501,120 +315,66 @@ class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase):
km._fetcher.get = Mock(return_value=Response())
km.ca_cert_path = 'cacertpath'
- self.assertRaises(KeyAddressMismatch, km.fetch_key,
- ADDRESS_2, "http://site.domain/key", OpenPGPKey)
+ d = km.fetch_key(ADDRESS_2, "http://site.domain/key", OpenPGPKey)
+ return self.assertFailure(d, KeyAddressMismatch)
class KeyManagerCryptoTestCase(KeyManagerWithSoledadTestCase):
RAW_DATA = 'data'
+ @inlineCallbacks
def test_keymanager_openpgp_encrypt_decrypt(self):
km = self._key_manager()
# put raw private key
- km._wrapper_map[OpenPGPKey].put_ascii_key(PRIVATE_KEY, ADDRESS)
- km._wrapper_map[OpenPGPKey].put_ascii_key(PRIVATE_KEY_2, ADDRESS_2)
+ yield km._wrapper_map[OpenPGPKey].put_ascii_key(PRIVATE_KEY, ADDRESS)
+ yield km._wrapper_map[OpenPGPKey].put_ascii_key(
+ PRIVATE_KEY_2, ADDRESS_2)
# encrypt
- encdata = km.encrypt(self.RAW_DATA, ADDRESS, OpenPGPKey,
- sign=ADDRESS_2, fetch_remote=False)
+ encdata = yield km.encrypt(self.RAW_DATA, ADDRESS, OpenPGPKey,
+ sign=ADDRESS_2, fetch_remote=False)
self.assertNotEqual(self.RAW_DATA, encdata)
# decrypt
- rawdata, signingkey = km.decrypt(encdata, ADDRESS, OpenPGPKey,
- verify=ADDRESS_2, fetch_remote=False)
+ rawdata, signingkey = yield km.decrypt(
+ encdata, ADDRESS, OpenPGPKey, verify=ADDRESS_2, fetch_remote=False)
self.assertEqual(self.RAW_DATA, rawdata)
- key = km.get_key(ADDRESS_2, OpenPGPKey, private=False,
- fetch_remote=False)
+ key = yield km.get_key(ADDRESS_2, OpenPGPKey, private=False,
+ fetch_remote=False)
self.assertEqual(signingkey.fingerprint, key.fingerprint)
+ @inlineCallbacks
def test_keymanager_openpgp_encrypt_decrypt_wrong_sign(self):
km = self._key_manager()
# put raw keys
- km._wrapper_map[OpenPGPKey].put_ascii_key(PRIVATE_KEY, ADDRESS)
- km._wrapper_map[OpenPGPKey].put_ascii_key(PRIVATE_KEY_2, ADDRESS_2)
+ yield km._wrapper_map[OpenPGPKey].put_ascii_key(PRIVATE_KEY, ADDRESS)
+ yield km._wrapper_map[OpenPGPKey].put_ascii_key(
+ PRIVATE_KEY_2, ADDRESS_2)
# encrypt
- encdata = km.encrypt(self.RAW_DATA, ADDRESS, OpenPGPKey,
- sign=ADDRESS_2, fetch_remote=False)
+ encdata = yield km.encrypt(self.RAW_DATA, ADDRESS, OpenPGPKey,
+ sign=ADDRESS_2, fetch_remote=False)
self.assertNotEqual(self.RAW_DATA, encdata)
# verify
- rawdata, signingkey = km.decrypt(encdata, ADDRESS, OpenPGPKey,
- verify=ADDRESS, fetch_remote=False)
+ rawdata, signingkey = yield km.decrypt(
+ encdata, ADDRESS, OpenPGPKey, verify=ADDRESS, fetch_remote=False)
self.assertEqual(self.RAW_DATA, rawdata)
self.assertTrue(signingkey is None)
+ @inlineCallbacks
def test_keymanager_openpgp_sign_verify(self):
km = self._key_manager()
# put raw private keys
- km._wrapper_map[OpenPGPKey].put_ascii_key(PRIVATE_KEY, ADDRESS)
- signdata = km.sign(self.RAW_DATA, ADDRESS, OpenPGPKey, detach=False)
+ yield km._wrapper_map[OpenPGPKey].put_ascii_key(PRIVATE_KEY, ADDRESS)
+ signdata = yield km.sign(self.RAW_DATA, ADDRESS, OpenPGPKey,
+ detach=False)
self.assertNotEqual(self.RAW_DATA, signdata)
# verify
- signingkey = km.verify(signdata, ADDRESS, OpenPGPKey,
+ signingkey = yield km.verify(signdata, ADDRESS, OpenPGPKey,
+ fetch_remote=False)
+ key = yield km.get_key(ADDRESS, OpenPGPKey, private=False,
fetch_remote=False)
- key = km.get_key(ADDRESS, OpenPGPKey, private=False,
- fetch_remote=False)
self.assertEqual(signingkey.fingerprint, key.fingerprint)
-# Key material for testing
-
-# key 7FEE575A: public key "anotheruser <anotheruser@leap.se>"
-PUBLIC_KEY_2 = """
------BEGIN PGP PUBLIC KEY BLOCK-----
-Version: GnuPG v1.4.10 (GNU/Linux)
-
-mI0EUYwJXgEEAMbTKHuPJ5/Gk34l9Z06f+0WCXTDXdte1UBoDtZ1erAbudgC4MOR
-gquKqoj3Hhw0/ILqJ88GcOJmKK/bEoIAuKaqlzDF7UAYpOsPZZYmtRfPC2pTCnXq
-Z1vdeqLwTbUspqXflkCkFtfhGKMq5rH8GV5a3tXZkRWZhdNwhVXZagC3ABEBAAG0
-IWFub3RoZXJ1c2VyIDxhbm90aGVydXNlckBsZWFwLnNlPoi4BBMBAgAiBQJRjAle
-AhsDBgsJCAcDAgYVCAIJCgsEFgIDAQIeAQIXgAAKCRB/nfpof+5XWotuA/4tLN4E
-gUr7IfLy2HkHAxzw7A4rqfMN92DIM9mZrDGaWRrOn3aVF7VU1UG7MDkHfPvp/cFw
-ezoCw4s4IoHVc/pVlOkcHSyt4/Rfh248tYEJmFCJXGHpkK83VIKYJAithNccJ6Q4
-JE/o06Mtf4uh/cA1HUL4a4ceqUhtpLJULLeKo7iNBFGMCV4BBADsyQI7GR0wSAxz
-VayLjuPzgT+bjbFeymIhjuxKIEwnIKwYkovztW+4bbOcQs785k3Lp6RzvigTpQQt
-Z/hwcLOqZbZw8t/24+D+Pq9mMP2uUvCFFqLlVvA6D3vKSQ/XNN+YB919WQ04jh63
-yuRe94WenT1RJd6xU1aaUff4rKizuQARAQABiJ8EGAECAAkFAlGMCV4CGwwACgkQ
-f536aH/uV1rPZQQAqCzRysOlu8ez7PuiBD4SebgRqWlxa1TF1ujzfLmuPivROZ2X
-Kw5aQstxgGSjoB7tac49s0huh4X8XK+BtJBfU84JS8Jc2satlfwoyZ35LH6sDZck
-I+RS/3we6zpMfHs3vvp9xgca6ZupQxivGtxlJs294TpJorx+mFFqbV17AzQ=
-=Thdu
------END PGP PUBLIC KEY BLOCK-----
-"""
-
-PRIVATE_KEY_2 = """
------BEGIN PGP PRIVATE KEY BLOCK-----
-Version: GnuPG v1.4.10 (GNU/Linux)
-
-lQHYBFGMCV4BBADG0yh7jyefxpN+JfWdOn/tFgl0w13bXtVAaA7WdXqwG7nYAuDD
-kYKriqqI9x4cNPyC6ifPBnDiZiiv2xKCALimqpcwxe1AGKTrD2WWJrUXzwtqUwp1
-6mdb3Xqi8E21LKal35ZApBbX4RijKuax/BleWt7V2ZEVmYXTcIVV2WoAtwARAQAB
-AAP7BLuSAx7tOohnimEs74ks8l/L6dOcsFQZj2bqs4AoY3jFe7bV0tHr4llypb/8
-H3/DYvpf6DWnCjyUS1tTnXSW8JXtx01BUKaAufSmMNg9blKV6GGHlT/Whe9uVyks
-7XHk/+9mebVMNJ/kNlqq2k+uWqJohzC8WWLRK+d1tBeqDsECANZmzltPaqUsGV5X
-C3zszE3tUBgptV/mKnBtopKi+VH+t7K6fudGcG+bAcZDUoH/QVde52mIIjjIdLje
-uajJuHUCAO1mqh+vPoGv4eBLV7iBo3XrunyGXiys4a39eomhxTy3YktQanjjx+ty
-GltAGCs5PbWGO6/IRjjvd46wh53kzvsCAO0J97gsWhzLuFnkxFAJSPk7RRlyl7lI
-1XS/x0Og6j9XHCyY1OYkfBm0to3UlCfkgirzCYlTYObCofzdKFIPDmSqHbQhYW5v
-dGhlcnVzZXIgPGFub3RoZXJ1c2VyQGxlYXAuc2U+iLgEEwECACIFAlGMCV4CGwMG
-CwkIBwMCBhUIAgkKCwQWAgMBAh4BAheAAAoJEH+d+mh/7ldai24D/i0s3gSBSvsh
-8vLYeQcDHPDsDiup8w33YMgz2ZmsMZpZGs6fdpUXtVTVQbswOQd8++n9wXB7OgLD
-izgigdVz+lWU6RwdLK3j9F+Hbjy1gQmYUIlcYemQrzdUgpgkCK2E1xwnpDgkT+jT
-oy1/i6H9wDUdQvhrhx6pSG2kslQst4qjnQHYBFGMCV4BBADsyQI7GR0wSAxzVayL
-juPzgT+bjbFeymIhjuxKIEwnIKwYkovztW+4bbOcQs785k3Lp6RzvigTpQQtZ/hw
-cLOqZbZw8t/24+D+Pq9mMP2uUvCFFqLlVvA6D3vKSQ/XNN+YB919WQ04jh63yuRe
-94WenT1RJd6xU1aaUff4rKizuQARAQABAAP9EyElqJ3dq3EErXwwT4mMnbd1SrVC
-rUJrNWQZL59mm5oigS00uIyR0SvusOr+UzTtd8ysRuwHy5d/LAZsbjQStaOMBILx
-77TJveOel0a1QK0YSMF2ywZMCKvquvjli4hAtWYz/EwfuzQN3t23jc5ny+GqmqD2
-3FUxLJosFUfLNmECAO9KhVmJi+L9dswIs+2Dkjd1eiRQzNOEVffvYkGYZyKxNiXF
-UA5kvyZcB4iAN9sWCybE4WHZ9jd4myGB0MPDGxkCAP1RsXJbbuD6zS7BXe5gwunO
-2q4q7ptdSl/sJYQuTe1KNP5d/uGsvlcFfsYjpsopasPjFBIncc/2QThMKlhoEaEB
-/0mVAxpT6SrEvUbJ18z7kna24SgMPr3OnPMxPGfvNLJY/Xv/A17YfoqjmByCvsKE
-JCDjopXtmbcrZyoEZbEht9mko4ifBBgBAgAJBQJRjAleAhsMAAoJEH+d+mh/7lda
-z2UEAKgs0crDpbvHs+z7ogQ+Enm4EalpcWtUxdbo83y5rj4r0TmdlysOWkLLcYBk
-o6Ae7WnOPbNIboeF/FyvgbSQX1POCUvCXNrGrZX8KMmd+Sx+rA2XJCPkUv98Hus6
-THx7N776fcYHGumbqUMYrxrcZSbNveE6SaK8fphRam1dewM0
-=a5gs
------END PGP PRIVATE KEY BLOCK-----
-"""
import unittest
if __name__ == "__main__":
unittest.main()
diff --git a/keymanager/src/leap/keymanager/tests/test_openpgp.py b/keymanager/src/leap/keymanager/tests/test_openpgp.py
new file mode 100644
index 00000000..01cf3417
--- /dev/null
+++ b/keymanager/src/leap/keymanager/tests/test_openpgp.py
@@ -0,0 +1,250 @@
+# -*- coding: utf-8 -*-
+# test_keymanager.py
+# Copyright (C) 2014 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 the OpenPGP support on Key Manager.
+"""
+
+
+from twisted.internet.defer import inlineCallbacks
+
+from leap.keymanager import (
+ KeyNotFound,
+ openpgp,
+)
+from leap.keymanager.openpgp import OpenPGPKey
+from leap.keymanager.tests import (
+ KeyManagerWithSoledadTestCase,
+ ADDRESS,
+ ADDRESS_2,
+ KEY_FINGERPRINT,
+ PUBLIC_KEY,
+ PUBLIC_KEY_2,
+ PRIVATE_KEY,
+ PRIVATE_KEY_2,
+ GPG_BINARY_PATH
+)
+
+
+class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase):
+
+ @inlineCallbacks
+ def _test_openpgp_gen_key(self):
+ pgp = openpgp.OpenPGPScheme(
+ self._soledad, gpgbinary=GPG_BINARY_PATH)
+ yield self._assert_key_not_found(pgp, 'user@leap.se')
+ key = yield pgp.gen_key('user@leap.se')
+ self.assertIsInstance(key, openpgp.OpenPGPKey)
+ self.assertEqual(
+ ['user@leap.se'], key.address, 'Wrong address bound to key.')
+ self.assertEqual(
+ 4096, key.length, 'Wrong key length.')
+
+ @inlineCallbacks
+ def test_openpgp_put_delete_key(self):
+ pgp = openpgp.OpenPGPScheme(
+ self._soledad, gpgbinary=GPG_BINARY_PATH)
+ yield self._assert_key_not_found(pgp, ADDRESS)
+ yield pgp.put_ascii_key(PUBLIC_KEY, ADDRESS)
+ key = yield pgp.get_key(ADDRESS, private=False)
+ yield pgp.delete_key(key)
+ yield self._assert_key_not_found(pgp, ADDRESS)
+
+ @inlineCallbacks
+ def test_openpgp_put_ascii_key(self):
+ pgp = openpgp.OpenPGPScheme(
+ self._soledad, gpgbinary=GPG_BINARY_PATH)
+ yield self._assert_key_not_found(pgp, ADDRESS)
+ yield pgp.put_ascii_key(PUBLIC_KEY, ADDRESS)
+ key = yield pgp.get_key(ADDRESS, private=False)
+ self.assertIsInstance(key, openpgp.OpenPGPKey)
+ self.assertTrue(
+ ADDRESS in key.address, 'Wrong address bound to key.')
+ self.assertEqual(
+ 4096, key.length, 'Wrong key length.')
+ yield pgp.delete_key(key)
+ yield self._assert_key_not_found(pgp, ADDRESS)
+
+ @inlineCallbacks
+ def test_get_public_key(self):
+ pgp = openpgp.OpenPGPScheme(
+ self._soledad, gpgbinary=GPG_BINARY_PATH)
+ yield self._assert_key_not_found(pgp, ADDRESS)
+ yield pgp.put_ascii_key(PUBLIC_KEY, ADDRESS)
+ yield self._assert_key_not_found(pgp, ADDRESS, private=True)
+ key = yield pgp.get_key(ADDRESS, private=False)
+ self.assertTrue(ADDRESS in key.address)
+ self.assertFalse(key.private)
+ self.assertEqual(KEY_FINGERPRINT, key.fingerprint)
+ yield pgp.delete_key(key)
+ yield self._assert_key_not_found(pgp, ADDRESS)
+
+ @inlineCallbacks
+ def test_openpgp_encrypt_decrypt(self):
+ data = 'data'
+ pgp = openpgp.OpenPGPScheme(
+ self._soledad, gpgbinary=GPG_BINARY_PATH)
+
+ # encrypt
+ yield pgp.put_ascii_key(PUBLIC_KEY, ADDRESS)
+ pubkey = yield pgp.get_key(ADDRESS, private=False)
+ cyphertext = pgp.encrypt(data, pubkey)
+
+ self.assertTrue(cyphertext is not None)
+ self.assertTrue(cyphertext != '')
+ self.assertTrue(cyphertext != data)
+ self.assertTrue(pgp.is_encrypted(cyphertext))
+ self.assertTrue(pgp.is_encrypted(cyphertext))
+
+ # decrypt
+ 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)
+ self.assertEqual(decrypted, data)
+
+ yield pgp.delete_key(pubkey)
+ yield pgp.delete_key(privkey)
+ yield self._assert_key_not_found(pgp, ADDRESS, private=False)
+ yield self._assert_key_not_found(pgp, ADDRESS, private=True)
+
+ @inlineCallbacks
+ def test_verify_with_private_raises(self):
+ data = 'data'
+ pgp = openpgp.OpenPGPScheme(
+ self._soledad, gpgbinary=GPG_BINARY_PATH)
+ yield pgp.put_ascii_key(PRIVATE_KEY, ADDRESS)
+ privkey = yield pgp.get_key(ADDRESS, private=True)
+ signed = pgp.sign(data, privkey)
+ self.assertRaises(
+ AssertionError,
+ pgp.verify, signed, privkey)
+
+ @inlineCallbacks
+ def test_sign_with_public_raises(self):
+ data = 'data'
+ pgp = openpgp.OpenPGPScheme(
+ self._soledad, gpgbinary=GPG_BINARY_PATH)
+ yield pgp.put_ascii_key(PUBLIC_KEY, ADDRESS)
+ self.assertRaises(
+ AssertionError,
+ pgp.sign, data, ADDRESS, OpenPGPKey)
+
+ @inlineCallbacks
+ def test_verify_with_wrong_key_raises(self):
+ data = 'data'
+ pgp = openpgp.OpenPGPScheme(
+ self._soledad, gpgbinary=GPG_BINARY_PATH)
+ yield pgp.put_ascii_key(PRIVATE_KEY, ADDRESS)
+ privkey = yield pgp.get_key(ADDRESS, private=True)
+ signed = pgp.sign(data, privkey)
+ yield pgp.put_ascii_key(PUBLIC_KEY_2, ADDRESS_2)
+ wrongkey = yield pgp.get_key(ADDRESS_2)
+ self.assertFalse(pgp.verify(signed, wrongkey))
+
+ @inlineCallbacks
+ def test_encrypt_sign_with_public_raises(self):
+ data = 'data'
+ pgp = openpgp.OpenPGPScheme(
+ self._soledad, gpgbinary=GPG_BINARY_PATH)
+ 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)
+
+ @inlineCallbacks
+ def test_decrypt_verify_with_private_raises(self):
+ data = 'data'
+ pgp = openpgp.OpenPGPScheme(
+ self._soledad, gpgbinary=GPG_BINARY_PATH)
+ 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)
+ self.assertRaises(
+ AssertionError,
+ pgp.decrypt,
+ encrypted_and_signed, privkey, verify=privkey)
+
+ @inlineCallbacks
+ def test_decrypt_verify_with_wrong_key(self):
+ data = 'data'
+ pgp = openpgp.OpenPGPScheme(
+ self._soledad, gpgbinary=GPG_BINARY_PATH)
+ 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)
+ 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)
+ self.assertEqual(decrypted, data)
+ self.assertFalse(validsign)
+
+ @inlineCallbacks
+ def test_sign_verify(self):
+ data = 'data'
+ pgp = openpgp.OpenPGPScheme(
+ self._soledad, gpgbinary=GPG_BINARY_PATH)
+ yield pgp.put_ascii_key(PRIVATE_KEY, ADDRESS)
+ privkey = yield pgp.get_key(ADDRESS, private=True)
+ signed = pgp.sign(data, privkey, detach=False)
+ pubkey = yield pgp.get_key(ADDRESS, private=False)
+ validsign = pgp.verify(signed, pubkey)
+ self.assertTrue(validsign)
+
+ @inlineCallbacks
+ def test_encrypt_sign_decrypt_verify(self):
+ pgp = openpgp.OpenPGPScheme(
+ self._soledad, gpgbinary=GPG_BINARY_PATH)
+
+ yield pgp.put_ascii_key(PRIVATE_KEY, ADDRESS)
+ pubkey = yield pgp.get_key(ADDRESS, private=False)
+ privkey = yield pgp.get_key(ADDRESS, private=True)
+
+ yield pgp.put_ascii_key(PRIVATE_KEY_2, ADDRESS_2)
+ pubkey2 = yield pgp.get_key(ADDRESS_2, private=False)
+ privkey2 = yield pgp.get_key(ADDRESS_2, private=True)
+
+ data = 'data'
+ encrypted_and_signed = pgp.encrypt(
+ data, pubkey2, sign=privkey)
+ res, validsign = pgp.decrypt(
+ encrypted_and_signed, privkey2, verify=pubkey)
+ self.assertEqual(data, res)
+ self.assertTrue(validsign)
+
+ @inlineCallbacks
+ def test_sign_verify_detached_sig(self):
+ data = 'data'
+ pgp = openpgp.OpenPGPScheme(
+ self._soledad, gpgbinary=GPG_BINARY_PATH)
+ yield pgp.put_ascii_key(PRIVATE_KEY, ADDRESS)
+ privkey = yield pgp.get_key(ADDRESS, private=True)
+ signature = yield pgp.sign(data, privkey, detach=True)
+ pubkey = yield pgp.get_key(ADDRESS, private=False)
+ validsign = pgp.verify(data, pubkey, detached_sig=signature)
+ self.assertTrue(validsign)
+
+ 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/keymanager/src/leap/keymanager/tests/test_validation.py b/keymanager/src/leap/keymanager/tests/test_validation.py
index 400d36e8..83a02e00 100644
--- a/keymanager/src/leap/keymanager/tests/test_validation.py
+++ b/keymanager/src/leap/keymanager/tests/test_validation.py
@@ -19,6 +19,7 @@ Tests for the Validation Levels
"""
from datetime import datetime
+from twisted.internet.defer import inlineCallbacks
from leap.keymanager.openpgp import OpenPGPKey
from leap.keymanager.errors import (
@@ -35,51 +36,57 @@ from leap.keymanager.validation import ValidationLevel
class ValidationLevelTestCase(KeyManagerWithSoledadTestCase):
+ @inlineCallbacks
def test_none_old_key(self):
km = self._key_manager()
- km.put_raw_key(PUBLIC_KEY, OpenPGPKey, ADDRESS)
- key = km.get_key(ADDRESS, OpenPGPKey, fetch_remote=False)
+ yield km.put_raw_key(PUBLIC_KEY, OpenPGPKey, ADDRESS)
+ key = yield km.get_key(ADDRESS, OpenPGPKey, fetch_remote=False)
self.assertEqual(key.fingerprint, KEY_FINGERPRINT)
+ @inlineCallbacks
def test_cant_upgrade(self):
km = self._key_manager()
- km.put_raw_key(PUBLIC_KEY, OpenPGPKey, ADDRESS,
- validation=ValidationLevel.Provider_Trust)
- self.assertRaises(KeyNotValidUpgrade, km.put_raw_key, UNRELATED_KEY,
- OpenPGPKey, ADDRESS)
+ yield km.put_raw_key(PUBLIC_KEY, OpenPGPKey, ADDRESS,
+ validation=ValidationLevel.Provider_Trust)
+ d = km.put_raw_key(UNRELATED_KEY, OpenPGPKey, ADDRESS)
+ yield self.assertFailure(d, KeyNotValidUpgrade)
+ @inlineCallbacks
def test_fingerprint_level(self):
km = self._key_manager()
- km.put_raw_key(PUBLIC_KEY, OpenPGPKey, ADDRESS)
- km.put_raw_key(UNRELATED_KEY, OpenPGPKey, ADDRESS,
- validation=ValidationLevel.Fingerprint)
- key = km.get_key(ADDRESS, OpenPGPKey, fetch_remote=False)
+ yield km.put_raw_key(PUBLIC_KEY, OpenPGPKey, ADDRESS)
+ yield km.put_raw_key(UNRELATED_KEY, OpenPGPKey, ADDRESS,
+ validation=ValidationLevel.Fingerprint)
+ key = yield km.get_key(ADDRESS, OpenPGPKey, fetch_remote=False)
self.assertEqual(key.fingerprint, UNRELATED_FINGERPRINT)
+ @inlineCallbacks
def test_expired_key(self):
km = self._key_manager()
- km.put_raw_key(EXPIRED_KEY, OpenPGPKey, ADDRESS)
- km.put_raw_key(UNRELATED_KEY, OpenPGPKey, ADDRESS)
- key = km.get_key(ADDRESS, OpenPGPKey, fetch_remote=False)
+ yield km.put_raw_key(EXPIRED_KEY, OpenPGPKey, ADDRESS)
+ yield km.put_raw_key(UNRELATED_KEY, OpenPGPKey, ADDRESS)
+ key = yield km.get_key(ADDRESS, OpenPGPKey, fetch_remote=False)
self.assertEqual(key.fingerprint, UNRELATED_FINGERPRINT)
+ @inlineCallbacks
def test_expired_fail_lower_level(self):
km = self._key_manager()
- km.put_raw_key(EXPIRED_KEY, OpenPGPKey, ADDRESS,
- validation=ValidationLevel.Third_Party_Endorsement)
- self.assertRaises(
- KeyNotValidUpgrade,
- km.put_raw_key,
+ yield km.put_raw_key(
+ EXPIRED_KEY, OpenPGPKey, ADDRESS,
+ validation=ValidationLevel.Third_Party_Endorsement)
+ d = km.put_raw_key(
UNRELATED_KEY,
OpenPGPKey,
ADDRESS,
validation=ValidationLevel.Provider_Trust)
+ yield self.assertFailure(d, KeyNotValidUpgrade)
+ @inlineCallbacks
def test_roll_back(self):
km = self._key_manager()
- km.put_raw_key(EXPIRED_KEY_UPDATED, OpenPGPKey, ADDRESS)
- km.put_raw_key(EXPIRED_KEY, OpenPGPKey, ADDRESS)
- key = km.get_key(ADDRESS, OpenPGPKey, fetch_remote=False)
+ yield km.put_raw_key(EXPIRED_KEY_UPDATED, OpenPGPKey, ADDRESS)
+ yield km.put_raw_key(EXPIRED_KEY, OpenPGPKey, ADDRESS)
+ key = yield km.get_key(ADDRESS, OpenPGPKey, fetch_remote=False)
self.assertEqual(key.expiry_date, EXPIRED_KEY_NEW_EXPIRY_DATE)
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-09 01:13:32 +0000