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

# -*- 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 <http://www.gnu.org/licenses/>.
"""
Key Manager is a Nicknym agent for LEAP client.
"""
# let's do a little sanity check to see if we're using the wrong gnupg
import sys

try:
    from gnupg.gnupg import GPGUtilities
    assert(GPGUtilities)  # pyflakes happy
    from gnupg import __version__
    from distutils.version import LooseVersion as V
    assert(V(__version__) >= V('1.2.3'))

except (ImportError, AssertionError):
    print "*******"
    print "Ooops! It looks like there is a conflict in the installed version "
    print "of gnupg."
    print
    print "Disclaimer: Ideally, we would need to work a patch and propose the "
    print "merge to upstream. But until then do: "
    print
    print "% pip uninstall python-gnupg"
    print "% pip install gnupg"
    print "*******"
    sys.exit(1)

import logging
import requests

from leap.common.check import leap_assert, leap_assert_type
from leap.common.events import signal
from leap.common.events import events_pb2 as proto
from leap.common.decorators import memoized_method

from leap.keymanager.errors import KeyNotFound

from leap.keymanager.keys import (
    EncryptionKey,
    build_key_from_dict,
    KEYMANAGER_KEY_TAG,
    TAGS_PRIVATE_INDEX,
)
from leap.keymanager.openpgp import (
    OpenPGPKey,
    OpenPGPScheme,
)

logger = logging.getLogger(__name__)


#
# The Key Manager
#

