Return signing key on signature verification

Don't throw an exception if verification fails

5 files changed, 65 insertions, 56 deletions

diff --git a/src/leap/keymanager/__init__.py b/src/leap/keymanager/__init__.py
index 1704e0b4..b2b05f4a 100644
--- a/src/leap/keymanager/__init__.py
+++ b/src/leap/keymanager/__init__.py
@@ -452,22 +452,21 @@ class KeyManager(object):
                              to fetch from nickserver
         :type fetch_remote: bool
 
-        :return: The decrypted data.
-        :rtype: str
+        :return: The decrypted data and the signing key if signature verifies
+        :rtype: (unicode, EncryptionKey)
 
         :raise KeyNotFound: If any of the keys was not found both locally and
                             in keyserver.
         :raise DecryptError: Raised if failed decrypting for some reason.
-        :raise InvalidSignature: Raised if unable to verify the signature with
-                                 C{verify} address.
         """
         privkey = self.get_key(address, ktype, private=True)
         pubkey = None
         if verify is not None:
             pubkey = self.get_key(verify, ktype, private=False,
                                   fetch_remote=fetch_remote)
-        return self._wrapper_map[ktype].decrypt(
+        decrypted, signed = self._wrapper_map[ktype].decrypt(
             data, privkey, passphrase, pubkey)
+        return (decrypted, pubkey if signed else None)
 
     def sign(self, data, address, ktype, digest_algo='SHA512', clearsign=False,
              detach=True, binary=False):
@@ -520,18 +519,17 @@ class KeyManager(object):
                              to fetch from nickserver
         :type fetch_remote: bool
 
-        :return: signature matches
-        :rtype: bool
+        :return: The signing key if signature verifies else None
+        :rtype: EncryptionKey
 
         :raise KeyNotFound: If the key was not found both locally and
                             in keyserver.
-        :raise InvalidSignature: Raised if unable to verify the signature with
-                                 C{verify} address.
         """
         pubkey = self.get_key(address, ktype, private=False,
                               fetch_remote=fetch_remote)
-        return self._wrapper_map[ktype].verify(
+        signed = self._wrapper_map[ktype].verify(
             data, pubkey, detached_sig=detached_sig)
+        return pubkey if signed else None
 
     def delete_key(self, key):
         """
diff --git a/src/leap/keymanager/errors.py b/src/leap/keymanager/errors.py
index f8965823..c068b27f 100644
--- a/src/leap/keymanager/errors.py
+++ b/src/leap/keymanager/errors.py
@@ -51,13 +51,6 @@ class NoPasswordGiven(Exception):
     pass
 
 
-class InvalidSignature(Exception):
-    """
-    Raised when signature could not be verified.
-    """
-    pass
-
-
 class EncryptError(Exception):
     """
     Raised upon failures of encryption.
diff --git a/src/leap/keymanager/keys.py b/src/leap/keymanager/keys.py
index 7c732e35..0e243ba0 100644
--- a/src/leap/keymanager/keys.py
+++ b/src/leap/keymanager/keys.py
@@ -373,11 +373,10 @@ class EncryptionScheme(object):
         :param verify: The key used to verify a signature.
         :type verify: OpenPGPKey
 
-        :return: The decrypted data.
-        :rtype: str
+        :return: The decrypted data and if signature verifies
+        :rtype: (unicode, bool)
 
-        @raise InvalidSignature: Raised if unable to verify the signature with
-            C{verify} key.
+        :raise DecryptError: Raised if failed decrypting for some reason.
         """
         pass
 
diff --git a/src/leap/keymanager/openpgp.py b/src/leap/keymanager/openpgp.py
index 3f298f71..1d1de989 100644
--- a/src/leap/keymanager/openpgp.py
+++ b/src/leap/keymanager/openpgp.py
@@ -640,12 +640,10 @@ class OpenPGPScheme(EncryptionScheme):
         :param verify: The key used to verify a signature.
         :type verify: OpenPGPKey
 
-        :return: The decrypted data.
-        :rtype: unicode
+        :return: The decrypted data and if signature verifies
+        :rtype: (unicode, bool)
 
         :raise DecryptError: Raised if failed decrypting for some reason.
-        :raise InvalidSignature: Raised if unable to verify the signature with
-                                 C{verify} key.
         """
         leap_assert(privkey.private is True, 'Key is not private.')
         keys = [privkey]
@@ -658,15 +656,15 @@ class OpenPGPScheme(EncryptionScheme):
                 result = gpg.decrypt(
                     data, passphrase=passphrase, always_trust=True)
                 self._assert_gpg_result_ok(result)
+
                 # verify signature
-                if (verify is not None):
-                    if result.valid is False or \
-                            verify.fingerprint != result.pubkey_fingerprint:
-                        raise errors.InvalidSignature(
-                            'Failed to verify signature with key %s: %s' %
-                            (verify.key_id, result.stderr))
+                sign_valid = False
+                if (verify is not None and
+                        result.valid is True and
+                        verify.fingerprint == result.pubkey_fingerprint):
+                    sign_valid = True
 
-                return result.data
+                return (result.data, sign_valid)
             except errors.GPGError as e:
                 logger.error('Failed to decrypt: %s.' % str(e))
                 raise errors.DecryptError(str(e))
@@ -764,9 +762,4 @@ class OpenPGPScheme(EncryptionScheme):
             valid = result.valid
             rfprint = result.fingerprint
             kfprint = gpgpubkey['fingerprint']
-            # raise in case sig is invalid
-            if valid is False or rfprint != kfprint:
-                raise errors.InvalidSignature(
-                    'Failed to verify signature '
-                    'with key %s.' % gpgpubkey['keyid'])
-            return True
+            return valid and rfprint == kfprint
diff --git a/src/leap/keymanager/tests/test_keymanager.py b/src/leap/keymanager/tests/test_keymanager.py
index 8ae12bff..ee4462a1 100644
--- a/src/leap/keymanager/tests/test_keymanager.py
+++ b/src/leap/keymanager/tests/test_keymanager.py
@@ -183,11 +183,12 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase):
             self._soledad, gpgbinary=GPG_BINARY_PATH)
         pgp.put_ascii_key(PUBLIC_KEY, ADDRESS)
         pubkey = pgp.get_key(ADDRESS, private=False)
