Make X.509 certificate and key handling more alike for RSA and EC keys.

Author: rohe

src/cryptojwt/jwk/ec.py

@@ -2,13 +2,15 @@
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric import ec
 
+from .asym import AsymmetricKey
+from .x509 import import_public_key_from_pem_data
+from .x509 import import_public_key_from_pem_file
 from ..exception import DeSerializationNotPossible
 from ..exception import JWKESTException
 from ..exception import UnsupportedECurve
 from ..utils import as_unicode
 from ..utils import deser
 from ..utils import long_to_base64
-from .asym import AsymmetricKey
 
 # This is used to translate between the curve representation in
 # Cryptography and the one used by NIST (and in RFC 7518)
@@ -123,7 +125,7 @@ class ECKey(AsymmetricKey):
     required = ["kty", "crv", "x", "y"]
 
     def __init__(
-        self, kty="EC", alg="", use="", kid="", crv="", x="", y="", d="", **kwargs
+            self, kty="EC", alg="", use="", kid="", crv="", x="", y="", d="", **kwargs
     ):
         AsymmetricKey.__init__(self, kty, alg, use, kid, **kwargs)
         self.crv = crv
@@ -315,3 +317,32 @@ def new_ec_key(crv, kid="", **kwargs):
         _rk.add_kid()
 
     return _rk
+
+
+def import_public_ec_key_from_file(filename):
+    """
+    Read a public Elliptic Curve key from a PEM file.
+
+    :param filename: The name of the file
+    :param passphrase: A pass phrase to use to unpack the PEM file.
+    :return: A cryptography.hazmat.primitives.asymmetric.ec.EllipticCurvePublicKey instance
+    """
+    public_key = import_public_key_from_pem_file(filename)
+    if isinstance(public_key, ec.EllipticCurvePublicKey):
+        return public_key
+    else:
+        return ValueError('Not a Elliptic Curve key')
+
+
+def import_ec_key(pem_data):
+    """
+    Extract an Elliptic Curve key from a PEM-encoded X.509 certificate
+
+    :param pem_data: Elliptic Curve key encoded in standard form
+    :return: ec.EllipticCurvePublicKey
+    """
+    public_key = import_public_key_from_pem_data(pem_data)
+    if isinstance(public_key, ec.EllipticCurvePublicKey):
+        return public_key
+    else:
+        return ValueError('Not a Elliptic Curve key')

src/cryptojwt/jwk/jwk.py

@@ -2,7 +2,10 @@
 import json
 import os
 
+from cryptography import x509
 from cryptography.hazmat import backends
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric import ec
 from cryptography.hazmat.primitives.asymmetric import rsa
 from cryptography.hazmat.primitives.asymmetric.rsa import rsa_crt_dmp1

src/cryptojwt/jwk/rsa.py

@@ -1,22 +1,24 @@
 import base64
-import hashlib
 import logging
 
-from cryptography import x509
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric import rsa
 
+from . import JWK
+from .asym import AsymmetricKey
+from .x509 import der_cert
+from .x509 import import_private_key_from_pem_file
+from .x509 import import_public_key_from_pem_data
+from .x509 import import_public_key_from_pem_file
+from .x509 import x5t_calculation
 from ..exception import DeSerializationNotPossible
 from ..exception import JWKESTException
 from ..exception import SerializationNotPossible
 from ..exception import UnsupportedKeyType
 from ..utils import as_unicode
-from ..utils import b64e
 from ..utils import deser
 from ..utils import long_to_base64
-from . import JWK
-from .asym import AsymmetricKey
 
 logger = logging.getLogger(__name__)
 
@@ -67,11 +69,11 @@ def import_private_rsa_key_from_file(filename, passphrase=None):
     :return: A
         cryptography.hazmat.primitives.asymmetric.rsa.RSAPrivateKey instance
     """
-    with open(filename, "rb") as key_file:
-        private_key = serialization.load_pem_private_key(
-            key_file.read(), password=passphrase, backend=default_backend()
-        )
-    return private_key
+    private_key = import_private_key_from_pem_file(filename, passphrase)
+    if isinstance(private_key, rsa.RSAPrivateKey):
+        return private_key
+    else:
+        return ValueError('Not a RSA key')
 
 
 def import_public_rsa_key_from_file(filename):
@@ -80,14 +82,13 @@ def import_public_rsa_key_from_file(filename):
 
     :param filename: The name of the file
     :param passphrase: A pass phrase to use to unpack the PEM file.
-    :return: A
-        cryptography.hazmat.primitives.asymmetric.rsa.RSAPublicKey instance
+    :return: A cryptography.hazmat.primitives.asymmetric.rsa.RSAPublicKey instance
     """
