[feature] Added support for importing ECDSA certificates #118

django_x509/base/models.py

@@ -19,12 +19,18 @@
 from .. import settings as app_settings
 
 KEY_LENGTH_CHOICES = (
-    ("512", "512"),
-    ("1024", "1024"),
-    ("2048", "2048"),
-    ("4096", "4096"),
+    ("256", "256 (ECDSA)"),
+    ("384", "384 (ECDSA)"),
+    ("521", "521 (ECDSA)"),
+    ("512", "512 (RSA)"),
+    ("1024", "1024 (RSA)"),
+    ("2048", "2048 (RSA)"),
+    ("4096", "4096 (RSA)"),
 )
 
+RSA_KEY_LENGTHS = ("512", "1024", "2048", "4096")
+EC_KEY_LENGTHS = ("256", "384", "521")
+
 DIGEST_CHOICES = (
     ("sha1", "SHA1"),
     ("sha224", "SHA224"),
@@ -178,6 +184,54 @@ def clean(self):
         if self.serial_number:
             self._validate_serial_number()
         self._verify_extension_format()
+        if self.certificate:
+            try:
+                public_key = self.x509.public_key()
+                is_ec_length = self.key_length in EC_KEY_LENGTHS
+                is_rsa_length = self.key_length in RSA_KEY_LENGTHS
+                if isinstance(public_key, rsa.RSAPublicKey):
+                    if is_ec_length:
+                        raise ValidationError(
+                            {
+                                "key_length": _(
+                                    "Selected length is only valid for ECDSA, "
+                                    "but an RSA key was provided."
+                                )
+                            }
+                        )
+                    if str(public_key.key_size) != self.key_length:
+                        raise ValidationError(
+                            {
+                                "key_length": _(
+                                    "The provided RSA key size (%s) "
+                                    "does not match the selected length."
+                                )
+                                % public_key.key_size
+                            }
+                        )
+                elif isinstance(public_key, ec.EllipticCurvePublicKey):
+                    if is_rsa_length:
+                        raise ValidationError(
+                            {
+                                "key_length": _(
+                                    "Selected length is only valid for RSA, "
+                                    "but an ECDSA key was provided."
+                                )
+                            }
+                        )
+                    actual_ec_length = str(public_key.curve.key_size)
+                    if actual_ec_length != self.key_length:
+                        raise ValidationError(
+                            {
+                                "key_length": _(
+                                    "The provided ECDSA curve size (%s) "
+                                    "does not match the selected length."
+                                )
+                                % actual_ec_length
+                            }
+                        )
+            except (ValueError, UnsupportedAlgorithm):
+                pass
 
     def save(self, *args, **kwargs):
         if self._state.adding and not self.certificate and not self.private_key:
@@ -280,10 +334,24 @@ def _generate(self):
         for attr in ["x509", "pkey"]:
             if attr in self.__dict__:
                 del self.__dict__[attr]
-        key = rsa.generate_private_key(
-            public_exponent=65537,
-            key_size=int(self.key_length),
-        )
+        is_ec = self.key_length in EC_KEY_LENGTHS
+        if is_ec:
+            curves = {
+                "256": ec.SECP256R1(),
+                "384": ec.SECP384R1(),
+                "521": ec.SECP521R1(),
+            }
+            curve = curves.get(self.key_length)
+            if not curve:
+                raise ValidationError(
+                    _("Unsupported EC key length: %s") % self.key_length
+                )
+            key = ec.generate_private_key(curve)
+        else:
+            key = rsa.generate_private_key(
+                public_exponent=65537,
+                key_size=int(self.key_length),
+            )
         if hasattr(self, "ca"):
             signing_key = self.ca.pkey
             issuer_name = self.ca.x509.subject
@@ -307,7 +375,12 @@ def _generate(self):
             "sha384": hashes.SHA384,
             "sha512": hashes.SHA512,
         }
-        digest_name = self.digest.lower().replace("withrsaencryption", "")
+        digest_name = (
+            self.digest.lower()
+            .replace("withrsaencryption", "")
+            .replace("ecdsa-with-", "")
+            .replace("withsha", "sha")
+        )
         digest_alg = HASH_MAP.get(digest_name, hashes.SHA256)()
         cert = builder.sign(signing_key, digest_alg)
         self.certificate = cert.public_bytes(serialization.Encoding.PEM).decode("utf-8")