-        cyphertext = pgp.encrypt('data', pubkey)
+        data = 'data'
+        cyphertext = pgp.encrypt(data, pubkey)
         # assert
         self.assertTrue(cyphertext is not None)
         self.assertTrue(cyphertext != '')
-        self.assertTrue(cyphertext != 'data')
+        self.assertTrue(cyphertext != data)
         self.assertTrue(pgp.is_encrypted(cyphertext))
         self.assertTrue(pgp.is_encrypted(cyphertext))
         # decrypt
@@ -195,6 +196,8 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase):
             KeyNotFound, pgp.get_key, ADDRESS, private=True)
         pgp.put_ascii_key(PRIVATE_KEY, ADDRESS)
         privkey = pgp.get_key(ADDRESS, private=True)
+        decrypted, _ = pgp.decrypt(cyphertext, privkey)
+        self.assertEqual(decrypted, data)
         pgp.delete_key(pubkey)
         pgp.delete_key(privkey)
         self.assertRaises(
@@ -231,9 +234,7 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase):
         signed = pgp.sign(data, privkey)
         pgp.put_ascii_key(PUBLIC_KEY_2, ADDRESS_2)
         wrongkey = pgp.get_key(ADDRESS_2)
-        self.assertRaises(
-            errors.InvalidSignature,
-            pgp.verify, signed, wrongkey)
+        self.assertFalse(pgp.verify(signed, wrongkey))
 
     def test_encrypt_sign_with_public_raises(self):
         pgp = openpgp.OpenPGPScheme(
@@ -260,7 +261,7 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase):
             pgp.decrypt,
             encrypted_and_signed, privkey, verify=privkey)
 
-    def test_decrypt_verify_with_wrong_key_raises(self):
+    def test_decrypt_verify_with_wrong_key(self):
         pgp = openpgp.OpenPGPScheme(
             self._soledad, gpgbinary=GPG_BINARY_PATH)
         pgp.put_ascii_key(PRIVATE_KEY, ADDRESS)
@@ -270,9 +271,10 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase):
         encrypted_and_signed = pgp.encrypt(data, pubkey, sign=privkey)
         pgp.put_ascii_key(PUBLIC_KEY_2, ADDRESS_2)
         wrongkey = pgp.get_key(ADDRESS_2)
