[feat] use PGPy instead of leap.keymanager to encrypt

In PGPy 0.4.0 keys are not accepted if they have empty lines before the
ascii armored text. I addapt the tests around that for now. This is fixed
already in PGPy repo, so it will not be an issue in comming releases.

- Resolves: #8558, #8663

2 files changed, 32 insertions, 63 deletions

diff --git a/src/leap/mx/mail_receiver.py b/src/leap/mx/mail_receiver.py
index c8113c1..e376554 100644
--- a/src/leap/mx/mail_receiver.py
+++ b/src/leap/mx/mail_receiver.py
@@ -42,6 +42,8 @@ import email.utils
 
 from datetime import datetime, timedelta
 from email import message_from_string
+from pgpy import PGPKey, PGPMessage
+from pgpy.errors import PGPEncryptionError
 
 from twisted.application.service import Service, IService
 from twisted.internet import inotify, defer, task, reactor
@@ -54,8 +56,6 @@ from leap.soledad.common.crypto import ENC_JSON_KEY
 from leap.soledad.common.crypto import ENC_SCHEME_KEY
 from leap.soledad.common.document import ServerDocument
 
-from leap.keymanager import openpgp
-
 from leap.mx.bounce import bounce_message
 from leap.mx.bounce import InvalidReturnPathError
 
