1 files changed, 384 insertions, 0 deletions
diff --git a/tests/client/test_crypto.py b/tests/client/test_crypto.py
new file mode 100644
index 00000000..5b647b73
--- /dev/null
+++ b/tests/client/test_crypto.py
@@ -0,0 +1,384 @@
+# -*- coding: utf-8 -*-
+# test_crypto.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/>.
+"""
+Tests for cryptographic related stuff.
+"""
+import binascii
+import base64
+import json
+import os
+
+from io import BytesIO
+
+import pytest
+
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+from cryptography.hazmat.backends import default_backend
+from cryptography.exceptions import InvalidTag
+
+from leap.soledad.common.document import SoledadDocument
+from test_soledad.util import BaseSoledadTest
+from leap.soledad.client import _crypto
+from leap.soledad.client import _scrypt
+from leap.soledad.common.blobs import preamble as _preamble
+
+from twisted.trial import unittest
+from twisted.internet import defer
+
+
+snowden1 = (
+    "You can't come up against "
+    "the world's most powerful intelligence "
+    "agencies and not accept the risk. "
+    "If they want to get you, over time "
+    "they will.")
+
+
+class ScryptTest(unittest.TestCase):
+
+    def test_scrypt(self):
+        secret = 'supersikret'
+        salt = 'randomsalt'
+        key = _scrypt.hash(secret, salt, buflen=32)
+        expected = ('47996b569ea58d51ccbcc318d710'
+                    'a537acd28bb7a94615ab8d061d4b2a920f01')
+        assert binascii.b2a_hex(key) == expected
+
+
+class AESTest(unittest.TestCase):
+
+    def test_chunked_encryption(self):
+        key = 'A' * 32
+
+        fd = BytesIO()
+        aes = _crypto.AESWriter(key, _buffer=fd)
+        iv = aes.iv
+
+        data = snowden1
+        block = 16
+
+        for i in range(len(data) / block):
+            chunk = data[i * block:(i + 1) * block]
+            aes.write(chunk)
+        aes.end()
+
+        ciphertext_chunked = fd.getvalue()
+        ciphertext, tag = _aes_encrypt(key, iv, data)
+
+        assert ciphertext_chunked == ciphertext
+
+    def test_decrypt(self):
+        key = 'A' * 32
+        iv = 'A' * 16
+
+        data = snowden1
+        block = 16
+
+        ciphertext, tag = _aes_encrypt(key, iv, data)
+
+        fd = BytesIO()
+        aes = _crypto.AESWriter(key, iv, fd, tag=tag)
+
+        for i in range(len(ciphertext) / block):
+            chunk = ciphertext[i * block:(i + 1) * block]
+            aes.write(chunk)
+        aes.end()
+
+        cleartext_chunked = fd.getvalue()
+        assert cleartext_chunked == data
+
+
+class BlobTestCase(unittest.TestCase):
+
+    class doc_info:
+        doc_id = 'D-deadbeef'
+        rev = '397932e0c77f45fcb7c3732930e7e9b2:1'
+
+    def setUp(self):
+        self.inf = BytesIO(snowden1)
+        self.blob = _crypto.BlobEncryptor(
+            self.doc_info, self.inf,
+            armor=True,
+            secret='A' * 96)
+
+    @defer.inlineCallbacks
+    def test_unarmored_blob_encrypt(self):
+        self.blob.armor = False
+        encrypted = yield self.blob.encrypt()
+
+        decryptor = _crypto.BlobDecryptor(
+            self.doc_info, encrypted, armor=False,
+            secret='A' * 96)
+        decrypted = yield decryptor.decrypt()
+        assert decrypted.getvalue() == snowden1
+
+    @defer.inlineCallbacks
+    def test_default_armored_blob_encrypt(self):
+        encrypted = yield self.blob.encrypt()
+        decode = base64.urlsafe_b64decode
+        assert map(decode, encrypted.getvalue().split())
+
+    @defer.inlineCallbacks
+    def test_blob_encryptor(self):
+        encrypted = yield self.blob.encrypt()
+        preamble, ciphertext = encrypted.getvalue().split()
+        preamble = base64.urlsafe_b64decode(preamble)
+        ciphertext = base64.urlsafe_b64decode(ciphertext)
+        ciphertext = ciphertext[:-16]
+
+        assert len(preamble) == _preamble.PACMAN.size
+        unpacked_data = _preamble.PACMAN.unpack(preamble)
+        magic, sch, meth, ts, iv, doc_id, rev, _ = unpacked_data
+        assert magic == _crypto.MAGIC
+        assert sch == 1
+        assert meth == _crypto.ENC_METHOD.aes_256_gcm
+        assert iv == self.blob.iv
+        assert doc_id == 'D-deadbeef'
+        assert rev == self.doc_info.rev
+
+        aes_key = _crypto._get_sym_key_for_doc(
+            self.doc_info.doc_id, 'A' * 96)
+        assert ciphertext == _aes_encrypt(aes_key, self.blob.iv, snowden1)[0]
+
+        decrypted = _aes_decrypt(aes_key, self.blob.iv, self.blob.tag,
+                                 ciphertext, preamble)
+        assert str(decrypted) == snowden1
+
+    @defer.inlineCallbacks
+    def test_init_with_preamble_alone(self):
+        ciphertext = yield self.blob.encrypt()
+        preamble = ciphertext.getvalue().split()[0]
+        decryptor = _crypto.BlobDecryptor(
+            self.doc_info, BytesIO(preamble),
+            start_stream=False,
+            secret='A' * 96)
+        assert decryptor._consume_preamble()
+
+    @defer.inlineCallbacks
+    def test_incremental_blob_decryptor(self):
+        ciphertext = yield self.blob.encrypt()
+        preamble, ciphertext = ciphertext.getvalue().split()
+        ciphertext = base64.urlsafe_b64decode(ciphertext)
+
+        decryptor = _crypto.BlobDecryptor(
+            self.doc_info, BytesIO(preamble),
+            start_stream=False,
+            secret='A' * 96,
+            tag=ciphertext[-16:])
+        ciphertext = BytesIO(ciphertext[:-16])
+        chunk = ciphertext.read(10)
+        while chunk:
+            decryptor.write(chunk)
+            chunk = ciphertext.read(10)
+        decrypted = decryptor._end_stream()
+        assert decrypted.getvalue() == snowden1
+
+    @defer.inlineCallbacks
+    def test_blob_decryptor(self):
+        ciphertext = yield self.blob.encrypt()
+
+        decryptor = _crypto.BlobDecryptor(
+            self.doc_info, ciphertext,
+            secret='A' * 96)
+        decrypted = yield decryptor.decrypt()
+        assert decrypted.getvalue() == snowden1
+
+    @defer.inlineCallbacks
+    def test_unarmored_blob_decryptor(self):
+        self.blob.armor = False
+        ciphertext = yield self.blob.encrypt()
+
+        decryptor = _crypto.BlobDecryptor(
+            self.doc_info, ciphertext,
+            armor=False,
+            secret='A' * 96)
+        decrypted = yield decryptor.decrypt()
+        assert decrypted.getvalue() == snowden1
+
+    @defer.inlineCallbacks
+    def test_encrypt_and_decrypt(self):
+        """
+        Check that encrypting and decrypting gives same doc.
+        """
+        crypto = _crypto.SoledadCrypto('A' * 96)
+        payload = {'key': 'someval'}
+        doc1 = SoledadDocument('id1', '1', json.dumps(payload))
+
+        encrypted = yield crypto.encrypt_doc(doc1)
+        assert encrypted != payload
+        assert 'raw' in encrypted
+        doc2 = SoledadDocument('id1', '1')
+        doc2.set_json(encrypted)
+        assert _crypto.is_symmetrically_encrypted(encrypted)
+        decrypted = (yield crypto.decrypt_doc(doc2)).getvalue()
+        assert len(decrypted) != 0
+        assert json.loads(decrypted) == payload
+
+    @defer.inlineCallbacks
+    def test_decrypt_with_wrong_tag_raises(self):
+        """
+        Trying to decrypt a document with wrong MAC should raise.
+        """
+        crypto = _crypto.SoledadCrypto('A' * 96)
+        payload = {'key': 'someval'}
+        doc1 = SoledadDocument('id1', '1', json.dumps(payload))
+
+        encrypted = yield crypto.encrypt_doc(doc1)
+        encdict = json.loads(encrypted)
+        preamble, raw = str(encdict['raw']).split()
+        preamble = base64.urlsafe_b64decode(preamble)
+        raw = base64.urlsafe_b64decode(raw)
+        # mess with tag
+        messed = raw[:-16] + '0' * 16
+
+        preamble = base64.urlsafe_b64encode(preamble)
+        newraw = preamble + ' ' + base64.urlsafe_b64encode(str(messed))
+        doc2 = SoledadDocument('id1', '1')
+        doc2.set_json(json.dumps({"raw": str(newraw)}))
+
+        with pytest.raises(_crypto.InvalidBlob):
+            yield crypto.decrypt_doc(doc2)
+
+
+class SoledadSecretsTestCase(BaseSoledadTest):
+
+    def test_generated_secrets_have_correct_length(self):
+        expected = self._soledad.secrets.lengths
+        for name, length in expected.iteritems():
+            secret = getattr(self._soledad.secrets, name)
+            self.assertEqual(length, len(secret))
+
+
+class SoledadCryptoAESTestCase(BaseSoledadTest):
+
+    def test_encrypt_decrypt_sym(self):
+        # generate 256-bit key
+        key = os.urandom(32)
+        iv, cyphertext = _crypto.encrypt_sym('data', key)
+        self.assertTrue(cyphertext is not None)
+        self.assertTrue(cyphertext != '')
+        self.assertTrue(cyphertext != 'data')
+        plaintext = _crypto.decrypt_sym(cyphertext, key, iv)
+        self.assertEqual('data', plaintext)
+
+    def test_decrypt_with_wrong_iv_raises(self):
+        key = os.urandom(32)
+        iv, cyphertext = _crypto.encrypt_sym('data', key)
+        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:]
+        with pytest.raises(InvalidTag):
+            _crypto.decrypt_sym(
+                cyphertext, key, iv=binascii.b2a_base64(wrongiv))
+
+    def test_decrypt_with_wrong_key_raises(self):
+        key = os.urandom(32)
+        iv, cyphertext = _crypto.encrypt_sym('data', key)
+        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)
+        with pytest.raises(InvalidTag):
+            _crypto.decrypt_sym(cyphertext, wrongkey, iv)
+
+
+class PreambleTestCase(unittest.TestCase):
+    class doc_info:
+        doc_id = 'D-deadbeef'
+        rev = '397932e0c77f45fcb7c3732930e7e9b2:1'
+
+    def setUp(self):
+        self.cleartext = BytesIO(snowden1)
+        self.blob = _crypto.BlobEncryptor(
+            self.doc_info, self.cleartext,
+            secret='A' * 96)
+
+    def test_preamble_starts_with_magic_signature(self):
+        preamble = self.blob._encode_preamble()
+        assert preamble.startswith(_crypto.MAGIC)
+
+    def test_preamble_has_cipher_metadata(self):
+        preamble = self.blob._encode_preamble()
+        unpacked = _preamble.PACMAN.unpack(preamble)
+        encryption_scheme, encryption_method = unpacked[1:3]
+        assert encryption_scheme in _crypto.ENC_SCHEME
+        assert encryption_method in _crypto.ENC_METHOD
+        assert unpacked[4] == self.blob.iv
+
+    def test_preamble_has_document_sync_metadata(self):
+        preamble = self.blob._encode_preamble()
+        unpacked = _preamble.PACMAN.unpack(preamble)
+        doc_id, doc_rev = unpacked[5:7]
+        assert doc_id == self.doc_info.doc_id
+        assert doc_rev == self.doc_info.rev
+
+    def test_preamble_has_document_size(self):
+        preamble = self.blob._encode_preamble()
+        unpacked = _preamble.PACMAN.unpack(preamble)
+        size = unpacked[7]
+        assert size == _crypto._ceiling(len(snowden1))
+
+    @defer.inlineCallbacks
+    def test_preamble_can_come_without_size(self):
+        # XXX: This test case is here only to test backwards compatibility!
+        preamble = self.blob._encode_preamble()
+        # repack preamble using legacy format, without doc size
+        unpacked = _preamble.PACMAN.unpack(preamble)
+        preamble_without_size = _preamble.LEGACY_PACMAN.pack(*unpacked[0:7])
+        # encrypt it manually for custom tag
+        ciphertext, tag = _aes_encrypt(self.blob.sym_key, self.blob.iv,
+                                       self.cleartext.getvalue(),
+                                       aead=preamble_without_size)
+        ciphertext = ciphertext + tag
+        # encode it
+        ciphertext = base64.urlsafe_b64encode(ciphertext)
+        preamble_without_size = base64.urlsafe_b64encode(preamble_without_size)
+        # decrypt it
+        ciphertext = preamble_without_size + ' ' + ciphertext
+        cleartext = yield _crypto.BlobDecryptor(
+            self.doc_info, BytesIO(ciphertext),
+            secret='A' * 96).decrypt()
+        assert cleartext.getvalue() == self.cleartext.getvalue()
+        warnings = self.flushWarnings()
+        assert len(warnings) == 1
+        assert 'legacy preamble without size' in warnings[0]['message']
+
+
+def _aes_encrypt(key, iv, data, aead=''):
+    backend = default_backend()
+    cipher = Cipher(algorithms.AES(key), modes.GCM(iv), backend=backend)
+    encryptor = cipher.encryptor()
+    if aead:
+        encryptor.authenticate_additional_data(aead)
+    return encryptor.update(data) + encryptor.finalize(), encryptor.tag
+
+
+def _aes_decrypt(key, iv, tag, data, aead=''):
+    backend = default_backend()
+    cipher = Cipher(algorithms.AES(key), modes.GCM(iv, tag), backend=backend)
+    decryptor = cipher.decryptor()
+    if aead:
+        decryptor.authenticate_additional_data(aead)
+    return decryptor.update(data) + decryptor.finalize()