class KeyManager(object):

    #
    # server's key storage constants
    #

    OPENPGP_KEY = 'openpgp'
    PUBKEY_KEY = "user[public_key]"

    def __init__(self, address, nickserver_uri, soledad, session_id=None,
                 ca_cert_path=None, api_uri=None, api_version=None, uid=None,
                 gpgbinary=None):
        """
        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 nickserver.
        :type url: str
        :param soledad: A Soledad instance for local storage of keys.
        :type soledad: leap.soledad.Soledad
        :param session_id: The session ID for interacting with the webapp API.
        :type session_id: str
        :param ca_cert_path: The path to the CA certificate.
        :type ca_cert_path: str
        :param api_uri: The URI of the webapp API.
        :type api_uri: str
        :param api_version: The version of the webapp API.
        :type api_version: str
        :param uid: The users' UID.
        :type uid: str
        :param gpgbinary: Name for GnuPG binary executable.
        :type gpgbinary: C{str}
        """
        self._address = address
        self._nickserver_uri = nickserver_uri
        self._soledad = soledad
        self._session_id = session_id
        self.ca_cert_path = ca_cert_path
        self.api_uri = api_uri
        self.api_version = api_version
        self.uid = uid
        # a dict to map key types to their handlers
        self._wrapper_map = {
            OpenPGPKey: OpenPGPScheme(soledad, gpgbinary=gpgbinary),
            # other types of key will be added to this mapper.
        }
        # the following are used to perform https requests
        self._fetcher = requests
        self._session = self._fetcher.session()

    #
    # 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 _get(self, uri, data=None):
        """
        Send a GET request to C{uri} containing C{data}.

        :param uri: The URI of the request.
        :type uri: str
        :param data: The body of the request.
        :type data: dict, str or file

        :return: The response to the request.
        :rtype: requests.Response
        """
        leap_assert(
            self._ca_cert_path is not None,
            'We need the CA certificate path!')
        res = self._fetcher.get(uri, data=data, verify=self._ca_cert_path)
        # Nickserver now returns 404 for key not found and 500 for
        # other cases (like key too small), so we are skipping this
        # check for the time being
        # res.raise_for_status()

        # Responses are now text/plain, although it's json anyway, but
        # this will fail when it shouldn't
        # leap_assert(
        #     res.headers['content-type'].startswith('application/json'),
        #     'Content-type is not JSON.')
        return res

    def _put(self, uri, data=None):
        """
        Send a PUT request to C{uri} containing C{data}.

        The request will be sent using the configured CA certificate path to
        verify the server certificate and the configured session id for
        authentication.

        :param uri: The URI of the request.
        :type uri: str
        :param data: The body of the request.
        :type data: dict, str or file

        :return: The response to the request.
        :rtype: requests.Response
        """
        leap_assert(
            self._ca_cert_path is not None,
            'We need the CA certificate path!')
        leap_assert(
            self._session_id is not None,
            'We need a session_id to interact with webapp!')
        res = self._fetcher.put(
            uri, data=data, verify=self._ca_cert_path,
            cookies={'_session_id': self._session_id})
        # assert that the response is valid
        res.raise_for_status()
        return res

    def _fetch_keys_from_server(self, address):
        """
        Fetch keys bound to C{address} from nickserver and insert them in
        local database.

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

        @raise KeyNotFound: If the key was not found on nickserver.
        """
        # request keys from the nickserver
        res = None
        try:
            res = self._get(self._nickserver_uri, {'address': address})
            server_keys = res.json()
            # insert keys in local database
            if self.OPENPGP_KEY in server_keys:
                self._wrapper_map[OpenPGPKey].put_ascii_key(
                    server_keys['openpgp'])
        except Exception as e:
            logger.warning("Error retrieving the keys: %r" % (e,))
            if res:
                logger.warning("%s" % (res.content,))

    #
    # key management
    #

    def send_key(self, ktype):
        """
        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 ktype: The type of the key.
        :type ktype: KeyType

        @raise KeyNotFound: If the key was not found in local database.
        """
        leap_assert(
            ktype is OpenPGPKey,
            'For now we only know how to send OpenPGP public keys.')
        # prepare the public key bound to address
        pubkey = self.get_key(
            self._address, ktype, private=False, fetch_remote=False)
        data = {
            self.PUBKEY_KEY: pubkey.key_data
        }
        uri = "%s/%s/users/%s.json" % (
            self._api_uri,
            self._api_version,
            self._uid)
        self._put(uri, data)
        signal(proto.KEYMANAGER_DONE_UPLOADING_KEYS, self._address)

    @memoized_method
    def get_key_from_cache(self, *args, **kwargs):
        """
        Public interface to `get_key`, that is memoized.
        """
        return self.get_key(*args, **kwargs)

    def get_key(self, address, ktype, private=False, fetch_remote=True):
        """
        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
        :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.
        """
        logger.debug("getting key for %s" % (address,))
        leap_assert(
            ktype in self._wrapper_map,
            'Unkown key type: %s.' % str(ktype))
        try:
            signal(proto.KEYMANAGER_LOOKING_FOR_KEY, address)
            # return key if it exists in local database
            key = self._wrapper_map[ktype].get_key(address, private=private)
            signal(proto.KEYMANAGER_KEY_FOUND, address)

            return key
        except KeyNotFound:
            signal(proto.KEYMANAGER_KEY_NOT_FOUND, address)

            # we will only try to fetch a key from nickserver if fetch_remote
            # is True and the key is not private.
            if fetch_remote is False or private is True:
                raise

            signal(proto.KEYMANAGER_LOOKING_FOR_KEY, address)
            self._fetch_keys_from_server(address)
            key = self._wrapper_map[ktype].get_key(address, private=False)
            signal(proto.KEYMANAGER_KEY_FOUND, address)

            return key

    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_PRIVATE_INDEX,
                KEYMANAGER_KEY_TAG,
                '1' if private else '0'))

    def refresh_keys(self):
        """
        Fetch keys from nickserver and update them locally.
        """
        addresses = set(map(
            lambda doc: doc.address,
            self.get_all_keys_in_local_db(private=False)))
        for address in addresses:
            # do not attempt to refresh our own key
            if address == self._address:
                continue
            self._fetch_keys_from_server(address)

    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
        """
        signal(proto.KEYMANAGER_STARTED_KEY_GENERATION, self._address)
        key = self._wrapper_map[ktype].gen_key(self._address)
        signal(proto.KEYMANAGER_FINISHED_KEY_GENERATION, self._address)

        return key

    #
    # Setters/getters
    #

    def _get_session_id(self):
        return self._session_id

    def _set_session_id(self, session_id):
        self._session_id = session_id

    session_id = property(
        _get_session_id, _set_session_id, doc='The session id.')

    def _get_ca_cert_path(self):
        return self._ca_cert_path

    def _set_ca_cert_path(self, ca_cert_path):
        self._ca_cert_path = ca_cert_path

    ca_cert_path = property(
        _get_ca_cert_path, _set_ca_cert_path,
        doc='The path to the CA certificate.')

    def _get_api_uri(self):
        return self._api_uri

    def _set_api_uri(self, api_uri):
        self._api_uri = api_uri

    api_uri = property(
        _get_api_uri, _set_api_uri, doc='The webapp API URI.')

    def _get_api_version(self):
        return self._api_version

    def _set_api_version(self, api_version):
        self._api_version = api_version

    api_version = property(
        _get_api_version, _set_api_version, doc='The webapp API version.')

    def _get_uid(self):
        return self._uid

    def _set_uid(self, uid):
        self._uid = uid

    uid = property(
        _get_uid, _set_uid, doc='The uid of the user.')

    #
    # encrypt/decrypt and sign/verify API
    #

    def encrypt(self, data, pubkey, passphrase=None, sign=None,
                cipher_algo='AES256'):
        """
        Encrypt C{data} using public @{key} 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: EncryptionKey
        :param sign: The key used for signing.
        :type sign: EncryptionKey
        :param cipher_algo: The cipher algorithm to use.
        :type cipher_algo: str

        :return: The encrypted data.
        :rtype: str
        """
        leap_assert_type(pubkey, EncryptionKey)
        leap_assert(pubkey.__class__ in self._wrapper_map, 'Unknown key type.')
        leap_assert(pubkey.private is False, 'Key is not public.')
        return self._wrapper_map[pubkey.__class__].encrypt(
            data, pubkey, passphrase, sign)

    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: The decrypted data.
        :rtype: str

        @raise InvalidSignature: Raised if unable to verify the signature with
            C{verify} key.
        """
        leap_assert_type(privkey, EncryptionKey)
        leap_assert(
            privkey.__class__ in self._wrapper_map,
            'Unknown key type.')
        leap_assert(privkey.private is True, 'Key is not private.')
        return self._wrapper_map[privkey.__class__].decrypt(
            data, privkey, passphrase, verify)

    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: EncryptionKey
        :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 signed data.
        :rtype: str
        """
        leap_assert_type(privkey, EncryptionKey)
        leap_assert(
            privkey.__class__ in self._wrapper_map,
            'Unknown key type.')
        leap_assert(privkey.private is True, 'Key is not private.')
        return self._wrapper_map[privkey.__class__].sign(
            data, privkey, digest_algo=digest_algo, clearsign=clearsign,
            detach=detach, binary=binary)

    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: EncryptionKey
        :param detached_sig: A detached signature. If given, C{data} is
                             verified using this detached signature.
        :type detached_sig: str

        :return: The signed data.
        :rtype: str
        """
        leap_assert_type(pubkey, EncryptionKey)
        leap_assert(pubkey.__class__ in self._wrapper_map, 'Unknown key type.')
        leap_assert(pubkey.private is False, 'Key is not public.')
        return self._wrapper_map[pubkey.__class__].verify(
            data, pubkey, detached_sig=detached_sig)

    def parse_openpgp_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
        """
        return self._wrapper_map[OpenPGPKey].parse_ascii_key(key_data)

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

        May raise:
            openpgp.errors.KeyNotFound
            openpgp.errors.KeyAttributesDiffer

        :param key: The key to be removed.
        :type key: EncryptionKey
        """
        try:
            self._wrapper_map[type(key)].delete_key(key)
        except IndexError as e:
            leap_assert(False, "Unsupported key type. Error {0!r}".format(e))

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

        :param key: The key to be stored.
        :type key: OpenPGPKey
        """
        try:
            self._wrapper_map[type(key)].put_key(key)
        except IndexError as e:
            leap_assert(False, "Unsupported key type. Error {0!r}".format(e))

from ._version import get_versions
__version__ = get_versions()['version']
del get_versions