Fix invalid RSA private key PKCS8 encoding

jose/backends/rsa_backend.py

@@ -1,6 +1,9 @@
+import binascii
+
 import six
-from pyasn1.codec.der import encoder
-from pyasn1.type import univ
+from pyasn1.codec.der import decoder, encoder
+from pyasn1.error import PyAsn1Error
+from pyasn1.type import namedtype, univ
 
 import rsa as pyrsa
 import rsa.pem as pyrsa_pem
@@ -12,7 +15,18 @@
 from jose.utils import base64_to_long, long_to_base64
 
 
-PKCS8_RSA_HEADER = b'0\x82\x04\xbd\x02\x01\x000\r\x06\t*\x86H\x86\xf7\r\x01\x01\x01\x05\x00'
+LEGACY_INVALID_PKCS8_RSA_HEADER = binascii.unhexlify(
+    "30"  # sequence
+    "8204BD"  # DER-encoded sequence contents length of 1213 bytes -- INCORRECT STATIC LENGTH
+    "020100"  # integer: 0 -- Version
+    "30"  # sequence
+    "0D"  # DER-encoded sequence contents length of 13 bytes -- PrivateKeyAlgorithmIdentifier
+    "06092A864886F70D010101"  # OID -- rsaEncryption
+    "0500"  # NULL -- parameters
+)
+ASN1_SEQUENCE_ID = binascii.unhexlify("30")
+RSA_ENCRYPTION_ASN1_OID = "1.2.840.113549.1.1.1"
+
 # Functions gcd and rsa_recover_prime_factors were copied from cryptography 1.9
 # to enable pure python rsa module to be in compliance with section 6.3.1 of RFC7518
 # which requires only private exponent (d) for private key.
@@ -83,6 +97,65 @@ def pem_to_spki(pem, fmt='PKCS8'):
     return key.to_pem(fmt)
 
 
+def _legacy_private_key_pkcs8_to_pkcs1(pkcs8_key):
+    """Legacy RSA private key PKCS8-to-PKCS1 conversion.
+
+    .. warning::
+
+        This is incorrect parsing and only works because the legacy PKCS1-to-PKCS8
+        encoding was also incorrect.
+    """
+    # Only allow this processing if the prefix matches
+    # AND the following byte indicates an ASN1 sequence,
+    # as we would expect with the legacy encoding.
+    if not pkcs8_key.startswith(LEGACY_INVALID_PKCS8_RSA_HEADER + ASN1_SEQUENCE_ID):
+        raise ValueError("Invalid private key encoding")
+
+    return pkcs8_key[len(LEGACY_INVALID_PKCS8_RSA_HEADER):]
+
+
+class PKCS8RsaPrivateKeyAlgorithm(univ.Sequence):
+    """ASN1 structure for recording RSA PrivateKeyAlgorithm identifiers."""
+    componentType = namedtype.NamedTypes(
+        namedtype.NamedType("rsaEncryption", univ.ObjectIdentifier()),
+        namedtype.NamedType("parameters", univ.Null())
+    )
+
+
+class PKCS8PrivateKey(univ.Sequence):
+    """ASN1 structure for recording PKCS8 private keys."""
+    componentType = namedtype.NamedTypes(
+        namedtype.NamedType("version", univ.Integer()),
+        namedtype.NamedType("privateKeyAlgorithm", PKCS8RsaPrivateKeyAlgorithm()),
+        namedtype.NamedType("privateKey", univ.OctetString())
+    )
+
+
+def _private_key_pkcs8_to_pkcs1(pkcs8_key):
+    """Convert a PKCS8-encoded RSA private key to PKCS1."""
+    decoded_values = decoder.decode(pkcs8_key, asn1Spec=PKCS8PrivateKey())
+
+    try:
+        decoded_key = decoded_values[0]
+    except IndexError:
+        raise ValueError("Invalid private key encoding")
+
+    return decoded_key["privateKey"]
+
+
+def _private_key_pkcs1_to_pkcs8(pkcs1_key):
+    """Convert a PKCS1-encoded RSA private key to PKCS8."""
+    algorithm = PKCS8RsaPrivateKeyAlgorithm()
+    algorithm["rsaEncryption"] = RSA_ENCRYPTION_ASN1_OID
+
+    pkcs8_key = PKCS8PrivateKey()
+    pkcs8_key["version"] = 0
+    pkcs8_key["privateKeyAlgorithm"] = algorithm
+    pkcs8_key["privateKey"] = pkcs1_key
+
+    return encoder.encode(pkcs8_key)
+
+
 class RSAKey(Key):
     SHA256 = 'SHA-256'
     SHA384 = 'SHA-384'
@@ -121,12 +194,15 @@ def __init__(self, key, algorithm):
                         self._prepared_key = pyrsa.PrivateKey.load_pkcs1(key)
                     except ValueError:
                         try:
-                            # python-rsa does not support PKCS8 yet so we have to manually remove OID
                             der = pyrsa_pem.load_pem(key, b'PRIVATE KEY')
-                            header, der = der[:22], der[22:]
-                            if header != PKCS8_RSA_HEADER:
-                                raise ValueError("Invalid PKCS8 header")
-                            self._prepared_key = pyrsa.PrivateKey._load_pkcs1_der(der)
+                            try:
+                                pkcs1_key = _private_key_pkcs8_to_pkcs1(der)
+                            except PyAsn1Error:
+                                # If the key was encoded using the old, invalid,
+                                # encoding then pyasn1 will throw an error attempting
+                                # to parse the key.
+                                pkcs1_key = _legacy_private_key_pkcs8_to_pkcs1(der)
+                            self._prepared_key = pyrsa.PrivateKey.load_pkcs1(pkcs1_key, format="DER")
                         except ValueError as e:
                             raise JWKError(e)
             return