@@ -168,7 +168,6 @@ class MailReceiver(Service):
         """
         Given a public key and a message, it encrypts the message to
         that public key.
-        The address is needed in order to build the OpenPGPKey object.
 
         :param pubkey: public key for the owner of the message
         :type pubkey: str
@@ -182,52 +181,38 @@ class MailReceiver(Service):
         if pubkey is None or len(pubkey) == 0:
             log.msg("_encrypt_message: Something went wrong, here's all "
                     "I know: %r" % (pubkey,))
-            return None
+            raise Exception('Not valid key')
 
         doc = ServerDocument(doc_id=str(pyuuid.uuid4()))
 
         # store plain text if pubkey is not available
         data = {'incoming': True, 'content': message}
-        if pubkey is None or len(pubkey) == 0:
+        json_dump = json.dumps(data, ensure_ascii=False)
+        try:
+            key, _ = PGPKey.from_blob(pubkey)
+            if key.expires_at and key.expires_at < datetime.now():
+                log.msg("_encrypt_message: the key is expired (%s)"
+                        % str(key.expires_at))
+
+            message = PGPMessage.new(json_dump)
+            encryption_result = key.encrypt(message)
+        except (ValueError, PGPEncryptionError) as e:
+            log.msg("_encrypt_message: Encryption failed with status: %s"
+                    % (e,))
             doc.content = {
                 self.INCOMING_KEY: True,
                 self.ERROR_DECRYPTING_KEY: False,
                 ENC_SCHEME_KEY: EncryptionSchemes.NONE,
-                ENC_JSON_KEY: json.dumps(data,
-                                         ensure_ascii=False)
+                ENC_JSON_KEY: json_dump
             }
             return doc
 
-        # otherwise, encrypt
-        with openpgp.TempGPGWrapper(gpgbinary='/usr/bin/gpg') as gpg:
-            gpg.import_keys(pubkey)
-            key = gpg.list_keys().pop()
-
-            if key['expires']:
-                expires = datetime.fromtimestamp(int(key['expires']))
-                if expires < datetime.now():
-                    log.msg("_encrypt_message: the key is expired (%s), "
-                            "can't encrypt" % (str(expires),))
-                    raise Exception("Expired key")
-
-            encryption_result = gpg.encrypt(
-                json.dumps(data, ensure_ascii=False),
-                key["fingerprint"],
-                symmetric=False)
-
-            if not encryption_result.ok:
-                log.msg("_encrypt_message: Encryption failed with status: %r"
-                        % (encryption_result.status,))
-                raise Exception("Encryption failed: %r"
-                                % (encryption_result.status,))
-
-            doc.content = {
-                self.INCOMING_KEY: True,
-                self.ERROR_DECRYPTING_KEY: False,
-                ENC_SCHEME_KEY: EncryptionSchemes.PUBKEY,
-                ENC_JSON_KEY: str(encryption_result)
-            }
-
+        doc.content = {
+            self.INCOMING_KEY: True,
+            self.ERROR_DECRYPTING_KEY: False,
+            ENC_SCHEME_KEY: EncryptionSchemes.PUBKEY,
+            ENC_JSON_KEY: str(encryption_result)
+        }
         return doc
 
     @defer.inlineCallbacks
diff --git a/src/leap/mx/tests/test_mail_receiver.py b/src/leap/mx/tests/test_mail_receiver.py
index 68f8d0b..abe5a71 100644
--- a/src/leap/mx/tests/test_mail_receiver.py
+++ b/src/leap/mx/tests/test_mail_receiver.py
@@ -26,10 +26,10 @@ import shutil
 import tempfile
 
 from email.message import Message
+from pgpy import PGPKey, PGPMessage
 from twisted.internet import defer, reactor
 from twisted.trial import unittest
 
-from leap.keymanager import openpgp
 from leap.mx.mail_receiver import MailReceiver
 
 
@@ -58,6 +58,7 @@ class MailReceiverTestCase(unittest.TestCase):
 
     def usersCdb(self):
         self.pubKey = PUBLIC_KEY
+        self.privKey = PRIVATE_KEY
         self.docs = []
         self.defer_put_doc = defer.Deferred()
 
@@ -98,22 +99,10 @@ class MailReceiverTestCase(unittest.TestCase):
         yield defer_called
         self.assertTrue(os.path.exists(path))
 
-    @defer.inlineCallbacks
     def test_expired_key(self):
-        defer_bounce = defer.Deferred()
-
-        def bounce_mock(*args):
-            defer_bounce.callback(args)
-            return defer.succeed(None)
-
         self.pubKey = EXPIRED_KEY
-        self.receiver._bounce_with_timeout = bounce_mock
-
-        msg, path = self.addMail("foo bar")
-        bpath, bmsg, berr = yield defer_bounce
-        self.assertEqual(path, bpath.path)
-        self.assertEqual(msg, bmsg.as_string())
-        self.assertEqual("Expired key", str(berr))
+        self.privKey = EXPIRED_PRIVATE
+        return self.test_single_mail()
 
     @defer.inlineCallbacks
     def test_misleading_encoding(self):
@@ -145,13 +134,10 @@ class MailReceiverTestCase(unittest.TestCase):
         return msg.as_string(), path
 
     def decryptDoc(self, doc):
-        encdoc = doc.content['_enc_json']
-        decdoc = {}
-
-        with openpgp.TempGPGWrapper(gpgbinary='/usr/bin/gpg') as gpg:
-            gpg.import_keys(PRIVATE_KEY)
-            decstr = gpg.decrypt(encdoc)
-            decdoc = json.loads(decstr.data)
+        key, _ = PGPKey.from_blob(self.privKey)
+        message = PGPMessage.from_blob(doc.content['_enc_json'])
+        decmsg = key.decrypt(message)
+        decdoc = json.loads(str(decmsg.message))
 
         self.assertTrue(decdoc['incoming'])
         return decdoc['content']
@@ -159,8 +145,7 @@ class MailReceiverTestCase(unittest.TestCase):
 
 # key 24D18DDF: public key "Leap Test Key <leap@leap.se>"
 KEY_FINGERPRINT = "E36E738D69173C13D709E44F2F455E2824D18DDF"
-PUBLIC_KEY = """
------BEGIN PGP PUBLIC KEY BLOCK-----
+PUBLIC_KEY = """-----BEGIN PGP PUBLIC KEY BLOCK-----
 Version: GnuPG v1.4.10 (GNU/Linux)
 
 mQINBFC9+dkBEADNRfwV23TWEoGc/x0wWH1P7PlXt8MnC2Z1kKaKKmfnglVrpOiz
@@ -212,8 +197,7 @@ ZtQ/VymwFL3XdUWV6B/hU4PVAFvO3qlOtdJ6TpE+nEWgcWjCv5g7RjXX
 =MuOY
 -----END PGP PUBLIC KEY BLOCK-----
 """
-PRIVATE_KEY = """
------BEGIN PGP PRIVATE KEY BLOCK-----
+PRIVATE_KEY = """-----BEGIN PGP PRIVATE KEY BLOCK-----
 Version: GnuPG v1.4.10 (GNU/Linux)
 
 lQcYBFC9+dkBEADNRfwV23TWEoGc/x0wWH1P7PlXt8MnC2Z1kKaKKmfnglVrpOiz