Add support for RSA signature recovery

CHANGELOG.rst

@@ -27,6 +27,11 @@ Changelog
   in any application outside of testing.
 * Python 2 support is deprecated in ``cryptography``. This is the last release
   that will support Python 2.
+* Added the
+  :meth:`~cryptography.hazmat.primitives.asymmetric.rsa.RSAPublicKey.recover_data_from_signature`
+  function to
+  :class:`~cryptography.hazmat.primitives.asymmetric.rsa.RSAPublicKey`
+  for recovering the signed data from an RSA signature.
 
 .. _v3-2-1:
 

docs/hazmat/primitives/asymmetric/rsa.rst

@@ -709,6 +709,55 @@ Key interfaces
         :raises cryptography.exceptions.InvalidSignature: If the signature does
             not validate.
 
+    .. method:: recover_data_from_signature(signature, padding, algorithm)
+
+        .. versionadded:: 3.3
+
+        Recovers the signed data from the signature. The data contains the
+        digest of the original message string. The ``padding`` and
+        ``algorithm`` parameters must match the ones used when the signature
+        was created for the recovery to succeed.
+
+        The ``algorithm`` parameter can also be set to ``None`` to recover all
+        the data present in the signature, without regard to its format or the
+        hash algorithm used for its creation.
+
+        For
+        :class:`~cryptography.hazmat.primitives.asymmetric.padding.PKCS1v15`
+        padding, this returns the data after removing the padding layer. For
+        standard signatures the data contains the full ``DigestInfo`` structure.
+        For non-standard signatures, any data can be returned, including zero-
+        length data.
+
+        Normally you should use the
+        :meth:`~cryptography.hazmat.primitives.asymmetric.rsa.RSAPublicKey.verify`
+        function to validate the signature. But for some non-standard signature
+        formats you may need to explicitly recover and validate the signed
+        data. Following are some examples:
+
+        - Some old Thawte and Verisign timestamp certificates without ``DigestInfo``.
+        - Signed MD5/SHA1 hashes in TLS 1.1 or earlier (RFC 4346, section 4.7).
+        - IKE version 1 signatures without ``DigestInfo`` (RFC 2409, section 5.1).
+
+        :param bytes signature: The signature.
+
+        :param padding: An instance of
+            :class:`~cryptography.hazmat.primitives.asymmetric.padding.AsymmetricPadding`.
+            Recovery is only supported with some of the padding types. (Currently
+            only with
+            :class:`~cryptography.hazmat.primitives.asymmetric.padding.PKCS1v15`).
+
+        :param algorithm: An instance of
+            :class:`~cryptography.hazmat.primitives.hashes.HashAlgorithm`.
+            Can be ``None`` to return the all the data present in the signature.
+
+        :return bytes: The signed data.
+
+        :raises cryptography.exceptions.InvalidSignature: If the signature is
+            invalid.
+
+        :raises cryptography.exceptions.UnsupportedAlgorithm: If signature
+            data recovery is not supported with the provided ``padding`` type.
 
 .. class:: RSAPublicKeyWithSerialization
 

docs/spelling_wordlist.txt

@@ -103,6 +103,7 @@ Solaris
 syscall
 Tanja
 testability
+Thawte
 timestamp
 timestamps
 tunable

src/_cffi_src/openssl/evp.py

@@ -101,6 +101,9 @@
 int EVP_PKEY_verify_init(EVP_PKEY_CTX *);
 int EVP_PKEY_verify(EVP_PKEY_CTX *, const unsigned char *, size_t,
                     const unsigned char *, size_t);
+int EVP_PKEY_verify_recover_init(EVP_PKEY_CTX *);
+int EVP_PKEY_verify_recover(EVP_PKEY_CTX *, unsigned char *,
+                            size_t *, const unsigned char *, size_t);
 int EVP_PKEY_encrypt_init(EVP_PKEY_CTX *);
 int EVP_PKEY_decrypt_init(EVP_PKEY_CTX *);
 

src/cryptography/hazmat/backends/openssl/rsa.py

@@ -142,6 +142,7 @@ def _rsa_sig_determine_padding(backend, key, padding, algorithm):
     backend.openssl_assert(pkey_size > 0)
 
     if isinstance(padding, PKCS1v15):
+        # Hash algorithm is ignored for PKCS1v15-padding, may be None.
         padding_enum = backend._lib.RSA_PKCS1_PADDING
     elif isinstance(padding, PSS):
         if not isinstance(padding._mgf, MGF1):
@@ -150,6 +151,10 @@ def _rsa_sig_determine_padding(backend, key, padding, algorithm):
                 _Reasons.UNSUPPORTED_MGF,
             )
 
+        # PSS padding requires a hash algorithm
+        if not isinstance(algorithm, hashes.HashAlgorithm):
+            raise TypeError("Expected instance of hashes.HashAlgorithm.")
+
         # Size of key in bytes - 2 is the maximum
         # PSS signature length (salt length is checked later)
         if pkey_size - algorithm.digest_size - 2 < 0:
@@ -168,25 +173,37 @@ def _rsa_sig_determine_padding(backend, key, padding, algorithm):
     return padding_enum
 
 