-        self.assertRaises(
-            errors.InvalidSignature,
-            pgp.verify, encrypted_and_signed, wrongkey)
+        decrypted, validsign = pgp.decrypt(encrypted_and_signed, privkey,
+                                           verify=wrongkey)
+        self.assertEqual(decrypted, data)
+        self.assertFalse(validsign)
 
     def test_sign_verify(self):
         pgp = openpgp.OpenPGPScheme(
@@ -296,9 +298,10 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase):
         data = 'data'
         encrypted_and_signed = pgp.encrypt(
             data, pubkey2, sign=privkey)
-        res = pgp.decrypt(
+        res, validsign = pgp.decrypt(
             encrypted_and_signed, privkey2, verify=pubkey)
-        self.assertTrue(data, res)
+        self.assertEqual(data, res)
+        self.assertTrue(validsign)
 
     def test_sign_verify_detached_sig(self):
         pgp = openpgp.OpenPGPScheme(
@@ -308,7 +311,8 @@ class OpenPGPCryptoTestCase(KeyManagerWithSoledadTestCase):
         privkey = pgp.get_key(ADDRESS, private=True)
         signature = pgp.sign(data, privkey, detach=True)
         pubkey = pgp.get_key(ADDRESS, private=False)
-        self.assertTrue(pgp.verify(data, pubkey, detached_sig=signature))
+        validsign = pgp.verify(data, pubkey, detached_sig=signature)
+        self.assertTrue(validsign)
 
 
 class KeyManagerKeyManagementTestCase(KeyManagerWithSoledadTestCase):
@@ -509,24 +513,46 @@ class KeyManagerCryptoTestCase(KeyManagerWithSoledadTestCase):
         km = self._key_manager()
         # put raw private key
         km._wrapper_map[OpenPGPKey].put_ascii_key(PRIVATE_KEY, ADDRESS)
+        km._wrapper_map[OpenPGPKey].put_ascii_key(PRIVATE_KEY_2, ADDRESS_2)
         # encrypt
         encdata = km.encrypt(self.RAW_DATA, ADDRESS, OpenPGPKey,
-                             fetch_remote=False)
+                             sign=ADDRESS_2, fetch_remote=False)
         self.assertNotEqual(self.RAW_DATA, encdata)
         # decrypt
-        rawdata = km.decrypt(encdata, ADDRESS, OpenPGPKey)
+        rawdata, signingkey = km.decrypt(encdata, ADDRESS, OpenPGPKey,
+                                         verify=ADDRESS_2, fetch_remote=False)
         self.assertEqual(self.RAW_DATA, rawdata)
+        key = km.get_key(ADDRESS_2, OpenPGPKey, private=False,
+                         fetch_remote=False)
+        self.assertEqual(signingkey.fingerprint, key.fingerprint)
+
+    def test_keymanager_openpgp_encrypt_decrypt_wrong_sign(self):
+        km = self._key_manager()
+        # put raw keys
+        km._wrapper_map[OpenPGPKey].put_ascii_key(PRIVATE_KEY, ADDRESS)
+        km._wrapper_map[OpenPGPKey].put_ascii_key(PRIVATE_KEY_2, ADDRESS_2)
+        # encrypt
+        encdata = km.encrypt(self.RAW_DATA, ADDRESS, OpenPGPKey,
+                             sign=ADDRESS_2, fetch_remote=False)
+        self.assertNotEqual(self.RAW_DATA, encdata)
+        # verify
+        rawdata, signingkey = km.decrypt(encdata, ADDRESS, OpenPGPKey,
+                                         verify=ADDRESS, fetch_remote=False)
+        self.assertEqual(self.RAW_DATA, rawdata)
+        self.assertTrue(signingkey is None)
 
     def test_keymanager_openpgp_sign_verify(self):
         km = self._key_manager()
         # put raw private keys
         km._wrapper_map[OpenPGPKey].put_ascii_key(PRIVATE_KEY, ADDRESS)
-        # encrypt
         signdata = km.sign(self.RAW_DATA, ADDRESS, OpenPGPKey, detach=False)
         self.assertNotEqual(self.RAW_DATA, signdata)
-        # decrypt
-        self.assertTrue(km.verify(signdata, ADDRESS, OpenPGPKey,
-                                  fetch_remote=False))
+        # verify
+        signingkey = km.verify(signdata, ADDRESS, OpenPGPKey,
+                               fetch_remote=False)
+        key = km.get_key(ADDRESS, OpenPGPKey, private=False,
+                         fetch_remote=False)
+        self.assertEqual(signingkey.fingerprint, key.fingerprint)
 
 
 # Key material for testing