summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorVictor Shyba <victor1984@riseup.net>2017-03-15 08:26:56 -0300
committerdrebs <drebs@leap.se>2017-04-04 18:27:33 +0200
commit02cd9d60641cd40cfc6c7b8216d5861f40f92494 (patch)
tree6af9baeb5dec123b61dd059473291ce33756824a
parentb8d021a4f3c68da35a66f5c1f52ff4eb1f9ff6db (diff)
[feature] ascii armor disable option
-rw-r--r--client/src/leap/soledad/client/_crypto.py30
-rw-r--r--testing/tests/client/test_crypto.py52
2 files changed, 49 insertions, 33 deletions
diff --git a/client/src/leap/soledad/client/_crypto.py b/client/src/leap/soledad/client/_crypto.py
index 8dbdc094..923891fc 100644
--- a/client/src/leap/soledad/client/_crypto.py
+++ b/client/src/leap/soledad/client/_crypto.py
@@ -300,8 +300,7 @@ class BlobEncryptor(object):
result.write(
base64.urlsafe_b64encode(encrypted + self.tag))
else:
- result.write(encrypted)
- result.write(self.tag)
+ result.write(encrypted + self.tag)
result.seek(0)
return defer.succeed(result)
@@ -319,28 +318,25 @@ class BlobDecryptor(object):
# TODO enable the ascii armor = False
def __init__(self, doc_info, ciphertext_fd, result=None,
- secret=None, armor=True, start_stream=True):
+ secret=None, armor=True, start_stream=True, tag=None):
if not secret:
raise EncryptionDecryptionError('no secret given')
- if armor is False:
- raise NotImplementedError
self.doc_id = doc_info.doc_id
self.rev = doc_info.rev
self.fd = ciphertext_fd
self.armor = armor
self._producer = None
+ self.result = result or BytesIO()
+ self.sym_key = _get_sym_key_for_doc(doc_info.doc_id, secret)
self.size = None
-
+ self.tag = tag
preamble, iv = self._consume_preamble()
assert preamble
assert iv
-
- self.result = result or BytesIO()
- sym_key = _get_sym_key_for_doc(doc_info.doc_id, secret)
-
- self._aes = AESWriter(sym_key, iv, self.result, tag=self.tag)
+ self._aes = AESWriter(self.sym_key, iv, self.result, tag=self.tag)
self._aes.authenticate(preamble)
+
if start_stream:
self._start_stream()
@@ -348,14 +344,14 @@ class BlobDecryptor(object):
self._producer = FileBodyProducer(self.fd, readSize=2**16)
def _consume_preamble(self):
-
self.fd.seek(0)
try:
- parts = self.fd.getvalue().split()
- encoded_preamble = parts[0]
- preamble = base64.urlsafe_b64decode(encoded_preamble)
- ciphertext = base64.urlsafe_b64decode(parts[1])
- self.tag, ciphertext = ciphertext[-16:], ciphertext[:-16]
+ preamble, ciphertext = self.fd.getvalue().split(SEPARATOR, 1)
+ preamble = base64.urlsafe_b64decode(preamble)
+ if self.armor:
+ ciphertext = base64.urlsafe_b64decode(ciphertext)
+ tag, ciphertext = ciphertext[-16:], ciphertext[:-16]
+ self.tag = self.tag or tag
except (TypeError, ValueError):
raise InvalidBlob
diff --git a/testing/tests/client/test_crypto.py b/testing/tests/client/test_crypto.py
index e268428d..44e1b2a5 100644
--- a/testing/tests/client/test_crypto.py
+++ b/testing/tests/client/test_crypto.py
@@ -95,16 +95,30 @@ class BlobTestCase(unittest.TestCase):
doc_id = 'D-deadbeef'
rev = '397932e0c77f45fcb7c3732930e7e9b2:1'
- @defer.inlineCallbacks
- def test_blob_encryptor(self):
+ def setUp(self):
+ self.inf = BytesIO(snowden1)
+ self.blob = _crypto.BlobEncryptor(
+ self.doc_info, self.inf,
+ armor=True,
+ secret='A' * 96)
- inf = BytesIO(snowden1)
+ @defer.inlineCallbacks
+ def test_unarmored_blob_encrypt(self):
+ self.blob.armor = False
+ encrypted = yield self.blob.encrypt()
+ decode = base64.urlsafe_b64decode
+ with pytest.raises(TypeError):
+ assert map(decode, encrypted.getvalue().split())
- blob = _crypto.BlobEncryptor(
- self.doc_info, inf,
- secret='A' * 96)
+ @defer.inlineCallbacks
+ def test_default_armored_blob_encrypt(self):
+ encrypted = yield self.blob.encrypt()
+ decode = base64.urlsafe_b64decode
+ assert map(decode, encrypted.getvalue().split())
- encrypted = yield blob.encrypt()
+ @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)
@@ -116,30 +130,36 @@ class BlobTestCase(unittest.TestCase):
assert magic == _crypto.BLOB_SIGNATURE_MAGIC
assert sch == 1
assert meth == _crypto.ENC_METHOD.aes_256_gcm
- assert iv == blob.iv
+ 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, blob.iv, snowden1)[0]
+ assert ciphertext == _aes_encrypt(aes_key, self.blob.iv, snowden1)[0]
- decrypted = _aes_decrypt(aes_key, blob.iv, blob.tag, ciphertext,
- preamble)
+ decrypted = _aes_decrypt(aes_key, self.blob.iv, self.blob.tag,
+ ciphertext, preamble)
assert str(decrypted) == snowden1
@defer.inlineCallbacks
def test_blob_decryptor(self):
+ ciphertext = yield self.blob.encrypt()
- inf = BytesIO(snowden1)
-
- blob = _crypto.BlobEncryptor(
- self.doc_info, inf,
+ decryptor = _crypto.BlobDecryptor(
+ self.doc_info, ciphertext,
secret='A' * 96)
- ciphertext = yield blob.encrypt()
+ 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