-def _rsa_sig_setup(backend, padding, algorithm, key, data, init_func):
+# Hash algorithm can be absent (None) to initialize the context without setting
+# any message digest algorithm. This is currently only valid for the PKCS1v15
+# padding type, where it means that the signature data is encoded/decoded
+# as provided, without being wrapped in a DigestInfo structure.
+def _rsa_sig_setup(backend, padding, algorithm, key, init_func):
     padding_enum = _rsa_sig_determine_padding(backend, key, padding, algorithm)
-    evp_md = backend._evp_md_non_null_from_algorithm(algorithm)
     pkey_ctx = backend._lib.EVP_PKEY_CTX_new(key._evp_pkey, backend._ffi.NULL)
     backend.openssl_assert(pkey_ctx != backend._ffi.NULL)
     pkey_ctx = backend._ffi.gc(pkey_ctx, backend._lib.EVP_PKEY_CTX_free)
     res = init_func(pkey_ctx)
     backend.openssl_assert(res == 1)
-    res = backend._lib.EVP_PKEY_CTX_set_signature_md(pkey_ctx, evp_md)
-    if res == 0:
+    if algorithm is not None:
+        evp_md = backend._evp_md_non_null_from_algorithm(algorithm)
+        res = backend._lib.EVP_PKEY_CTX_set_signature_md(pkey_ctx, evp_md)
+        if res == 0:
+            backend._consume_errors()
+            raise UnsupportedAlgorithm(
+                "{} is not supported by this backend for RSA signing.".format(
+                    algorithm.name
+                ),
+                _Reasons.UNSUPPORTED_HASH,
+            )
+    res = backend._lib.EVP_PKEY_CTX_set_rsa_padding(pkey_ctx, padding_enum)
+    if res <= 0:
         backend._consume_errors()
         raise UnsupportedAlgorithm(
-            "{} is not supported by this backend for RSA signing.".format(
-                algorithm.name
+            "{} is not supported for the RSA signature operation.".format(
+                padding.name
             ),
-            _Reasons.UNSUPPORTED_HASH,
+            _Reasons.UNSUPPORTED_PADDING,
         )
-    res = backend._lib.EVP_PKEY_CTX_set_rsa_padding(pkey_ctx, padding_enum)
-    backend.openssl_assert(res > 0)
     if isinstance(padding, PSS):
         res = backend._lib.EVP_PKEY_CTX_set_rsa_pss_saltlen(
             pkey_ctx, _get_rsa_pss_salt_length(padding, key, algorithm)
@@ -208,7 +225,6 @@ def _rsa_sig_sign(backend, padding, algorithm, private_key, data):
         padding,
         algorithm,
         private_key,
-        data,
         backend._lib.EVP_PKEY_sign_init,
     )
     buflen = backend._ffi.new("size_t *")
@@ -235,7 +251,6 @@ def _rsa_sig_verify(backend, padding, algorithm, public_key, signature, data):
         padding,
         algorithm,
         public_key,
-        data,
         backend._lib.EVP_PKEY_verify_init,
     )
     res = backend._lib.EVP_PKEY_verify(
@@ -250,6 +265,36 @@ def _rsa_sig_verify(backend, padding, algorithm, public_key, signature, data):
         raise InvalidSignature
 
 
+def _rsa_sig_recover(backend, padding, algorithm, public_key, signature):
+    pkey_ctx = _rsa_sig_setup(
+        backend,
+        padding,
+        algorithm,
+        public_key,
+        backend._lib.EVP_PKEY_verify_recover_init,
+    )
+
+    # Attempt to keep the rest of the code in this function as constant/time
+    # as possible. See the comment in _enc_dec_rsa_pkey_ctx. Note that the
+    # outlen parameter is used even though its value may be undefined in the
+    # error case. Due to the tolerant nature of Python slicing this does not
+    # trigger any exceptions.
+    maxlen = backend._lib.EVP_PKEY_size(public_key._evp_pkey)
+    backend.openssl_assert(maxlen > 0)
+    buf = backend._ffi.new("unsigned char[]", maxlen)
+    buflen = backend._ffi.new("size_t *", maxlen)
+    res = backend._lib.EVP_PKEY_verify_recover(
+        pkey_ctx, buf, buflen, signature, len(signature)
+    )
+    resbuf = backend._ffi.buffer(buf)[: buflen[0]]
+    backend._lib.ERR_clear_error()
+    # Assume that all parameter errors are handled during the setup phase and
+    # any error here is due to invalid signature.
+    if res != 1:
+        raise InvalidSignature
+    return resbuf
+
+
 @utils.register_interface(AsymmetricSignatureContext)
 class _RSASignatureContext(object):
     def __init__(self, backend, private_key, padding, algorithm):
@@ -463,3 +508,9 @@ def verify(self, signature, data, padding, algorithm):
         return _rsa_sig_verify(
             self._backend, padding, algorithm, self, signature, data
         )
+
+    def recover_data_from_signature(self, signature, padding, algorithm):
+        _check_not_prehashed(algorithm)
+        return _rsa_sig_recover(
+            self._backend, padding, algorithm, self, signature
+        )

src/cryptography/hazmat/backends/openssl/utils.py

@@ -52,7 +52,8 @@ def _check_not_prehashed(signature_algorithm):
     if isinstance(signature_algorithm, Prehashed):
         raise TypeError(
             "Prehashed is only supported in the sign and verify methods. "
-            "It cannot be used with signer or verifier."
+            "It cannot be used with signer, verifier or "
+            "recover_data_from_signature."
         )
 
 

src/cryptography/hazmat/primitives/asymmetric/rsa.py

@@ -106,6 +106,12 @@ def verify(self, signature, data, padding, algorithm):
         Verifies the signature of the data.
         """
 
+    @abc.abstractmethod
+    def recover_data_from_signature(self, signature, padding, algorithm):
+        """
+        Recovers the original data from the signature.
+        """
+
 
 RSAPublicKeyWithSerialization = RSAPublicKey
 

tests/hazmat/primitives/test_rsa.py

@@ -730,6 +730,18 @@ def test_prehashed_unsupported_in_verifier_ctx(self, backend):
                 asym_utils.Prehashed(hashes.SHA1()),
             )
 
+    def test_prehashed_unsupported_in_signature_recover(self, backend):
+        private_key = RSA_KEY_512.private_key(backend)
+        public_key = private_key.public_key()
+        signature = private_key.sign(
+            b"sign me", padding.PKCS1v15(), hashes.SHA1()
+        )
+        prehashed_alg = asym_utils.Prehashed(hashes.SHA1())
+        with pytest.raises(TypeError):
+            public_key.recover_data_from_signature(
+                signature, padding.PKCS1v15(), prehashed_alg
+            )
+
     def test_corrupted_private_key(self, backend):
         with pytest.raises(ValueError):
             serialization.load_pem_private_key(
@@ -759,13 +771,28 @@ def test_pkcs1v15_verification(self, pkcs1_example, backend):
         public_key = rsa.RSAPublicNumbers(
             e=public["public_exponent"], n=public["modulus"]
         ).public_key(backend)
+        signature = binascii.unhexlify(example["signature"])
+        message = binascii.unhexlify(example["message"])
         public_key.verify(
-            binascii.unhexlify(example["signature"]),
-            binascii.unhexlify(example["message"]),
-            padding.PKCS1v15(),
-            hashes.SHA1(),
+            signature, message, padding.PKCS1v15(), hashes.SHA1()
         )
 
+        # Test digest recovery by providing hash
+        digest = hashes.Hash(hashes.SHA1())
+        digest.update(message)
+        msg_digest = digest.finalize()
+        rec_msg_digest = public_key.recover_data_from_signature(
+            signature, padding.PKCS1v15(), hashes.SHA1()
+        )
+        assert msg_digest == rec_msg_digest
+
+        # Test recovery of all data (full DigestInfo) with hash alg. as None
+        rec_sig_data = public_key.recover_data_from_signature(
+            signature, padding.PKCS1v15(), None
+        )
+        assert len(rec_sig_data) > len(msg_digest)
+        assert msg_digest == rec_sig_data[-len(msg_digest) :]
+
     @pytest.mark.supported(
         only_if=lambda backend: backend.rsa_padding_supported(
             padding.PKCS1v15()
@@ -783,6 +810,17 @@ def test_invalid_pkcs1v15_signature_wrong_data(self, backend):
                 signature, b"incorrect data", padding.PKCS1v15(), hashes.SHA1()
             )
 
+    def test_invalid_pkcs1v15_signature_recover_wrong_hash_alg(self, backend):
+        private_key = RSA_KEY_512.private_key(backend)
+        public_key = private_key.public_key()
+        signature = private_key.sign(
+            b"sign me", padding.PKCS1v15(), hashes.SHA1()
+        )
+        with pytest.raises(InvalidSignature):
+            public_key.recover_data_from_signature(
+                signature, padding.PKCS1v15(), hashes.SHA256()
+            )
+
     def test_invalid_signature_sequence_removed(self, backend):
         """
         This test comes from wycheproof
@@ -970,6 +1008,27 @@ def test_invalid_pss_signature_data_too_large_for_modulus(self, backend):
                 hashes.SHA1(),
             )
 
+    def test_invalid_pss_signature_recover(self, backend):
+        private_key = RSA_KEY_1024.private_key(backend)
+        public_key = private_key.public_key()
+        pss_padding = padding.PSS(
+            mgf=padding.MGF1(algorithm=hashes.SHA1()),
+            salt_length=padding.PSS.MAX_LENGTH,
+        )
+        signature = private_key.sign(b"sign me", pss_padding, hashes.SHA1())
+
+        # Hash algorithm can not be absent for PSS padding
+        with pytest.raises(TypeError):
+            public_key.recover_data_from_signature(
+                signature, pss_padding, None
+            )
+
+        # Signature data recovery not supported with PSS
+        with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_PADDING):
+            public_key.recover_data_from_signature(
+                signature, pss_padding, hashes.SHA1()
+            )
+
     @pytest.mark.supported(
         only_if=lambda backend: backend.rsa_padding_supported(
             padding.PKCS1v15()