diff options
| -rw-r--r-- | client/changes/feat-7338_refactor_crypto | 1 | ||||
| -rw-r--r-- | client/src/leap/soledad/client/api.py | 6 | ||||
| -rw-r--r-- | client/src/leap/soledad/client/crypto.py | 180 | ||||
| -rw-r--r-- | client/src/leap/soledad/client/http_target.py | 15 | ||||
| -rw-r--r-- | client/src/leap/soledad/client/secrets.py | 10 | ||||
| -rw-r--r-- | common/src/leap/soledad/common/crypto.py | 1 | ||||
| -rw-r--r-- | common/src/leap/soledad/common/tests/test_crypto.py | 89 |
7 files changed, 82 insertions, 220 deletions
diff --git a/client/changes/feat-7338_refactor_crypto b/client/changes/feat-7338_refactor_crypto new file mode 100644 index 00000000..d5afa96c --- /dev/null +++ b/client/changes/feat-7338_refactor_crypto @@ -0,0 +1 @@ +- refactor SoledadCrypto to remove circular dependency with SoledadSecrets (Closes: 7338) diff --git a/client/src/leap/soledad/client/api.py b/client/src/leap/soledad/client/api.py index 3e9f6bd4..57df4021 100644 --- a/client/src/leap/soledad/client/api.py +++ b/client/src/leap/soledad/client/api.py @@ -183,9 +183,7 @@ class Soledad(object): global SOLEDAD_CERT SOLEDAD_CERT = cert_file - # init crypto variables self._set_token(auth_token) - self._crypto = SoledadCrypto(self) self._init_config_with_defaults() self._init_working_dirs() @@ -200,6 +198,8 @@ class Soledad(object): self._init_secrets() self._init_u1db_sqlcipher_backend() + self._crypto = SoledadCrypto(self._secrets.remote_storage_secret) + if syncable: self._init_u1db_syncer() @@ -240,7 +240,7 @@ class Soledad(object): """ self._secrets = SoledadSecrets( self.uuid, self._passphrase, self._secrets_path, - self.shared_db, self._crypto) + self.shared_db) self._secrets.bootstrap() def _init_u1db_sqlcipher_backend(self): diff --git a/client/src/leap/soledad/client/crypto.py b/client/src/leap/soledad/client/crypto.py index 6328fd2b..90ad656e 100644 --- a/client/src/leap/soledad/client/crypto.py +++ b/client/src/leap/soledad/client/crypto.py @@ -25,7 +25,6 @@ import json import logging from pycryptopp.cipher.aes import AES -from pycryptopp.cipher.xsalsa20 import XSalsa20 from leap.soledad.common import soledad_assert from leap.soledad.common import soledad_assert_type @@ -38,100 +37,52 @@ logger = logging.getLogger(__name__) MAC_KEY_LENGTH = 64 -def _assert_known_encryption_method(method): +def encrypt_sym(data, key): """ - Assert that we can encrypt/decrypt the given C{method} - - :param method: The encryption method to assert. - :type method: str - - :raise UnknownEncryptionMethodError: Raised when C{method} is unknown. - """ - valid_methods = [ - crypto.EncryptionMethods.AES_256_CTR, - crypto.EncryptionMethods.XSALSA20, - ] - try: - soledad_assert(method in valid_methods) - except AssertionError: - raise crypto.UnknownEncryptionMethodError - - -def encrypt_sym(data, key, method): - """ - Encrypt C{data} using a {password}. - - Currently, the only encryption methods supported are AES-256 in CTR - mode and XSalsa20. + Encrypt data using AES-256 cipher in CTR mode. :param data: The data to be encrypted. :type data: str - :param key: The key used to encrypt C{data} (must be 256 bits long). + :param key: The key used to encrypt data (must be 256 bits long). :type key: str - :param method: The encryption method to use. - :type method: str - :return: A tuple with the initial value and the encrypted data. + :return: A tuple with the initialization vector and the encrypted data. :rtype: (long, str) - - :raise AssertionError: Raised if C{method} is unknown. """ soledad_assert_type(key, str) soledad_assert( len(key) == 32, # 32 x 8 = 256 bits. 'Wrong key size: %s bits (must be 256 bits long).' % (len(key) * 8)) - _assert_known_encryption_method(method) - - iv = None - # AES-256 in CTR mode - if method == crypto.EncryptionMethods.AES_256_CTR: - iv = os.urandom(16) - ciphertext = AES(key=key, iv=iv).process(data) - # XSalsa20 - elif method == crypto.EncryptionMethods.XSALSA20: - iv = os.urandom(24) - ciphertext = XSalsa20(key=key, iv=iv).process(data) + + iv = os.urandom(16) + ciphertext = AES(key=key, iv=iv).process(data) return binascii.b2a_base64(iv), ciphertext -def decrypt_sym(data, key, method, **kwargs): +def decrypt_sym(data, key, iv): """ - Decrypt data using symmetric secret. - - Currently, the only encryption method supported is AES-256 CTR mode. + Decrypt some data previously encrypted using AES-256 cipher in CTR mode. :param data: The data to be decrypted. :type data: str - :param key: The key used to decrypt C{data} (must be 256 bits long). + :param key: The symmetric key used to decrypt data (must be 256 bits + long). :type key: str - :param method: The encryption method to use. - :type method: str - :param kwargs: Other parameters specific to each encryption method. - :type kwargs: dict + :param iv: The initialization vector. + :type iv: long :return: The decrypted data. :rtype: str - - :raise UnknownEncryptionMethodError: Raised when C{method} is unknown. """ soledad_assert_type(key, str) # assert params soledad_assert( len(key) == 32, # 32 x 8 = 256 bits. 'Wrong key size: %s (must be 256 bits long).' % len(key)) - soledad_assert( - 'iv' in kwargs, - '%s needs an initial value.' % method) - _assert_known_encryption_method(method) - # AES-256 in CTR mode - if method == crypto.EncryptionMethods.AES_256_CTR: - return AES( - key=key, iv=binascii.a2b_base64(kwargs['iv'])).process(data) - elif method == crypto.EncryptionMethods.XSALSA20: - return XSalsa20( - key=key, iv=binascii.a2b_base64(kwargs['iv'])).process(data) + return AES( + key=key, iv=binascii.a2b_base64(iv)).process(data) def doc_mac_key(doc_id, secret): @@ -164,25 +115,17 @@ class SoledadCrypto(object): General cryptographic functionality encapsulated in a object that can be passed along. """ - def __init__(self, soledad): + def __init__(self, secret): """ Initialize the crypto object. - :param soledad: A Soledad instance for key lookup. - :type soledad: leap.soledad.Soledad + :param secret: The Soledad remote storage secret. + :type secret: str """ - self._soledad = soledad - - def encrypt_sym(self, data, key, - method=crypto.EncryptionMethods.AES_256_CTR): - return encrypt_sym(data, key, method) - - def decrypt_sym(self, data, key, - method=crypto.EncryptionMethods.AES_256_CTR, **kwargs): - return decrypt_sym(data, key, method, **kwargs) + self._secret = secret - def doc_mac_key(self, doc_id, secret): - return doc_mac_key(doc_id, self.secret) + def doc_mac_key(self, doc_id): + return doc_mac_key(doc_id, self._secret) def doc_passphrase(self, doc_id): """ @@ -201,21 +144,41 @@ class SoledadCrypto(object): :return: The passphrase. :rtype: str """ - soledad_assert(self.secret is not None) + soledad_assert(self._secret is not None) return hmac.new( - self.secret[MAC_KEY_LENGTH:], + self._secret[MAC_KEY_LENGTH:], doc_id, hashlib.sha256).digest() - # - # secret setters/getters - # + def encrypt_doc(self, doc): + """ + Wrapper around encrypt_docstr that accepts the document as argument. - def _get_secret(self): - return self._soledad.secrets.remote_storage_secret + :param doc: the document. + :type doc: SoledadDocument + """ + key = self.doc_passphrase(doc.doc_id) - secret = property( - _get_secret, doc='The secret used for symmetric encryption') + return encrypt_docstr( + doc.get_json(), doc.doc_id, doc.rev, key, self._secret) + + def decrypt_doc(self, doc): + """ + Wrapper around decrypt_doc_dict that accepts the document as argument. + + :param doc: the document. + :type doc: SoledadDocument + + :return: json string with the decrypted document + :rtype: str + """ + key = self.doc_passphrase(doc.doc_id) + return decrypt_doc_dict( + doc.content, doc.doc_id, doc.rev, key, self._secret) + + @property + def secret(self): + return self._secret # @@ -273,23 +236,6 @@ def mac_doc(doc_id, doc_rev, ciphertext, enc_scheme, enc_method, enc_iv, hashlib.sha256).digest() -def encrypt_doc(crypto, doc): - """ - Wrapper around encrypt_docstr that accepts a crypto object and the document - as arguments. - - :param crypto: a soledad crypto object. - :type crypto: SoledadCrypto - :param doc: the document. - :type doc: SoledadDocument - """ - key = crypto.doc_passphrase(doc.doc_id) - secret = crypto.secret - - return encrypt_docstr( - doc.get_json(), doc.doc_id, doc.rev, key, secret) - - def encrypt_docstr(docstr, doc_id, doc_rev, key, secret): """ Encrypt C{doc}'s content. @@ -330,7 +276,7 @@ def encrypt_docstr(docstr, doc_id, doc_rev, key, secret): mac_method = crypto.MacMethods.HMAC enc_iv, ciphertext = encrypt_sym( str(docstr), # encryption/decryption routines expect str - key, method=enc_method) + key) mac = binascii.b2a_hex( # store the mac as hex. mac_doc( doc_id, @@ -356,24 +302,6 @@ def encrypt_docstr(docstr, doc_id, doc_rev, key, secret): }) -def decrypt_doc(crypto, doc): - """ - Wrapper around decrypt_doc_dict that accepts a crypto object and the - document as arguments. - - :param crypto: a soledad crypto object. - :type crypto: SoledadCrypto - :param doc: the document. - :type doc: SoledadDocument - - :return: json string with the decrypted document - :rtype: str - """ - key = crypto.doc_passphrase(doc.doc_id) - secret = crypto.secret - return decrypt_doc_dict(doc.content, doc.doc_id, doc.rev, key, secret) - - def _verify_doc_mac(doc_id, doc_rev, ciphertext, enc_scheme, enc_method, enc_iv, mac_method, secret, doc_mac): """ @@ -458,8 +386,8 @@ def decrypt_doc_dict(doc_dict, doc_id, doc_rev, key, secret): :param key: The key used to encrypt ``data`` (must be 256 bits long). :type key: str - :param secret: - :type secret: + :param secret: The Soledad storage secret. + :type secret: str :return: The JSON serialization of the decrypted content. :rtype: str @@ -491,7 +419,7 @@ def decrypt_doc_dict(doc_dict, doc_id, doc_rev, key, secret): doc_id, doc_rev, ciphertext, enc_scheme, enc_method, enc_iv, mac_method, secret, doc_mac) - return decrypt_sym(ciphertext, key, method=enc_method, iv=enc_iv) + return decrypt_sym(ciphertext, key, enc_iv) def is_symmetrically_encrypted(doc): diff --git a/client/src/leap/soledad/client/http_target.py b/client/src/leap/soledad/client/http_target.py index 3dc45e20..f7f50af6 100644 --- a/client/src/leap/soledad/client/http_target.py +++ b/client/src/leap/soledad/client/http_target.py @@ -41,8 +41,6 @@ from leap.soledad.common.document import SoledadDocument from leap.soledad.common.errors import InvalidAuthTokenError from leap.soledad.client.crypto import is_symmetrically_encrypted -from leap.soledad.client.crypto import encrypt_doc -from leap.soledad.client.crypto import decrypt_doc from leap.soledad.client.events import SOLEDAD_SYNC_SEND_STATUS from leap.soledad.client.events import SOLEDAD_SYNC_RECEIVE_STATUS from leap.soledad.client.events import emit @@ -89,9 +87,10 @@ class SoledadHTTPSyncTarget(SyncTarget): instead of retreiving it from the dedicated database. :type sync_db: Sqlite handler - :param verify_ssl: Whether we should perform SSL server certificate - verification. - :type verify_ssl: bool + :param sync_enc_pool: The encryption pool to use to defer encryption. + If None is passed the encryption will not be + deferred. + :type sync_enc_pool: leap.soledad.client.encdecpool.SyncEncrypterPool """ if url.endswith("/"): url = url[:-1] @@ -347,7 +346,7 @@ class SoledadHTTPSyncTarget(SyncTarget): d = defer.succeed(None) elif not self._defer_encryption: # fallback case, for tests - d = defer.succeed(encrypt_doc(self._crypto, doc)) + d = defer.succeed(self._crypto.encrypt_doc(doc)) else: def _maybe_encrypt_doc_inline(doc_json): @@ -355,7 +354,7 @@ class SoledadHTTPSyncTarget(SyncTarget): # the document is not marked as tombstone, but we got # nothing from the sync db. As it is not encrypted # yet, we force inline encryption. - return encrypt_doc(self._crypto, doc) + return self._crypto.encrypt_doc(doc) return doc_json d = self._sync_enc_pool.get_encrypted_doc(doc.doc_id, doc.rev) @@ -491,7 +490,7 @@ class SoledadHTTPSyncTarget(SyncTarget): idx) else: # defer_decryption is False or no-sync-db fallback - doc.set_json(decrypt_doc(self._crypto, doc)) + doc.set_json(self._crypto.decrypt_doc(doc)) self._insert_doc_cb(doc, gen, trans_id) else: # not symmetrically encrypted doc, insert it directly diff --git a/client/src/leap/soledad/client/secrets.py b/client/src/leap/soledad/client/secrets.py index 97dbbaca..e55d64c6 100644 --- a/client/src/leap/soledad/client/secrets.py +++ b/client/src/leap/soledad/client/secrets.py @@ -37,6 +37,7 @@ from leap.soledad.common import soledad_assert_type from leap.soledad.common import document from leap.soledad.common import errors from leap.soledad.client import events +from leap.soledad.client.crypto import encrypt_sym, decrypt_sym logger = logging.getLogger(name=__name__) @@ -148,7 +149,7 @@ class SoledadSecrets(object): Keys used to access storage secrets in recovery documents. """ - def __init__(self, uuid, passphrase, secrets_path, shared_db, crypto): + def __init__(self, uuid, passphrase, secrets_path, shared_db): """ Initialize the secrets manager. @@ -162,8 +163,6 @@ class SoledadSecrets(object): :type secrets_path: str :param shared_db: The shared database that stores user secrets. :type shared_db: leap.soledad.client.shared_db.SoledadSharedDatabase - :param crypto: A soledad crypto object. - :type crypto: SoledadCrypto """ # XXX removed since not in use # We will pick the first secret available. @@ -173,7 +172,6 @@ class SoledadSecrets(object): self._passphrase = passphrase self._secrets_path = secrets_path self._shared_db = shared_db - self._crypto = crypto self._secrets = {} self._secret_id = None @@ -511,7 +509,7 @@ class SoledadSecrets(object): iv, ciphertext = encrypted_secret_dict[self.SECRET_KEY].split( self.IV_SEPARATOR, 1) ciphertext = binascii.a2b_base64(ciphertext) - decrypted_secret = self._crypto.decrypt_sym(ciphertext, key, iv=iv) + decrypted_secret = decrypt_sym(ciphertext, key, iv) if encrypted_secret_dict[self.LENGTH_KEY] != len(decrypted_secret): raise SecretsException("Wrong length of decrypted secret.") return decrypted_secret @@ -543,7 +541,7 @@ class SoledadSecrets(object): salt = os.urandom(self.SALT_LENGTH) # get a 256-bit key key = scrypt.hash(self._passphrase_as_string(), salt, buflen=32) - iv, ciphertext = self._crypto.encrypt_sym(decrypted_secret, key) + iv, ciphertext = encrypt_sym(decrypted_secret, key) encrypted_secret_dict = { # leap.soledad.crypto submodule uses AES256 for symmetric # encryption. diff --git a/common/src/leap/soledad/common/crypto.py b/common/src/leap/soledad/common/crypto.py index b4f3234f..77a5a93b 100644 --- a/common/src/leap/soledad/common/crypto.py +++ b/common/src/leap/soledad/common/crypto.py @@ -48,7 +48,6 @@ class EncryptionMethods(object): """ AES_256_CTR = 'aes-256-ctr' - XSALSA20 = 'xsalsa20' class UnknownEncryptionMethodError(Exception): diff --git a/common/src/leap/soledad/common/tests/test_crypto.py b/common/src/leap/soledad/common/tests/test_crypto.py index fdad8aac..55ecbdd7 100644 --- a/common/src/leap/soledad/common/tests/test_crypto.py +++ b/common/src/leap/soledad/common/tests/test_crypto.py @@ -47,7 +47,7 @@ class EncryptedSyncTestCase(BaseSoledadTest): doc1.content = simpledoc # encrypt doc - doc1.set_json(crypto.encrypt_doc(self._soledad._crypto, doc1)) + doc1.set_json(self._soledad._crypto.encrypt_doc(doc1)) # assert content is different and includes keys self.assertNotEqual( simpledoc, doc1.content, @@ -55,7 +55,7 @@ class EncryptedSyncTestCase(BaseSoledadTest): self.assertTrue(ENC_JSON_KEY in doc1.content) self.assertTrue(ENC_SCHEME_KEY in doc1.content) # decrypt doc - doc1.set_json(crypto.decrypt_doc(self._soledad._crypto, doc1)) + doc1.set_json(self._soledad._crypto.decrypt_doc(doc1)) self.assertEqual( simpledoc, doc1.content, 'incorrect document encryption') @@ -153,7 +153,7 @@ class MacAuthTestCase(BaseSoledadTest): doc = SoledadDocument(doc_id='id') doc.content = simpledoc # encrypt doc - doc.set_json(crypto.encrypt_doc(self._soledad._crypto, doc)) + doc.set_json(self._soledad._crypto.encrypt_doc(doc)) self.assertTrue(MAC_KEY in doc.content) self.assertTrue(MAC_METHOD_KEY in doc.content) # mess with MAC @@ -161,7 +161,7 @@ class MacAuthTestCase(BaseSoledadTest): # try to decrypt doc self.assertRaises( WrongMacError, - crypto.decrypt_doc, self._soledad._crypto, doc) + self._soledad._crypto.decrypt_doc, doc) def test_decrypt_with_unknown_mac_method_raises(self): """ @@ -171,7 +171,7 @@ class MacAuthTestCase(BaseSoledadTest): doc = SoledadDocument(doc_id='id') doc.content = simpledoc # encrypt doc - doc.set_json(crypto.encrypt_doc(self._soledad._crypto, doc)) + doc.set_json(self._soledad._crypto.encrypt_doc(doc)) self.assertTrue(MAC_KEY in doc.content) self.assertTrue(MAC_METHOD_KEY in doc.content) # mess with MAC method @@ -179,7 +179,7 @@ class MacAuthTestCase(BaseSoledadTest): # try to decrypt doc self.assertRaises( UnknownMacMethodError, - crypto.decrypt_doc, self._soledad._crypto, doc) + self._soledad._crypto.decrypt_doc, doc) class SoledadCryptoAESTestCase(BaseSoledadTest): @@ -187,22 +187,16 @@ class SoledadCryptoAESTestCase(BaseSoledadTest): def test_encrypt_decrypt_sym(self): # generate 256-bit key key = os.urandom(32) - iv, cyphertext = self._soledad._crypto.encrypt_sym( - 'data', key, - method=EncryptionMethods.AES_256_CTR) + iv, cyphertext = crypto.encrypt_sym('data', key) self.assertTrue(cyphertext is not None) self.assertTrue(cyphertext != '') self.assertTrue(cyphertext != 'data') - plaintext = self._soledad._crypto.decrypt_sym( - cyphertext, key, iv=iv, - method=EncryptionMethods.AES_256_CTR) + plaintext = crypto.decrypt_sym(cyphertext, key, iv) self.assertEqual('data', plaintext) def test_decrypt_with_wrong_iv_fails(self): key = os.urandom(32) - iv, cyphertext = self._soledad._crypto.encrypt_sym( - 'data', key, - method=EncryptionMethods.AES_256_CTR) + iv, cyphertext = crypto.encrypt_sym('data', key) self.assertTrue(cyphertext is not None) self.assertTrue(cyphertext != '') self.assertTrue(cyphertext != 'data') @@ -211,16 +205,13 @@ class SoledadCryptoAESTestCase(BaseSoledadTest): wrongiv = rawiv while wrongiv == rawiv: wrongiv = os.urandom(1) + rawiv[1:] - plaintext = self._soledad._crypto.decrypt_sym( - cyphertext, key, iv=binascii.b2a_base64(wrongiv), - method=EncryptionMethods.AES_256_CTR) + plaintext = crypto.decrypt_sym( + cyphertext, key, iv=binascii.b2a_base64(wrongiv)) self.assertNotEqual('data', plaintext) def test_decrypt_with_wrong_key_fails(self): key = os.urandom(32) - iv, cyphertext = self._soledad._crypto.encrypt_sym( - 'data', key, - method=EncryptionMethods.AES_256_CTR) + iv, cyphertext = crypto.encrypt_sym('data', key) self.assertTrue(cyphertext is not None) self.assertTrue(cyphertext != '') self.assertTrue(cyphertext != 'data') @@ -228,59 +219,5 @@ class SoledadCryptoAESTestCase(BaseSoledadTest): # ensure keys are different in case we are extremely lucky while wrongkey == key: wrongkey = os.urandom(32) - plaintext = self._soledad._crypto.decrypt_sym( - cyphertext, wrongkey, iv=iv, - method=EncryptionMethods.AES_256_CTR) - self.assertNotEqual('data', plaintext) - - -class SoledadCryptoXSalsa20TestCase(BaseSoledadTest): - - def test_encrypt_decrypt_sym(self): - # generate 256-bit key - key = os.urandom(32) - iv, cyphertext = self._soledad._crypto.encrypt_sym( - 'data', key, - method=EncryptionMethods.XSALSA20) - self.assertTrue(cyphertext is not None) - self.assertTrue(cyphertext != '') - self.assertTrue(cyphertext != 'data') - plaintext = self._soledad._crypto.decrypt_sym( - cyphertext, key, iv=iv, - method=EncryptionMethods.XSALSA20) - self.assertEqual('data', plaintext) - - def test_decrypt_with_wrong_iv_fails(self): - key = os.urandom(32) - iv, cyphertext = self._soledad._crypto.encrypt_sym( - 'data', key, - method=EncryptionMethods.XSALSA20) - self.assertTrue(cyphertext is not None) - self.assertTrue(cyphertext != '') - self.assertTrue(cyphertext != 'data') - # 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 = self._soledad._crypto.decrypt_sym( - cyphertext, key, iv=binascii.b2a_base64(wrongiv), - method=EncryptionMethods.XSALSA20) - self.assertNotEqual('data', plaintext) - - def test_decrypt_with_wrong_key_fails(self): - key = os.urandom(32) - iv, cyphertext = self._soledad._crypto.encrypt_sym( - 'data', key, - method=EncryptionMethods.XSALSA20) - self.assertTrue(cyphertext is not None) - self.assertTrue(cyphertext != '') - self.assertTrue(cyphertext != 'data') - wrongkey = os.urandom(32) # 256-bits key - # ensure keys are different in case we are extremely lucky - while wrongkey == key: - wrongkey = os.urandom(32) - plaintext = self._soledad._crypto.decrypt_sym( - cyphertext, wrongkey, iv=iv, - method=EncryptionMethods.XSALSA20) + plaintext = crypto.decrypt_sym(cyphertext, wrongkey, iv) self.assertNotEqual('data', plaintext) |