path: root/src/leap/keymanager/openpgp.py

# -*- coding: utf-8 -*-
# openpgp.py
# Copyright (C) 2013-2015 LEAP
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
Infrastructure for using OpenPGP keys in Key Manager.
"""
import logging
import os
import re
import shutil
import tempfile
import io


from datetime import datetime
from multiprocessing import cpu_count
from gnupg import GPG
from gnupg.gnupg import GPGUtilities
from twisted.internet import defer
from twisted.internet.threads import deferToThread

from leap.common.check import leap_assert, leap_assert_type, leap_check
from leap.keymanager import errors
from leap.keymanager.keys import (
    EncryptionKey,
    EncryptionScheme,
    is_address,
    build_key_from_dict,
    TYPE_FINGERPRINT_PRIVATE_INDEX,
    TYPE_ADDRESS_PRIVATE_INDEX,
    KEY_ADDRESS_KEY,
    KEY_FINGERPRINT_KEY,
    KEYMANAGER_ACTIVE_TYPE,
)


logger = logging.getLogger(__name__)


#
# A temporary GPG keyring wrapped to provide OpenPGP functionality.
#

# This function will be used to call blocking GPG functions outside
# of Twisted reactor and match the concurrent calls to the amount of CPU cores
cpu_core_semaphore = defer.DeferredSemaphore(cpu_count())


def from_thread(func, *args, **kwargs):
    call = lambda: deferToThread(func, *args, **kwargs)
    return cpu_core_semaphore.run(call)


class TempGPGWrapper(object):
    """
    A context manager that wraps a temporary GPG keyring which only contains
    the keys given at object creation.
    """

    def __init__(self, keys=None, gpgbinary=None):
        """
        Create an empty temporary keyring and import any given C{keys} into
        it.

        :param keys: OpenPGP key, or list of.
        :type keys: OpenPGPKey or list of OpenPGPKeys
        :param gpgbinary: Name for GnuPG binary executable.
        :type gpgbinary: C{str}
        """
        self._gpg = None
        self._gpgbinary = gpgbinary
        if not keys:
            keys = list()
        if not isinstance(keys, list):
            keys = [keys]
        self._keys = keys
        for key in keys:
            leap_assert_type(key, OpenPGPKey)

    def __enter__(self):
        """
        Build and return a GPG keyring containing the keys given on
        object creation.

        :return: A GPG instance containing the keys given on object creation.
        :rtype: gnupg.GPG
        """
        self._build_keyring()
        return self._gpg

    def __exit__(self, exc_type, exc_value, traceback):
        """
        Ensure the gpg is properly destroyed.
        """
        # TODO handle exceptions and log here
        self._destroy_keyring()

    def _build_keyring(self):
        """
        Create a GPG keyring containing the keys given on object creation.

        :return: A GPG instance containing the keys given on object creation.
        :rtype: gnupg.GPG
        """
        privkeys = [key for key in self._keys if key and key.private is True]
        publkeys = [key for key in self._keys if key and key.private is False]
        # here we filter out public keys that have a correspondent
        # private key in the list because the private key_data by
        # itself is enough to also have the public key in the keyring,
        # and we want to count the keys afterwards.

        privfps = map(lambda privkey: privkey.fingerprint, privkeys)
        publkeys = filter(
            lambda pubkey: pubkey.fingerprint not in privfps, publkeys)

        listkeys = lambda: self._gpg.list_keys()
        listsecretkeys = lambda: self._gpg.list_keys(secret=True)

        self._gpg = GPG(binary=self._gpgbinary,
                        homedir=tempfile.mkdtemp())
        leap_assert(len(listkeys()) is 0, 'Keyring not empty.')

        # import keys into the keyring:
        # concatenating ascii-armored keys, which is correctly
        # understood by GPG.

        self._gpg.import_keys("".join(
            [x.key_data for x in publkeys + privkeys]))

        # assert the number of keys in the keyring
        leap_assert(
            len(listkeys()) == len(publkeys) + len(privkeys),
            'Wrong number of public keys in keyring: %d, should be %d)' %
            (len(listkeys()), len(publkeys) + len(privkeys)))
        leap_assert(
            len(listsecretkeys()) == len(privkeys),
            'Wrong number of private keys in keyring: %d, should be %d)' %
            (len(listsecretkeys()), len(privkeys)))

    def _destroy_keyring(self):
        """
        Securely erase the keyring.
        """
        # TODO: implement some kind of wiping of data or a more
        # secure way that
        # does not write to disk.

        try:
            for secret in [True, False]:
                for key in self._gpg.list_keys(secret=secret):
                    self._gpg.delete_keys(
                        key['fingerprint'],
                        secret=secret)
            leap_assert(len(self._gpg.list_keys()) is 0, 'Keyring not empty!')

        except:
            raise

        finally:
            leap_assert(self._gpg.homedir != os.path.expanduser('~/.gnupg'),
                        "watch out! Tried to remove default gnupg home!")
            shutil.rmtree(self._gpg.homedir)


def _parse_address(address):
    """
    Remove name, '<', '>' and the identity suffix after the '+' until the '@'
    e.g.: test_user+something@provider.com becomes test_user@provider.com
    since the key belongs to the identity without the '+' suffix.

    :type address: str
    :rtype: str
    """
    mail_regex = '(.*<)?([\w.-]+)(\+.*)?(@[\w.-]+)(>.*)?'
    match = re.match(mail_regex, address)
    if match is None:
        return None
    return ''.join(match.group(2, 4))


#
# The OpenPGP wrapper
#

class OpenPGPKey(EncryptionKey):
    """
    Base class for OpenPGP keys.
    """

    def __init__(self, address, gpgbinary=None, **kwargs):
        self._gpgbinary = gpgbinary
        super(OpenPGPKey, self).__init__(address, **kwargs)

    @property
    def signatures(self):
        """
        Get the key signatures

        :return: the key IDs that have signed the key
        :rtype: list(str)
        """
        with TempGPGWrapper(keys=[self], gpgbinary=self._gpgbinary) as gpg:
            res = gpg.list_sigs(self.fingerprint)
            for uid, sigs in res.sigs.iteritems():
                if _parse_address(uid) in self.address:
                    return sigs

        return []

    def merge(self, newkey):
        if newkey.fingerprint != self.fingerprint:
            logger.critical(
                "Can't put a key whith the same key_id and different "
                "fingerprint: %s, %s"
                % (newkey.fingerprint, self.fingerprint))
            raise errors.KeyFingerprintMismatch(newkey.fingerprint)

        with TempGPGWrapper(gpgbinary=self._gpgbinary) as gpg:
            gpg.import_keys(self.key_data)
            gpg.import_keys(newkey.key_data)
            gpgkey = gpg.list_keys(secret=newkey.private).pop()

            if gpgkey['expires']:
                self.expiry_date = datetime.fromtimestamp(
                    int(gpgkey['expires']))
            else:
                self.expiry_date = None

            self.uids = []
            for uid in gpgkey['uids']:
                self.uids.append(_parse_address(uid))

            self.length = int(gpgkey['length'])
            self.key_data = gpg.export_keys(gpgkey['fingerprint'],
                                            secret=self.private)

        if newkey.validation > self.validation:
            self.validation = newkey.validation
        if newkey.last_audited_at > self.last_audited_at:
            self.validation = newkey.last_audited_at
        self.encr_used = newkey.encr_used or self.encr_used
        self.sign_used = newkey.sign_used or self.sign_used
        self.refreshed_at = datetime.now()


class OpenPGPScheme(EncryptionScheme):
    """
    A wrapper for OpenPGP keys management and use (encryption, decyption,
    signing and verification).
    """

    # type used on the soledad documents
    KEY_TYPE = OpenPGPKey.__name__
    ACTIVE_TYPE = KEY_TYPE + KEYMANAGER_ACTIVE_TYPE

    def __init__(self, soledad, gpgbinary=None):
        """
        Initialize the OpenPGP wrapper.

        :param soledad: A Soledad instance for key storage.
        :type soledad: leap.soledad.Soledad
        :param gpgbinary: Name for GnuPG binary executable.
        :type gpgbinary: C{str}
        """
        EncryptionScheme.__init__(self, soledad)
        self._wait_indexes("get_key", "put_key")
        self._gpgbinary = gpgbinary

    #
    # Keys management
    #

    def gen_key(self, address):
        """
        Generate an OpenPGP keypair bound to C{address}.

        :param address: The address bound to the key.
        :type address: str

        :return: A Deferred which fires with the key bound to address, or fails
                 with KeyAlreadyExists if key already exists in local database.
        :rtype: Deferred
        """
        # make sure the key does not already exist
        leap_assert(is_address(address), 'Not an user address: %s' % address)

        @defer.inlineCallbacks
        def _gen_key(_):
            with TempGPGWrapper(gpgbinary=self._gpgbinary) as gpg:
                # TODO: inspect result, or use decorator
                params = gpg.gen_key_input(
                    key_type='RSA',
                    key_length=4096,
                    name_real=address,
                    name_email=address,
                    name_comment='')
                logger.info("About to generate keys... "
                            "This might take SOME time.")
                yield from_thread(gpg.gen_key, params)
                logger.info("Keys for %s have been successfully "
                            "generated." % (address,))
                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()
                leap_assert(
                    len(key['uids']) is 1,  # with just one uid!
                    'Wrong number of uids for key: %d.' % len(key['uids']))
                uid_match = False
                for uid in key['uids']:
                    if re.match('.*<%s>$' % address, uid) is not None:
                        uid_match = True
                        break
                leap_assert(uid_match, 'Key not correctly bound to address.')

                # insert both public and private keys in storage
                deferreds = []
                for secret in [True, False]:
                    key = gpg.list_keys(secret=secret).pop()
                    openpgp_key = self._build_key_from_gpg(
                        key,
                        gpg.export_keys(key['fingerprint'], secret=secret))
                    d = self.put_key(openpgp_key, address)
                    deferreds.append(d)
                yield defer.gatherResults(deferreds)

        def key_already_exists(_):
            raise errors.KeyAlreadyExists(address)

        d = self.get_key(address)
        d.addCallbacks(key_already_exists, _gen_key)
        d.addCallback(lambda _: self.get_key(address, private=True))
        return d

    def get_key(self, address, private=False):
        """
        Get key bound to C{address} 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: A Deferred which fires with the OpenPGPKey bound to address,
                 or which fails with KeyNotFound if the key was not found on
                 local storage.
        :rtype: Deferred
        """
        address = _parse_address(address)

        def build_key((keydoc, activedoc)):
            if keydoc is None:
                raise errors.KeyNotFound(address)
            leap_assert(
                address in keydoc.content[KEY_ADDRESS_KEY],
                'Wrong address in key %s. Expected %s, found %s.'
                % (keydoc.content[KEY_FINGERPRINT_KEY], address,
                   keydoc.content[KEY_ADDRESS_KEY]))
            key = build_key_from_dict(OpenPGPKey, keydoc.content,
                                      activedoc.content)
            key._gpgbinary = self._gpgbinary
            return key

        d = self._get_key_doc(address, private)
        d.addCallback(build_key)
        return d

    def parse_ascii_key(self, key_data):
        """
        Parses an ascii armored key (or key pair) data and returns
        the OpenPGPKey keys.

        :param key_data: the key data to be parsed.
        :type key_data: str or unicode

        :returns: the public key and private key (if applies) for that data.
        :rtype: (public, private) -> tuple(OpenPGPKey, OpenPGPKey)
                the tuple may have one or both components None
        """
        leap_assert_type(key_data, (str, unicode))
        # TODO: add more checks for correct key data.
        leap_assert(key_data is not None, 'Data does not represent a key.')

        priv_info, privkey = process_ascii_key(
            key_data, self._gpgbinary, secret=True)
        pub_info, pubkey = process_ascii_key(
            key_data, self._gpgbinary, secret=False)

        if not pubkey:
            return (None, None)

        openpgp_privkey = None
        if privkey:
            # build private key
            openpgp_privkey = self._build_key_from_gpg(priv_info, privkey)
            leap_check(pub_info['fingerprint'] == priv_info['fingerprint'],
                       'Fingerprints for public and private key differ.',
                       errors.KeyFingerprintMismatch)
        # build public key
        openpgp_pubkey = self._build_key_from_gpg(pub_info, pubkey)

        return (openpgp_pubkey, openpgp_privkey)

    def put_ascii_key(self, key_data, address):
        """
        Put key contained in ascii-armored C{key_data} in local storage.

        :param key_data: The key data to be stored.
        :type key_data: str or unicode
        :param address: address for which this key will be active
        :type address: str

        :return: A Deferred which fires when the OpenPGPKey is in the storage.
        :rtype: Deferred
        """
        leap_assert_type(key_data, (str, unicode))

        openpgp_privkey = None
        try:
            openpgp_pubkey, openpgp_privkey = self.parse_ascii_key(key_data)
        except (errors.KeyAddressMismatch, errors.KeyFingerprintMismatch) as e:
            return defer.fail(e)

        def put_key(_, key):
            return self.put_key(key, address)

        d = defer.succeed(None)
        if openpgp_pubkey is not None:
            d.addCallback(put_key, openpgp_pubkey)
        if openpgp_privkey is not None:
            d.addCallback(put_key, openpgp_privkey)
        return d

    def put_key(self, key, address):
        """
        Put C{key} in local storage.

        :param key: The key to be stored.
        :type key: OpenPGPKey
        :param address: address for which this key will be active.
        :type address: str

        :return: A Deferred which fires when the key is in the storage.
        :rtype: Deferred
        """
        def merge_and_put((keydoc, activedoc)):
            if not keydoc:
                return put_new_key(activedoc)

            active_content = None
            if activedoc:
                active_content = activedoc.content
            oldkey = build_key_from_dict(OpenPGPKey, keydoc.content,
                                         active_content)

            key.merge(oldkey)
            keydoc.set_json(key.get_json())
            deferred_key = self._soledad.put_doc(keydoc)

            active_json = key.get_active_json(address)
            if activedoc:
                activedoc.set_json(active_json)
                deferred_active = self._soledad.put_doc(activedoc)
            else:
                deferred_active = self._soledad.create_doc_from_json(
                    active_json)

            return defer.gatherResults([deferred_key, deferred_active])

        def put_new_key(activedoc):
            deferreds = []
            if activedoc:
                d = self._soledad.delete_doc(activedoc)
                deferreds.append(d)
            for json in [key.get_json(), key.get_active_json(address)]:
                d = self._soledad.create_doc_from_json(json)
                deferreds.append(d)
            return defer.gatherResults(deferreds)

        dk = self._get_key_doc_from_fingerprint(key.fingerprint, key.private)
        da = self._get_active_doc_from_address(address, key.private)
        d = defer.gatherResults([dk, da])
        d.addCallback(merge_and_put)
        return d

    def _get_key_doc(self, address, private=False):
        """
        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 address: The address bound to the key.
        :type address: str
        :param private: Whether to look for a private key.
        :type private: bool

        :return: A Deferred which fires with a touple of two SoledadDocument
                 (keydoc, activedoc) or None if it does not exist.
        :rtype: Deferred
        """
        def get_key_from_active_doc(activedoc):
            if not activedoc:
                return (None, None)
            fingerprint = activedoc.content[KEY_FINGERPRINT_KEY]
            d = self._get_key_doc_from_fingerprint(fingerprint, private)
            d.addCallback(delete_active_if_no_key, activedoc)
            return d

        def delete_active_if_no_key(keydoc, activedoc):
            if not keydoc:
                d = self._soledad.delete_doc(activedoc)
                d.addCallback(lambda _: (None, None))
                return d
            return (keydoc, activedoc)

        d = self._get_active_doc_from_address(address, private)
        d.addCallback(get_key_from_active_doc)
        return d

    def _build_key_from_gpg(self, 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 key: Key obtained from GPG storage.
        :type key: dict
        :param key_data: Key data obtained from GPG storage.
        :type key_data: str
        :return: An instance of the key.
        :rtype: OpenPGPKey
        """
        return build_gpg_key(key, key_data, self._gpgbinary)

    def delete_key(self, key):
        """
        Remove C{key} from storage.

        :param key: The key to be removed.
        :type key: EncryptionKey

        :return: A Deferred which fires when the key is deleted, or which
                 fails with KeyNotFound if the key was not found on local
                 storage.
        :rtype: Deferred
        """
        leap_assert_type(key, OpenPGPKey)

        def delete_docs(activedocs):
            deferreds = []
            for doc in activedocs:
                d = self._soledad.delete_doc(doc)
                deferreds.append(d)
            return defer.gatherResults(deferreds)

        def get_key_docs(_):
            return self._soledad.get_from_index(
                TYPE_FINGERPRINT_PRIVATE_INDEX,
                self.KEY_TYPE,
                key.fingerprint,
                '1' if key.private else '0')

        def delete_key(docs):
            if len(docs) == 0:
                raise errors.KeyNotFound(key)
            elif len(docs) > 1:
                logger.warning("There is more than one key for fingerprint %s"
                               % key.fingerprint)

            has_deleted = False
            deferreds = []
            for doc in docs:
                if doc.content['fingerprint'] == key.fingerprint:
                    d = self._soledad.delete_doc(doc)
                    deferreds.append(d)
                    has_deleted = True
            if not has_deleted:
                raise errors.KeyNotFound(key)
            return defer.gatherResults(deferreds)

        d = self._soledad.get_from_index(
            TYPE_FINGERPRINT_PRIVATE_INDEX,
            self.ACTIVE_TYPE,
            key.fingerprint,
            '1' if key.private else '0')
        d.addCallback(delete_docs)
        d.addCallback(get_key_docs)
        d.addCallback(delete_key)
        return d

    #
    # Data encryption, decryption, signing and verifying
    #

    @staticmethod
    def _assert_gpg_result_ok(result):
        """
        Check if GPG result is 'ok' and log stderr outputs.

        :param result: GPG results, which have a field calld 'ok' that states
                       whether the gpg operation was successful or not.
        :type result: object

        :raise GPGError: Raised when the gpg operation was not successful.
        """
        stderr = getattr(result, 'stderr', None)
        if stderr:
            logger.debug("%s" % (stderr,))
        if getattr(result, 'ok', None) is not True:
            raise errors.GPGError(
                'Failed to encrypt/decrypt: %s' % stderr)

    @defer.inlineCallbacks
    def encrypt(self, data, pubkey, passphrase=None, sign=None,
                cipher_algo='AES256'):
        """
        Encrypt C{data} using public @{pubkey} and sign with C{sign} key.

        :param data: The data to be encrypted.
        :type data: str
        :param pubkey: The key used to encrypt.
        :type pubkey: OpenPGPKey
        :param sign: The key used for signing.
        :type sign: OpenPGPKey
        :param cipher_algo: The cipher algorithm to use.
        :type cipher_algo: str

        :return: A Deferred that will be fired with the encrypted data.
        :rtype: defer.Deferred

        :raise EncryptError: Raised if failed encrypting for some reason.
        """
        leap_assert_type(pubkey, OpenPGPKey)
        leap_assert(pubkey.private is False, 'Key is not public.')
        keys = [pubkey]
        if sign is not None:
            leap_assert_type(sign, OpenPGPKey)
            leap_assert(sign.private is True)
            keys.append(sign)
        with TempGPGWrapper(keys, self._gpgbinary) as gpg:
            result = yield from_thread(
                gpg.encrypt,
                data, pubkey.fingerprint,
                default_key=sign.fingerprint if sign else None,
                passphrase=passphrase, symmetric=False,
                cipher_algo=cipher_algo)
            # Here we cannot assert for correctness of sig because the sig is
            # in the ciphertext.
            # result.ok    - (bool) indicates if the operation succeeded
            # result.data  - (bool) contains the result of the operation
            try:
                self._assert_gpg_result_ok(result)
                defer.returnValue(result.data)
            except errors.GPGError as e:
                logger.error('Failed to decrypt: %s.' % str(e))
                raise errors.EncryptError()

    @defer.inlineCallbacks
    def decrypt(self, data, privkey, passphrase=None, verify=None):
        """
        Decrypt C{data} using private @{privkey} and verify with C{verify} key.

        :param data: The data to be decrypted.
        :type data: str
        :param privkey: The key used to decrypt.
        :type privkey: OpenPGPKey
        :param passphrase: The passphrase for the secret key used for
                           decryption.
        :type passphrase: str
        :param verify: The key used to verify a signature.
        :type verify: OpenPGPKey

        :return: Deferred that will fire with the decrypted data and
                 if signature verifies (unicode, bool)
        :rtype: Deferred

        :raise DecryptError: Raised if failed decrypting for some reason.
        """
        leap_assert(privkey.private is True, 'Key is not private.')
        keys = [privkey]
        if verify is not None:
            leap_assert_type(verify, OpenPGPKey)
            leap_assert(verify.private is False)
            keys.append(verify)
        with TempGPGWrapper(keys, self._gpgbinary) as gpg:
            try:
                result = yield from_thread(gpg.decrypt,
                                           data, passphrase=passphrase,
                                           always_trust=True)
                self._assert_gpg_result_ok(result)

                # verify signature
                sign_valid = False
                if (verify is not None and
                        result.valid is True and
                        verify.fingerprint == result.pubkey_fingerprint):
                    sign_valid = True

                defer.returnValue((result.data, sign_valid))
            except errors.GPGError as e:
                logger.error('Failed to decrypt: %s.' % str(e))
                raise errors.DecryptError(str(e))

    def is_encrypted(self, 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
        """
        with TempGPGWrapper(gpgbinary=self._gpgbinary) as gpg:
            gpgutil = GPGUtilities(gpg)
            return gpgutil.is_encrypted_asym(data)

    def sign(self, data, privkey, digest_algo='SHA512', clearsign=False,
             detach=True, binary=False):
        """
        Sign C{data} with C{privkey}.

        :param data: The data to be signed.
        :type data: str

        :param privkey: The private key to be used to sign.
        :type privkey: OpenPGPKey
        :param digest_algo: The hash digest to use.
        :type digest_algo: str
        :param clearsign: If True, create a cleartext signature.
        :type clearsign: bool
        :param detach: If True, create a detached signature.
        :type detach: bool
        :param binary: If True, do not ascii armour the output.
        :type binary: bool

        :return: The ascii-armored signed data.
        :rtype: str
        """
        leap_assert_type(privkey, OpenPGPKey)
        leap_assert(privkey.private is True)

        # result.fingerprint - contains the fingerprint of the key used to
        #                      sign.
        with TempGPGWrapper(privkey, self._gpgbinary) as gpg:
            result = gpg.sign(data, default_key=privkey.fingerprint,
                              digest_algo=digest_algo, clearsign=clearsign,
                              detach=detach, binary=binary)
            rfprint = privkey.fingerprint
            privkey = gpg.list_keys(secret=True).pop()
            kfprint = privkey['fingerprint']
            if result.fingerprint is None:
                raise errors.SignFailed(
                    'Failed to sign with key %s: %s' %
                    (privkey['fingerprint'], result.stderr))
            leap_assert(
                result.fingerprint == kfprint,
                'Signature and private key fingerprints mismatch: '
                '%s != %s' % (rfprint, kfprint))
        return result.data

    def verify(self, data, pubkey, detached_sig=None):
        """
        Verify signed C{data} with C{pubkey}, eventually using
        C{detached_sig}.

        :param data: The data to be verified.
        :type data: str
        :param pubkey: The public key to be used on verification.
        :type pubkey: OpenPGPKey
        :param detached_sig: A detached signature. If given, C{data} is
                             verified against this detached signature.
        :type detached_sig: str

        :return: signature matches
        :rtype: bool
        """
        leap_assert_type(pubkey, OpenPGPKey)
        leap_assert(pubkey.private is False)
        with TempGPGWrapper(pubkey, self._gpgbinary) as gpg:
            result = None
            if detached_sig is None:
                result = gpg.verify(data)
            else:
                # to verify using a detached sig we have to use
                # gpg.verify_file(), which receives the data as a binary
                # stream and the name of a file containing the signature.
                sf, sfname = tempfile.mkstemp()
                with os.fdopen(sf, 'w') as sfd:
                    sfd.write(detached_sig)
                result = gpg.verify_file(io.BytesIO(data), sig_file=sfname)
                os.unlink(sfname)
            gpgpubkey = gpg.list_keys().pop()
            valid = result.valid
            rfprint = result.fingerprint
            kfprint = gpgpubkey['fingerprint']
            return valid and rfprint == kfprint

    def _get_active_doc_from_address(self, address, private):
        d = self._soledad.get_from_index(
            TYPE_ADDRESS_PRIVATE_INDEX,
            self.ACTIVE_TYPE,
            address,
            '1' if private else '0')
        d.addCallback(self._repair_and_get_doc, self._repair_active_docs)
        return d

    def _get_key_doc_from_fingerprint(self, fingerprint, private):
        d = self._soledad.get_from_index(
            TYPE_FINGERPRINT_PRIVATE_INDEX,
            self.KEY_TYPE,
            fingerprint,
            '1' if private else '0')
        d.addCallback(self._repair_and_get_doc, self._repair_key_docs)
        return d

    def _repair_and_get_doc(self, doclist, repair_func):
        if len(doclist) is 0:
            return None
        elif len(doclist) > 1:
            return repair_func(doclist)
        return doclist[0]


def process_ascii_key(key_data, gpgbinary, secret=False):
    with TempGPGWrapper(gpgbinary=gpgbinary) as gpg:
        try:
            gpg.import_keys(key_data)
            info = gpg.list_keys(secret=secret).pop()
            key = gpg.export_keys(info['fingerprint'], secret=secret)
        except IndexError:
            info = {}
            key = None
    return info, key


def build_gpg_key(key_info, key_data, gpgbinary=None):
    expiry_date = None
    if key_info['expires']:
        expiry_date = datetime.fromtimestamp(int(key_info['expires']))
    address = []
    for uid in key_info['uids']:
        address.append(_parse_address(uid))

    return OpenPGPKey(
        address,
        gpgbinary=gpgbinary,
        fingerprint=key_info['fingerprint'],
        key_data=key_data,
        private=True if key_info['type'] == 'sec' else False,
        length=int(key_info['length']),
        expiry_date=expiry_date,
        refreshed_at=datetime.now())