@@ -183,7 +259,8 @@ def to_pem(self, pem_format='PKCS8'):
         if isinstance(self._prepared_key, pyrsa.PrivateKey):
             der = self._prepared_key.save_pkcs1(format='DER')
             if pem_format == 'PKCS8':
-                pem = pyrsa_pem.save_pem(PKCS8_RSA_HEADER + der, pem_marker='PRIVATE KEY')
+                pkcs8_der = _private_key_pkcs1_to_pkcs8(der)
+                pem = pyrsa_pem.save_pem(pkcs8_der, pem_marker='PRIVATE KEY')
             elif pem_format == 'PKCS1':
                 pem = pyrsa_pem.save_pem(der, pem_marker='RSA PRIVATE KEY')
             else:
@@ -196,7 +273,7 @@ def to_pem(self, pem_format='PKCS8'):
                 der = encoder.encode(asn_key)
 
                 header = PubKeyHeader()
-                header['oid'] = univ.ObjectIdentifier('1.2.840.113549.1.1.1')
+                header['oid'] = univ.ObjectIdentifier(RSA_ENCRYPTION_ASN1_OID)
                 pub_key = OpenSSLPubKey()
                 pub_key['header'] = header
                 pub_key['key'] = univ.BitString.fromOctetString(der)

requirements.txt

@@ -3,3 +3,4 @@ six
 future
 rsa
 ecdsa
+pyasn1

setup.py

@@ -26,6 +26,11 @@ def get_packages(package):
     'pycrypto': ['pycrypto >=2.6.0, <2.7.0'],
     'pycryptodome': ['pycryptodome >=3.3.1, <4.0.0'],
 }
+legacy_backend_requires = ['ecdsa <1.0', 'rsa', 'pyasn1']
+install_requires = ['six <2.0', 'future <1.0']
+
+# TODO: work this into the extras selection instead.
+install_requires += legacy_backend_requires
 
 
 setup(
@@ -64,5 +69,5 @@ def get_packages(package):
         'pytest-cov',
         'pytest-runner',
     ],
-    install_requires=['six <2.0', 'ecdsa <1.0', 'rsa', 'future <1.0']
+    install_requires=install_requires
 )

tests/algorithms/test_EC_compat.py

@@ -15,7 +15,7 @@
     None in (ECDSAECKey, CryptographyECKey),
     reason="Multiple crypto backends not available for backend compatibility tests"
 )
-class TestBackendRsaCompatibility(object):
+class TestBackendEcdsaCompatibility(object):
 
     @pytest.mark.parametrize("BackendSign", [ECDSAECKey, CryptographyECKey])
     @pytest.mark.parametrize("BackendVerify", [ECDSAECKey, CryptographyECKey])
@@ -31,3 +31,42 @@ def test_signing_parity(self, BackendSign, BackendVerify):
 
         # invalid signature
         assert not key_verify.verify(msg, b'n' * 64)
+
+    @pytest.mark.parametrize("BackendFrom", [ECDSAECKey, CryptographyECKey])
+    @pytest.mark.parametrize("BackendTo", [ECDSAECKey, CryptographyECKey])
+    def test_public_key_to_pem(self, BackendFrom, BackendTo):
+        key = BackendFrom(private_key, ALGORITHMS.ES256)
+        key2 = BackendTo(private_key, ALGORITHMS.ES256)
+
+        assert key.public_key().to_pem().strip() == key2.public_key().to_pem().strip()
+
+    @pytest.mark.parametrize("BackendFrom", [ECDSAECKey, CryptographyECKey])
+    @pytest.mark.parametrize("BackendTo", [ECDSAECKey, CryptographyECKey])
+    def test_private_key_to_pem(self, BackendFrom, BackendTo):
+        key = BackendFrom(private_key, ALGORITHMS.ES256)
+        key2 = BackendTo(private_key, ALGORITHMS.ES256)
+
+        assert key.to_pem().strip() == key2.to_pem().strip()
+
+    @pytest.mark.parametrize("BackendFrom", [ECDSAECKey, CryptographyECKey])
+    @pytest.mark.parametrize("BackendTo", [ECDSAECKey, CryptographyECKey])
+    def test_public_key_load_cycle(self, BackendFrom, BackendTo):
+        key = BackendFrom(private_key, ALGORITHMS.ES256)
+        pubkey = key.public_key()
+
+        pub_pem_source = pubkey.to_pem().strip()
+
+        pub_target = BackendTo(pub_pem_source, ALGORITHMS.ES256)
+
+        assert pub_pem_source == pub_target.to_pem().strip()
+
+    @pytest.mark.parametrize("BackendFrom", [ECDSAECKey, CryptographyECKey])
+    @pytest.mark.parametrize("BackendTo", [ECDSAECKey, CryptographyECKey])
+    def test_private_key_load_cycle(self, BackendFrom, BackendTo):
+        key = BackendFrom(private_key, ALGORITHMS.ES256)
+
+        pem_source = key.to_pem().strip()
+
+        target = BackendTo(pem_source, ALGORITHMS.ES256)
+
+        assert pem_source == target.to_pem().strip()