-    with open(filename, "rb") as key_file:
-        public_key = serialization.load_pem_public_key(
-            key_file.read(), backend=default_backend()
-        )
-    return public_key
+    public_key = import_public_key_from_pem_file(filename)
+    if isinstance(public_key, rsa.RSAPublicKey):
+        return public_key
+    else:
+        return ValueError('Not a RSA key')
 
 
 def import_rsa_key(pem_data):
@@ -97,12 +98,11 @@ def import_rsa_key(pem_data):
     :param pem_data: RSA key encoded in standard form
     :return: rsa.RSAPublicKey instance
     """
-    if not pem_data.startswith(PREFIX):
-        pem_data = bytes("{}\n{}\n{}".format(PREFIX, pem_data, POSTFIX), "utf-8")
+    public_key = import_public_key_from_pem_data(pem_data)
+    if isinstance(public_key, rsa.RSAPublicKey):
+        return public_key
     else:
-        pem_data = bytes(pem_data, "utf-8")
-    cert = x509.load_pem_x509_certificate(pem_data, default_backend())
-    return cert.public_key()
+        return ValueError('Not a RSA key')
 
 
 def import_rsa_key_from_cert_file(pem_file):
@@ -182,46 +182,6 @@ def rsa_construct_private(numbers):
     return rprivn.private_key(default_backend())
 
 
-def der_cert(der_data):
-    """
-    Load a DER encoded certificate
-
-    :param der_data: DER-encoded certificate
-    :return: A cryptography.x509.certificate instance
-    """
-    if isinstance(der_data, str):
-        der_data = bytes(der_data, "utf-8")
-    return x509.load_der_x509_certificate(der_data, default_backend())
-
-
-def load_x509_cert(url, httpc, spec2key, **get_args):
-    """
-    Get and transform a X509 cert into a key.
-
-    :param url: Where the X509 cert can be found
-    :param httpc: HTTP client to use for fetching
-    :param spec2key: A dictionary over keys already seen
-    :param get_args: Extra key word arguments to the HTTP GET request
-    :return: List of 2-tuples (keytype, key)
-    """
-    try:
-        r = httpc("GET", url, allow_redirects=True, **get_args)
-        if r.status_code == 200:
-            cert = str(r.text)
-            try:
-                public_key = spec2key[cert]  # If I've already seen it
-            except KeyError:
-                public_key = import_rsa_key(cert)
-                spec2key[cert] = public_key
-            if isinstance(public_key, rsa.RSAPublicKey):
-                return {"rsa": public_key}
-        else:
-            raise Exception("HTTP Get error: %s" % r.status_code)
-    except Exception as err:  # not a RSA key
-        logger.warning("Can't load key: %s" % err)
-        return []
-
-
 def cmp_public_numbers(pn1, pn2):
     """
     Compare 2 sets of public numbers. These is a way to compare
@@ -255,22 +215,6 @@ def cmp_private_numbers(pn1, pn2):
     return True
 
 
