summaryrefslogtreecommitdiff
path: root/src/leap
diff options
context:
space:
mode:
Diffstat (limited to 'src/leap')
-rw-r--r--src/leap/common/keymanager/__init__.py37
-rw-r--r--src/leap/common/keymanager/keys.py64
-rw-r--r--src/leap/common/keymanager/openpgp.py26
-rw-r--r--src/leap/common/tests/test_crypto.py12
-rw-r--r--src/leap/common/tests/test_keymanager.py17
5 files changed, 69 insertions, 87 deletions
diff --git a/src/leap/common/keymanager/__init__.py b/src/leap/common/keymanager/__init__.py
index f5c5f5e..9435cea 100644
--- a/src/leap/common/keymanager/__init__.py
+++ b/src/leap/common/keymanager/__init__.py
@@ -35,6 +35,7 @@ from leap.common.keymanager.errors import (
from leap.common.keymanager.keys import (
build_key_from_dict,
KEYMANAGER_KEY_TAG,
+ TAGS_PRIVATE_INDEX,
)
from leap.common.keymanager.openpgp import (
OpenPGPKey,
@@ -43,18 +44,6 @@ from leap.common.keymanager.openpgp import (
#
-# key indexing constants.
-#
-
-TAGS_INDEX = 'by-tags'
-TAGS_AND_PRIVATE_INDEX = 'by-tags-and-private'
-INDEXES = {
- TAGS_INDEX: ['tags'],
- TAGS_AND_PRIVATE_INDEX: ['tags', 'bool(private)'],
-}
-
-
-#
# The Key Manager
#
@@ -103,8 +92,6 @@ class KeyManager(object):
OpenPGPKey: OpenPGPScheme(soledad),
# other types of key will be added to this mapper.
}
- # initialize the indexes needed to query the database
- self._init_indexes()
# the following are used to perform https requests
self._fetcher = requests
self._session = self._fetcher.session()
@@ -121,26 +108,6 @@ class KeyManager(object):
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 _get(self, uri, data=None):
"""
Send a GET request to C{uri} containing C{data}.
@@ -294,7 +261,7 @@ class KeyManager(object):
doc.content['address'],
doc.content),
self._soledad.get_from_index(
- TAGS_AND_PRIVATE_INDEX,
+ TAGS_PRIVATE_INDEX,
KEYMANAGER_KEY_TAG,
'1' if private else '0'))
diff --git a/src/leap/common/keymanager/keys.py b/src/leap/common/keymanager/keys.py
index f2c1beb..a3c8537 100644
--- a/src/leap/common/keymanager/keys.py
+++ b/src/leap/common/keymanager/keys.py
@@ -28,7 +28,6 @@ except ImportError:
import re
-from hashlib import sha256
from abc import ABCMeta, abstractmethod
from leap.common.check import leap_assert
@@ -57,6 +56,26 @@ KEY_TAGS_KEY = 'tags'
KEYMANAGER_KEY_TAG = 'keymanager-key'
+
+#
+# key indexing constants.
+#
+
+TAGS_PRIVATE_INDEX = 'by-tags-private'
+TAGS_ADDRESS_PRIVATE_INDEX = 'by-tags-address-private'
+INDEXES = {
+ TAGS_PRIVATE_INDEX: [
+ KEY_TAGS_KEY,
+ 'bool(%s)' % KEY_PRIVATE_KEY,
+ ],
+ TAGS_ADDRESS_PRIVATE_INDEX: [
+ KEY_TAGS_KEY,
+ KEY_ADDRESS_KEY,
+ 'bool(%s)' % KEY_PRIVATE_KEY,
+ ]
+}
+
+
#
# Key handling utilities
#
@@ -100,28 +119,6 @@ def build_key_from_dict(kClass, address, kdict):
validation=kdict[KEY_VALIDATION_KEY], # 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 = KEY_PRIVATE_KEY if private else 'public'
- return sha256('keymanager-'+address+'-'+ktype+'-'+visibility).hexdigest()
-
-
#
# Abstraction for encryption keys
#
@@ -215,6 +212,27 @@ class EncryptionScheme(object):
:type soledad: leap.soledad.Soledad
"""
self._soledad = soledad
+ self._init_indexes()
+
+ 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)
@abstractmethod
def get_key(self, address, private=False):
diff --git a/src/leap/common/keymanager/openpgp.py b/src/leap/common/keymanager/openpgp.py
index b08bff6..dd11157 100644
--- a/src/leap/common/keymanager/openpgp.py
+++ b/src/leap/common/keymanager/openpgp.py
@@ -26,13 +26,13 @@ import tempfile
from leap.common.check import leap_assert, leap_assert_type
from leap.common.keymanager import errors
-
from leap.common.keymanager.keys import (
EncryptionKey,
EncryptionScheme,
is_address,
- keymanager_doc_id,
build_key_from_dict,
+ KEYMANAGER_KEY_TAG,
+ TAGS_ADDRESS_PRIVATE_INDEX,
)
from leap.common.keymanager.gpg import GPGWrapper
@@ -588,10 +588,7 @@ class OpenPGPScheme(EncryptionScheme):
"""
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(
- OpenPGPKey, key.address, key.private))
+ self._soledad.create_doc_from_json(key.get_json())
else:
doc.set_json(key.get_json())
self._soledad.put_doc(doc)
@@ -609,8 +606,18 @@ class OpenPGPScheme(EncryptionScheme):
: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(OpenPGPKey, address, private))
+ doclist = self._soledad.get_from_index(
+ TAGS_ADDRESS_PRIVATE_INDEX,
+ KEYMANAGER_KEY_TAG,
+ address,
+ '1' if private else '0')
+ if len(doclist) is 0:
+ return None
+ leap_assert(
+ len(doclist) is 1,
+ 'Found more than one %s key for address!' %
+ 'private' if private else 'public')
+ return doclist.pop()
def delete_key(self, key):
"""
@@ -625,6 +632,5 @@ class OpenPGPScheme(EncryptionScheme):
raise errors.KeyNotFound(key)
if stored_key.__dict__ != key.__dict__:
raise errors.KeyAttributesDiffer(key)
- doc = self._soledad.get_doc(
- keymanager_doc_id(OpenPGPKey, key.address, key.private))
+ doc = self._get_key_doc(key.address, key.private)
self._soledad.delete_doc(doc)
diff --git a/src/leap/common/tests/test_crypto.py b/src/leap/common/tests/test_crypto.py
index b704c05..ae7dc71 100644
--- a/src/leap/common/tests/test_crypto.py
+++ b/src/leap/common/tests/test_crypto.py
@@ -21,6 +21,10 @@ Tests for the crypto submodule.
"""
+import os
+import binascii
+
+
from leap.common.testing.basetest import BaseLeapTest
from leap.common import crypto
from Crypto import Random
@@ -56,9 +60,13 @@ class CryptoTestCase(BaseLeapTest):
self.assertTrue(cyphertext is not None)
self.assertTrue(cyphertext != '')
self.assertTrue(cyphertext != 'data')
- iv += 1
+ # get a different iv by changing the first byte
+ rawiv = binascii.a2b_base64(iv)
+ wrongiv = rawiv
+ while wrongiv == rawiv:
+ wrongiv = os.urandom(1) + rawiv[1:]
plaintext = crypto.decrypt_sym(
- cyphertext, key, iv=iv,
+ cyphertext, key, iv=binascii.b2a_base64(wrongiv),
method=crypto.EncryptionMethods.AES_256_CTR)
self.assertNotEqual('data', plaintext)
diff --git a/src/leap/common/tests/test_keymanager.py b/src/leap/common/tests/test_keymanager.py
index cffa073..73611b6 100644
--- a/src/leap/common/tests/test_keymanager.py
+++ b/src/leap/common/tests/test_keymanager.py
@@ -44,7 +44,6 @@ from leap.common.keymanager.openpgp import OpenPGPKey
from leap.common.keymanager.keys import (
is_address,
build_key_from_dict,
- keymanager_doc_id,
)
from leap.common.keymanager import errors
@@ -120,22 +119,6 @@ class KeyManagerUtilTestCase(BaseLeapTest):
kdict['validation'], key.validation,
'Wrong data in key.')
- def test_keymanager_doc_id(self):
- doc_id1 = keymanager_doc_id(
- OpenPGPKey, ADDRESS, private=False)
- doc_id2 = keymanager_doc_id(
- OpenPGPKey, ADDRESS, 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!')
- 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 KeyManagerWithSoledadTestCase(BaseLeapTest):
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-01 02:01:32 +0000