summaryrefslogtreecommitdiff
path: root/src/leap/bitmask/mail/incoming/tests
diff options
context:
space:
mode:
Diffstat (limited to 'src/leap/bitmask/mail/incoming/tests')
-rw-r--r--src/leap/bitmask/mail/incoming/tests/rfc822.multi-encrypt-signed.message61
-rw-r--r--src/leap/bitmask/mail/incoming/tests/test_incoming_mail.py391
2 files changed, 452 insertions, 0 deletions
diff --git a/src/leap/bitmask/mail/incoming/tests/rfc822.multi-encrypt-signed.message b/src/leap/bitmask/mail/incoming/tests/rfc822.multi-encrypt-signed.message
new file mode 100644
index 0000000..98304f2
--- /dev/null
+++ b/src/leap/bitmask/mail/incoming/tests/rfc822.multi-encrypt-signed.message
@@ -0,0 +1,61 @@
+Content-Type: multipart/encrypted;
+ boundary="Apple-Mail=_C01A1464-6C43-43BF-8F62-157335B7E25B";
+ protocol="application/pgp-encrypted";
+Subject: Enc signed
+Mime-Version: 1.0 (Mac OS X Mail 9.3 \(3124\))
+From: Leap Test Key <leap@leap.se>
+Date: Tue, 24 May 2016 11:47:24 -0300
+Content-Description: OpenPGP encrypted message
+To: leap@leap.se
+
+This is an OpenPGP/MIME encrypted message (RFC 2440 and 3156)
+--Apple-Mail=_C01A1464-6C43-43BF-8F62-157335B7E25B
+Content-Type: application/pgp-encrypted
+Content-Description: PGP/MIME Versions Identification
+
+--Apple-Mail=_C01A1464-6C43-43BF-8F62-157335B7E25B
+Content-Disposition: inline;
+ filename=encrypted.asc
+Content-Type: application/octet-stream;
+ name=encrypted.asc
+Content-Description: OpenPGP encrypted message
+
+-----BEGIN PGP MESSAGE-----
+Version: GnuPG v2
+
+hQIMAyj9aG/xtZOwAQ/9Gft0KmOpgzL6z4wmVlLm2aeAvHolXmxWb7N/ByL/dZ4n
+YZd/GPRj42X3BwUrDEL5aO3Mcp+rqq8ACh9hsZXiau0Q9cs1K7Gr55Y06qLrIjom
+2fLqwLFBxCL2sAX1dvClgStyfsRFk9Y/+5tX+IjWaD8dAoRdxCO8IbUDuYGnaKld
+bB9h0NMfKVddCAvuQvX1Zc1Nx0Yb3Hd+ocDD7i9BVgX1BBiGu4/ElS3d32TAVCFs
+Na3tjitWB2G472CYu1O6exY7h1F5V4FHfXH6iMRJSYnvV2Jr+oPZENzNdEEA5H/H
+fUbpWrpKzPafjho9S5rJBBM/tqtmBQFBIdgFVcBVb+bXO6DJ8SMTLiiGcVUvvm1b
+9N2VQIhsxtZ8DpcHHSqFVgT2Gt4UkSrEleSoReg36TzS1s8Uw0oU068PwTe3K0Gx
+2pLMdT9NA6X/t7movpXP6tih1l6P5z62dxFl6W12J9OcegISCt0Q7gex1gk/a8zM
+rzBJC3mVxRiFlvHPBgD6oUKarnTJPQx5f5dFXg8DXBWR1Eh/aFjPQIzhZBYpmOi8
+HqgjcAA+WhMQ7v5c0enJoJJS+8Xfai/MK2vTUGsfAT6HqHLw1HSIn6XQGEf4sQ/U
+NfLeFHHbe9rTk8QhyjrSl2vvek2H4EBQVLF08/FUrAfPELUttOFtysQfC3+M0+PS
+6QGyeIlUjKpBJG7HBd4ibuKMQ5vnA+ACsg/TySYeCO6P85xsN+Lmqlr8cAICn/hR
+ezFSzlibaIelRgfDEDJdjVyCsa7qBMjhRCvGYBdkyTzIRq53qwD9pkhrQ6nwWQrv
+bBzyLrl+NVR8CTEOwbeFLI6qf68kblojk3lwo3Qi3psmeMJdiaV9uevsHrgmEFTH
+lZ3rFECPWzmrkMSfVjWu5d8jJqMcqa4lnGzFQKaB76I8BzGhCWrnuvHPB9c9SVhI
+AnAwNw3gY5xgsbXMxZhnPgYeBSViPkQkgRCWl8Jz41eiAJ3Gtj8QSSFWGHpX+MgP
+ohBaPHz6Fnkhz7Lok97e2AcuRZrDVKV6i28r8mizI3B2Mah6ZV0Yuv0EYNtzBv/v
+yV3nu4DWuOOU0301CXBayxJGX0h07z1Ycv7jWD6LNiBXa1vahtbU4WSYNkF0OJaz
+nf8O3CZy5twMq5kQYoPacdNNLregAmWquvE1nxqWbtHFMjtXitP7czxzUTU/DE+C
+jr+irDoYEregEKg9xov91UCRPZgxL+TML71+tSYOMO3JG6lbGw77PQ8s2So7xore
+8+FeDFPaaJqh6uhF5LETRSx8x/haZiXLd+WtO7wF8S3+Vz7AJIFIe8MUadZrYwnH
+wfMAktQKbep3iHCeZ5jHYA461AOhnCca2y+GoyHZUDDFwS1pC1RN4lMkafSE1AgH
+cmEcjLYsw1gqT0+DfqrvjbXmMjGgkgnkMybJH7df5TKu36Q0Nqvcbc2XLFkalr5V
+Vk0SScqKYnKL+cJjabqA8rKkeAh22E2FBCpKPqxSS3te2bRb3XBX26bP0LshkJuy
+GPu6LKvwmUn0obPKCnLJvb9ImIGZToXu6Fb/Cd2c3DG1IK5PptQz4f7ZRW98huPO
+2w59Bswwt5q4lQqsMEzVRnIDH45MmnhEUeS4NaxqLTO7eJpMpb4VxT2u/Ac3XWKp
+o2RE6CbqTyJ+n8tY9OwBRMKzdVd9RFAMqMHTzWTAuU4BgW2vT2sHYZdAsX8sktBr
+5mo9P3MqvgdPNpg8+AOB03JlIv0dzrAFWCZxxLLGIIIz0eXsjghHzQ9QjGfr0xFH
+Z79AKDjsoRisWyWCnadS2oM9fdAg4T/h1STnfxc44o7N1+ym7u58ODICFi+Kg8IR
+JBHIp3CK02JLTLd/WFhUVyWgc6l8gn+oBK+r7Dw+FTWhqX2/ZHCO8qKK1ZK3NIMn
+MBcSVvHSnTPtppb+oND5nk38xazVVHnwxNHaIh7g3NxDB4hl5rBhrWsgTNuqDDRU
+w7ufvMYr1AOV+8e92cHCEKPM19nFKEgaBFECEptEObesGI3QZPAESlojzQ3cDeBa
+=tEyc
+-----END PGP MESSAGE-----
+
+--Apple-Mail=_C01A1464-6C43-43BF-8F62-157335B7E25B-- \ No newline at end of file
diff --git a/src/leap/bitmask/mail/incoming/tests/test_incoming_mail.py b/src/leap/bitmask/mail/incoming/tests/test_incoming_mail.py
new file mode 100644
index 0000000..29422ec
--- /dev/null
+++ b/src/leap/bitmask/mail/incoming/tests/test_incoming_mail.py
@@ -0,0 +1,391 @@
+# -*- coding: utf-8 -*-
+# test_incoming_mail.py
+# Copyright (C) 2015 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/>.
+"""
+Test case for leap.mail.incoming.service
+
+@authors: Ruben Pollan, <meskio@sindominio.net>
+
+@license: GPLv3, see included LICENSE file
+"""
+
+import json
+import os
+import tempfile
+import uuid
+
+from email.mime.application import MIMEApplication
+from email.mime.multipart import MIMEMultipart
+from email.parser import Parser
+from mock import Mock
+
+from twisted.internet import defer
+from twisted.python import log
+
+from leap.keymanager.errors import KeyAddressMismatch
+from leap.mail.adaptors import soledad_indexes as fields
+from leap.mail.adaptors.soledad import cleanup_deferred_locks
+from leap.mail.adaptors.soledad import SoledadMailAdaptor
+from leap.mail.mail import MessageCollection
+from leap.mail.mailbox_indexer import MailboxIndexer
+
+from leap.mail.incoming.service import IncomingMail
+from leap.mail.rfc3156 import MultipartEncrypted, PGPEncrypted
+from leap.mail.testing import KeyManagerWithSoledadTestCase
+from leap.mail.testing import ADDRESS, ADDRESS_2
+from leap.soledad.common.document import SoledadDocument
+from leap.soledad.common.crypto import (
+ EncryptionSchemes,
+ ENC_JSON_KEY,
+ ENC_SCHEME_KEY,
+)
+
+HERE = os.path.split(os.path.abspath(__file__))[0]
+
+# TODO: add some tests for encrypted, unencrypted, signed and unsgined messages
+
+
+class IncomingMailTestCase(KeyManagerWithSoledadTestCase):
+ """
+ Tests for the incoming mail parser
+ """
+ NICKSERVER = "http://domain"
+ BODY = """
+Governments of the Industrial World, you weary giants of flesh and steel, I
+come from Cyberspace, the new home of Mind. On behalf of the future, I ask
+you of the past to leave us alone. You are not welcome among us. You have
+no sovereignty where we gather.
+ """
+ EMAIL = """from: Test from SomeDomain <%(from)s>
+to: %(to)s
+subject: independence of cyberspace
+
+%(body)s
+ """ % {
+ "from": ADDRESS_2,
+ "to": ADDRESS,
+ "body": BODY
+ }
+
+ def setUp(self):
+ cleanup_deferred_locks()
+ try:
+ del self._soledad
+ del self.km
+ except AttributeError:
+ pass
+
+ # pytest handles correctly the setupEnv for the class,
+ # but trial ignores it.
+ if not getattr(self, 'tempdir', None):
+ self.tempdir = tempfile.mkdtemp()
+
+ def getCollection(_):
+ adaptor = SoledadMailAdaptor()
+ store = self._soledad
+ adaptor.store = store
+ mbox_indexer = MailboxIndexer(store)
+ mbox_name = "INBOX"
+ mbox_uuid = str(uuid.uuid4())
+
+ def get_collection_from_mbox_wrapper(wrapper):
+ wrapper.uuid = mbox_uuid
+ return MessageCollection(
+ adaptor, store,
+ mbox_indexer=mbox_indexer, mbox_wrapper=wrapper)
+
+ d = adaptor.initialize_store(store)
+ d.addCallback(lambda _: mbox_indexer.create_table(mbox_uuid))
+ d.addCallback(
+ lambda _: adaptor.get_or_create_mbox(store, mbox_name))
+ d.addCallback(get_collection_from_mbox_wrapper)
+ return d
+
+ def setUpFetcher(inbox_collection):
+ self.fetcher = IncomingMail(
+ self.km,
+ self._soledad,
+ inbox_collection,
+ ADDRESS)
+
+ # The messages don't exist on soledad will fail on deletion
+ self.fetcher._delete_incoming_message = Mock(
+ return_value=defer.succeed(None))
+
+ d = KeyManagerWithSoledadTestCase.setUp(self)
+ d.addCallback(getCollection)
+ d.addCallback(setUpFetcher)
+ d.addErrback(log.err)
+ return d
+
+ def tearDown(self):
+ d = KeyManagerWithSoledadTestCase.tearDown(self)
+ return d
+
+ def testExtractOpenPGPHeader(self):
+ """
+ Test the OpenPGP header key extraction
+ """
+ KEYURL = "https://leap.se/key.txt"
+ OpenPGP = "id=12345678; url=\"%s\"; preference=signencrypt" % (KEYURL,)
+
+ message = Parser().parsestr(self.EMAIL)
+ message.add_header("OpenPGP", OpenPGP)
+ self.fetcher._keymanager.fetch_key = Mock(
+ return_value=defer.succeed(None))
+
+ def fetch_key_called(ret):
+ self.fetcher._keymanager.fetch_key.assert_called_once_with(
+ ADDRESS_2, KEYURL)
+
+ d = self._create_incoming_email(message.as_string())
+ d.addCallback(
+ lambda email:
+ self._mock_soledad_get_from_index(fields.JUST_MAIL_IDX, [email]))
+ d.addCallback(lambda _: self.fetcher.fetch())
+ d.addCallback(fetch_key_called)
+ return d
+
+ def testExtractOpenPGPHeaderInvalidUrl(self):
+ """
+ Test the OpenPGP header key extraction
+ """
+ KEYURL = "https://someotherdomain.com/key.txt"
+ OpenPGP = "id=12345678; url=\"%s\"; preference=signencrypt" % (KEYURL,)
+
+ message = Parser().parsestr(self.EMAIL)
+ message.add_header("OpenPGP", OpenPGP)
+ self.fetcher._keymanager.fetch_key = Mock()
+
+ def fetch_key_called(ret):
+ self.assertFalse(self.fetcher._keymanager.fetch_key.called)
+
+ d = self._create_incoming_email(message.as_string())
+ d.addCallback(
+ lambda email:
+ self._mock_soledad_get_from_index(fields.JUST_MAIL_IDX, [email]))
+ d.addCallback(lambda _: self.fetcher.fetch())
+ d.addCallback(fetch_key_called)
+ return d
+
+ def testExtractAttachedKey(self):
+ KEY = "-----BEGIN PGP PUBLIC KEY BLOCK-----\n..."
+
+ message = MIMEMultipart()
+ message.add_header("from", ADDRESS_2)
+ key = MIMEApplication("", "pgp-keys")
+ key.set_payload(KEY)
+ message.attach(key)
+ self.fetcher._keymanager.put_raw_key = Mock(
+ return_value=defer.succeed(None))
+
+ def put_raw_key_called(_):
+ self.fetcher._keymanager.put_raw_key.assert_called_once_with(
+ KEY, address=ADDRESS_2)
+
+ d = self._do_fetch(message.as_string())
+ d.addCallback(put_raw_key_called)
+ return d
+
+ def testExtractInvalidAttachedKey(self):
+ KEY = "-----BEGIN PGP PUBLIC KEY BLOCK-----\n..."
+
+ message = MIMEMultipart()
+ message.add_header("from", ADDRESS_2)
+ key = MIMEApplication("", "pgp-keys")
+ key.set_payload(KEY)
+ message.attach(key)
+ self.fetcher._keymanager.put_raw_key = Mock(
+ return_value=defer.fail(KeyAddressMismatch()))
+
+ def put_raw_key_called(_):
+ self.fetcher._keymanager.put_raw_key.assert_called_once_with(
+ KEY, address=ADDRESS_2)
+
+ d = self._do_fetch(message.as_string())
+ d.addCallback(put_raw_key_called)
+ d.addErrback(log.err)
+ return d
+
+ def testExtractAttachedKeyAndNotOpenPGPHeader(self):
+ KEY = "-----BEGIN PGP PUBLIC KEY BLOCK-----\n..."
+ KEYURL = "https://leap.se/key.txt"
+ OpenPGP = "id=12345678; url=\"%s\"; preference=signencrypt" % (KEYURL,)
+
+ message = MIMEMultipart()
+ message.add_header("from", ADDRESS_2)
+ message.add_header("OpenPGP", OpenPGP)
+ key = MIMEApplication("", "pgp-keys")
+ key.set_payload(KEY)
+ message.attach(key)
+
+ self.fetcher._keymanager.put_raw_key = Mock(
+ return_value=defer.succeed(None))
+ self.fetcher._keymanager.fetch_key = Mock()
+
+ def put_raw_key_called(_):
+ self.fetcher._keymanager.put_raw_key.assert_called_once_with(
+ KEY, address=ADDRESS_2)
+ self.assertFalse(self.fetcher._keymanager.fetch_key.called)
+
+ d = self._do_fetch(message.as_string())
+ d.addCallback(put_raw_key_called)
+ return d
+
+ def testExtractOpenPGPHeaderIfInvalidAttachedKey(self):
+ KEY = "-----BEGIN PGP PUBLIC KEY BLOCK-----\n..."
+ KEYURL = "https://leap.se/key.txt"
+ OpenPGP = "id=12345678; url=\"%s\"; preference=signencrypt" % (KEYURL,)
+
+ message = MIMEMultipart()
+ message.add_header("from", ADDRESS_2)
+ message.add_header("OpenPGP", OpenPGP)
+ key = MIMEApplication("", "pgp-keys")
+ key.set_payload(KEY)
+ message.attach(key)
+
+ self.fetcher._keymanager.put_raw_key = Mock(
+ return_value=defer.fail(KeyAddressMismatch()))
+ self.fetcher._keymanager.fetch_key = Mock()
+
+ def put_raw_key_called(_):
+ self.fetcher._keymanager.put_raw_key.assert_called_once_with(
+ KEY, address=ADDRESS_2)
+ self.fetcher._keymanager.fetch_key.assert_called_once_with(
+ ADDRESS_2, KEYURL)
+
+ d = self._do_fetch(message.as_string())
+ d.addCallback(put_raw_key_called)
+ return d
+
+ def testAddDecryptedHeader(self):
+ class DummyMsg():
+
+ def __init__(self):
+ self.headers = {}
+
+ def add_header(self, k, v):
+ self.headers[k] = v
+
+ msg = DummyMsg()
+ self.fetcher._add_decrypted_header(msg)
+
+ self.assertEquals(msg.headers['X-Leap-Encryption'], 'decrypted')
+
+ def testDecryptEmail(self):
+
+ self.fetcher._decryption_error = Mock()
+ self.fetcher._add_decrypted_header = Mock()
+
+ def create_encrypted_message(encstr):
+ message = Parser().parsestr(self.EMAIL)
+ newmsg = MultipartEncrypted('application/pgp-encrypted')
+ for hkey, hval in message.items():
+ newmsg.add_header(hkey, hval)
+
+ encmsg = MIMEApplication(
+ encstr, _subtype='octet-stream', _encoder=lambda x: x)
+ encmsg.add_header('content-disposition', 'attachment',
+ filename='msg.asc')
+ # create meta message
+ metamsg = PGPEncrypted()
+ metamsg.add_header('Content-Disposition', 'attachment')
+ # attach pgp message parts to new message
+ newmsg.attach(metamsg)
+ newmsg.attach(encmsg)
+ return newmsg
+
+ def decryption_error_not_called(_):
+ self.assertFalse(self.fetcher._decryption_error.called,
+ "There was some errors with decryption")
+
+ def add_decrypted_header_called(_):
+ self.assertTrue(self.fetcher._add_decrypted_header.called,
+ "There was some errors with decryption")
+
+ d = self.km.encrypt(self.EMAIL, ADDRESS, sign=ADDRESS_2)
+ d.addCallback(create_encrypted_message)
+ d.addCallback(
+ lambda message:
+ self._do_fetch(message.as_string()))
+ d.addCallback(decryption_error_not_called)
+ d.addCallback(add_decrypted_header_called)
+ return d
+
+ def testValidateSignatureFromEncryptedEmailFromAppleMail(self):
+ enc_signed_file = os.path.join(
+ HERE, 'rfc822.multi-encrypt-signed.message')
+ self.fetcher._add_verified_signature_header = Mock()
+
+ def add_verified_signature_header_called(_):
+ self.assertTrue(self.fetcher._add_verified_signature_header.called,
+ "There was some errors verifying signature")
+
+ with open(enc_signed_file) as f:
+ enc_signed_raw = f.read()
+
+ d = self._do_fetch(enc_signed_raw)
+ d.addCallback(add_verified_signature_header_called)
+ return d
+
+ def testListener(self):
+ self.called = False
+
+ def listener(uid):
+ self.called = True
+
+ def listener_called(_):
+ self.assertTrue(self.called)
+
+ self.fetcher.add_listener(listener)
+ d = self._do_fetch(self.EMAIL)
+ d.addCallback(listener_called)
+ return d
+
+ def _do_fetch(self, message):
+ d = self._create_incoming_email(message)
+ d.addCallback(
+ lambda email:
+ self._mock_soledad_get_from_index(fields.JUST_MAIL_IDX, [email]))
+ d.addCallback(lambda _: self.fetcher.fetch())
+ return d
+
+ def _create_incoming_email(self, email_str):
+ email = SoledadDocument()
+ data = json.dumps(
+ {"incoming": True, "content": email_str},
+ ensure_ascii=False)
+
+ def set_email_content(encr_data):
+ email.content = {
+ fields.INCOMING_KEY: True,
+ fields.ERROR_DECRYPTING_KEY: False,
+ ENC_SCHEME_KEY: EncryptionSchemes.PUBKEY,
+ ENC_JSON_KEY: encr_data
+ }
+ return email
+ d = self.km.encrypt(data, ADDRESS, fetch_remote=False)
+ d.addCallback(set_email_content)
+ return d
+
+ def _mock_soledad_get_from_index(self, index_name, value):
+ get_from_index = self._soledad.get_from_index
+
+ def soledad_mock(idx_name, *key_values):
+ if index_name == idx_name:
+ return defer.succeed(value)
+ return get_from_index(idx_name, *key_values)
+ self.fetcher._soledad.get_from_index = Mock(side_effect=soledad_mock)