-def x5t_calculation(cert):
-    """
-    base64url-encoded SHA-1 thumbprint (a.k.a. digest) of the DER
-    encoding of an X.509 certificate.
-
-    :param cert: DER encoded X.509 certificate
-    :return: x5t value
-    """
-    if isinstance(cert, str):
-        der_cert = base64.b64decode(cert.encode("ascii"))
-    else:
-        der_cert = base64.b64decode(cert)
-
-    return b64e(hashlib.sha1(der_cert).digest())
-
-
 class RSAKey(AsymmetricKey):
     """
     JSON Web key representation of a RSA key
@@ -303,24 +247,24 @@ class RSAKey(AsymmetricKey):
     required = ["kty", "n", "e"]
 
     def __init__(
-        self,
-        kty="RSA",
-        alg="",
-        use="",
-        kid="",
-        x5c=None,
-        x5t="",
-        x5u="",
-        n="",
-        e="",
-        d="",
-        p="",
-        q="",
-        dp="",
-        dq="",
-        di="",
-        qi="",
-        **kwargs
+            self,
+            kty="RSA",
+            alg="",
+            use="",
+            kid="",
+            x5c=None,
+            x5t="",
+            x5u="",
+            n="",
+            e="",
+            d="",
+            p="",
+            q="",
+            dp="",
+            dq="",
+            di="",
+            qi="",
+            **kwargs
     ):
         AsymmetricKey.__init__(self, kty, alg, use, kid, x5c, x5t, x5u, **kwargs)
         self.n = n

src/cryptojwt/jwk/x509.py

@@ -0,0 +1,122 @@
+import base64
+import hashlib
+import logging
+
+from cryptography import x509
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import ec
+from cryptography.hazmat.primitives.asymmetric import rsa
+
+from cryptojwt.utils import b64e
+
+logger = logging.getLogger(__name__)
+
+
+def import_public_key_from_pem_file(filename):
+    """
+    Read a public RSA key from a PEM file.
+
+    :param filename: The name of the file
+    :param passphrase: A pass phrase to use to unpack the PEM file.
+    :return: A public key instance
+    """
+    with open(filename, "rb") as key_file:
+        public_key = serialization.load_pem_public_key(
+            key_file.read(), backend=default_backend()
+        )
+    return public_key
+
+
+def import_private_key_from_pem_file(filename, passphrase=None):
+    """
+    Read a private RSA key from a PEM file.
+
+    :param filename: The name of the file
+    :param passphrase: A pass phrase to use to unpack the PEM file.
+    :return: A private key instance
+    """
+    with open(filename, "rb") as key_file:
+        private_key = serialization.load_pem_private_key(
+            key_file.read(), password=passphrase, backend=default_backend()
+        )
+    return private_key
+
+
+PREFIX = "-----BEGIN CERTIFICATE-----"
+POSTFIX = "-----END CERTIFICATE-----"
+
+
+def import_public_key_from_pem_data(pem_data):
+    """
+    Extract an RSA key from a PEM-encoded X.509 certificate
+
+    :param pem_data: RSA key encoded in standard form
+    :return: rsa.RSAPublicKey instance
+    """
+    if not pem_data.startswith(PREFIX):
+        pem_data = bytes("{}\n{}\n{}".format(PREFIX, pem_data, POSTFIX), "utf-8")
+    else:
+        pem_data = bytes(pem_data, "utf-8")
+    cert = x509.load_pem_x509_certificate(pem_data, default_backend())
+    return cert.public_key()
+
+
+def der_cert(der_data):
+    """
+    Load a DER encoded certificate
+
+    :param der_data: DER-encoded certificate
+    :return: A cryptography.x509.certificate instance
+    """
+    if isinstance(der_data, str):
+        der_data = bytes(der_data, "utf-8")
+    return x509.load_der_x509_certificate(der_data, default_backend())
+
+
+def load_x509_cert(url, httpc, spec2key, **get_args):
+    """
+    Get and transform a X509 cert into a key.
+
+    :param url: Where the X509 cert can be found
+    :param httpc: HTTP client to use for fetching
+    :param spec2key: A dictionary over keys already seen
+    :param get_args: Extra key word arguments to the HTTP GET request
+    :return: List of 2-tuples (keytype, key)
+    """
+    try:
+        r = httpc("GET", url, allow_redirects=True, **get_args)
+        if r.status_code == 200:
+            cert = str(r.text)
+            try:
+                public_key = spec2key[cert]  # If I've already seen it
+            except KeyError:
+                public_key = import_public_key_from_pem_data(cert)
+                spec2key[cert] = public_key
+
+            if isinstance(public_key, rsa.RSAPublicKey):
+                return {"rsa": public_key}
+            elif isinstance(public_key, ec.EllipticCurvePublicKey):
+                return {'ec': public_key}
+        else:
+            raise Exception("HTTP Get error: %s" % r.status_code)
+    except Exception as err:  # not a RSA key
+        logger.warning("Can't load key: %s" % err)
+        return []
+
+
+def x5t_calculation(cert):
+    """
+    base64url-encoded SHA-1 thumbprint (a.k.a. digest) of the DER
+    encoding of an X.509 certificate.
+
+    :param cert: DER encoded X.509 certificate
+    :return: x5t value
+    """
+    if isinstance(cert, str):
+        der_cert = base64.b64decode(cert.encode("ascii"))
+    else:
+        der_cert = base64.b64decode(cert)
+
+    return b64e(hashlib.sha1(der_cert).digest())
+

src/cryptojwt/jwx.py

@@ -11,7 +11,7 @@
 from .jwk.jwk import key_from_jwk_dict
 from .jwk.rsa import RSAKey
 from .jwk.rsa import import_rsa_key
-from .jwk.rsa import load_x509_cert
+from .jwk.x509 import load_x509_cert
 from .utils import as_bytes
 from .utils import as_unicode
 from .utils import b64d