@@ -318,7 +391,11 @@ def _generate(self):
         )
         self.private_key = key.private_bytes(
             encoding=serialization.Encoding.PEM,
-            format=serialization.PrivateFormat.TraditionalOpenSSL,
+            format=(
+                serialization.PrivateFormat.PKCS8
+                if is_ec
+                else serialization.PrivateFormat.TraditionalOpenSSL
+            ),
             encryption_algorithm=encryption,
         ).decode("utf-8")
 
@@ -352,6 +429,36 @@ def _import(self):
         imports existing x509 certificates
         """
         cert = self.x509
+        public_key = cert.public_key()
+        if isinstance(public_key, rsa.RSAPublicKey):
+            actual_length = str(public_key.key_size)
+            actual_is_ec = False
+        elif isinstance(public_key, ec.EllipticCurvePublicKey):
+            actual_length = str(public_key.curve.key_size)
+            actual_is_ec = True
+        else:
+            raise ValidationError(
+                _(
+                    "Unsupported key type in certificate. "
+                    "Only RSA and EC keys are supported."
+                )
+            )
+        selected_is_ec = self.key_length in EC_KEY_LENGTHS
+        if selected_is_ec != actual_is_ec:
+            algorithm_expected = "ECDSA" if selected_is_ec else "RSA"
+            algorithm_provided = "ECDSA" if actual_is_ec else "RSA"
+            raise ValidationError(
+                {
+                    "key_length": _(
+                        "Algorithm mismatch: You selected a length for %s, "
+                        "but the provided certificate contains an %s key."
+                    )
+                    % (algorithm_expected, algorithm_provided)
+                }
+            )
+        if actual_is_ec and actual_length not in EC_KEY_LENGTHS:
+            raise ValidationError(_("Unsupported EC curve size: %s") % actual_length)
+        self.key_length = actual_length
         # when importing an end entity certificate
         if hasattr(self, "ca"):
             self._verify_ca()
@@ -366,7 +473,6 @@ def _import(self):
             email = str(attrs[0].value) if attrs else ""
 
         self.email = email
-        self.key_length = str(cert.public_key().key_size)
         self.digest = cert.signature_hash_algorithm.name.lower()
         self.validity_start = cert.not_valid_before_utc
         self.validity_end = cert.not_valid_after_utc

django_x509/tests/test_ca.py

@@ -2,7 +2,8 @@
 from datetime import timezone as dt_timezone
 
 from cryptography import x509
-from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import ec, rsa
 from cryptography.x509.oid import NameOID
 from django.core.exceptions import ValidationError
 from django.test import TestCase
@@ -684,3 +685,57 @@ def test_renewal_serial_sync(self):
         cert_obj = x509.load_pem_x509_certificate(cert.certificate.encode())
         pem_serial = cert_obj.serial_number
         self.assertEqual(int(cert.serial_number), pem_serial)
+
+    def test_ca_ecdsa_full_lifecycle(self):
+        curves_to_test = [
+            ("256", ec.SECP256R1, hashes.SHA256()),
+            ("384", ec.SECP384R1, hashes.SHA384()),
+            ("521", ec.SECP521R1, hashes.SHA512()),
+        ]
+        for length, curve_class, digest in curves_to_test:
+            with self.subTest(key_length=length):
+                priv_key = ec.generate_private_key(curve_class())
+                key_pem = priv_key.private_bytes(
+                    encoding=serialization.Encoding.PEM,
+                    format=serialization.PrivateFormat.PKCS8,
+                    encryption_algorithm=serialization.NoEncryption(),
+                ).decode("utf-8")
+                now = datetime.now(dt_timezone.utc)
+                subject = issuer = x509.Name(
+                    [x509.NameAttribute(NameOID.COMMON_NAME, "test")]
+                )
+                cert = (
+                    x509.CertificateBuilder()
+                    .subject_name(subject)
+                    .issuer_name(issuer)
+                    .public_key(priv_key.public_key())
+                    .serial_number(x509.random_serial_number())
+                    .not_valid_before(now)
+                    .not_valid_after(now + timedelta(days=10))
+                    .sign(priv_key, digest)
+                )
+                cert_pem = cert.public_bytes(serialization.Encoding.PEM).decode("utf-8")
+                ca = Ca(
+                    name=f"EC-{length}",
+                    certificate=cert_pem,
+                    private_key=key_pem,
+                    key_length=length,
+                )
+                ca.full_clean()
+                ca.save()
+                self.assertEqual(ca.key_length, length)
+                self.assertIsInstance(ca.pkey, ec.EllipticCurvePrivateKey)
+                gen_ca = Ca(
+                    name=f"Gen-EC-{length}",
+                    key_length=length,
+                )
+                gen_ca.full_clean()
+                gen_ca.save()
+                self.assertIsInstance(gen_ca.pkey, ec.EllipticCurvePrivateKey)
+                original_cert = gen_ca.certificate
+                original_key = gen_ca.private_key
+                gen_ca.renew()
+                gen_ca.refresh_from_db()
+                self.assertEqual(gen_ca.key_length, length)
+                self.assertNotEqual(gen_ca.private_key, original_key)
+                self.assertNotEqual(original_cert, gen_ca.certificate)

django_x509/tests/test_cert.py

@@ -2,7 +2,8 @@
 from datetime import timezone as dt_timezone
 
 from cryptography import x509
-from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import ec, rsa
 from cryptography.x509.oid import NameOID
 from django.core.exceptions import ValidationError
 from django.test import TestCase
@@ -490,3 +491,58 @@ def test_cert_common_name_length(self):
         message_dict = context_manager.exception.message_dict
         self.assertIn("common_name", message_dict)
         self.assertEqual(message_dict["common_name"][0], msg)
+
+    def test_cert_ecdsa_full_lifecycle(self):
+        curves_to_test = [
+            ("256", ec.SECP256R1, hashes.SHA256()),
+            ("384", ec.SECP384R1, hashes.SHA384()),
+            ("521", ec.SECP521R1, hashes.SHA512()),
+        ]
+        for length, curve_class, digest in curves_to_test:
+            with self.subTest(key_length=length):
+                ca = Ca(name=f"CA-{length}", key_length=length)
+                ca.full_clean()
+                ca.save()
+                priv_key = ec.generate_private_key(curve_class())
+                key_pem = priv_key.private_bytes(
+                    encoding=serialization.Encoding.PEM,
+                    format=serialization.PrivateFormat.PKCS8,
+                    encryption_algorithm=serialization.NoEncryption(),
+                ).decode("utf-8")
+                now = datetime.now(dt_timezone.utc)
+                subject = x509.Name(
+                    [x509.NameAttribute(NameOID.COMMON_NAME, "test-cert")]
+                )
+                cert = (
+                    x509.CertificateBuilder()
+                    .subject_name(subject)
+                    .issuer_name(ca.x509.subject)
+                    .public_key(priv_key.public_key())
+                    .serial_number(x509.random_serial_number())
+                    .not_valid_before(now)
+                    .not_valid_after(now + timedelta(days=10))
+                    .sign(ca.pkey, digest)
+                )
+                cert_pem = cert.public_bytes(serialization.Encoding.PEM).decode("utf-8")
+                entity_cert = Cert(
+                    name=f"EC-{length}-Import",
+                    ca=ca,
+                    certificate=cert_pem,
+                    private_key=key_pem,
+                    key_length=length,
+                )
+                entity_cert.full_clean()
+                entity_cert.save()
+                self.assertEqual(entity_cert.key_length, length)
+                gen_cert = Cert(
+                    name=f"Gen-EC-{length}",
+                    ca=ca,
+                    key_length=length,
+                )
+                gen_cert.full_clean()
+                gen_cert.save()
+                self.assertIsInstance(gen_cert.pkey, ec.EllipticCurvePrivateKey)
+                original_pem = gen_cert.certificate
+                gen_cert.renew()
+                gen_cert.refresh_from_db()
+                self.assertNotEqual(original_pem, gen_cert.certificate)