summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/leap/common/keymanager/__init__.py54
-rw-r--r--src/leap/common/keymanager/errors.py5
-rw-r--r--src/leap/common/keymanager/http.py78
-rw-r--r--src/leap/common/keymanager/keys.py4
-rw-r--r--src/leap/common/keymanager/openpgp.py112
-rw-r--r--src/leap/common/keymanager/util.py97
-rw-r--r--src/leap/common/tests/test_keymanager.py81
7 files changed, 362 insertions, 69 deletions
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 <http://www.gnu.org/licenses/>.
+"""
+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://www.gnu.org/licenses/>.
+
+
+"""
+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 <http://www.gnu.org/licenses/>.
+
+
+"""
+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(