From 1858a50809dc48bfcc4ad2d96dd5641f1de6b9eb Mon Sep 17 00:00:00 2001 From: drebs Date: Sun, 14 Apr 2013 14:15:10 -0300 Subject: Add key manager basic API docstrings. --- src/leap/common/keymanager/__init__.py | 232 +++++++++++++++++++++++++++++++++ 1 file changed, 232 insertions(+) create mode 100644 src/leap/common/keymanager/__init__.py (limited to 'src') diff --git a/src/leap/common/keymanager/__init__.py b/src/leap/common/keymanager/__init__.py new file mode 100644 index 0000000..71aaddd --- /dev/null +++ b/src/leap/common/keymanager/__init__.py @@ -0,0 +1,232 @@ +# -*- coding: utf-8 -*- +# __init__.py +# Copyright (C) 2013 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 . + + +""" +Key Manager is a Nicknym agent for LEAP client. +""" + + +try: + import simplejson as json +except ImportError: + import json # noqa + + +from abc import ABCMeta, abstractmethod +from u1db.errors import HTTPError + + +# +# Key types +# + +class EncryptionKey(object): + """ + Abstract class for encryption keys. + + A key is "validated" if the nicknym agent has bound the user address to a + public key. Nicknym supports three different levels of key validation: + + * Level 3 - path trusted: A path of cryptographic signatures can be traced + from a trusted key to the key under evaluation. By default, only the + provider key from the user's provider is a "trusted key". + * level 2 - provider signed: The key has been signed by a provider key for + the same domain, but the provider key is not validated using a trust + path (i.e. it is only registered) + * level 1 - registered: The key has been encountered and saved, it has no + signatures (that are meaningful to the nicknym agent). + """ + + __metaclass__ = ABCMeta + + def __init__(self, address, key_id=None, fingerprint=None, + key_data=None, length=None, expiry_date=None, + validation=None, first_seen_at=None, + last_audited_at=None): + self.address = address + self.key_id = key_id + self.fingerprint = fingerprint + self.key_data = key_data + self.length = length + self.expiry_date = expiry_date + self.validation = validation + self.first_seen_at = first_seen_at + self.last_audited_at = last_audited_at + + @abstractmethod + def get_json(self): + """ + Return a JSON string describing this key. + + @return: The JSON string describing this key. + @rtype: str + """ + + +# +# Key wrappers +# + +class KeyTypeWrapper(object): + """ + Abstract class for Key Type Wrappers. + + A wrapper for a certain key type should know how to get and put keys in + local storage using Soledad and also how to generate new keys. + """ + + __metaclass__ = ABCMeta + + @abstractmethod + def get_key(self, address): + """ + Get key from local storage. + + @param address: The address bound to the key. + @type address: str + + @return: The key bound to C{address}. + @rtype: EncryptionKey + @raise KeyNotFound: If the key was not found on local storage. + """ + + @abstractmethod + def put_key(self, key): + """ + Put a key in local storage. + + @param key: The key to be stored. + @type key: EncryptionKey + """ + + @abstractmethod + def gen_key(self, address): + """ + Generate a new key. + + @param address: The address bound to the key. + @type address: str + @return: The key bound to C{address}. + @rtype: EncryptionKey + """ + + +# +# Key manager +# + + +class KeyNotFound(Exception): + """ + Raised when key was no found on keyserver. + """ + + +class KeyManager(object): + + def __init__(self, address, url): + """ + Initialize a Key Manager for user's C{address} with provider's + nickserver reachable in C{url}. + + @param address: The address of the user of this Key Manager. + @type address: str + @param url: The URL of the key manager. + @type url: str + """ + self.address = address + self.url = url + + def send_key(self, ktype, send_private=False, password=None): + """ + Send user's key of type C{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. + + If C{send_private} is True, then the private key is encrypted with + C{password} and sent to server in the same request, together with a + hash string of user's address and password. The encrypted private key + will be saved in the server in a way it is publicly retrievable + through the hash string. + + @param address: The address bound to the key. + @type address: str + @param ktype: The type of the key. + @type ktype: KeyType + + @raise httplib.HTTPException: + """ + + def get_key(self, address, ktype): + """ + Return a key of type C{ktype} bound to C{address}. + + First, search for the key in local storage. If it is not available, + then try to fetch from nickserver. + + @param address: The address bound to the key. + @type address: str + @param ktype: The type of the key. + @type ktype: KeyType + + @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. + """ + try: + return wrapper_map[ktype].get_key(address) + except KeyNotFound: + key = filter(lambda k: isinstance(k, ktype), + self._fetch_keys(address)) + if key is None + raise KeyNotFound() + wrapper_map[ktype].put_key(key) + return key + + + def _fetch_keys(self, address): + """ + Fetch keys bound to C{address} from nickserver. + + @param address: The address bound to the keys. + @type address: str + + @return: A list of keys bound to C{address}. + @rtype: list of EncryptionKey + @raise KeyNotFound: If the key was not found on nickserver. + @raise httplib.HTTPException: + """ + + def refresh_keys(self): + """ + Update the user's db of validated keys to see if there are changes. + """ + + def gen_key(self, ktype): + """ + Generate a key of type C{ktype} bound to the user's address. + + @param ktype: The type of the key. + @type ktype: KeyType + + @return: The generated key. + @rtype: EncryptionKey + """ + return wrapper_map[ktype].gen_key(self.address) -- cgit v1.2.3 From 314bc876d564cd6265cc8eb4095e423f1140349a Mon Sep 17 00:00:00 2001 From: drebs Date: Mon, 15 Apr 2013 10:41:56 -0300 Subject: Add basic openpgp key handling to Key Manager --- src/leap/common/keymanager/__init__.py | 116 +-------- src/leap/common/keymanager/errors.py | 29 +++ src/leap/common/keymanager/gpg.py | 398 +++++++++++++++++++++++++++++++ src/leap/common/keymanager/keys.py | 127 ++++++++++ src/leap/common/keymanager/openpgp.py | 126 ++++++++++ src/leap/common/tests/test_keymanager.py | 230 ++++++++++++++++++ 6 files changed, 922 insertions(+), 104 deletions(-) create mode 100644 src/leap/common/keymanager/errors.py create mode 100644 src/leap/common/keymanager/gpg.py create mode 100644 src/leap/common/keymanager/keys.py create mode 100644 src/leap/common/keymanager/openpgp.py create mode 100644 src/leap/common/tests/test_keymanager.py (limited to 'src') diff --git a/src/leap/common/keymanager/__init__.py b/src/leap/common/keymanager/__init__.py index 71aaddd..10acb36 100644 --- a/src/leap/common/keymanager/__init__.py +++ b/src/leap/common/keymanager/__init__.py @@ -27,114 +27,22 @@ except ImportError: import json # noqa -from abc import ABCMeta, abstractmethod from u1db.errors import HTTPError -# -# Key types -# - -class EncryptionKey(object): - """ - Abstract class for encryption keys. - - A key is "validated" if the nicknym agent has bound the user address to a - public key. Nicknym supports three different levels of key validation: - - * Level 3 - path trusted: A path of cryptographic signatures can be traced - from a trusted key to the key under evaluation. By default, only the - provider key from the user's provider is a "trusted key". - * level 2 - provider signed: The key has been signed by a provider key for - the same domain, but the provider key is not validated using a trust - path (i.e. it is only registered) - * level 1 - registered: The key has been encountered and saved, it has no - signatures (that are meaningful to the nicknym agent). - """ - - __metaclass__ = ABCMeta - - def __init__(self, address, key_id=None, fingerprint=None, - key_data=None, length=None, expiry_date=None, - validation=None, first_seen_at=None, - last_audited_at=None): - self.address = address - self.key_id = key_id - self.fingerprint = fingerprint - self.key_data = key_data - self.length = length - self.expiry_date = expiry_date - self.validation = validation - self.first_seen_at = first_seen_at - self.last_audited_at = last_audited_at - - @abstractmethod - def get_json(self): - """ - Return a JSON string describing this key. - - @return: The JSON string describing this key. - @rtype: str - """ - - -# -# Key wrappers -# - -class KeyTypeWrapper(object): - """ - Abstract class for Key Type Wrappers. - - A wrapper for a certain key type should know how to get and put keys in - local storage using Soledad and also how to generate new keys. - """ - - __metaclass__ = ABCMeta - - @abstractmethod - def get_key(self, address): - """ - Get key from local storage. - - @param address: The address bound to the key. - @type address: str - - @return: The key bound to C{address}. - @rtype: EncryptionKey - @raise KeyNotFound: If the key was not found on local storage. - """ - - @abstractmethod - def put_key(self, key): - """ - Put a key in local storage. - - @param key: The key to be stored. - @type key: EncryptionKey - """ - - @abstractmethod - def gen_key(self, address): - """ - Generate a new key. - - @param address: The address bound to the key. - @type address: str - @return: The key bound to C{address}. - @rtype: EncryptionKey - """ - - -# -# Key manager -# +from leap.common.keymanager.errors import ( + KeyNotFound, + KeyAlreadyExists, +) +from leap.common.keymanager.openpgp import ( + OpenPGPKey, + OpenPGPWrapper, +) -class KeyNotFound(Exception): - """ - Raised when key was no found on keyserver. - """ +wrapper_map = { + OpenPGPKey: OpenPGPWrapper(), +} class KeyManager(object): @@ -195,7 +103,7 @@ class KeyManager(object): except KeyNotFound: key = filter(lambda k: isinstance(k, ktype), self._fetch_keys(address)) - if key is None + if key is None: raise KeyNotFound() wrapper_map[ktype].put_key(key) return key diff --git a/src/leap/common/keymanager/errors.py b/src/leap/common/keymanager/errors.py new file mode 100644 index 0000000..f5bb1ab --- /dev/null +++ b/src/leap/common/keymanager/errors.py @@ -0,0 +1,29 @@ +# -*- coding: utf-8 -*- +# errors.py +# Copyright (C) 2013 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 . + + + +class KeyNotFound(Exception): + """ + Raised when key was no found on keyserver. + """ + + +class KeyAlreadyExists(Exception): + """ + Raised when attempted to create a key that already exists. + """ diff --git a/src/leap/common/keymanager/gpg.py b/src/leap/common/keymanager/gpg.py new file mode 100644 index 0000000..dc5d791 --- /dev/null +++ b/src/leap/common/keymanager/gpg.py @@ -0,0 +1,398 @@ +# -*- coding: utf-8 -*- +# gpgwrapper.py +# Copyright (C) 2013 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 . + + +""" +A GPG wrapper used to handle OpenPGP keys. + +This is a temporary class that will be superseded by the a revised version of +python-gnupg. +""" + + +import os +import gnupg +import re +from gnupg import ( + logger, + _is_sequence, + _make_binary_stream, +) + + +class ListPackets(): + """ + Handle status messages for --list-packets. + """ + + def __init__(self, gpg): + """ + Initialize the packet listing handling class. + + @param gpg: GPG object instance. + @type gpg: gnupg.GPG + """ + self.gpg = gpg + self.nodata = None + self.key = None + self.need_passphrase = None + self.need_passphrase_sym = None + self.userid_hint = None + + def handle_status(self, key, value): + """ + Handle one line of the --list-packets status message. + + @param key: The status message key. + @type key: str + @param value: The status message value. + @type value: str + """ + # TODO: write tests for handle_status + if key == 'NODATA': + self.nodata = True + if key == 'ENC_TO': + # This will only capture keys in our keyring. In the future we + # may want to include multiple unknown keys in this list. + self.key, _, _ = value.split() + if key == 'NEED_PASSPHRASE': + self.need_passphrase = True + if key == 'NEED_PASSPHRASE_SYM': + self.need_passphrase_sym = True + if key == 'USERID_HINT': + self.userid_hint = value.strip().split() + + +class GPGWrapper(gnupg.GPG): + """ + This is a temporary class for handling GPG requests, and should be + replaced by a more general class used throughout the project. + """ + + GNUPG_HOME = os.environ['HOME'] + "/.config/leap/gnupg" + GNUPG_BINARY = "/usr/bin/gpg" # this has to be changed based on OS + + def __init__(self, gpgbinary=GNUPG_BINARY, gnupghome=GNUPG_HOME, + verbose=False, use_agent=False, keyring=None, options=None): + """ + Initialize a GnuPG process wrapper. + + @param gpgbinary: Name for GnuPG binary executable. + @type gpgbinary: C{str} + @param gpghome: Full pathname to directory containing the public and + private keyrings. + @type gpghome: C{str} + @param keyring: Name of alternative keyring file to use. If specified, + the default keyring is not used. + @param verbose: Should some verbose info be output? + @type verbose: bool + @param use_agent: Should pass `--use-agent` to GPG binary? + @type use_agent: bool + @param keyring: Path for the keyring to use. + @type keyring: str + @options: A list of additional options to pass to the GPG binary. + @type options: list + + @raise: RuntimeError with explanation message if there is a problem + invoking gpg. + """ + gnupg.GPG.__init__(self, gnupghome=gnupghome, gpgbinary=gpgbinary, + verbose=verbose, use_agent=use_agent, + keyring=keyring, options=options) + self.result_map['list-packets'] = ListPackets + + def find_key_by_email(self, email, secret=False): + """ + Find user's key based on their email. + + @param email: Email address of key being searched for. + @type email: str + @param secret: Should we search for a secret key? + @type secret: bool + + @return: The fingerprint of the found key. + @rtype: str + """ + for key in self.list_keys(secret=secret): + for uid in key['uids']: + if re.search(email, uid): + return key + raise LookupError("GnuPG public key for email %s not found!" % email) + + def find_key_by_subkey(self, subkey, secret=False): + """ + Find user's key based on a subkey fingerprint. + + @param email: Subkey fingerprint of the key being searched for. + @type email: str + @param secret: Should we search for a secret key? + @type secret: bool + + @return: The fingerprint of the found key. + @rtype: str + """ + for key in self.list_keys(secret=secret): + for sub in key['subkeys']: + if sub[0] == subkey: + return key + raise LookupError( + "GnuPG public key for subkey %s not found!" % subkey) + + def find_key_by_keyid(self, keyid, secret=False): + """ + Find user's key based on the key ID. + + @param email: The key ID of the key being searched for. + @type email: str + @param secret: Should we search for a secret key? + @type secret: bool + + @return: The fingerprint of the found key. + @rtype: str + """ + for key in self.list_keys(secret=secret): + if keyid == key['keyid']: + return key + raise LookupError( + "GnuPG public key for keyid %s not found!" % keyid) + + def find_key_by_fingerprint(self, fingerprint, secret=False): + """ + Find user's key based on the key fingerprint. + + @param email: The fingerprint of the key being searched for. + @type email: str + @param secret: Should we search for a secret key? + @type secret: bool + + @return: The fingerprint of the found key. + @rtype: str + """ + for key in self.list_keys(secret=secret): + if fingerprint == key['fingerprint']: + return key + raise LookupError( + "GnuPG public key for fingerprint %s not found!" % fingerprint) + + def encrypt(self, data, recipient, sign=None, always_trust=True, + passphrase=None, symmetric=False): + """ + Encrypt data using GPG. + + @param data: The data to be encrypted. + @type data: str + @param recipient: The address of the public key to be used. + @type recipient: str + @param sign: Should the encrypted content be signed? + @type sign: bool + @param always_trust: Skip key validation and assume that used keys + are always fully trusted? + @type always_trust: bool + @param passphrase: The passphrase to be used if symmetric encryption + is desired. + @type passphrase: str + @param symmetric: Should we encrypt to a password? + @type symmetric: bool + + @return: An object with encrypted result in the `data` field. + @rtype: gnupg.Crypt + """ + # TODO: devise a way so we don't need to "always trust". + return gnupg.GPG.encrypt(self, data, recipient, sign=sign, + always_trust=always_trust, + passphrase=passphrase, + symmetric=symmetric, + cipher_algo='AES256') + + def decrypt(self, data, always_trust=True, passphrase=None): + """ + Decrypt data using GPG. + + @param data: The data to be decrypted. + @type data: str + @param always_trust: Skip key validation and assume that used keys + are always fully trusted? + @type always_trust: bool + @param passphrase: The passphrase to be used if symmetric encryption + is desired. + @type passphrase: str + + @return: An object with decrypted result in the `data` field. + @rtype: gnupg.Crypt + """ + # TODO: devise a way so we don't need to "always trust". + return gnupg.GPG.decrypt(self, data, always_trust=always_trust, + passphrase=passphrase) + + def send_keys(self, keyserver, *keyids): + """ + Send keys to a keyserver + + @param keyserver: The keyserver to send the keys to. + @type keyserver: str + @param keyids: The key ids to send. + @type keyids: list + + @return: A list of keys sent to server. + @rtype: gnupg.ListKeys + """ + # TODO: write tests for this. + # TODO: write a SendKeys class to handle status for this. + result = self.result_map['list'](self) + gnupg.logger.debug('send_keys: %r', keyids) + data = gnupg._make_binary_stream("", self.encoding) + args = ['--keyserver', keyserver, '--send-keys'] + args.extend(keyids) + self._handle_io(args, data, result, binary=True) + gnupg.logger.debug('send_keys result: %r', result.__dict__) + data.close() + return result + + def encrypt_file(self, file, recipients, sign=None, + always_trust=False, passphrase=None, + armor=True, output=None, symmetric=False, + cipher_algo=None): + """ + Encrypt the message read from the file-like object 'file'. + + @param file: The file to be encrypted. + @type data: file + @param recipient: The address of the public key to be used. + @type recipient: str + @param sign: Should the encrypted content be signed? + @type sign: bool + @param always_trust: Skip key validation and assume that used keys + are always fully trusted? + @type always_trust: bool + @param passphrase: The passphrase to be used if symmetric encryption + is desired. + @type passphrase: str + @param armor: Create ASCII armored output? + @type armor: bool + @param output: Path of file to write results in. + @type output: str + @param symmetric: Should we encrypt to a password? + @type symmetric: bool + @param cipher_algo: Algorithm to use. + @type cipher_algo: str + + @return: An object with encrypted result in the `data` field. + @rtype: gnupg.Crypt + """ + args = ['--encrypt'] + if symmetric: + args = ['--symmetric'] + if cipher_algo: + args.append('--cipher-algo %s' % cipher_algo) + else: + args = ['--encrypt'] + if not _is_sequence(recipients): + recipients = (recipients,) + for recipient in recipients: + args.append('--recipient "%s"' % recipient) + if armor: # create ascii-armored output - set to False for binary + args.append('--armor') + if output: # write the output to a file with the specified name + if os.path.exists(output): + os.remove(output) # to avoid overwrite confirmation message + args.append('--output "%s"' % output) + if sign: + args.append('--sign --default-key "%s"' % sign) + if always_trust: + args.append("--always-trust") + result = self.result_map['crypt'](self) + self._handle_io(args, file, result, passphrase=passphrase, binary=True) + logger.debug('encrypt result: %r', result.data) + return result + + def list_packets(self, data): + """ + List the sequence of packets. + + @param data: The data to extract packets from. + @type data: str + + @return: An object with packet info. + @rtype ListPackets + """ + args = ["--list-packets"] + result = self.result_map['list-packets'](self) + self._handle_io( + args, + _make_binary_stream(data, self.encoding), + result, + ) + return result + + def encrypted_to(self, data): + """ + Return the key to which data is encrypted to. + + @param data: The data to be examined. + @type data: str + + @return: The fingerprint of the key to which data is encrypted to. + @rtype: str + """ + # TODO: make this support multiple keys. + result = self.list_packets(data) + if not result.key: + raise LookupError( + "Content is not encrypted to a GnuPG key!") + try: + return self.find_key_by_keyid(result.key) + except: + return self.find_key_by_subkey(result.key) + + def is_encrypted_sym(self, data): + """ + Say whether some chunk of data is encrypted to a symmetric key. + + @param data: The data to be examined. + @type data: str + + @return: Whether data is encrypted to a symmetric key. + @rtype: bool + """ + result = self.list_packets(data) + return bool(result.need_passphrase_sym) + + def is_encrypted_asym(self, data): + """ + Say whether some chunk of data is encrypted to a private key. + + @param data: The data to be examined. + @type data: str + + @return: Whether data is encrypted to a private key. + @rtype: bool + """ + result = self.list_packets(data) + return bool(result.key) + + def is_encrypted(self, data): + """ + Say whether some chunk of data is encrypted to a key. + + @param data: The data to be examined. + @type data: str + + @return: Whether data is encrypted to a key. + @rtype: bool + """ + self.is_encrypted_asym() or self.is_encrypted_sym() + diff --git a/src/leap/common/keymanager/keys.py b/src/leap/common/keymanager/keys.py new file mode 100644 index 0000000..13e3c0b --- /dev/null +++ b/src/leap/common/keymanager/keys.py @@ -0,0 +1,127 @@ +# -*- coding: utf-8 -*- +# keys.py +# Copyright (C) 2013 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 . + + +""" +Abstact key type and wrapper representations. +""" + + +from abc import ABCMeta, abstractmethod + + +class EncryptionKey(object): + """ + Abstract class for encryption keys. + + A key is "validated" if the nicknym agent has bound the user address to a + public key. Nicknym supports three different levels of key validation: + + * Level 3 - path trusted: A path of cryptographic signatures can be traced + from a trusted key to the key under evaluation. By default, only the + provider key from the user's provider is a "trusted key". + * level 2 - provider signed: The key has been signed by a provider key for + the same domain, but the provider key is not validated using a trust + path (i.e. it is only registered) + * level 1 - registered: The key has been encountered and saved, it has no + signatures (that are meaningful to the nicknym agent). + """ + + __metaclass__ = ABCMeta + + def __init__(self, address, key_id=None, fingerprint=None, + key_data=None, length=None, expiry_date=None, + validation=None, first_seen_at=None, + last_audited_at=None): + self.address = address + self.key_id = key_id + self.fingerprint = fingerprint + self.key_data = key_data + self.length = length + self.expiry_date = expiry_date + self.validation = validation + self.first_seen_at = first_seen_at + self.last_audited_at = last_audited_at + + def get_json(self): + """ + Return a JSON string describing this key. + + @return: The JSON string describing this key. + @rtype: str + """ + return json.dumps({ + 'address': self.address, + 'type': str(self.__type__), + 'key_id': self.key_id, + 'fingerprint': self.fingerprint, + 'key_data': self.key_data, + 'length': self.length, + 'expiry_date': self.expiry_date, + 'validation': self.validation, + 'first_seen_at': self.first_seen_at, + 'last_audited_at': self.last_audited_at, + }) + + +# +# Key wrappers +# + +class KeyTypeWrapper(object): + """ + Abstract class for Key Type Wrappers. + + A wrapper for a certain key type should know how to get and put keys in + local storage using Soledad and also how to generate new keys. + """ + + __metaclass__ = ABCMeta + + @abstractmethod + def get_key(self, address): + """ + Get key from local storage. + + @param address: The address bound to the key. + @type address: str + + @return: The key bound to C{address}. + @rtype: EncryptionKey + @raise KeyNotFound: If the key was not found on local storage. + """ + + @abstractmethod + def put_key(self, key): + """ + Put a key in local storage. + + @param key: The key to be stored. + @type key: EncryptionKey + """ + + @abstractmethod + def gen_key(self, address): + """ + Generate a new key. + + @param address: The address bound to the key. + @type address: str + @return: The key bound to C{address}. + @rtype: EncryptionKey + """ + diff --git a/src/leap/common/keymanager/openpgp.py b/src/leap/common/keymanager/openpgp.py new file mode 100644 index 0000000..bb73089 --- /dev/null +++ b/src/leap/common/keymanager/openpgp.py @@ -0,0 +1,126 @@ +# -*- coding: utf-8 -*- +# openpgpwrapper.py +# Copyright (C) 2013 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 . + + +""" +Infrastructure for using OpenPGP keys in Key Manager. +""" + + +import re + +from leap.common.keymanager.errors import ( + KeyNotFound, + KeyAlreadyExists, +) +from leap.common.keymanager.keys import ( + EncryptionKey, + KeyTypeWrapper, +) +from leap.common.keymanager.gpg import GPGWrapper + + +class OpenPGPKey(EncryptionKey): + """ + Base class for OpenPGP keys. + """ + + +class OpenPGPWrapper(KeyTypeWrapper): + """ + A wrapper for OpenPGP keys. + """ + + def __init__(self, gnupghome=None): + self._gpg = GPGWrapper(gnupghome=gnupghome) + + def _build_key(self, address, result): + """ + Build an OpenPGPWrapper key for C{address} based on C{result} from + local storage. + + @param address: The address bound to the key. + @type address: str + @param result: Result obtained from GPG storage. + @type result: dict + """ + key_data = self._gpg.export_keys(result['fingerprint'], secret=False) + return OpenPGPKey( + address, + key_id=result['keyid'], + fingerprint=result['fingerprint'], + key_data=key_data, + length=result['length'], + expiry_date=result['expires'], + validation=None, # TODO: verify for validation. + ) + + def gen_key(self, address): + """ + Generate an OpenPGP keypair for C{address}. + + @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. + """ + try: + self.get_key(address) + raise KeyAlreadyExists() + except KeyNotFound: + pass + params = self._gpg.gen_key_input( + key_type='RSA', + key_length=4096, + name_real=address, + name_email=address, + name_comment='Generated by LEAP Key Manager.') + self._gpg.gen_key(params) + return self.get_key(address) + + def get_key(self, address): + """ + Get key bound to C{address} from local storage. + + @param address: The address bound to the key. + @type address: str + + @return: The key bound to C{address}. + @rtype: OpenPGPKey + @raise KeyNotFound: If the key was not found on local storage. + """ + m = re.compile('.*<%s>$' % address) + keys = self._gpg.list_keys(secret=False) + + def bound_to_address(key): + return bool(filter(lambda u: m.match(u), key['uids'])) + + try: + bound_key = filter(bound_to_address, keys).pop() + return self._build_key(address, bound_key) + except IndexError: + raise KeyNotFound(address) + + def put_key(self, data): + """ + Put key contained in {data} in local storage. + + @param key: The key data to be stored. + @type key: str + """ + self._gpg.import_keys(data) diff --git a/src/leap/common/tests/test_keymanager.py b/src/leap/common/tests/test_keymanager.py new file mode 100644 index 0000000..4189aac --- /dev/null +++ b/src/leap/common/tests/test_keymanager.py @@ -0,0 +1,230 @@ +## -*- coding: utf-8 -*- +# test_keymanager.py +# Copyright (C) 2013 LEAP +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + + +""" +Tests for the Key Manager. +""" + + +import unittest + + +from leap.common.testing.basetest import BaseLeapTest +from leap.common.keymanager import KeyManager, openpgp, KeyNotFound + + +class KeyManagerTestCase(BaseLeapTest): + + def setUp(self): + pass + + def tearDown(self): + pass + + def _key_manager(user='user@leap.se', url='https://domain.org:6425'): + return KeyManager(user, url) + + def test_openpgp_gen_key(self): + pgp = openpgp.OpenPGPWrapper(self.tempdir+'/gnupg') + try: + pgp.get_key('user@leap.se') + except KeyNotFound: + 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_key(self): + pgp = openpgp.OpenPGPWrapper(self.tempdir+'/gnupg2') + try: + pgp.get_key('leap@leap.se') + except KeyNotFound: + pgp.put_key(PUBLIC_KEY) + key = pgp.get_key('leap@leap.se') + self.assertIsInstance(key, openpgp.OpenPGPKey) + self.assertEqual( + 'leap@leap.se', key.address, 'Wrong address bound to key.') + self.assertEqual( + '4096', key.length, 'Wrong key length.') + + + +# Key material for testing +KEY_FINGERPRINT = "E36E738D69173C13D709E44F2F455E2824D18DDF" +PUBLIC_KEY = """ +-----BEGIN PGP PUBLIC KEY BLOCK----- +Version: GnuPG v1.4.10 (GNU/Linux) + +mQINBFC9+dkBEADNRfwV23TWEoGc/x0wWH1P7PlXt8MnC2Z1kKaKKmfnglVrpOiz +iLWoiU58sfZ0L5vHkzXHXCBf6Eiy/EtUIvdiWAn+yASJ1mk5jZTBKO/WMAHD8wTO +zpMsFmWyg3xc4DkmFa9KQ5EVU0o/nqPeyQxNMQN7px5pPwrJtJFmPxnxm+aDkPYx +irDmz/4DeDNqXliazGJKw7efqBdlwTHkl9Akw2gwy178pmsKwHHEMOBOFFvX61AT +huKqHYmlCGSliwbrJppTG7jc1/ls3itrK+CWTg4txREkSpEVmfcASvw/ZqLbjgfs +d/INMwXnR9U81O8+7LT6yw/ca4ppcFoJD7/XJbkRiML6+bJ4Dakiy6i727BzV17g +wI1zqNvm5rAhtALKfACha6YO43aJzairO4II1wxVHvRDHZn2IuKDDephQ3Ii7/vb +hUOf6XCSmchkAcpKXUOvbxm1yfB1LRa64mMc2RcZxf4mW7KQkulBsdV5QG2276lv +U2UUy2IutXcGP5nXC+f6sJJGJeEToKJ57yiO/VWJFjKN8SvP+7AYsQSqINUuEf6H +T5gCPCraGMkTUTPXrREvu7NOohU78q6zZNaL3GW8ai7eSeANSuQ8Vzffx7Wd8Y7i +Pw9sYj0SMFs1UgjbuL6pO5ueHh+qyumbtAq2K0Bci0kqOcU4E9fNtdiovQARAQAB +tBxMZWFwIFRlc3QgS2V5IDxsZWFwQGxlYXAuc2U+iQI3BBMBCAAhBQJQvfnZAhsD +BQsJCAcDBRUKCQgLBRYCAwEAAh4BAheAAAoJEC9FXigk0Y3fT7EQAKH3IuRniOpb +T/DDIgwwjz3oxB/W0DDMyPXowlhSOuM0rgGfntBpBb3boezEXwL86NPQxNGGruF5 +hkmecSiuPSvOmQlqlS95NGQp6hNG0YaKColh+Q5NTspFXCAkFch9oqUje0LdxfSP +QfV9UpeEvGyPmk1I9EJV/YDmZ4+Djge1d7qhVZInz4Rx1NrSyF/Tc2EC0VpjQFsU +Y9Kb2YBBR7ivG6DBc8ty0jJXi7B4WjkFcUEJviQpMF2dCLdonCehYs1PqsN1N7j+ +eFjQd+hqVMJgYuSGKjvuAEfClM6MQw7+FmFwMyLgK/Ew/DttHEDCri77SPSkOGSI +txCzhTg6798f6mJr7WcXmHX1w1Vcib5FfZ8vTDFVhz/XgAgArdhPo9V6/1dgSSiB +KPQ/spsco6u5imdOhckERE0lnAYvVT6KE81TKuhF/b23u7x+Wdew6kK0EQhYA7wy +7LmlaNXc7rMBQJ9Z60CJ4JDtatBWZ0kNrt2VfdDHVdqBTOpl0CraNUjWE5YMDasr +K2dF5IX8D3uuYtpZnxqg0KzyLg0tzL0tvOL1C2iudgZUISZNPKbS0z0v+afuAAnx +2pTC3uezbh2Jt8SWTLhll4i0P4Ps5kZ6HQUO56O+/Z1cWovX+mQekYFmERySDR9n +3k1uAwLilJmRmepGmvYbB8HloV8HqwgguQINBFC9+dkBEAC0I/xn1uborMgDvBtf +H0sEhwnXBC849/32zic6udB6/3Efk9nzbSpL3FSOuXITZsZgCHPkKarnoQ2ztMcS +sh1ke1C5gQGms75UVmM/nS+2YI4vY8OX/GC/on2vUyncqdH+bR6xH5hx4NbWpfTs +iQHmz5C6zzS/kuabGdZyKRaZHt23WQ7JX/4zpjqbC99DjHcP9BSk7tJ8wI4bkMYD +uFVQdT9O6HwyKGYwUU4sAQRAj7XCTGvVbT0dpgJwH4RmrEtJoHAx4Whg8mJ710E0 +GCmzf2jqkNuOw76ivgk27Kge+Hw00jmJjQhHY0yVbiaoJwcRrPKzaSjEVNgrpgP3 +lXPRGQArgESsIOTeVVHQ8fhK2YtTeCY9rIiO+L0OX2xo9HK7hfHZZWL6rqymXdyS +fhzh/f6IPyHFWnvj7Brl7DR8heMikygcJqv+ed2yx7iLyCUJ10g12I48+aEj1aLe +dP7lna32iY8/Z0SHQLNH6PXO9SlPcq2aFUgKqE75A/0FMk7CunzU1OWr2ZtTLNO1 +WT/13LfOhhuEq9jTyTosn0WxBjJKq18lnhzCXlaw6EAtbA7CUwsD3CTPR56aAXFK +3I7KXOVAqggrvMe5Tpdg5drfYpI8hZovL5aAgb+7Y5ta10TcJdUhS5K3kFAWe/td +U0cmWUMDP1UMSQ5Jg6JIQVWhSwARAQABiQIfBBgBCAAJBQJQvfnZAhsMAAoJEC9F +Xigk0Y3fRwsP/i0ElYCyxeLpWJTwo1iCLkMKz2yX1lFVa9nT1BVTPOQwr/IAc5OX +NdtbJ14fUsKL5pWgW8OmrXtwZm1y4euI1RPWWubG01ouzwnGzv26UcuHeqC5orZj +cOnKtL40y8VGMm8LoicVkRJH8blPORCnaLjdOtmA3rx/v2EXrJpSa3AhOy0ZSRXk +ZSrK68AVNwamHRoBSYyo0AtaXnkPX4+tmO8X8BPfj125IljubvwZPIW9VWR9UqCE +VPfDR1XKegVb6VStIywF7kmrknM1C5qUY28rdZYWgKorw01hBGV4jTW0cqde3N51 +XT1jnIAa+NoXUM9uQoGYMiwrL7vNsLlyyiW5ayDyV92H/rIuiqhFgbJsHTlsm7I8 +oGheR784BagAA1NIKD1qEO9T6Kz9lzlDaeWS5AUKeXrb7ZJLI1TTCIZx5/DxjLqM +Tt/RFBpVo9geZQrvLUqLAMwdaUvDXC2c6DaCPXTh65oCZj/hqzlJHH+RoTWWzKI+ +BjXxgUWF9EmZUBrg68DSmI+9wuDFsjZ51BcqvJwxyfxtTaWhdoYqH/UQS+D1FP3/ +diZHHlzwVwPICzM9ooNTgbrcDzyxRkIVqsVwBq7EtzcvgYUyX53yG25Giy6YQaQ2 +ZtQ/VymwFL3XdUWV6B/hU4PVAFvO3qlOtdJ6TpE+nEWgcWjCv5g7RjXX +=MuOY +-----END PGP PUBLIC KEY BLOCK----- +""" +PRIVATE_KEY = """ +-----BEGIN PGP PRIVATE KEY BLOCK----- +Version: GnuPG v1.4.10 (GNU/Linux) + +lQcYBFC9+dkBEADNRfwV23TWEoGc/x0wWH1P7PlXt8MnC2Z1kKaKKmfnglVrpOiz +iLWoiU58sfZ0L5vHkzXHXCBf6Eiy/EtUIvdiWAn+yASJ1mk5jZTBKO/WMAHD8wTO +zpMsFmWyg3xc4DkmFa9KQ5EVU0o/nqPeyQxNMQN7px5pPwrJtJFmPxnxm+aDkPYx +irDmz/4DeDNqXliazGJKw7efqBdlwTHkl9Akw2gwy178pmsKwHHEMOBOFFvX61AT +huKqHYmlCGSliwbrJppTG7jc1/ls3itrK+CWTg4txREkSpEVmfcASvw/ZqLbjgfs +d/INMwXnR9U81O8+7LT6yw/ca4ppcFoJD7/XJbkRiML6+bJ4Dakiy6i727BzV17g +wI1zqNvm5rAhtALKfACha6YO43aJzairO4II1wxVHvRDHZn2IuKDDephQ3Ii7/vb +hUOf6XCSmchkAcpKXUOvbxm1yfB1LRa64mMc2RcZxf4mW7KQkulBsdV5QG2276lv +U2UUy2IutXcGP5nXC+f6sJJGJeEToKJ57yiO/VWJFjKN8SvP+7AYsQSqINUuEf6H +T5gCPCraGMkTUTPXrREvu7NOohU78q6zZNaL3GW8ai7eSeANSuQ8Vzffx7Wd8Y7i +Pw9sYj0SMFs1UgjbuL6pO5ueHh+qyumbtAq2K0Bci0kqOcU4E9fNtdiovQARAQAB +AA/+JHtlL39G1wsH9R6UEfUQJGXR9MiIiwZoKcnRB2o8+DS+OLjg0JOh8XehtuCs +E/8oGQKtQqa5bEIstX7IZoYmYFiUQi9LOzIblmp2vxOm+HKkxa4JszWci2/ZmC3t +KtaA4adl9XVnshoQ7pijuCMUKB3naBEOAxd8s9d/JeReGIYkJErdrnVfNk5N71Ds +FmH5Ll3XtEDvgBUQP3nkA6QFjpsaB94FHjL3gDwum/cxzj6pCglcvHOzEhfY0Ddb +J967FozQTaf2JW3O+w3LOqtcKWpq87B7+O61tVidQPSSuzPjCtFF0D2LC9R/Hpky +KTMQ6CaKja4MPhjwywd4QPcHGYSqjMpflvJqi+kYIt8psUK/YswWjnr3r4fbuqVY +VhtiHvnBHQjz135lUqWvEz4hM3Xpnxydx7aRlv5NlevK8+YIO5oFbWbGNTWsPZI5 +jpoFBpSsnR1Q5tnvtNHauvoWV+XN2qAOBTG+/nEbDYH6Ak3aaE9jrpTdYh0CotYF +q7csANsDy3JvkAzeU6WnYpsHHaAjqOGyiZGsLej1UcXPFMosE/aUo4WQhiS8Zx2c +zOVKOi/X5vQ2GdNT9Qolz8AriwzsvFR+bxPzyd8V6ALwDsoXvwEYinYBKK8j0OPv +OOihSR6HVsuP9NUZNU9ewiGzte/+/r6pNXHvR7wTQ8EWLcEIAN6Zyrb0bHZTIlxt +VWur/Ht2mIZrBaO50qmM5RD3T5oXzWXi/pjLrIpBMfeZR9DWfwQwjYzwqi7pxtYx +nJvbMuY505rfnMoYxb4J+cpRXV8MS7Dr1vjjLVUC9KiwSbM3gg6emfd2yuA93ihv +Pe3mffzLIiQa4mRE3wtGcioC43nWuV2K2e1KjxeFg07JhrezA/1Cak505ab/tmvP +4YmjR5c44+yL/YcQ3HdFgs4mV+nVbptRXvRcPpolJsgxPccGNdvHhsoR4gwXMS3F +RRPD2z6x8xeN73Q4KH3bm01swQdwFBZbWVfmUGLxvN7leCdfs9+iFJyqHiCIB6Iv +mQfp8F0IAOwSo8JhWN+V1dwML4EkIrM8wUb4yecNLkyR6TpPH/qXx4PxVMC+vy6x +sCtjeHIwKE+9vqnlhd5zOYh7qYXEJtYwdeDDmDbL8oks1LFfd+FyAuZXY33DLwn0 +cRYsr2OEZmaajqUB3NVmj3H4uJBN9+paFHyFSXrH68K1Fk2o3n+RSf2EiX+eICwI +L6rqoF5sSVUghBWdNegV7qfy4anwTQwrIMGjgU5S6PKW0Dr/3iO5z3qQpGPAj5OW +ATqPWkDICLbObPxD5cJlyyNE2wCA9VVc6/1d6w4EVwSq9h3/WTpATEreXXxTGptd +LNiTA1nmakBYNO2Iyo3djhaqBdWjk+EIAKtVEnJH9FAVwWOvaj1RoZMA5DnDMo7e +SnhrCXl8AL7Z1WInEaybasTJXn1uQ8xY52Ua4b8cbuEKRKzw/70NesFRoMLYoHTO +dyeszvhoDHberpGRTciVmpMu7Hyi33rM31K9epA4ib6QbbCHnxkWOZB+Bhgj1hJ8 +xb4RBYWiWpAYcg0+DAC3w9gfxQhtUlZPIbmbrBmrVkO2GVGUj8kH6k4UV6kUHEGY +HQWQR0HcbKcXW81ZXCCD0l7ROuEWQtTe5Jw7dJ4/QFuqZnPutXVRNOZqpl6eRShw +7X2/a29VXBpmHA95a88rSQsL+qm7Fb3prqRmuMCtrUZgFz7HLSTuUMR867QcTGVh +cCBUZXN0IEtleSA8bGVhcEBsZWFwLnNlPokCNwQTAQgAIQUCUL352QIbAwULCQgH +AwUVCgkICwUWAgMBAAIeAQIXgAAKCRAvRV4oJNGN30+xEACh9yLkZ4jqW0/wwyIM +MI896MQf1tAwzMj16MJYUjrjNK4Bn57QaQW926HsxF8C/OjT0MTRhq7heYZJnnEo +rj0rzpkJapUveTRkKeoTRtGGigqJYfkOTU7KRVwgJBXIfaKlI3tC3cX0j0H1fVKX +hLxsj5pNSPRCVf2A5mePg44HtXe6oVWSJ8+EcdTa0shf03NhAtFaY0BbFGPSm9mA +QUe4rxugwXPLctIyV4uweFo5BXFBCb4kKTBdnQi3aJwnoWLNT6rDdTe4/nhY0Hfo +alTCYGLkhio77gBHwpTOjEMO/hZhcDMi4CvxMPw7bRxAwq4u+0j0pDhkiLcQs4U4 +Ou/fH+pia+1nF5h19cNVXIm+RX2fL0wxVYc/14AIAK3YT6PVev9XYEkogSj0P7Kb +HKOruYpnToXJBERNJZwGL1U+ihPNUyroRf29t7u8flnXsOpCtBEIWAO8Muy5pWjV +3O6zAUCfWetAieCQ7WrQVmdJDa7dlX3Qx1XagUzqZdAq2jVI1hOWDA2rKytnReSF +/A97rmLaWZ8aoNCs8i4NLcy9Lbzi9QtornYGVCEmTTym0tM9L/mn7gAJ8dqUwt7n +s24dibfElky4ZZeItD+D7OZGeh0FDuejvv2dXFqL1/pkHpGBZhEckg0fZ95NbgMC +4pSZkZnqRpr2GwfB5aFfB6sIIJ0HGARQvfnZARAAtCP8Z9bm6KzIA7wbXx9LBIcJ +1wQvOPf99s4nOrnQev9xH5PZ820qS9xUjrlyE2bGYAhz5Cmq56ENs7THErIdZHtQ +uYEBprO+VFZjP50vtmCOL2PDl/xgv6J9r1Mp3KnR/m0esR+YceDW1qX07IkB5s+Q +us80v5LmmxnWcikWmR7dt1kOyV/+M6Y6mwvfQ4x3D/QUpO7SfMCOG5DGA7hVUHU/ +Tuh8MihmMFFOLAEEQI+1wkxr1W09HaYCcB+EZqxLSaBwMeFoYPJie9dBNBgps39o +6pDbjsO+or4JNuyoHvh8NNI5iY0IR2NMlW4mqCcHEazys2koxFTYK6YD95Vz0RkA +K4BErCDk3lVR0PH4StmLU3gmPayIjvi9Dl9saPRyu4Xx2WVi+q6spl3ckn4c4f3+ +iD8hxVp74+wa5ew0fIXjIpMoHCar/nndsse4i8glCddINdiOPPmhI9Wi3nT+5Z2t +9omPP2dEh0CzR+j1zvUpT3KtmhVICqhO+QP9BTJOwrp81NTlq9mbUyzTtVk/9dy3 +zoYbhKvY08k6LJ9FsQYySqtfJZ4cwl5WsOhALWwOwlMLA9wkz0eemgFxStyOylzl +QKoIK7zHuU6XYOXa32KSPIWaLy+WgIG/u2ObWtdE3CXVIUuSt5BQFnv7XVNHJllD +Az9VDEkOSYOiSEFVoUsAEQEAAQAP/1AagnZQZyzHDEgw4QELAspYHCWLXE5aZInX +wTUJhK31IgIXNn9bJ0hFiSpQR2xeMs9oYtRuPOu0P8oOFMn4/z374fkjZy8QVY3e +PlL+3EUeqYtkMwlGNmVw5a/NbNuNfm5Darb7pEfbYd1gPcni4MAYw7R2SG/57GbC +9gucvspHIfOSfBNLBthDzmK8xEKe1yD2eimfc2T7IRYb6hmkYfeds5GsqvGI6mwI +85h4uUHWRc5JOlhVM6yX8hSWx0L60Z3DZLChmc8maWnFXd7C8eQ6P1azJJbW71Ih +7CoK0XW4LE82vlQurSRFgTwfl7wFYszW2bOzCuhHDDtYnwH86Nsu0DC78ZVRnvxn +E8Ke/AJgrdhIOo4UAyR+aZD2+2mKd7/waOUTUrUtTzc7i8N3YXGi/EIaNReBXaq+ +ZNOp24BlFzRp+FCF/pptDW9HjPdiV09x0DgICmeZS4Gq/4vFFIahWctg52NGebT0 +Idxngjj+xDtLaZlLQoOz0n5ByjO/Wi0ANmMv1sMKCHhGvdaSws2/PbMR2r4caj8m +KXpIgdinM/wUzHJ5pZyF2U/qejsRj8Kw8KH/tfX4JCLhiaP/mgeTuWGDHeZQERAT +xPmRFHaLP9/ZhvGNh6okIYtrKjWTLGoXvKLHcrKNisBLSq+P2WeFrlme1vjvJMo/ +jPwLT5o9CADQmcbKZ+QQ1ZM9v99iDZol7SAMZX43JC019sx6GK0u6xouJBcLfeB4 +OXacTgmSYdTa9RM9fbfVpti01tJ84LV2SyL/VJq/enJF4XQPSynT/tFTn1PAor6o +tEAAd8fjKdJ6LnD5wb92SPHfQfXqI84rFEO8rUNIE/1ErT6DYifDzVCbfD2KZdoF +cOSp7TpD77sY1bs74ocBX5ejKtd+aH99D78bJSMM4pSDZsIEwnomkBHTziubPwJb +OwnATy0LmSMAWOw5rKbsh5nfwCiUTM20xp0t5JeXd+wPVWbpWqI2EnkCEN+RJr9i +7dp/ymDQ+Yt5wrsN3NwoyiexPOG91WQVCADdErHsnglVZZq9Z8Wx7KwecGCUurJ2 +H6lKudv5YOxPnAzqZS5HbpZd/nRTMZh2rdXCr5m2YOuewyYjvM757AkmUpM09zJX +MQ1S67/UX2y8/74TcRF97Ncx9HeELs92innBRXoFitnNguvcO6Esx4BTe1OdU6qR +ER3zAmVf22Le9ciXbu24DN4mleOH+OmBx7X2PqJSYW9GAMTsRB081R6EWKH7romQ +waxFrZ4DJzZ9ltyosEJn5F32StyLrFxpcrdLUoEaclZCv2qka7sZvi0EvovDVEBU +e10jOx9AOwf8Gj2ufhquQ6qgVYCzbP+YrodtkFrXRS3IsljIchj1M2ffB/0bfoUs +rtER9pLvYzCjBPg8IfGLw0o754Qbhh/ReplCRTusP/fQMybvCvfxreS3oyEriu/G +GufRomjewZ8EMHDIgUsLcYo2UHZsfF7tcazgxMGmMvazp4r8vpgrvW/8fIN/6Adu +tF+WjWDTvJLFJCe6O+BFJOWrssNrrra1zGtLC1s8s+Wfpe+bGPL5zpHeebGTwH1U +22eqgJArlEKxrfarz7W5+uHZJHSjF/K9ZvunLGD0n9GOPMpji3UO3zeM8IYoWn7E +/EWK1XbjnssNemeeTZ+sDh+qrD7BOi+vCX1IyBxbfqnQfJZvmcPWpruy1UsO+aIC +0GY8Jr3OL69dDQ21jueJAh8EGAEIAAkFAlC9+dkCGwwACgkQL0VeKCTRjd9HCw/+ +LQSVgLLF4ulYlPCjWIIuQwrPbJfWUVVr2dPUFVM85DCv8gBzk5c121snXh9Swovm +laBbw6ate3BmbXLh64jVE9Za5sbTWi7PCcbO/bpRy4d6oLmitmNw6cq0vjTLxUYy +bwuiJxWREkfxuU85EKdouN062YDevH+/YResmlJrcCE7LRlJFeRlKsrrwBU3BqYd +GgFJjKjQC1peeQ9fj62Y7xfwE9+PXbkiWO5u/Bk8hb1VZH1SoIRU98NHVcp6BVvp +VK0jLAXuSauSczULmpRjbyt1lhaAqivDTWEEZXiNNbRyp17c3nVdPWOcgBr42hdQ +z25CgZgyLCsvu82wuXLKJblrIPJX3Yf+si6KqEWBsmwdOWybsjygaF5HvzgFqAAD +U0goPWoQ71PorP2XOUNp5ZLkBQp5etvtkksjVNMIhnHn8PGMuoxO39EUGlWj2B5l +Cu8tSosAzB1pS8NcLZzoNoI9dOHrmgJmP+GrOUkcf5GhNZbMoj4GNfGBRYX0SZlQ +GuDrwNKYj73C4MWyNnnUFyq8nDHJ/G1NpaF2hiof9RBL4PUU/f92JkceXPBXA8gL +Mz2ig1OButwPPLFGQhWqxXAGrsS3Ny+BhTJfnfIbbkaLLphBpDZm1D9XKbAUvdd1 +RZXoH+FTg9UAW87eqU610npOkT6cRaBxaMK/mDtGNdc= +=JTFu +-----END PGP PRIVATE KEY BLOCK----- +""" -- cgit v1.2.3 From 32999ef8d08b6e94d356ea5fbce43ceebbf5247c Mon Sep 17 00:00:00 2001 From: drebs Date: Fri, 19 Apr 2013 12:47:22 -0300 Subject: Make the key wrapper map an object property. --- src/leap/common/keymanager/__init__.py | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) (limited to 'src') diff --git a/src/leap/common/keymanager/__init__.py b/src/leap/common/keymanager/__init__.py index 10acb36..8296b92 100644 --- a/src/leap/common/keymanager/__init__.py +++ b/src/leap/common/keymanager/__init__.py @@ -40,11 +40,6 @@ from leap.common.keymanager.openpgp import ( ) -wrapper_map = { - OpenPGPKey: OpenPGPWrapper(), -} - - class KeyManager(object): def __init__(self, address, url): @@ -59,6 +54,9 @@ class KeyManager(object): """ self.address = address self.url = url + self.wrapper_map = { + OpenPGPKey: OpenPGPWrapper(), + } def send_key(self, ktype, send_private=False, password=None): """ @@ -99,13 +97,13 @@ class KeyManager(object): keyserver. """ try: - return wrapper_map[ktype].get_key(address) + return self.wrapper_map[ktype].get_key(address) except KeyNotFound: key = filter(lambda k: isinstance(k, ktype), self._fetch_keys(address)) if key is None: raise KeyNotFound() - wrapper_map[ktype].put_key(key) + self.wrapper_map[ktype].put_key(key) return key @@ -137,4 +135,4 @@ class KeyManager(object): @return: The generated key. @rtype: EncryptionKey """ - return wrapper_map[ktype].gen_key(self.address) + return self.wrapper_map[ktype].gen_key(self.address) -- cgit v1.2.3 From b833d9042da3a1650fde3354f38998a2e497672b Mon Sep 17 00:00:00 2001 From: drebs Date: Fri, 19 Apr 2013 21:48:57 -0300 Subject: Make keymanager OpenPGP wrapper store using Soledad. --- src/leap/common/keymanager/__init__.py | 29 ++-- src/leap/common/keymanager/errors.py | 1 - src/leap/common/keymanager/gpg.py | 1 - src/leap/common/keymanager/keys.py | 25 ++- src/leap/common/keymanager/openpgp.py | 282 ++++++++++++++++++++++++++----- src/leap/common/tests/test_keymanager.py | 26 ++- 6 files changed, 289 insertions(+), 75 deletions(-) (limited to 'src') diff --git a/src/leap/common/keymanager/__init__.py b/src/leap/common/keymanager/__init__.py index 8296b92..d197e4c 100644 --- a/src/leap/common/keymanager/__init__.py +++ b/src/leap/common/keymanager/__init__.py @@ -21,12 +21,6 @@ Key Manager is a Nicknym agent for LEAP client. """ -try: - import simplejson as json -except ImportError: - import json # noqa - - from u1db.errors import HTTPError @@ -42,20 +36,22 @@ from leap.common.keymanager.openpgp import ( class KeyManager(object): - def __init__(self, address, url): + def __init__(self, address, url, soledad): """ Initialize a Key Manager for user's C{address} with provider's nickserver reachable in C{url}. @param address: The address of the user of this Key Manager. @type address: str - @param url: The URL of the key manager. + @param url: The URL of the nickserver. @type url: str + @param soledad: A Soledad instance for local storage of keys. + @type soledad: leap.soledad.Soledad """ - self.address = address - self.url = url - self.wrapper_map = { - OpenPGPKey: OpenPGPWrapper(), + self._address = address + self._url = url + self._wrapper_map = { + OpenPGPKey: OpenPGPWrapper(soledad), } def send_key(self, ktype, send_private=False, password=None): @@ -97,16 +93,15 @@ class KeyManager(object): keyserver. """ try: - return self.wrapper_map[ktype].get_key(address) + return self._wrapper_map[ktype].get_key(address) except KeyNotFound: key = filter(lambda k: isinstance(k, ktype), self._fetch_keys(address)) if key is None: raise KeyNotFound() - self.wrapper_map[ktype].put_key(key) + self._wrapper_map[ktype].put_key(key) return key - def _fetch_keys(self, address): """ Fetch keys bound to C{address} from nickserver. @@ -119,11 +114,13 @@ class KeyManager(object): @raise KeyNotFound: If the key was not found on nickserver. @raise httplib.HTTPException: """ + raise NotImplementedError(self._fetch_keys) def refresh_keys(self): """ Update the user's db of validated keys to see if there are changes. """ + raise NotImplementedError(self.refresh_keys) def gen_key(self, ktype): """ @@ -135,4 +132,4 @@ class KeyManager(object): @return: The generated key. @rtype: EncryptionKey """ - return self.wrapper_map[ktype].gen_key(self.address) + return self._wrapper_map[ktype].gen_key(self._address) diff --git a/src/leap/common/keymanager/errors.py b/src/leap/common/keymanager/errors.py index f5bb1ab..4853869 100644 --- a/src/leap/common/keymanager/errors.py +++ b/src/leap/common/keymanager/errors.py @@ -16,7 +16,6 @@ # along with this program. If not, see . - class KeyNotFound(Exception): """ Raised when key was no found on keyserver. diff --git a/src/leap/common/keymanager/gpg.py b/src/leap/common/keymanager/gpg.py index dc5d791..5571ace 100644 --- a/src/leap/common/keymanager/gpg.py +++ b/src/leap/common/keymanager/gpg.py @@ -395,4 +395,3 @@ class GPGWrapper(gnupg.GPG): @rtype: bool """ self.is_encrypted_asym() or self.is_encrypted_sym() - diff --git a/src/leap/common/keymanager/keys.py b/src/leap/common/keymanager/keys.py index 13e3c0b..2e1ed89 100644 --- a/src/leap/common/keymanager/keys.py +++ b/src/leap/common/keymanager/keys.py @@ -21,6 +21,12 @@ Abstact key type and wrapper representations. """ +try: + import simplejson as json +except ImportError: + import json # noqa + + from abc import ABCMeta, abstractmethod @@ -44,13 +50,13 @@ class EncryptionKey(object): __metaclass__ = ABCMeta def __init__(self, address, key_id=None, fingerprint=None, - key_data=None, length=None, expiry_date=None, - validation=None, first_seen_at=None, - last_audited_at=None): + key_data=None, private=None, length=None, expiry_date=None, + validation=None, first_seen_at=None, last_audited_at=None): self.address = address self.key_id = key_id self.fingerprint = fingerprint self.key_data = key_data + self.private = private self.length = length self.expiry_date = expiry_date self.validation = validation @@ -66,10 +72,11 @@ class EncryptionKey(object): """ return json.dumps({ 'address': self.address, - 'type': str(self.__type__), + 'type': str(self.__class__), 'key_id': self.key_id, 'fingerprint': self.fingerprint, 'key_data': self.key_data, + 'private': self.private, 'length': self.length, 'expiry_date': self.expiry_date, 'validation': self.validation, @@ -92,6 +99,15 @@ class KeyTypeWrapper(object): __metaclass__ = ABCMeta + def __init__(self, soledad): + """ + Initialize the Key Type Wrapper. + + @param soledad: A Soledad instance for local storage of keys. + @type soledad: leap.soledad.Soledad + """ + self._soledad = soledad + @abstractmethod def get_key(self, address): """ @@ -124,4 +140,3 @@ class KeyTypeWrapper(object): @return: The key bound to C{address}. @rtype: EncryptionKey """ - diff --git a/src/leap/common/keymanager/openpgp.py b/src/leap/common/keymanager/openpgp.py index bb73089..1c51d94 100644 --- a/src/leap/common/keymanager/openpgp.py +++ b/src/leap/common/keymanager/openpgp.py @@ -22,7 +22,10 @@ Infrastructure for using OpenPGP keys in Key Manager. import re +import tempfile +import shutil +from hashlib import sha256 from leap.common.keymanager.errors import ( KeyNotFound, KeyAlreadyExists, @@ -34,6 +37,153 @@ from leap.common.keymanager.keys import ( from leap.common.keymanager.gpg import GPGWrapper +# +# Utility functions +# + +def _is_address(address): + """ + Return whether the given C{address} is in the form user@provider. + + @param address: The address to be tested. + @type address: str + @return: Whether C{address} is in the form user@provider. + @rtype: bool + """ + return bool(re.match('[\w.-]+@[\w.-]+', address)) + + +def _build_key_from_doc(address, doc): + """ + Build an OpenPGPKey for C{address} based on C{doc} from local storage. + + @param address: The address bound to the key. + @type address: str + @param doc: Document obtained from Soledad storage. + @type doc: leap.soledad.backends.leap_backend.LeapDocument + @return: The built key. + @rtype: OpenPGPKey + """ + return OpenPGPKey( + address, + key_id=doc.content['key_id'], + fingerprint=doc.content['fingerprint'], + key_data=doc.content['key_data'], + private=doc.content['private'], + length=doc.content['length'], + expiry_date=doc.content['expiry_date'], + validation=None, # TODO: verify for validation. + ) + + +def _build_key_from_gpg(address, key, key_data): + """ + Build an OpenPGPKey for C{address} based on C{key} from + local gpg storage. + + ASCII armored GPG key data has to be queried independently in this + wrapper, so we receive it in C{key_data}. + + @param address: The address bound to the key. + @type address: str + @param key: Key obtained from GPG storage. + @type key: dict + @param key_data: Key data obtained from GPG storage. + @type key_data: str + @return: The built key. + @rtype: OpenPGPKey + """ + return OpenPGPKey( + address, + key_id=key['keyid'], + fingerprint=key['fingerprint'], + key_data=key_data, + private=True if key['type'] == 'sec' else False, + length=key['length'], + expiry_date=key['expires'], + validation=None, # TODO: verify for validation. + ) + + +def _keymanager_doc_id(address, private=False): + """ + Return the document id for the document containing a key for + C{address}. + + @param address: The address bound to the key. + @type address: str + @param private: Whether the key is private or not. + @type private: bool + @return: The document id for the document that stores a key bound to + C{address}. + @rtype: str + """ + assert _is_address(address) + ktype = 'private' if private else 'public' + return sha256('key-manager-'+address+'-'+ktype).hexdigest() + + +def _build_unitary_gpgwrapper(key_data=None): + """ + Return a temporary GPG wrapper keyring containing exactly zero or one + keys. + + Temporary unitary keyrings allow the to use GPG's facilities for exactly + one key. This function creates an empty temporary keyring and imports + C{key_data} if it is not None. + + @param key_data: ASCII armored key data. + @type key_data: str + @return: A GPG wrapper with a unitary keyring. + @rtype: gnupg.GPG + """ + tmpdir = tempfile.mkdtemp() + gpg = GPGWrapper(gnupghome=tmpdir) + assert len(gpg.list_keys()) is 0 + if key_data: + gpg.import_keys(key_data) + assert len(gpg.list_keys()) is 1 + return gpg + + +def _destroy_unitary_gpgwrapper(gpg): + """ + Securely erase a unitary keyring. + + @param gpg: A GPG wrapper instance. + @type gpg: gnupg.GPG + """ + for secret in [True, False]: + for key in gpg.list_keys(secret=secret): + gpg.delete_keys( + key['fingerprint'], + secret=secret) + assert len(gpg.list_keys()) == 0 + # TODO: implement some kind of wiping of data or a more secure way that + # does not write to disk. + shutil.rmtree(gpg.gnupghome) + + +def _safe_call(callback, key_data=None, **kwargs): + """ + Run C{callback} in an unitary keyring containing C{key_data}. + + @param callback: Function whose first argument is the gpg keyring. + @type callback: function(gnupg.GPG) + @param key_data: ASCII armored key data. + @type key_data: str + @param **kwargs: Other eventual parameters for the callback. + @type **kwargs: **dict + """ + gpg = _build_unitary_gpgwrapper(key_data) + callback(gpg, **kwargs) + _destroy_unitary_gpgwrapper(gpg) + + +# +# The OpenPGP wrapper +# + class OpenPGPKey(EncryptionKey): """ Base class for OpenPGP keys. @@ -45,33 +195,19 @@ class OpenPGPWrapper(KeyTypeWrapper): A wrapper for OpenPGP keys. """ - def __init__(self, gnupghome=None): - self._gpg = GPGWrapper(gnupghome=gnupghome) - - def _build_key(self, address, result): + def __init__(self, soledad): """ - Build an OpenPGPWrapper key for C{address} based on C{result} from - local storage. + Initialize the OpenPGP wrapper. - @param address: The address bound to the key. - @type address: str - @param result: Result obtained from GPG storage. - @type result: dict + @param soledad: A Soledad instance for key storage. + @type soledad: leap.soledad.Soledad """ - key_data = self._gpg.export_keys(result['fingerprint'], secret=False) - return OpenPGPKey( - address, - key_id=result['keyid'], - fingerprint=result['fingerprint'], - key_data=key_data, - length=result['length'], - expiry_date=result['expires'], - validation=None, # TODO: verify for validation. - ) + KeyTypeWrapper.__init__(self, soledad) + self._soledad = soledad def gen_key(self, address): """ - Generate an OpenPGP keypair for C{address}. + Generate an OpenPGP keypair bound to C{address}. @param address: The address bound to the key. @type address: str @@ -79,21 +215,36 @@ class OpenPGPWrapper(KeyTypeWrapper): @rtype: OpenPGPKey @raise KeyAlreadyExists: If key already exists in local database. """ + # make sure the key does not already exist + assert _is_address(address) try: self.get_key(address) - raise KeyAlreadyExists() + raise KeyAlreadyExists(address) except KeyNotFound: pass - params = self._gpg.gen_key_input( - key_type='RSA', - key_length=4096, - name_real=address, - name_email=address, - name_comment='Generated by LEAP Key Manager.') - self._gpg.gen_key(params) - return self.get_key(address) - - def get_key(self, address): + + def _gen_key_cb(gpg): + params = gpg.gen_key_input( + key_type='RSA', + key_length=4096, + name_real=address, + name_email=address, + name_comment='Generated by LEAP Key Manager.') + gpg.gen_key(params) + assert len(gpg.list_keys()) is 1 # a unitary keyring! + key = gpg.list_keys(secret=True).pop() + assert len(key['uids']) is 1 # with just one uid! + # assert for correct address + assert re.match('.*<%s>$' % address, key['uids'][0]) is not None + openpgp_key = _build_key_from_gpg( + address, key, + gpg.export_keys(key['fingerprint'])) + self.put_key(openpgp_key) + + _safe_call(_gen_key_cb) + return self.get_key(address, private=True) + + def get_key(self, address, private=False): """ Get key bound to C{address} from local storage. @@ -104,23 +255,62 @@ class OpenPGPWrapper(KeyTypeWrapper): @rtype: OpenPGPKey @raise KeyNotFound: If the key was not found on local storage. """ - m = re.compile('.*<%s>$' % address) - keys = self._gpg.list_keys(secret=False) + assert _is_address(address) + doc = self._get_key_doc(address, private) + if doc is None: + raise KeyNotFound(address) + return _build_key_from_doc(address, doc) - def bound_to_address(key): - return bool(filter(lambda u: m.match(u), key['uids'])) + def put_key_raw(self, data): + """ + Put key contained in raw C{data} in local storage. - try: - bound_key = filter(bound_to_address, keys).pop() - return self._build_key(address, bound_key) - except IndexError: - raise KeyNotFound(address) + @param data: The key data to be stored. + @type data: str + """ + assert data is not None + + def _put_key_raw_cb(gpg): + + key = gpg.list_keys(secret=False).pop() # unitary keyring + # extract adress from first uid on key + match = re.match('.*<([\w.-]+@[\w.-]+)>.*', key['uids'].pop()) + assert match is not None + address = match.group(1) + openpgp_key = _build_key_from_gpg( + address, key, + gpg.export_keys(key['fingerprint'])) + self.put_key(openpgp_key) - def put_key(self, data): + _safe_call(_put_key_raw_cb, data) + + def put_key(self, key): + """ + Put C{key} in local storage. + + @param key: The key to be stored. + @type key: OpenPGPKey + """ + doc = self._get_key_doc(key.address, private=key.private) + if doc is None: + self._soledad.create_doc_from_json( + key.get_json(), + doc_id=_keymanager_doc_id(key.address, key.private)) + else: + doc.set_json(key.get_json()) + self._soledad.put_doc(doc) + + def _get_key_doc(self, address, private=False): """ - Put key contained in {data} in local storage. + Get the document with a key (public, by default) bound to C{address}. + + If C{private} is True, looks for a private key instead of a public. - @param key: The key data to be stored. - @type key: str + @param address: The address bound to the key. + @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.backends.leap_backend.LeapDocument """ - self._gpg.import_keys(data) + return self._soledad.get_doc(_keymanager_doc_id(address, private)) diff --git a/src/leap/common/tests/test_keymanager.py b/src/leap/common/tests/test_keymanager.py index 4189aac..23d702b 100644 --- a/src/leap/common/tests/test_keymanager.py +++ b/src/leap/common/tests/test_keymanager.py @@ -26,12 +26,26 @@ import unittest from leap.common.testing.basetest import BaseLeapTest from leap.common.keymanager import KeyManager, openpgp, KeyNotFound - +from leap.soledad import Soledad +from leap.common.keymanager.gpg import GPGWrapper class KeyManagerTestCase(BaseLeapTest): def setUp(self): - pass + self._soledad = Soledad( + "leap@leap.se", + "123456", + gnupg_home=self.tempdir+"/gnupg", + secret_path=self.tempdir+"/secret.gpg", + local_db_path=self.tempdir+"/soledad.u1db", + bootstrap=False, + ) + # initialize solead by hand for testing purposes + self._soledad._init_dirs() + self._soledad._gpg = GPGWrapper(gnupghome=self.tempdir+"/gnupg") + self._soledad._shared_db = None + self._soledad._init_keys() + self._soledad._init_db() def tearDown(self): pass @@ -40,7 +54,7 @@ class KeyManagerTestCase(BaseLeapTest): return KeyManager(user, url) def test_openpgp_gen_key(self): - pgp = openpgp.OpenPGPWrapper(self.tempdir+'/gnupg') + pgp = openpgp.OpenPGPWrapper(self._soledad) try: pgp.get_key('user@leap.se') except KeyNotFound: @@ -51,12 +65,12 @@ class KeyManagerTestCase(BaseLeapTest): self.assertEqual( '4096', key.length, 'Wrong key length.') - def test_openpgp_put_key(self): - pgp = openpgp.OpenPGPWrapper(self.tempdir+'/gnupg2') + def test_openpgp_put_key_raw(self): + pgp = openpgp.OpenPGPWrapper(self._soledad) try: pgp.get_key('leap@leap.se') except KeyNotFound: - pgp.put_key(PUBLIC_KEY) + pgp.put_key_raw(PUBLIC_KEY) key = pgp.get_key('leap@leap.se') self.assertIsInstance(key, openpgp.OpenPGPKey) self.assertEqual( -- cgit v1.2.3 From b3ad976ec8aa64a00cc824dc57aa2135ab41deb6 Mon Sep 17 00:00:00 2001 From: drebs Date: Mon, 22 Apr 2013 10:39:58 -0300 Subject: Add send_keys() and refresh_keys() to Key Manager. --- src/leap/common/keymanager/__init__.py | 54 +++++++++++++-- src/leap/common/keymanager/errors.py | 5 ++ src/leap/common/keymanager/http.py | 78 +++++++++++++++++++++ src/leap/common/keymanager/keys.py | 4 +- src/leap/common/keymanager/openpgp.py | 112 ++++++++++++++----------------- src/leap/common/keymanager/util.py | 97 ++++++++++++++++++++++++++ src/leap/common/tests/test_keymanager.py | 81 +++++++++++++++++++++- 7 files changed, 362 insertions(+), 69 deletions(-) create mode 100644 src/leap/common/keymanager/http.py create mode 100644 src/leap/common/keymanager/util.py (limited to 'src') diff --git a/src/leap/common/keymanager/__init__.py b/src/leap/common/keymanager/__init__.py index d197e4c..a195724 100644 --- a/src/leap/common/keymanager/__init__.py +++ b/src/leap/common/keymanager/__init__.py @@ -20,10 +20,13 @@ Key Manager is a Nicknym agent for LEAP client. """ +import httplib + from u1db.errors import HTTPError +from leap.common.check import leap_assert from leap.common.keymanager.errors import ( KeyNotFound, KeyAlreadyExists, @@ -31,7 +34,9 @@ from leap.common.keymanager.errors import ( from leap.common.keymanager.openpgp import ( OpenPGPKey, OpenPGPWrapper, + _encrypt_symmetric, ) +from leap.common.keymanager.http import HTTPClient class KeyManager(object): @@ -49,9 +54,10 @@ class KeyManager(object): @type soledad: leap.soledad.Soledad """ self._address = address - self._url = url + self._http_client = HTTPClient(url) self._wrapper_map = { OpenPGPKey: OpenPGPWrapper(soledad), + # other types of key will be added to this mapper. } def send_key(self, ktype, send_private=False, password=None): @@ -73,9 +79,32 @@ class KeyManager(object): @type ktype: KeyType @raise httplib.HTTPException: + @raise KeyNotFound: If the key was not found both locally and in + keyserver. """ - - def get_key(self, address, ktype): + # prepare the public key bound to address + data = { + 'address': self._address, + 'keys': [ + json.loads( + self.get_key( + self._address, ktype, private=False).get_json()), + ] + } + # prepare the private key bound to address + if send_private: + privkey = json.loads( + self.get_key(self._address, ktype, private=True).get_json()) + privkey.key_data = _encrypt_symmetric(data, passphrase) + data['keys'].append(privkey) + headers = None # TODO: replace for token-based-auth + self._http_client.request( + 'PUT', + '/key/%s' % address, + json.dumps(data), + headers) + + def get_key(self, address, ktype, private=False): """ Return a key of type C{ktype} bound to C{address}. @@ -86,14 +115,19 @@ class KeyManager(object): @type address: str @param ktype: The type of the key. @type ktype: KeyType + @param private: Look for a private key instead of a public one? + @type private: 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. """ + leap_assert( + ktype in self._wrapper_map, + 'Unkown key type: %s.' % str(ktype)) try: - return self._wrapper_map[ktype].get_key(address) + return self._wrapper_map[ktype].get_key(address, private=private) except KeyNotFound: key = filter(lambda k: isinstance(k, ktype), self._fetch_keys(address)) @@ -114,7 +148,17 @@ class KeyManager(object): @raise KeyNotFound: If the key was not found on nickserver. @raise httplib.HTTPException: """ - raise NotImplementedError(self._fetch_keys) + self._http_client.request('GET', '/key/%s' % address, None, None) + keydata = json.loads(self._http_client.read_response()) + leap_assert( + keydata['address'] == address, + "Fetched key for wrong address.") + for key in keydata['keys']: + # find the key class in the mapper + keyCLass = filter( + lambda klass: str(klass) == key['type'], + self._wrapper_map).pop() + yield _build_key_from_dict(kClass, address, key) def refresh_keys(self): """ diff --git a/src/leap/common/keymanager/errors.py b/src/leap/common/keymanager/errors.py index 4853869..886c666 100644 --- a/src/leap/common/keymanager/errors.py +++ b/src/leap/common/keymanager/errors.py @@ -16,6 +16,11 @@ # along with this program. If not, see . +""" +Errors and exceptions used by the Key Manager. +""" + + class KeyNotFound(Exception): """ Raised when key was no found on keyserver. diff --git a/src/leap/common/keymanager/http.py b/src/leap/common/keymanager/http.py new file mode 100644 index 0000000..478137d --- /dev/null +++ b/src/leap/common/keymanager/http.py @@ -0,0 +1,78 @@ +# -*- coding: utf-8 -*- +# http.py +# Copyright (C) 2013 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 utilities. +""" + + +import urlparse +import httplib + + +def HTTPClient(object): + """ + A simple HTTP client for making requests. + """ + + def __init__(self, url): + """ + Initialize the HTTP client. + """ + self._url = urlparse.urlsplit(url) + self._conn = None + + def _ensure_connection(self): + """ + Ensure the creation of the connection object. + """ + if self._conn is not None: + return + if self._url.scheme == 'https': + connClass = httplib.HTTPSConnection + else: + connClass = httplib.HTTPConnection + self._conn = connClass(self._url.hostname, self._url.port) + + def request(method, url_query, body, headers): + """ + Make an HTTP request. + + @param method: The method of the request. + @type method: str + @param url_query: The URL query string of the request. + @type url_query: str + @param body: The body of the request. + @type body: str + @param headers: Headers to be sent on the request. + @type headers: list of str + """ + self._ensure_connection() + return self._conn.request(mthod, url_query, body, headers) + + def response(self): + """ + Return the response of an HTTP request. + """ + return self._conn.getresponse() + + def read_response(self): + """ + Get the contents of a response for an HTTP request. + """ + return self.response().read() diff --git a/src/leap/common/keymanager/keys.py b/src/leap/common/keymanager/keys.py index 2e1ed89..bed407c 100644 --- a/src/leap/common/keymanager/keys.py +++ b/src/leap/common/keymanager/keys.py @@ -109,12 +109,14 @@ class KeyTypeWrapper(object): self._soledad = soledad @abstractmethod - def get_key(self, address): + def get_key(self, address, private=False): """ Get key from local storage. @param address: The address bound to the key. @type address: str + @param private: Look for a private key instead of a public one? + @type private: bool @return: The key bound to C{address}. @rtype: EncryptionKey diff --git a/src/leap/common/keymanager/openpgp.py b/src/leap/common/keymanager/openpgp.py index 1c51d94..cd37138 100644 --- a/src/leap/common/keymanager/openpgp.py +++ b/src/leap/common/keymanager/openpgp.py @@ -1,5 +1,5 @@ # -*- coding: utf-8 -*- -# openpgpwrapper.py +# openpgp.py # Copyright (C) 2013 LEAP # # This program is free software: you can redistribute it and/or modify @@ -25,7 +25,7 @@ import re import tempfile import shutil -from hashlib import sha256 +from leap.common.check import leap_assert from leap.common.keymanager.errors import ( KeyNotFound, KeyAlreadyExists, @@ -35,45 +35,40 @@ from leap.common.keymanager.keys import ( KeyTypeWrapper, ) from leap.common.keymanager.gpg import GPGWrapper +from leap.common.keymanager.util import ( + _is_address, + _build_key_from_doc, + _keymanager_doc_id, +) # # Utility functions # -def _is_address(address): - """ - Return whether the given C{address} is in the form user@provider. - - @param address: The address to be tested. - @type address: str - @return: Whether C{address} is in the form user@provider. - @rtype: bool +def _encrypt_symmetric(data, password): """ - return bool(re.match('[\w.-]+@[\w.-]+', address)) + Encrypt C{data} with C{password}. + This function uses the OpenPGP wrapper to perform the encryption. -def _build_key_from_doc(address, doc): + @param data: The data to be encrypted. + @type data: str + @param password: The password used to encrypt C{data}. + @type password: str + @return: The encrypted data. + @rtype: str """ - Build an OpenPGPKey for C{address} based on C{doc} from local storage. + cyphertext = None - @param address: The address bound to the key. - @type address: str - @param doc: Document obtained from Soledad storage. - @type doc: leap.soledad.backends.leap_backend.LeapDocument - @return: The built key. - @rtype: OpenPGPKey - """ - return OpenPGPKey( - address, - key_id=doc.content['key_id'], - fingerprint=doc.content['fingerprint'], - key_data=doc.content['key_data'], - private=doc.content['private'], - length=doc.content['length'], - expiry_date=doc.content['expiry_date'], - validation=None, # TODO: verify for validation. - ) + def _encrypt_cb(gpg): + cyphertext = str( + gpg.encrypt( + data, None, passphrase=password, symmetric=True)) + data['keys'].append(privkey) + + _safe_call(_encrypt_cb) + return cyphertext def _build_key_from_gpg(address, key, key_data): @@ -90,7 +85,7 @@ def _build_key_from_gpg(address, key, key_data): @type key: dict @param key_data: Key data obtained from GPG storage. @type key_data: str - @return: The built key. + @return: An instance of the key. @rtype: OpenPGPKey """ return OpenPGPKey( @@ -105,24 +100,6 @@ def _build_key_from_gpg(address, key, key_data): ) -def _keymanager_doc_id(address, private=False): - """ - Return the document id for the document containing a key for - C{address}. - - @param address: The address bound to the key. - @type address: str - @param private: Whether the key is private or not. - @type private: bool - @return: The document id for the document that stores a key bound to - C{address}. - @rtype: str - """ - assert _is_address(address) - ktype = 'private' if private else 'public' - return sha256('key-manager-'+address+'-'+ktype).hexdigest() - - def _build_unitary_gpgwrapper(key_data=None): """ Return a temporary GPG wrapper keyring containing exactly zero or one @@ -139,10 +116,13 @@ def _build_unitary_gpgwrapper(key_data=None): """ tmpdir = tempfile.mkdtemp() gpg = GPGWrapper(gnupghome=tmpdir) - assert len(gpg.list_keys()) is 0 + leap_assert(len(gpg.list_keys()) is 0, 'Keyring not empty.') if key_data: gpg.import_keys(key_data) - assert len(gpg.list_keys()) is 1 + leap_assert( + len(gpg.list_keys()) is 1, + 'Unitary keyring has wrong number of keys: %d.' + % len(gpg.list_keys())) return gpg @@ -158,7 +138,7 @@ def _destroy_unitary_gpgwrapper(gpg): gpg.delete_keys( key['fingerprint'], secret=secret) - assert len(gpg.list_keys()) == 0 + leap_assert(len(gpg.list_keys()) is 0, 'Keyring not empty!') # TODO: implement some kind of wiping of data or a more secure way that # does not write to disk. shutil.rmtree(gpg.gnupghome) @@ -216,7 +196,7 @@ class OpenPGPWrapper(KeyTypeWrapper): @raise KeyAlreadyExists: If key already exists in local database. """ # make sure the key does not already exist - assert _is_address(address) + leap_assert(_is_address(address), 'Not an user address: %s' % address) try: self.get_key(address) raise KeyAlreadyExists(address) @@ -231,11 +211,18 @@ class OpenPGPWrapper(KeyTypeWrapper): name_email=address, name_comment='Generated by LEAP Key Manager.') gpg.gen_key(params) - assert len(gpg.list_keys()) is 1 # a unitary keyring! + pubkeys = gpg.list_keys() + # assert for new key characteristics + 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() - assert len(key['uids']) is 1 # with just one uid! - # assert for correct address - assert re.match('.*<%s>$' % address, key['uids'][0]) is not None + leap_assert( + len(key['uids']) is 1, # with just one uid! + 'Wrong number of uids for key: %d.' % len(key['uids'])) + leap_assert( + re.match('.*<%s>$' % address, key['uids'][0]) is not None, + 'Key not correctly bound to address.') openpgp_key = _build_key_from_gpg( address, key, gpg.export_keys(key['fingerprint'])) @@ -250,16 +237,18 @@ class OpenPGPWrapper(KeyTypeWrapper): @param address: The address bound to the key. @type address: str + @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. """ - assert _is_address(address) + leap_assert(_is_address(address), 'Not an user address: %s' % address) doc = self._get_key_doc(address, private) if doc is None: raise KeyNotFound(address) - return _build_key_from_doc(address, doc) + return _build_key_from_doc(OpenPGPKey, address, doc) def put_key_raw(self, data): """ @@ -268,14 +257,15 @@ class OpenPGPWrapper(KeyTypeWrapper): @param data: The key data to be stored. @type data: str """ - assert data is not None + # TODO: add more checks for correct key data. + leap_assert(data is not None, 'Data does not represent a key.') def _put_key_raw_cb(gpg): key = gpg.list_keys(secret=False).pop() # unitary keyring # extract adress from first uid on key match = re.match('.*<([\w.-]+@[\w.-]+)>.*', key['uids'].pop()) - assert match is not None + leap_assert(match is not None, 'No user address in key data.') address = match.group(1) openpgp_key = _build_key_from_gpg( address, key, diff --git a/src/leap/common/keymanager/util.py b/src/leap/common/keymanager/util.py new file mode 100644 index 0000000..42168c8 --- /dev/null +++ b/src/leap/common/keymanager/util.py @@ -0,0 +1,97 @@ +# -*- coding: utf-8 -*- +# util.py +# Copyright (C) 2013 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 . + + +""" +Utilities for the Key Manager. +""" + + +import re + + +from hashlib import sha256 +from leap.common.check import leap_assert + + +def _is_address(address): + """ + Return whether the given C{address} is in the form user@provider. + + @param address: The address to be tested. + @type address: str + @return: Whether C{address} is in the form user@provider. + @rtype: bool + """ + return bool(re.match('[\w.-]+@[\w.-]+', address)) + + +def _build_key_from_dict(kClass, address, kdict): + """ + Build an C{kClass} key bound to C{address} based on info in C{kdict}. + + @param address: The address bound to the key. + @type address: str + @param kdict: Dictionary with key data. + @type kdict: dict + @return: An instance of the key. + @rtype: C{kClass} + """ + return kClass( + address, + key_id=kdict['key_id'], + fingerprint=kdict['fingerprint'], + key_data=kdict['key_data'], + private=kdict['private'], + length=kdict['length'], + expiry_date=kdict['expiry_date'], + first_seen_at=kdict['first_seen_at'], + last_audited_at=kdict['last_audited_at'], + validation=kdict['validation'], # TODO: verify for validation. + ) + + +def _build_key_from_doc(kClass, address, doc): + """ + Build an C{kClass} for C{address} based on C{doc} from local storage. + + @param address: The address bound to the key. + @type address: str + @param doc: Document obtained from Soledad storage. + @type doc: leap.soledad.backends.leap_backend.LeapDocument + @return: An instance of the key. + @rtype: C{kClass} + """ + return _build_key_from_dict(kClass, address, doc.content) + + +def _keymanager_doc_id(address, private=False): + """ + Return the document id for the document containing a key for + C{address}. + + @param address: The address bound to the key. + @type address: str + @param private: Whether the key is private or not. + @type private: bool + @return: The document id for the document that stores a key bound to + C{address}. + @rtype: str + """ + leap_assert(_is_address(address), "Wrong address format: %s" % address) + ktype = 'private' if private else 'public' + return sha256('key-manager-'+address+'-'+ktype).hexdigest() diff --git a/src/leap/common/tests/test_keymanager.py b/src/leap/common/tests/test_keymanager.py index 23d702b..4a2693e 100644 --- a/src/leap/common/tests/test_keymanager.py +++ b/src/leap/common/tests/test_keymanager.py @@ -25,11 +25,88 @@ import unittest from leap.common.testing.basetest import BaseLeapTest -from leap.common.keymanager import KeyManager, openpgp, KeyNotFound from leap.soledad import Soledad +from leap.common.keymanager import KeyManager, openpgp, KeyNotFound +from leap.common.keymanager.openpgp import OpenPGPKey from leap.common.keymanager.gpg import GPGWrapper +from leap.common.keymanager.util import ( + _is_address, + _build_key_from_dict, + _keymanager_doc_id, +) + + +class KeyManagerUtilTestCase(BaseLeapTest): + + def setUp(self): + pass + + def tearDown(self): + pass + + def test__is_address(self): + self.assertTrue( + _is_address('user@leap.se'), + 'Incorrect address detection.') + self.assertFalse( + _is_address('userleap.se'), + 'Incorrect address detection.') + self.assertFalse( + _is_address('user@'), + 'Incorrect address detection.') + self.assertFalse( + _is_address('@leap.se'), + 'Incorrect address detection.') + + def test__build_key_from_dict(self): + kdict = { + 'address': 'leap@leap.se', + 'key_id': 'key_id', + 'fingerprint': 'fingerprint', + 'key_data': 'key_data', + 'private': 'private', + 'length': 'length', + 'expiry_date': 'expiry_date', + 'first_seen_at': 'first_seen_at', + 'last_audited_at': 'last_audited_at', + 'validation': 'validation', + } + key = _build_key_from_dict(OpenPGPKey, 'leap@leap.se', kdict) + self.assertEqual(kdict['address'], key.address, + 'Wrong data in key.') + self.assertEqual(kdict['key_id'], key.key_id, + 'Wrong data in key.') + self.assertEqual(kdict['fingerprint'], key.fingerprint, + 'Wrong data in key.') + self.assertEqual(kdict['key_data'], key.key_data, + 'Wrong data in key.') + self.assertEqual(kdict['private'], key.private, + 'Wrong data in key.') + self.assertEqual(kdict['length'], key.length, + 'Wrong data in key.') + self.assertEqual(kdict['expiry_date'], key.expiry_date, + 'Wrong data in key.') + self.assertEqual(kdict['first_seen_at'], key.first_seen_at, + 'Wrong data in key.') + self.assertEqual(kdict['last_audited_at'], key.last_audited_at, + 'Wrong data in key.') + self.assertEqual(kdict['validation'], key.validation, + 'Wrong data in key.') + + def test__keymanager_doc_id(self): + doc_id1 = _keymanager_doc_id('leap@leap.se', private=False) + doc_id2 = _keymanager_doc_id('leap@leap.se', private=True) + doc_id3 = _keymanager_doc_id('user@leap.se', private=False) + doc_id4 = _keymanager_doc_id('user@leap.se', private=True) + self.assertFalse(doc_id1 == doc_id2, 'Doc ids are equal!') + self.assertFalse(doc_id1 == doc_id3, 'Doc ids are equal!') + self.assertFalse(doc_id1 == doc_id4, 'Doc ids are equal!') + self.assertFalse(doc_id2 == doc_id3, 'Doc ids are equal!') + self.assertFalse(doc_id2 == doc_id4, 'Doc ids are equal!') + self.assertFalse(doc_id3 == doc_id4, 'Doc ids are equal!') + -class KeyManagerTestCase(BaseLeapTest): +class KeyManagerCryptoTestCase(BaseLeapTest): def setUp(self): self._soledad = Soledad( -- cgit v1.2.3 From 62b5a7798924188ba915a1c095917d8709e20ae7 Mon Sep 17 00:00:00 2001 From: drebs Date: Tue, 23 Apr 2013 20:50:02 -0300 Subject: Refactor, fixes, add api, tests. * Change KeyTypeWrapper to EncryptionScheme * Change OpenPGPWrapper to OpenPGPScheme * Add missing and standardized crypto API. * Add delete_key() * Fix put_key raw so it puts either public or private keys. * Fix gpg's is_encrypted() * Fix openpgp's safe callbacks so they return correctly. * Remove binascii because it generates invalid doc ids. * Add tests. --- src/leap/common/keymanager/__init__.py | 8 +- src/leap/common/keymanager/errors.py | 7 + src/leap/common/keymanager/gpg.py | 2 +- src/leap/common/keymanager/keys.py | 25 +++- src/leap/common/keymanager/openpgp.py | 212 +++++++++++++++++++++++++++---- src/leap/common/keymanager/util.py | 12 +- src/leap/common/tests/test_keymanager.py | 148 +++++++++++++++------ 7 files changed, 333 insertions(+), 81 deletions(-) (limited to 'src') diff --git a/src/leap/common/keymanager/__init__.py b/src/leap/common/keymanager/__init__.py index a195724..f939a4e 100644 --- a/src/leap/common/keymanager/__init__.py +++ b/src/leap/common/keymanager/__init__.py @@ -33,8 +33,8 @@ from leap.common.keymanager.errors import ( ) from leap.common.keymanager.openpgp import ( OpenPGPKey, - OpenPGPWrapper, - _encrypt_symmetric, + OpenPGPScheme, + encrypt_sym, ) from leap.common.keymanager.http import HTTPClient @@ -56,7 +56,7 @@ class KeyManager(object): self._address = address self._http_client = HTTPClient(url) self._wrapper_map = { - OpenPGPKey: OpenPGPWrapper(soledad), + OpenPGPKey: OpenPGPScheme(soledad), # other types of key will be added to this mapper. } @@ -95,7 +95,7 @@ class KeyManager(object): if send_private: privkey = json.loads( self.get_key(self._address, ktype, private=True).get_json()) - privkey.key_data = _encrypt_symmetric(data, passphrase) + privkey.key_data = encrypt_sym(data, passphrase) data['keys'].append(privkey) headers = None # TODO: replace for token-based-auth self._http_client.request( diff --git a/src/leap/common/keymanager/errors.py b/src/leap/common/keymanager/errors.py index 886c666..add6a38 100644 --- a/src/leap/common/keymanager/errors.py +++ b/src/leap/common/keymanager/errors.py @@ -31,3 +31,10 @@ class KeyAlreadyExists(Exception): """ Raised when attempted to create a key that already exists. """ + + +class KeyAttributesDiffer(Exception): + """ + Raised when trying to delete a key but the stored key differs from the key + passed to the delete_key() method. + """ diff --git a/src/leap/common/keymanager/gpg.py b/src/leap/common/keymanager/gpg.py index 5571ace..f3e6453 100644 --- a/src/leap/common/keymanager/gpg.py +++ b/src/leap/common/keymanager/gpg.py @@ -394,4 +394,4 @@ class GPGWrapper(gnupg.GPG): @return: Whether data is encrypted to a key. @rtype: bool """ - self.is_encrypted_asym() or self.is_encrypted_sym() + return self.is_encrypted_asym(data) or self.is_encrypted_sym(data) diff --git a/src/leap/common/keymanager/keys.py b/src/leap/common/keymanager/keys.py index bed407c..250c2fa 100644 --- a/src/leap/common/keymanager/keys.py +++ b/src/leap/common/keymanager/keys.py @@ -17,7 +17,7 @@ """ -Abstact key type and wrapper representations. +Abstact key type and encryption scheme representations. """ @@ -86,22 +86,23 @@ class EncryptionKey(object): # -# Key wrappers +# Encryption schemes # -class KeyTypeWrapper(object): +class EncryptionScheme(object): """ - Abstract class for Key Type Wrappers. + Abstract class for Encryption Schemes. - A wrapper for a certain key type should know how to get and put keys in - local storage using Soledad and also how to generate new keys. + A wrapper for a certain encryption schemes should know how to get and put + keys in local storage using Soledad, how to generate new keys and how to + find out about possibly encrypted content. """ __metaclass__ = ABCMeta def __init__(self, soledad): """ - Initialize the Key Type Wrapper. + Initialize this Encryption Scheme. @param soledad: A Soledad instance for local storage of keys. @type soledad: leap.soledad.Soledad @@ -139,6 +140,16 @@ class KeyTypeWrapper(object): @param address: The address bound to the key. @type address: str + @return: The key bound to C{address}. @rtype: EncryptionKey """ + + @abstractmethod + def delete_key(self, key): + """ + Remove C{key} from storage. + + @param key: The key to be removed. + @type key: EncryptionKey + """ diff --git a/src/leap/common/keymanager/openpgp.py b/src/leap/common/keymanager/openpgp.py index cd37138..ace8c1e 100644 --- a/src/leap/common/keymanager/openpgp.py +++ b/src/leap/common/keymanager/openpgp.py @@ -32,7 +32,7 @@ from leap.common.keymanager.errors import ( ) from leap.common.keymanager.keys import ( EncryptionKey, - KeyTypeWrapper, + EncryptionScheme, ) from leap.common.keymanager.gpg import GPGWrapper from leap.common.keymanager.util import ( @@ -46,29 +46,137 @@ from leap.common.keymanager.util import ( # Utility functions # -def _encrypt_symmetric(data, password): +def encrypt_sym(data, passphrase): """ - Encrypt C{data} with C{password}. + Encrypt C{data} with C{passphrase}. - This function uses the OpenPGP wrapper to perform the encryption. + @param data: The data to be encrypted. + @type data: str + @param passphrase: The passphrase used to encrypt C{data}. + @type passphrase: str + + @return: The encrypted data. + @rtype: str + """ + + def _encrypt_cb(gpg): + return str( + gpg.encrypt( + data, None, passphrase=passphrase, symmetric=True)) + + return _safe_call(_encrypt_cb) + + +def decrypt_sym(data, passphrase): + """ + Decrypt C{data} with C{passphrase}. + + @param data: The data to be decrypted. + @type data: str + @param passphrase: The passphrase used to decrypt C{data}. + @type passphrase: str + + @return: The decrypted data. + @rtype: str + """ + + def _decrypt_cb(gpg): + return str(gpg.decrypt(data, passphrase=passphrase)) + + return _safe_call(_decrypt_cb) + + +def encrypt_asym(data, key): + """ + Encrypt C{data} using public @{key}. @param data: The data to be encrypted. @type data: str - @param password: The password used to encrypt C{data}. - @type password: str + @param key: The key used to encrypt. + @type key: OpenPGPKey + @return: The encrypted data. @rtype: str """ - cyphertext = None + leap_assert(key.private is False, 'Key is not public.') def _encrypt_cb(gpg): - cyphertext = str( + return str( gpg.encrypt( - data, None, passphrase=password, symmetric=True)) - data['keys'].append(privkey) + data, key.fingerprint, symmetric=False)) + + return _safe_call(_encrypt_cb, key.key_data) + + +def decrypt_asym(data, key): + """ + Decrypt C{data} using private @{key}. + + @param data: The data to be decrypted. + @type data: str + @param key: The key used to decrypt. + @type key: OpenPGPKey + + @return: The decrypted data. + @rtype: str + """ + leap_assert(key.private is True, 'Key is not private.') - _safe_call(_encrypt_cb) - return cyphertext + def _decrypt_cb(gpg): + return str(gpg.decrypt(data)) + + return _safe_call(_decrypt_cb, key.key_data) + + +def is_encrypted(data): + """ + Return whether C{data} was encrypted using OpenPGP. + + @param data: The data we want to know about. + @type data: str + + @return: Whether C{data} was encrypted using this wrapper. + @rtype: bool + """ + + def _is_encrypted_cb(gpg): + return gpg.is_encrypted(data) + + return _safe_call(_is_encrypted_cb) + + +def is_encrypted_sym(data): + """ + Return whether C{data} was encrypted using a public OpenPGP key. + + @param data: The data we want to know about. + @type data: str + + @return: Whether C{data} was encrypted using this wrapper. + @rtype: bool + """ + + def _is_encrypted_cb(gpg): + return gpg.is_encrypted_sym(data) + + return _safe_call(_is_encrypted_cb) + + +def is_encrypted_asym(data): + """ + Return whether C{data} was asymmetrically encrypted using OpenPGP. + + @param data: The data we want to know about. + @type data: str + + @return: Whether C{data} was encrypted using this wrapper. + @rtype: bool + """ + + def _is_encrypted_cb(gpg): + return gpg.is_encrypted_asym(data) + + return _safe_call(_is_encrypted_cb) def _build_key_from_gpg(address, key, key_data): @@ -154,10 +262,14 @@ def _safe_call(callback, key_data=None, **kwargs): @type key_data: str @param **kwargs: Other eventual parameters for the callback. @type **kwargs: **dict + + @return: The results of the callback. + @rtype: str or bool """ gpg = _build_unitary_gpgwrapper(key_data) - callback(gpg, **kwargs) + val = callback(gpg, **kwargs) _destroy_unitary_gpgwrapper(gpg) + return val # @@ -170,7 +282,7 @@ class OpenPGPKey(EncryptionKey): """ -class OpenPGPWrapper(KeyTypeWrapper): +class OpenPGPScheme(EncryptionScheme): """ A wrapper for OpenPGP keys. """ @@ -182,8 +294,7 @@ class OpenPGPWrapper(KeyTypeWrapper): @param soledad: A Soledad instance for key storage. @type soledad: leap.soledad.Soledad """ - KeyTypeWrapper.__init__(self, soledad) - self._soledad = soledad + EncryptionScheme.__init__(self, soledad) def gen_key(self, address): """ @@ -223,10 +334,13 @@ class OpenPGPWrapper(KeyTypeWrapper): leap_assert( re.match('.*<%s>$' % address, key['uids'][0]) is not None, 'Key not correctly bound to address.') - openpgp_key = _build_key_from_gpg( - address, key, - gpg.export_keys(key['fingerprint'])) - self.put_key(openpgp_key) + # 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( + address, key, + gpg.export_keys(key['fingerprint'], secret=secret)) + self.put_key(openpgp_key) _safe_call(_gen_key_cb) return self.get_key(address, private=True) @@ -262,15 +376,38 @@ class OpenPGPWrapper(KeyTypeWrapper): def _put_key_raw_cb(gpg): - key = gpg.list_keys(secret=False).pop() # unitary keyring + privkey = None + pubkey = None + try: + privkey = gpg.list_keys(secret=True).pop() + except IndexError: + pass + pubkey = gpg.list_keys(secret=False).pop() # unitary keyring # extract adress from first uid on key - match = re.match('.*<([\w.-]+@[\w.-]+)>.*', key['uids'].pop()) + match = re.match('.*<([\w.-]+@[\w.-]+)>.*', pubkey['uids'].pop()) leap_assert(match is not None, 'No user address in key data.') address = match.group(1) - openpgp_key = _build_key_from_gpg( - address, key, - gpg.export_keys(key['fingerprint'])) - self.put_key(openpgp_key) + if privkey is not None: + match = re.match( + '.*<([\w.-]+@[\w.-]+)>.*', privkey['uids'].pop()) + leap_assert(match is not None, 'No user address in key data.') + privaddress = match.group(1) + leap_assert( + address == privaddress, + 'Addresses in pub and priv key differ.') + leap_assert( + pubkey['fingerprint'] == privkey['fingerprint'], + 'Fingerprints for pub and priv key differ.') + # insert private key in storage + openpgp_privkey = _build_key_from_gpg( + address, privkey, + gpg.export_keys(privkey['fingerprint'], secret=True)) + self.put_key(openpgp_privkey) + # insert public key in storage + openpgp_pubkey = _build_key_from_gpg( + address, pubkey, + gpg.export_keys(pubkey['fingerprint'], secret=False)) + self.put_key(openpgp_pubkey) _safe_call(_put_key_raw_cb, data) @@ -285,7 +422,8 @@ class OpenPGPWrapper(KeyTypeWrapper): if doc is None: self._soledad.create_doc_from_json( key.get_json(), - doc_id=_keymanager_doc_id(key.address, key.private)) + doc_id=_keymanager_doc_id( + OpenPGPKey, key.address, key.private)) else: doc.set_json(key.get_json()) self._soledad.put_doc(doc) @@ -303,4 +441,22 @@ class OpenPGPWrapper(KeyTypeWrapper): @return: The document with the key or None if it does not exist. @rtype: leap.soledad.backends.leap_backend.LeapDocument """ - return self._soledad.get_doc(_keymanager_doc_id(address, private)) + return self._soledad.get_doc( + _keymanager_doc_id(OpenPGPKey, address, private)) + + def delete_key(self, key): + """ + Remove C{key} from storage. + + @param key: The key to be removed. + @type key: EncryptionKey + """ + leap_assert(key.__class__ is OpenPGPKey, 'Wrong key type.') + stored_key = self.get_key(key.address, private=key.private) + if stored_key is None: + raise KeyDoesNotExist(key) + if stored_key.__dict__ != key.__dict__: + raise KeyAttributesDiffer(key) + doc = self._soledad.get_doc( + _keymanager_doc_id(OpenPGPKey, key.address, key.private)) + self._soledad.delete_doc(doc) diff --git a/src/leap/common/keymanager/util.py b/src/leap/common/keymanager/util.py index 42168c8..667d2b2 100644 --- a/src/leap/common/keymanager/util.py +++ b/src/leap/common/keymanager/util.py @@ -25,6 +25,9 @@ import re from hashlib import sha256 +from binascii import b2a_base64 + + from leap.common.check import leap_assert @@ -79,11 +82,13 @@ def _build_key_from_doc(kClass, address, doc): return _build_key_from_dict(kClass, address, doc.content) -def _keymanager_doc_id(address, private=False): +def _keymanager_doc_id(ktype, address, private=False): """ Return the document id for the document containing a key for C{address}. + @param address: The type of the key. + @type address: KeyType @param address: The address bound to the key. @type address: str @param private: Whether the key is private or not. @@ -93,5 +98,6 @@ def _keymanager_doc_id(address, private=False): @rtype: str """ leap_assert(_is_address(address), "Wrong address format: %s" % address) - ktype = 'private' if private else 'public' - return sha256('key-manager-'+address+'-'+ktype).hexdigest() + ktype = str(ktype) + visibility = 'private' if private else 'public' + return sha256('key-manager-'+address+'-'+ktype+'-'+visibility).hexdigest() diff --git a/src/leap/common/tests/test_keymanager.py b/src/leap/common/tests/test_keymanager.py index 4a2693e..f9b478f 100644 --- a/src/leap/common/tests/test_keymanager.py +++ b/src/leap/common/tests/test_keymanager.py @@ -26,14 +26,17 @@ import unittest from leap.common.testing.basetest import BaseLeapTest from leap.soledad import Soledad +from leap.soledad.crypto import SoledadCrypto + + from leap.common.keymanager import KeyManager, openpgp, KeyNotFound from leap.common.keymanager.openpgp import OpenPGPKey -from leap.common.keymanager.gpg import GPGWrapper from leap.common.keymanager.util import ( _is_address, _build_key_from_dict, _keymanager_doc_id, ) +from leap.common.keymanager import errors class KeyManagerUtilTestCase(BaseLeapTest): @@ -72,32 +75,46 @@ class KeyManagerUtilTestCase(BaseLeapTest): 'validation': 'validation', } key = _build_key_from_dict(OpenPGPKey, 'leap@leap.se', kdict) - self.assertEqual(kdict['address'], key.address, + self.assertEqual( + kdict['address'], key.address, 'Wrong data in key.') - self.assertEqual(kdict['key_id'], key.key_id, + self.assertEqual( + kdict['key_id'], key.key_id, 'Wrong data in key.') - self.assertEqual(kdict['fingerprint'], key.fingerprint, + self.assertEqual( + kdict['fingerprint'], key.fingerprint, 'Wrong data in key.') - self.assertEqual(kdict['key_data'], key.key_data, + self.assertEqual( + kdict['key_data'], key.key_data, 'Wrong data in key.') - self.assertEqual(kdict['private'], key.private, + self.assertEqual( + kdict['private'], key.private, 'Wrong data in key.') - self.assertEqual(kdict['length'], key.length, + self.assertEqual( + kdict['length'], key.length, 'Wrong data in key.') - self.assertEqual(kdict['expiry_date'], key.expiry_date, + self.assertEqual( + kdict['expiry_date'], key.expiry_date, 'Wrong data in key.') - self.assertEqual(kdict['first_seen_at'], key.first_seen_at, + self.assertEqual( + kdict['first_seen_at'], key.first_seen_at, 'Wrong data in key.') - self.assertEqual(kdict['last_audited_at'], key.last_audited_at, + self.assertEqual( + kdict['last_audited_at'], key.last_audited_at, 'Wrong data in key.') - self.assertEqual(kdict['validation'], key.validation, + self.assertEqual( + kdict['validation'], key.validation, 'Wrong data in key.') def test__keymanager_doc_id(self): - doc_id1 = _keymanager_doc_id('leap@leap.se', private=False) - doc_id2 = _keymanager_doc_id('leap@leap.se', private=True) - doc_id3 = _keymanager_doc_id('user@leap.se', private=False) - doc_id4 = _keymanager_doc_id('user@leap.se', private=True) + doc_id1 = _keymanager_doc_id( + OpenPGPKey, 'leap@leap.se', private=False) + doc_id2 = _keymanager_doc_id( + OpenPGPKey, 'leap@leap.se', private=True) + doc_id3 = _keymanager_doc_id( + OpenPGPKey, 'user@leap.se', private=False) + doc_id4 = _keymanager_doc_id( + OpenPGPKey, 'user@leap.se', private=True) self.assertFalse(doc_id1 == doc_id2, 'Doc ids are equal!') self.assertFalse(doc_id1 == doc_id3, 'Doc ids are equal!') self.assertFalse(doc_id1 == doc_id4, 'Doc ids are equal!') @@ -119,7 +136,7 @@ class KeyManagerCryptoTestCase(BaseLeapTest): ) # initialize solead by hand for testing purposes self._soledad._init_dirs() - self._soledad._gpg = GPGWrapper(gnupghome=self.tempdir+"/gnupg") + self._soledad._crypto = SoledadCrypto(self._soledad) self._soledad._shared_db = None self._soledad._init_keys() self._soledad._init_db() @@ -130,31 +147,86 @@ class KeyManagerCryptoTestCase(BaseLeapTest): def _key_manager(user='user@leap.se', url='https://domain.org:6425'): return KeyManager(user, url) - def test_openpgp_gen_key(self): - pgp = openpgp.OpenPGPWrapper(self._soledad) - try: - pgp.get_key('user@leap.se') - except KeyNotFound: - 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_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) + self.assertRaises(KeyNotFound, pgp.get_key, 'leap@leap.se') + pgp.put_key_raw(PUBLIC_KEY) + key = pgp.get_key('leap@leap.se', private=False) + pgp.delete_key(key) + self.assertRaises(KeyNotFound, pgp.get_key, 'leap@leap.se') def test_openpgp_put_key_raw(self): - pgp = openpgp.OpenPGPWrapper(self._soledad) - try: - pgp.get_key('leap@leap.se') - except KeyNotFound: - pgp.put_key_raw(PUBLIC_KEY) - key = pgp.get_key('leap@leap.se') - self.assertIsInstance(key, openpgp.OpenPGPKey) - self.assertEqual( - 'leap@leap.se', key.address, 'Wrong address bound to key.') - self.assertEqual( - '4096', key.length, 'Wrong key length.') + pgp = openpgp.OpenPGPScheme(self._soledad) + self.assertRaises(KeyNotFound, pgp.get_key, 'leap@leap.se') + pgp.put_key_raw(PUBLIC_KEY) + key = pgp.get_key('leap@leap.se', private=False) + self.assertIsInstance(key, openpgp.OpenPGPKey) + self.assertEqual( + 'leap@leap.se', 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, 'leap@leap.se') + + def test_get_public_key(self): + pgp = openpgp.OpenPGPScheme(self._soledad) + self.assertRaises(KeyNotFound, pgp.get_key, 'leap@leap.se') + pgp.put_key_raw(PUBLIC_KEY) + self.assertRaises( + KeyNotFound, pgp.get_key, 'leap@leap.se', private=True) + key = pgp.get_key('leap@leap.se', private=False) + self.assertEqual('leap@leap.se', key.address) + self.assertFalse(key.private) + self.assertEqual(KEY_FINGERPRINT, key.fingerprint) + pgp.delete_key(key) + self.assertRaises(KeyNotFound, pgp.get_key, 'leap@leap.se') + + def test_openpgp_encrypt_decrypt_asym(self): + # encrypt + pgp = openpgp.OpenPGPScheme(self._soledad) + pgp.put_key_raw(PUBLIC_KEY) + pubkey = pgp.get_key('leap@leap.se', private=False) + cyphertext = openpgp.encrypt_asym('data', pubkey) + # assert + self.assertTrue(cyphertext is not None) + self.assertTrue(cyphertext != '') + self.assertTrue(cyphertext != 'data') + self.assertTrue(openpgp.is_encrypted_asym(cyphertext)) + self.assertFalse(openpgp.is_encrypted_sym(cyphertext)) + self.assertTrue(openpgp.is_encrypted(cyphertext)) + # decrypt + self.assertRaises( + KeyNotFound, pgp.get_key, 'leap@leap.se', private=True) + pgp.put_key_raw(PRIVATE_KEY) + privkey = pgp.get_key('leap@leap.se', private=True) + plaintext = openpgp.decrypt_asym(cyphertext, privkey) + pgp.delete_key(pubkey) + pgp.delete_key(privkey) + self.assertRaises( + KeyNotFound, pgp.get_key, 'leap@leap.se', private=False) + self.assertRaises( + KeyNotFound, pgp.get_key, 'leap@leap.se', private=True) + def test_openpgp_encrypt_decrypt_sym(self): + cyphertext = openpgp.encrypt_sym('data', 'pass') + self.assertTrue(cyphertext is not None) + self.assertTrue(cyphertext != '') + self.assertTrue(cyphertext != 'data') + self.assertTrue(openpgp.is_encrypted_sym(cyphertext)) + self.assertFalse(openpgp.is_encrypted_asym(cyphertext)) + self.assertTrue(openpgp.is_encrypted(cyphertext)) + plaintext = openpgp.decrypt_sym(cyphertext, 'pass') + self.assertEqual('data', plaintext) # Key material for testing -- cgit v1.2.3 From 4113dd985b9b5fc3b8e9839670ac5f7416f3f634 Mon Sep 17 00:00:00 2001 From: drebs Date: Sat, 27 Apr 2013 00:06:01 -0300 Subject: Add key refreshing for KeyManager. --- src/leap/common/keymanager/__init__.py | 103 +++++++++++++++++++++++++++---- src/leap/common/keymanager/keys.py | 71 +++++++++++++++++++++ src/leap/common/keymanager/openpgp.py | 20 +++--- src/leap/common/keymanager/util.py | 103 ------------------------------- src/leap/common/tests/test_keymanager.py | 68 ++++++++++++++------ 5 files changed, 218 insertions(+), 147 deletions(-) delete mode 100644 src/leap/common/keymanager/util.py (limited to 'src') diff --git a/src/leap/common/keymanager/__init__.py b/src/leap/common/keymanager/__init__.py index f939a4e..82fa99b 100644 --- a/src/leap/common/keymanager/__init__.py +++ b/src/leap/common/keymanager/__init__.py @@ -31,6 +31,9 @@ from leap.common.keymanager.errors import ( KeyNotFound, KeyAlreadyExists, ) +from leap.common.keymanager.keys import ( + build_key_from_dict, +) from leap.common.keymanager.openpgp import ( OpenPGPKey, OpenPGPScheme, @@ -39,6 +42,14 @@ from leap.common.keymanager.openpgp import ( from leap.common.keymanager.http import HTTPClient +TAGS_INDEX = 'by-tags' +TAGS_AND_PRIVATE_INDEX = 'by-tags-and-private' +INDEXES = { + TAGS_INDEX: ['tags'], + TAGS_AND_PRIVATE_INDEX: ['tags', 'bool(private)'], +} + + class KeyManager(object): def __init__(self, address, url, soledad): @@ -55,10 +66,45 @@ class KeyManager(object): """ self._address = address self._http_client = HTTPClient(url) + self._soledad = soledad self._wrapper_map = { OpenPGPKey: OpenPGPScheme(soledad), # other types of key will be added to this mapper. } + self._init_indexes() + + # + # utilities + # + + def _key_class_from_type(self, ktype): + """ + Return key class from string representation of key type. + """ + return filter( + lambda klass: str(klass) == ktype, + self._wrapper_map).pop() + + def _init_indexes(self): + """ + Initialize the database indexes. + """ + # 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 send_key(self, ktype, send_private=False, password=None): """ @@ -104,7 +150,7 @@ class KeyManager(object): json.dumps(data), headers) - def get_key(self, address, ktype, private=False): + def get_key(self, address, ktype, private=False, fetch_remote=True): """ Return a key of type C{ktype} bound to C{address}. @@ -129,14 +175,21 @@ class KeyManager(object): try: return self._wrapper_map[ktype].get_key(address, private=private) except KeyNotFound: - key = filter(lambda k: isinstance(k, ktype), - self._fetch_keys(address)) - if key is None: + if fetch_remote is False: + raise + # fetch keys from server and discard unwanted types. + keys = filter(lambda k: isinstance(k, ktype), + self.fetch_keys_from_server(address)) + if len(keys) is 0: raise KeyNotFound() - self._wrapper_map[ktype].put_key(key) + leap_assert( + len(keys) == 1, + 'Got more than one key of type %s for %s.' % + (str(ktype), address)) + self._wrapper_map[ktype].put_key(keys[0]) return key - def _fetch_keys(self, address): + def fetch_keys_from_server(self, address): """ Fetch keys bound to C{address} from nickserver. @@ -153,18 +206,42 @@ class KeyManager(object): leap_assert( keydata['address'] == address, "Fetched key for wrong address.") + keys = [] for key in keydata['keys']: - # find the key class in the mapper - keyCLass = filter( - lambda klass: str(klass) == key['type'], - self._wrapper_map).pop() - yield _build_key_from_dict(kClass, address, key) + keys.append( + build_key_from_dict( + self._key_class_from_type(key['type']), + address, + key)) + return keys + + def get_all_keys_in_local_db(self, private=False): + """ + Return all keys stored in local database. + + @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['address'], + doc.content), + self._soledad.get_from_index( + TAGS_AND_PRIVATE_INDEX, + 'keymanager-key', + '1' if private else '0')) def refresh_keys(self): """ - Update the user's db of validated keys to see if there are changes. + Fetch keys from nickserver and update them locally. """ - raise NotImplementedError(self.refresh_keys) + addresses = set(map( + lambda doc: doc.address, + self.get_all_keys_in_local_db(False))) + for address in addresses: + for key in self.fetch_keys_from_server(address): + self._wrapper_map[key.__class__].put_key(key) def gen_key(self, ktype): """ diff --git a/src/leap/common/keymanager/keys.py b/src/leap/common/keymanager/keys.py index 250c2fa..453e0ed 100644 --- a/src/leap/common/keymanager/keys.py +++ b/src/leap/common/keymanager/keys.py @@ -25,11 +25,81 @@ try: import simplejson as json except ImportError: import json # noqa +import re +from hashlib import sha256 from abc import ABCMeta, abstractmethod +from leap.common.check import leap_assert +# +# Key handling utilities +# + +def is_address(address): + """ + Return whether the given C{address} is in the form user@provider. + + @param address: The address to be tested. + @type address: str + @return: Whether C{address} is in the form user@provider. + @rtype: bool + """ + return bool(re.match('[\w.-]+@[\w.-]+', address)) + + +def build_key_from_dict(kClass, address, kdict): + """ + Build an C{kClass} key bound to C{address} based on info in C{kdict}. + + @param address: The address bound to the key. + @type address: str + @param kdict: Dictionary with key data. + @type kdict: dict + @return: An instance of the key. + @rtype: C{kClass} + """ + leap_assert(address == kdict['address'], 'Wrong address in key data.') + return kClass( + address, + key_id=kdict['key_id'], + fingerprint=kdict['fingerprint'], + key_data=kdict['key_data'], + private=kdict['private'], + length=kdict['length'], + expiry_date=kdict['expiry_date'], + first_seen_at=kdict['first_seen_at'], + last_audited_at=kdict['last_audited_at'], + validation=kdict['validation'], # TODO: verify for validation. + ) + + +def keymanager_doc_id(ktype, address, private=False): + """ + Return the document id for the document containing a key for + C{address}. + + @param address: The type of the key. + @type address: KeyType + @param address: The address bound to the key. + @type address: str + @param private: Whether the key is private or not. + @type private: bool + @return: The document id for the document that stores a key bound to + C{address}. + @rtype: str + """ + leap_assert(is_address(address), "Wrong address format: %s" % address) + ktype = str(ktype) + visibility = 'private' if private else 'public' + return sha256('keymanager-'+address+'-'+ktype+'-'+visibility).hexdigest() + + +# +# Abstraction for encryption keys +# + class EncryptionKey(object): """ Abstract class for encryption keys. @@ -82,6 +152,7 @@ class EncryptionKey(object): 'validation': self.validation, 'first_seen_at': self.first_seen_at, 'last_audited_at': self.last_audited_at, + 'tags': ['keymanager-key'], }) diff --git a/src/leap/common/keymanager/openpgp.py b/src/leap/common/keymanager/openpgp.py index ace8c1e..fa3f732 100644 --- a/src/leap/common/keymanager/openpgp.py +++ b/src/leap/common/keymanager/openpgp.py @@ -33,13 +33,11 @@ from leap.common.keymanager.errors import ( from leap.common.keymanager.keys import ( EncryptionKey, EncryptionScheme, + is_address, + keymanager_doc_id, + build_key_from_dict, ) from leap.common.keymanager.gpg import GPGWrapper -from leap.common.keymanager.util import ( - _is_address, - _build_key_from_doc, - _keymanager_doc_id, -) # @@ -307,7 +305,7 @@ class OpenPGPScheme(EncryptionScheme): @raise KeyAlreadyExists: If key already exists in local database. """ # make sure the key does not already exist - leap_assert(_is_address(address), 'Not an user address: %s' % address) + leap_assert(is_address(address), 'Not an user address: %s' % address) try: self.get_key(address) raise KeyAlreadyExists(address) @@ -358,11 +356,11 @@ class OpenPGPScheme(EncryptionScheme): @rtype: OpenPGPKey @raise KeyNotFound: If the key was not found on local storage. """ - leap_assert(_is_address(address), 'Not an user address: %s' % address) + leap_assert(is_address(address), 'Not an user address: %s' % address) doc = self._get_key_doc(address, private) if doc is None: raise KeyNotFound(address) - return _build_key_from_doc(OpenPGPKey, address, doc) + return build_key_from_dict(OpenPGPKey, address, doc.content) def put_key_raw(self, data): """ @@ -422,7 +420,7 @@ class OpenPGPScheme(EncryptionScheme): if doc is None: self._soledad.create_doc_from_json( key.get_json(), - doc_id=_keymanager_doc_id( + doc_id=keymanager_doc_id( OpenPGPKey, key.address, key.private)) else: doc.set_json(key.get_json()) @@ -442,7 +440,7 @@ class OpenPGPScheme(EncryptionScheme): @rtype: leap.soledad.backends.leap_backend.LeapDocument """ return self._soledad.get_doc( - _keymanager_doc_id(OpenPGPKey, address, private)) + keymanager_doc_id(OpenPGPKey, address, private)) def delete_key(self, key): """ @@ -458,5 +456,5 @@ class OpenPGPScheme(EncryptionScheme): if stored_key.__dict__ != key.__dict__: raise KeyAttributesDiffer(key) doc = self._soledad.get_doc( - _keymanager_doc_id(OpenPGPKey, key.address, key.private)) + keymanager_doc_id(OpenPGPKey, key.address, key.private)) self._soledad.delete_doc(doc) diff --git a/src/leap/common/keymanager/util.py b/src/leap/common/keymanager/util.py deleted file mode 100644 index 667d2b2..0000000 --- a/src/leap/common/keymanager/util.py +++ /dev/null @@ -1,103 +0,0 @@ -# -*- coding: utf-8 -*- -# util.py -# Copyright (C) 2013 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 . - - -""" -Utilities for the Key Manager. -""" - - -import re - - -from hashlib import sha256 -from binascii import b2a_base64 - - -from leap.common.check import leap_assert - - -def _is_address(address): - """ - Return whether the given C{address} is in the form user@provider. - - @param address: The address to be tested. - @type address: str - @return: Whether C{address} is in the form user@provider. - @rtype: bool - """ - return bool(re.match('[\w.-]+@[\w.-]+', address)) - - -def _build_key_from_dict(kClass, address, kdict): - """ - Build an C{kClass} key bound to C{address} based on info in C{kdict}. - - @param address: The address bound to the key. - @type address: str - @param kdict: Dictionary with key data. - @type kdict: dict - @return: An instance of the key. - @rtype: C{kClass} - """ - return kClass( - address, - key_id=kdict['key_id'], - fingerprint=kdict['fingerprint'], - key_data=kdict['key_data'], - private=kdict['private'], - length=kdict['length'], - expiry_date=kdict['expiry_date'], - first_seen_at=kdict['first_seen_at'], - last_audited_at=kdict['last_audited_at'], - validation=kdict['validation'], # TODO: verify for validation. - ) - - -def _build_key_from_doc(kClass, address, doc): - """ - Build an C{kClass} for C{address} based on C{doc} from local storage. - - @param address: The address bound to the key. - @type address: str - @param doc: Document obtained from Soledad storage. - @type doc: leap.soledad.backends.leap_backend.LeapDocument - @return: An instance of the key. - @rtype: C{kClass} - """ - return _build_key_from_dict(kClass, address, doc.content) - - -def _keymanager_doc_id(ktype, address, private=False): - """ - Return the document id for the document containing a key for - C{address}. - - @param address: The type of the key. - @type address: KeyType - @param address: The address bound to the key. - @type address: str - @param private: Whether the key is private or not. - @type private: bool - @return: The document id for the document that stores a key bound to - C{address}. - @rtype: str - """ - leap_assert(_is_address(address), "Wrong address format: %s" % address) - ktype = str(ktype) - visibility = 'private' if private else 'public' - return sha256('key-manager-'+address+'-'+ktype+'-'+visibility).hexdigest() diff --git a/src/leap/common/tests/test_keymanager.py b/src/leap/common/tests/test_keymanager.py index f9b478f..9bafb89 100644 --- a/src/leap/common/tests/test_keymanager.py +++ b/src/leap/common/tests/test_keymanager.py @@ -29,12 +29,18 @@ from leap.soledad import Soledad from leap.soledad.crypto import SoledadCrypto -from leap.common.keymanager import KeyManager, openpgp, KeyNotFound +from leap.common.keymanager import ( + KeyManager, + openpgp, + KeyNotFound, + TAGS_INDEX, + TAGS_AND_PRIVATE_INDEX, +) from leap.common.keymanager.openpgp import OpenPGPKey -from leap.common.keymanager.util import ( - _is_address, - _build_key_from_dict, - _keymanager_doc_id, +from leap.common.keymanager.keys import ( + is_address, + build_key_from_dict, + keymanager_doc_id, ) from leap.common.keymanager import errors @@ -47,21 +53,21 @@ class KeyManagerUtilTestCase(BaseLeapTest): def tearDown(self): pass - def test__is_address(self): + def test_is_address(self): self.assertTrue( - _is_address('user@leap.se'), + is_address('user@leap.se'), 'Incorrect address detection.') self.assertFalse( - _is_address('userleap.se'), + is_address('userleap.se'), 'Incorrect address detection.') self.assertFalse( - _is_address('user@'), + is_address('user@'), 'Incorrect address detection.') self.assertFalse( - _is_address('@leap.se'), + is_address('@leap.se'), 'Incorrect address detection.') - def test__build_key_from_dict(self): + def test_build_key_from_dict(self): kdict = { 'address': 'leap@leap.se', 'key_id': 'key_id', @@ -74,7 +80,7 @@ class KeyManagerUtilTestCase(BaseLeapTest): 'last_audited_at': 'last_audited_at', 'validation': 'validation', } - key = _build_key_from_dict(OpenPGPKey, 'leap@leap.se', kdict) + key = build_key_from_dict(OpenPGPKey, 'leap@leap.se', kdict) self.assertEqual( kdict['address'], key.address, 'Wrong data in key.') @@ -106,14 +112,14 @@ class KeyManagerUtilTestCase(BaseLeapTest): kdict['validation'], key.validation, 'Wrong data in key.') - def test__keymanager_doc_id(self): - doc_id1 = _keymanager_doc_id( + def test_keymanager_doc_id(self): + doc_id1 = keymanager_doc_id( OpenPGPKey, 'leap@leap.se', private=False) - doc_id2 = _keymanager_doc_id( + doc_id2 = keymanager_doc_id( OpenPGPKey, 'leap@leap.se', private=True) - doc_id3 = _keymanager_doc_id( + doc_id3 = keymanager_doc_id( OpenPGPKey, 'user@leap.se', private=False) - doc_id4 = _keymanager_doc_id( + doc_id4 = keymanager_doc_id( OpenPGPKey, 'user@leap.se', private=True) self.assertFalse(doc_id1 == doc_id2, 'Doc ids are equal!') self.assertFalse(doc_id1 == doc_id3, 'Doc ids are equal!') @@ -123,7 +129,7 @@ class KeyManagerUtilTestCase(BaseLeapTest): self.assertFalse(doc_id3 == doc_id4, 'Doc ids are equal!') -class KeyManagerCryptoTestCase(BaseLeapTest): +class KeyManagerWithSoledadTestCase(BaseLeapTest): def setUp(self): self._soledad = Soledad( @@ -144,8 +150,9 @@ class KeyManagerCryptoTestCase(BaseLeapTest): def tearDown(self): pass - def _key_manager(user='user@leap.se', url='https://domain.org:6425'): - return KeyManager(user, url) + + +class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase): def _test_openpgp_gen_key(self): pgp = openpgp.OpenPGPScheme(self._soledad) @@ -229,6 +236,27 @@ class KeyManagerCryptoTestCase(BaseLeapTest): self.assertEqual('data', plaintext) +class KeyManagerKeyManagementTestCase( + KeyManagerWithSoledadTestCase): + + def _key_manager(self, user='leap@leap.se', url=''): + return KeyManager(user, url, self._soledad) + + def test_get_all_keys_in_db(self): + km = self._key_manager() + km._wrapper_map[OpenPGPKey].put_key_raw(PRIVATE_KEY) + # get public keys + keys = km.get_all_keys_in_local_db(False) + self.assertEqual(len(keys), 1, 'Wrong number of keys') + self.assertEqual('leap@leap.se', keys[0].address) + self.assertFalse(keys[0].private) + # get private keys + keys = km.get_all_keys_in_local_db(True) + self.assertEqual(len(keys), 1, 'Wrong number of keys') + self.assertEqual('leap@leap.se', keys[0].address) + self.assertTrue(keys[0].private) + + # Key material for testing KEY_FINGERPRINT = "E36E738D69173C13D709E44F2F455E2824D18DDF" PUBLIC_KEY = """ -- cgit v1.2.3 From 365c318872e433ee13eff29e37039c10b22ee685 Mon Sep 17 00:00:00 2001 From: drebs Date: Sat, 27 Apr 2013 00:35:22 -0300 Subject: Use 'requests' module in KeyManager. --- src/leap/common/keymanager/__init__.py | 38 ++++++++++------ src/leap/common/keymanager/http.py | 78 -------------------------------- src/leap/common/tests/test_keymanager.py | 3 +- 3 files changed, 25 insertions(+), 94 deletions(-) delete mode 100644 src/leap/common/keymanager/http.py (limited to 'src') diff --git a/src/leap/common/keymanager/__init__.py b/src/leap/common/keymanager/__init__.py index 82fa99b..8db3b3c 100644 --- a/src/leap/common/keymanager/__init__.py +++ b/src/leap/common/keymanager/__init__.py @@ -21,9 +21,7 @@ Key Manager is a Nicknym agent for LEAP client. """ import httplib - - -from u1db.errors import HTTPError +import requests from leap.common.check import leap_assert @@ -39,7 +37,6 @@ from leap.common.keymanager.openpgp import ( OpenPGPScheme, encrypt_sym, ) -from leap.common.keymanager.http import HTTPClient TAGS_INDEX = 'by-tags' @@ -52,7 +49,7 @@ INDEXES = { class KeyManager(object): - def __init__(self, address, url, soledad): + def __init__(self, address, nickserver_url, soledad): """ Initialize a Key Manager for user's C{address} with provider's nickserver reachable in C{url}. @@ -65,7 +62,7 @@ class KeyManager(object): @type soledad: leap.soledad.Soledad """ self._address = address - self._http_client = HTTPClient(url) + self._nickserver_url = nickserver_url self._soledad = soledad self._wrapper_map = { OpenPGPKey: OpenPGPScheme(soledad), @@ -105,6 +102,22 @@ class KeyManager(object): self._soledad.delete_index(name) self._soledad.create_index(name, *expression) + def _get_dict_from_http_json(self, path): + """ + Make a GET HTTP request and return a dictionary containing the + response. + """ + response = requests.get(self._nickserver_url+path) + leap_assert(r.status_code == 200, 'Invalid response.') + leap_assert( + response.headers['content-type'].startswith('application/json') + is True, + 'Content-type is not JSON.') + return r.json() + + # + # key management + # def send_key(self, ktype, send_private=False, password=None): """ @@ -143,12 +156,10 @@ class KeyManager(object): self.get_key(self._address, ktype, private=True).get_json()) privkey.key_data = encrypt_sym(data, passphrase) data['keys'].append(privkey) - headers = None # TODO: replace for token-based-auth - self._http_client.request( - 'PUT', - '/key/%s' % address, - json.dumps(data), - headers) + requests.put( + self._nickserver_url + '/key/' + address, + data=data, + auth=(self._address, None)) # TODO: replace for token-based auth. def get_key(self, address, ktype, private=False, fetch_remote=True): """ @@ -201,8 +212,7 @@ class KeyManager(object): @raise KeyNotFound: If the key was not found on nickserver. @raise httplib.HTTPException: """ - self._http_client.request('GET', '/key/%s' % address, None, None) - keydata = json.loads(self._http_client.read_response()) + keydata = self._get_dict_from_http_json('/key/%s' % address) leap_assert( keydata['address'] == address, "Fetched key for wrong address.") diff --git a/src/leap/common/keymanager/http.py b/src/leap/common/keymanager/http.py deleted file mode 100644 index 478137d..0000000 --- a/src/leap/common/keymanager/http.py +++ /dev/null @@ -1,78 +0,0 @@ -# -*- coding: utf-8 -*- -# http.py -# Copyright (C) 2013 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 utilities. -""" - - -import urlparse -import httplib - - -def HTTPClient(object): - """ - A simple HTTP client for making requests. - """ - - def __init__(self, url): - """ - Initialize the HTTP client. - """ - self._url = urlparse.urlsplit(url) - self._conn = None - - def _ensure_connection(self): - """ - Ensure the creation of the connection object. - """ - if self._conn is not None: - return - if self._url.scheme == 'https': - connClass = httplib.HTTPSConnection - else: - connClass = httplib.HTTPConnection - self._conn = connClass(self._url.hostname, self._url.port) - - def request(method, url_query, body, headers): - """ - Make an HTTP request. - - @param method: The method of the request. - @type method: str - @param url_query: The URL query string of the request. - @type url_query: str - @param body: The body of the request. - @type body: str - @param headers: Headers to be sent on the request. - @type headers: list of str - """ - self._ensure_connection() - return self._conn.request(mthod, url_query, body, headers) - - def response(self): - """ - Return the response of an HTTP request. - """ - return self._conn.getresponse() - - def read_response(self): - """ - Get the contents of a response for an HTTP request. - """ - return self.response().read() diff --git a/src/leap/common/tests/test_keymanager.py b/src/leap/common/tests/test_keymanager.py index 9bafb89..9670f9b 100644 --- a/src/leap/common/tests/test_keymanager.py +++ b/src/leap/common/tests/test_keymanager.py @@ -151,10 +151,9 @@ class KeyManagerWithSoledadTestCase(BaseLeapTest): pass - class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase): - def _test_openpgp_gen_key(self): + 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') -- cgit v1.2.3 From 4e309e8be7dad025b7e30e99e10dbc5fb49f9bf5 Mon Sep 17 00:00:00 2001 From: drebs Date: Mon, 29 Apr 2013 16:06:59 -0300 Subject: Remove gpg reference on Soledad usage. --- src/leap/common/tests/test_keymanager.py | 1 - 1 file changed, 1 deletion(-) (limited to 'src') diff --git a/src/leap/common/tests/test_keymanager.py b/src/leap/common/tests/test_keymanager.py index 9670f9b..9bf394d 100644 --- a/src/leap/common/tests/test_keymanager.py +++ b/src/leap/common/tests/test_keymanager.py @@ -135,7 +135,6 @@ class KeyManagerWithSoledadTestCase(BaseLeapTest): self._soledad = Soledad( "leap@leap.se", "123456", - gnupg_home=self.tempdir+"/gnupg", secret_path=self.tempdir+"/secret.gpg", local_db_path=self.tempdir+"/soledad.u1db", bootstrap=False, -- cgit v1.2.3 From 852a0fa34a94b588f66e2af0aa628d058c243fd3 Mon Sep 17 00:00:00 2001 From: drebs Date: Mon, 29 Apr 2013 16:07:30 -0300 Subject: Remove string conversion for encryption/decryption results. --- src/leap/common/keymanager/openpgp.py | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) (limited to 'src') diff --git a/src/leap/common/keymanager/openpgp.py b/src/leap/common/keymanager/openpgp.py index fa3f732..94d55cc 100644 --- a/src/leap/common/keymanager/openpgp.py +++ b/src/leap/common/keymanager/openpgp.py @@ -58,9 +58,8 @@ def encrypt_sym(data, passphrase): """ def _encrypt_cb(gpg): - return str( - gpg.encrypt( - data, None, passphrase=passphrase, symmetric=True)) + return gpg.encrypt( + data, None, passphrase=passphrase, symmetric=True).data return _safe_call(_encrypt_cb) @@ -79,7 +78,7 @@ def decrypt_sym(data, passphrase): """ def _decrypt_cb(gpg): - return str(gpg.decrypt(data, passphrase=passphrase)) + return gpg.decrypt(data, passphrase=passphrase).data return _safe_call(_decrypt_cb) @@ -99,9 +98,8 @@ def encrypt_asym(data, key): leap_assert(key.private is False, 'Key is not public.') def _encrypt_cb(gpg): - return str( - gpg.encrypt( - data, key.fingerprint, symmetric=False)) + return gpg.encrypt( + data, key.fingerprint, symmetric=False).data return _safe_call(_encrypt_cb, key.key_data) @@ -121,7 +119,7 @@ def decrypt_asym(data, key): leap_assert(key.private is True, 'Key is not private.') def _decrypt_cb(gpg): - return str(gpg.decrypt(data)) + return gpg.decrypt(data).data return _safe_call(_decrypt_cb, key.key_data) -- cgit v1.2.3 From 170cd90f593a106ea7730babde310724410a585e Mon Sep 17 00:00:00 2001 From: Tomas Touceda Date: Thu, 2 May 2013 15:02:33 -0300 Subject: Various fixes --- src/leap/common/keymanager/__init__.py | 18 +++++++++--------- src/leap/common/keymanager/keys.py | 4 ++++ src/leap/common/keymanager/openpgp.py | 3 ++- src/leap/common/tests/test_keymanager.py | 3 --- 4 files changed, 15 insertions(+), 13 deletions(-) (limited to 'src') diff --git a/src/leap/common/keymanager/__init__.py b/src/leap/common/keymanager/__init__.py index 8db3b3c..01dc0da 100644 --- a/src/leap/common/keymanager/__init__.py +++ b/src/leap/common/keymanager/__init__.py @@ -20,14 +20,16 @@ Key Manager is a Nicknym agent for LEAP client. """ -import httplib import requests +try: + import simplejson as json +except ImportError: + import json # noqa from leap.common.check import leap_assert from leap.common.keymanager.errors import ( KeyNotFound, - KeyAlreadyExists, ) from leap.common.keymanager.keys import ( build_key_from_dict, @@ -108,12 +110,12 @@ class KeyManager(object): response. """ response = requests.get(self._nickserver_url+path) - leap_assert(r.status_code == 200, 'Invalid response.') + leap_assert(response.status_code == 200, 'Invalid response.') leap_assert( response.headers['content-type'].startswith('application/json') is True, 'Content-type is not JSON.') - return r.json() + return response.json() # # key management @@ -132,8 +134,6 @@ class KeyManager(object): will be saved in the server in a way it is publicly retrievable through the hash string. - @param address: The address bound to the key. - @type address: str @param ktype: The type of the key. @type ktype: KeyType @@ -154,10 +154,10 @@ class KeyManager(object): if send_private: privkey = json.loads( self.get_key(self._address, ktype, private=True).get_json()) - privkey.key_data = encrypt_sym(data, passphrase) + privkey.key_data = encrypt_sym(data, password) data['keys'].append(privkey) requests.put( - self._nickserver_url + '/key/' + address, + self._nickserver_url + '/key/' + self._address, data=data, auth=(self._address, None)) # TODO: replace for token-based auth. @@ -198,7 +198,7 @@ class KeyManager(object): 'Got more than one key of type %s for %s.' % (str(ktype), address)) self._wrapper_map[ktype].put_key(keys[0]) - return key + return self._wrapper_map[ktype].get_key(address, private=private) def fetch_keys_from_server(self, address): """ diff --git a/src/leap/common/keymanager/keys.py b/src/leap/common/keymanager/keys.py index 453e0ed..2e6bfe9 100644 --- a/src/leap/common/keymanager/keys.py +++ b/src/leap/common/keymanager/keys.py @@ -194,6 +194,7 @@ class EncryptionScheme(object): @rtype: EncryptionKey @raise KeyNotFound: If the key was not found on local storage. """ + pass @abstractmethod def put_key(self, key): @@ -203,6 +204,7 @@ class EncryptionScheme(object): @param key: The key to be stored. @type key: EncryptionKey """ + pass @abstractmethod def gen_key(self, address): @@ -215,6 +217,7 @@ class EncryptionScheme(object): @return: The key bound to C{address}. @rtype: EncryptionKey """ + pass @abstractmethod def delete_key(self, key): @@ -224,3 +227,4 @@ class EncryptionScheme(object): @param key: The key to be removed. @type key: EncryptionKey """ + pass diff --git a/src/leap/common/keymanager/openpgp.py b/src/leap/common/keymanager/openpgp.py index 94d55cc..e2ffe76 100644 --- a/src/leap/common/keymanager/openpgp.py +++ b/src/leap/common/keymanager/openpgp.py @@ -29,6 +29,7 @@ from leap.common.check import leap_assert from leap.common.keymanager.errors import ( KeyNotFound, KeyAlreadyExists, + KeyAttributesDiffer ) from leap.common.keymanager.keys import ( EncryptionKey, @@ -450,7 +451,7 @@ class OpenPGPScheme(EncryptionScheme): leap_assert(key.__class__ is OpenPGPKey, 'Wrong key type.') stored_key = self.get_key(key.address, private=key.private) if stored_key is None: - raise KeyDoesNotExist(key) + raise KeyNotFound(key) if stored_key.__dict__ != key.__dict__: raise KeyAttributesDiffer(key) doc = self._soledad.get_doc( diff --git a/src/leap/common/tests/test_keymanager.py b/src/leap/common/tests/test_keymanager.py index 9bf394d..32bd1fd 100644 --- a/src/leap/common/tests/test_keymanager.py +++ b/src/leap/common/tests/test_keymanager.py @@ -21,9 +21,6 @@ Tests for the Key Manager. """ -import unittest - - from leap.common.testing.basetest import BaseLeapTest from leap.soledad import Soledad from leap.soledad.crypto import SoledadCrypto -- cgit v1.2.3 From 71a3f21d3b72566efa6cf024317dfc96624a10f7 Mon Sep 17 00:00:00 2001 From: drebs Date: Thu, 2 May 2013 22:38:31 -0300 Subject: Add tests for key management remote methods. --- src/leap/common/keymanager/__init__.py | 42 ++++++-- src/leap/common/keymanager/errors.py | 6 ++ src/leap/common/tests/test_keymanager.py | 166 ++++++++++++++++++++++++++----- 3 files changed, 176 insertions(+), 38 deletions(-) (limited to 'src') diff --git a/src/leap/common/keymanager/__init__.py b/src/leap/common/keymanager/__init__.py index 01dc0da..d6dbb8a 100644 --- a/src/leap/common/keymanager/__init__.py +++ b/src/leap/common/keymanager/__init__.py @@ -30,6 +30,7 @@ except ImportError: from leap.common.check import leap_assert from leap.common.keymanager.errors import ( KeyNotFound, + NoPasswordGiven, ) from leap.common.keymanager.keys import ( build_key_from_dict, @@ -51,7 +52,7 @@ INDEXES = { class KeyManager(object): - def __init__(self, address, nickserver_url, soledad): + def __init__(self, address, nickserver_url, soledad, token=None): """ Initialize a Key Manager for user's C{address} with provider's nickserver reachable in C{url}. @@ -66,11 +67,13 @@ class KeyManager(object): self._address = address self._nickserver_url = nickserver_url self._soledad = soledad + self.token = token self._wrapper_map = { OpenPGPKey: OpenPGPScheme(soledad), # other types of key will be added to this mapper. } self._init_indexes() + self._fetcher = requests # # utilities @@ -109,7 +112,7 @@ class KeyManager(object): Make a GET HTTP request and return a dictionary containing the response. """ - response = requests.get(self._nickserver_url+path) + response = self._fetcher.get(self._nickserver_url+path) leap_assert(response.status_code == 200, 'Invalid response.') leap_assert( response.headers['content-type'].startswith('application/json') @@ -142,24 +145,27 @@ class KeyManager(object): keyserver. """ # prepare the public key bound to address + pubkey = self.get_key( + self._address, ktype, private=False, fetch_remote=False) data = { 'address': self._address, 'keys': [ - json.loads( - self.get_key( - self._address, ktype, private=False).get_json()), + json.loads(pubkey.get_json()), ] } # prepare the private key bound to address if send_private: - privkey = json.loads( - self.get_key(self._address, ktype, private=True).get_json()) - privkey.key_data = encrypt_sym(data, password) + if password is None or password == '': + raise NoPasswordGiven('Can\'t send unencrypted private keys!') + privkey = self.get_key( + self._address, ktype, private=True, fetch_remote=False) + privkey = json.loads(privkey.get_json()) + privkey.key_data = encrypt_sym(privkey.key_data, password) data['keys'].append(privkey) - requests.put( + self._fetcher.put( self._nickserver_url + '/key/' + self._address, data=data, - auth=(self._address, None)) # TODO: replace for token-based auth. + auth=(self._address, self._token)) def get_key(self, address, ktype, private=False, fetch_remote=True): """ @@ -248,7 +254,8 @@ class KeyManager(object): """ addresses = set(map( lambda doc: doc.address, - self.get_all_keys_in_local_db(False))) + self.get_all_keys_in_local_db(private=False))) + # TODO: maybe we should not attempt to refresh our own public key? for address in addresses: for key in self.fetch_keys_from_server(address): self._wrapper_map[key.__class__].put_key(key) @@ -264,3 +271,16 @@ class KeyManager(object): @rtype: EncryptionKey """ return self._wrapper_map[ktype].gen_key(self._address) + + # + # Token setter/getter + # + + def _get_token(self): + return self._token + + def _set_token(self, token): + self._token = token + + token = property( + _get_token, _set_token, doc='The auth token.') diff --git a/src/leap/common/keymanager/errors.py b/src/leap/common/keymanager/errors.py index add6a38..1cf506e 100644 --- a/src/leap/common/keymanager/errors.py +++ b/src/leap/common/keymanager/errors.py @@ -38,3 +38,9 @@ class KeyAttributesDiffer(Exception): Raised when trying to delete a key but the stored key differs from the key passed to the delete_key() method. """ + +class NoPasswordGiven(Exception): + """ + Raised when trying to perform some action that needs a password without + providing one. + """ diff --git a/src/leap/common/tests/test_keymanager.py b/src/leap/common/tests/test_keymanager.py index 32bd1fd..1d7a382 100644 --- a/src/leap/common/tests/test_keymanager.py +++ b/src/leap/common/tests/test_keymanager.py @@ -21,6 +21,13 @@ Tests for the Key Manager. """ +import mock +try: + import simplejson as json +except ImportError: + import json # noqa + + from leap.common.testing.basetest import BaseLeapTest from leap.soledad import Soledad from leap.soledad.crypto import SoledadCrypto @@ -30,6 +37,7 @@ from leap.common.keymanager import ( KeyManager, openpgp, KeyNotFound, + NoPasswordGiven, TAGS_INDEX, TAGS_AND_PRIVATE_INDEX, ) @@ -42,6 +50,9 @@ from leap.common.keymanager.keys import ( from leap.common.keymanager import errors +ADDRESS = 'leap@leap.se' + + class KeyManagerUtilTestCase(BaseLeapTest): def setUp(self): @@ -66,7 +77,7 @@ class KeyManagerUtilTestCase(BaseLeapTest): def test_build_key_from_dict(self): kdict = { - 'address': 'leap@leap.se', + 'address': ADDRESS, 'key_id': 'key_id', 'fingerprint': 'fingerprint', 'key_data': 'key_data', @@ -77,7 +88,7 @@ class KeyManagerUtilTestCase(BaseLeapTest): 'last_audited_at': 'last_audited_at', 'validation': 'validation', } - key = build_key_from_dict(OpenPGPKey, 'leap@leap.se', kdict) + key = build_key_from_dict(OpenPGPKey, ADDRESS, kdict) self.assertEqual( kdict['address'], key.address, 'Wrong data in key.') @@ -111,9 +122,9 @@ class KeyManagerUtilTestCase(BaseLeapTest): def test_keymanager_doc_id(self): doc_id1 = keymanager_doc_id( - OpenPGPKey, 'leap@leap.se', private=False) + OpenPGPKey, ADDRESS, private=False) doc_id2 = keymanager_doc_id( - OpenPGPKey, 'leap@leap.se', private=True) + OpenPGPKey, ADDRESS, private=True) doc_id3 = keymanager_doc_id( OpenPGPKey, 'user@leap.se', private=False) doc_id4 = keymanager_doc_id( @@ -134,6 +145,8 @@ class KeyManagerWithSoledadTestCase(BaseLeapTest): "123456", secret_path=self.tempdir+"/secret.gpg", local_db_path=self.tempdir+"/soledad.u1db", + server_url='', + cert_file=None, bootstrap=False, ) # initialize solead by hand for testing purposes @@ -144,7 +157,14 @@ class KeyManagerWithSoledadTestCase(BaseLeapTest): self._soledad._init_db() def tearDown(self): - pass + km = self._key_manager() + for key in km.get_all_keys_in_local_db(): + km._wrapper_map[key.__class__].delete_key(key) + for key in km.get_all_keys_in_local_db(private=True): + km._wrapper_map[key.__class__].delete_key(key) + + def _key_manager(self, user=ADDRESS, url=''): + return KeyManager(user, url, self._soledad) class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase): @@ -161,43 +181,43 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase): def test_openpgp_put_delete_key(self): pgp = openpgp.OpenPGPScheme(self._soledad) - self.assertRaises(KeyNotFound, pgp.get_key, 'leap@leap.se') + self.assertRaises(KeyNotFound, pgp.get_key, ADDRESS) pgp.put_key_raw(PUBLIC_KEY) - key = pgp.get_key('leap@leap.se', private=False) + key = pgp.get_key(ADDRESS, private=False) pgp.delete_key(key) - self.assertRaises(KeyNotFound, pgp.get_key, 'leap@leap.se') + self.assertRaises(KeyNotFound, pgp.get_key, ADDRESS) def test_openpgp_put_key_raw(self): pgp = openpgp.OpenPGPScheme(self._soledad) - self.assertRaises(KeyNotFound, pgp.get_key, 'leap@leap.se') + self.assertRaises(KeyNotFound, pgp.get_key, ADDRESS) pgp.put_key_raw(PUBLIC_KEY) - key = pgp.get_key('leap@leap.se', private=False) + key = pgp.get_key(ADDRESS, private=False) self.assertIsInstance(key, openpgp.OpenPGPKey) self.assertEqual( - 'leap@leap.se', key.address, 'Wrong address bound to key.') + ADDRESS, 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, 'leap@leap.se') + self.assertRaises(KeyNotFound, pgp.get_key, ADDRESS) def test_get_public_key(self): pgp = openpgp.OpenPGPScheme(self._soledad) - self.assertRaises(KeyNotFound, pgp.get_key, 'leap@leap.se') + self.assertRaises(KeyNotFound, pgp.get_key, ADDRESS) pgp.put_key_raw(PUBLIC_KEY) self.assertRaises( - KeyNotFound, pgp.get_key, 'leap@leap.se', private=True) - key = pgp.get_key('leap@leap.se', private=False) - self.assertEqual('leap@leap.se', key.address) + KeyNotFound, pgp.get_key, ADDRESS, private=True) + key = pgp.get_key(ADDRESS, private=False) + self.assertEqual(ADDRESS, key.address) self.assertFalse(key.private) self.assertEqual(KEY_FINGERPRINT, key.fingerprint) pgp.delete_key(key) - self.assertRaises(KeyNotFound, pgp.get_key, 'leap@leap.se') + self.assertRaises(KeyNotFound, pgp.get_key, ADDRESS) def test_openpgp_encrypt_decrypt_asym(self): # encrypt pgp = openpgp.OpenPGPScheme(self._soledad) pgp.put_key_raw(PUBLIC_KEY) - pubkey = pgp.get_key('leap@leap.se', private=False) + pubkey = pgp.get_key(ADDRESS, private=False) cyphertext = openpgp.encrypt_asym('data', pubkey) # assert self.assertTrue(cyphertext is not None) @@ -208,16 +228,16 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase): self.assertTrue(openpgp.is_encrypted(cyphertext)) # decrypt self.assertRaises( - KeyNotFound, pgp.get_key, 'leap@leap.se', private=True) + KeyNotFound, pgp.get_key, ADDRESS, private=True) pgp.put_key_raw(PRIVATE_KEY) - privkey = pgp.get_key('leap@leap.se', private=True) + privkey = pgp.get_key(ADDRESS, private=True) plaintext = openpgp.decrypt_asym(cyphertext, privkey) pgp.delete_key(pubkey) pgp.delete_key(privkey) self.assertRaises( - KeyNotFound, pgp.get_key, 'leap@leap.se', private=False) + KeyNotFound, pgp.get_key, ADDRESS, private=False) self.assertRaises( - KeyNotFound, pgp.get_key, 'leap@leap.se', private=True) + KeyNotFound, pgp.get_key, ADDRESS, private=True) def test_openpgp_encrypt_decrypt_sym(self): cyphertext = openpgp.encrypt_sym('data', 'pass') @@ -234,23 +254,115 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase): class KeyManagerKeyManagementTestCase( KeyManagerWithSoledadTestCase): - def _key_manager(self, user='leap@leap.se', url=''): - return KeyManager(user, url, self._soledad) - def test_get_all_keys_in_db(self): km = self._key_manager() km._wrapper_map[OpenPGPKey].put_key_raw(PRIVATE_KEY) # get public keys keys = km.get_all_keys_in_local_db(False) self.assertEqual(len(keys), 1, 'Wrong number of keys') - self.assertEqual('leap@leap.se', keys[0].address) + self.assertEqual(ADDRESS, keys[0].address) self.assertFalse(keys[0].private) # get private keys keys = km.get_all_keys_in_local_db(True) self.assertEqual(len(keys), 1, 'Wrong number of keys') - self.assertEqual('leap@leap.se', keys[0].address) + self.assertEqual(ADDRESS, keys[0].address) self.assertTrue(keys[0].private) + def test_get_public_key(self): + km = self._key_manager() + km._wrapper_map[OpenPGPKey].put_key_raw(PRIVATE_KEY) + # get the key + key = km.get_key(ADDRESS, OpenPGPKey, private=False, + fetch_remote=False) + self.assertTrue(key is not None) + self.assertEqual(key.address, ADDRESS) + self.assertEqual( + key.fingerprint.lower(), KEY_FINGERPRINT.lower()) + self.assertFalse(key.private) + + def test_get_private_key(self): + km = self._key_manager() + km._wrapper_map[OpenPGPKey].put_key_raw(PRIVATE_KEY) + # get the key + key = km.get_key(ADDRESS, OpenPGPKey, private=True, + fetch_remote=False) + self.assertTrue(key is not None) + self.assertEqual(key.address, ADDRESS) + self.assertEqual( + key.fingerprint.lower(), KEY_FINGERPRINT.lower()) + self.assertTrue(key.private) + + def test_send_key_raises_key_not_found(self): + km = self._key_manager() + self.assertRaises( + KeyNotFound, + km.send_key, OpenPGPKey, send_private=False) + + def test_send_private_key_raises_key_not_found(self): + km = self._key_manager() + km._wrapper_map[OpenPGPKey].put_key_raw(PUBLIC_KEY) + self.assertRaises( + KeyNotFound, + km.send_key, OpenPGPKey, send_private=True, + password='123') + + def test_send_private_key_without_password_raises(self): + km = self._key_manager() + km._wrapper_map[OpenPGPKey].put_key_raw(PUBLIC_KEY) + self.assertRaises( + NoPasswordGiven, + km.send_key, OpenPGPKey, send_private=True) + + def test_send_public_key(self): + km = self._key_manager() + km._wrapper_map[OpenPGPKey].put_key_raw(PUBLIC_KEY) + km._fetcher.put = mock.Mock() + km.token = '123' + km.send_key(OpenPGPKey, send_private=False) + # setup args + data = { + 'address': km._address, + 'keys': [ + json.loads( + km.get_key( + km._address, OpenPGPKey).get_json()), + ] + } + url = km._nickserver_url + '/key/' + km._address + + km._fetcher.put.assert_called_once_with( + url, data=data, auth=(km._address, '123') + ) + + def test_fetch_keys_from_server(self): + km = self._key_manager() + # setup mock + + class Response(object): + status_code = 200 + headers = {'content-type': 'application/json'} + def json(self): + return {'address': 'anotheruser@leap.se', 'keys': []} + + km._fetcher.get = mock.Mock( + return_value=Response()) + # do the fetch + km.fetch_keys_from_server('anotheruser@leap.se') + # and verify the call + km._fetcher.get.assert_called_once_with( + km._nickserver_url + '/key/' + 'anotheruser@leap.se', + ) + + def test_refresh_keys(self): + # TODO: maybe we should not attempt to refresh our own public key? + km = self._key_manager() + km._wrapper_map[OpenPGPKey].put_key_raw(PUBLIC_KEY) + km.fetch_keys_from_server = mock.Mock(return_value=[]) + km.refresh_keys() + km.fetch_keys_from_server.assert_called_once_with( + 'leap@leap.se' + ) + # Key material for testing KEY_FINGERPRINT = "E36E738D69173C13D709E44F2F455E2824D18DDF" -- cgit v1.2.3