Allow separate public key in CSR

docs/development/test-vectors.rst

@@ -755,6 +755,12 @@ Custom X.509 Request Vectors
   attribute whose value's tag is encoded in the long form.
 * ``zero-element-attribute.pem`` - A certificate signing request containing an
   attribute whose value has zero elements.
+* ``mldsa-mlkem768.pem`` and ``mldsa-mlkem768-signing-pubkey.pem`` - A 
+  certificate signing request signed using an ML-DSA private key, and 
+  containing a ML-KEM-768 public key. 
+* ``mldsa-mlkem1024.pem`` and ``mldsa-mlkem1024-signing-pubkey.pem`` - A 
+  certificate signing request signed using an ML-DSA private key, and 
+  containing a ML-KEM-1024 public key. 
 
 Custom X.509 Certificate Revocation List Vectors
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

docs/x509/reference.rst

@@ -1615,6 +1615,23 @@ X.509 CSR (Certificate Signing Request) Builder Object
         :returns: A new
             :class:`~cryptography.x509.CertificateSigningRequest`.
 
+    .. method:: verify_directly_signed_by(public_key)
+
+        .. versionadded:: 51.0.0
+
+        :param public_key: One of
+            :data:`~cryptography.hazmat.primitives.asymmetric.types.PublicKeyTypes`.
+
+        Validates that the request is signed by the private key belonging to 
+        provided public key. 
+
+        :return: None
+        :raise ValueError: If the signature algorithms of the request and 
+            provided public key do not match. 
+        :raise TypeError: If the signer does not have a supported public
+            key type.
+        :raise cryptography.exceptions.InvalidSignature: If the
+            signature fails to verify.
 
 .. class:: Name
     :canonical: cryptography.x509.name.Name

src/cryptography/hazmat/bindings/_rust/x509.pyi

@@ -14,6 +14,7 @@ from cryptography.hazmat.primitives.asymmetric.types import (
     CertificateIssuerPublicKeyTypes,
     CertificatePublicKeyTypes,
     PrivateKeyTypes,
+    PublicKeyTypes,
 )
 from cryptography.x509 import certificate_transparency
 
@@ -210,6 +211,10 @@ class CertificateSigningRequest:
     def tbs_certrequest_bytes(self) -> bytes: ...
     @property
     def is_signature_valid(self) -> bool: ...
+    def verify_directly_signed_by(
+        self,
+        public_key: PublicKeyTypes,
+    ) -> None: ...
 
 class PolicyBuilder:
     def time(self, time: datetime.datetime) -> PolicyBuilder: ...

src/cryptography/x509/base.py

@@ -183,13 +183,15 @@ class CertificateSigningRequestBuilder:
     def __init__(
         self,
         subject_name: Name | None = None,
+        public_key: CertificatePublicKeyTypes | None = None,
         extensions: list[Extension[ExtensionType]] = [],
         attributes: list[tuple[ObjectIdentifier, bytes, int | None]] = [],
     ):
         """
         Creates an empty X.509 certificate request (v1).
         """
         self._subject_name = subject_name
+        self._public_key = public_key
         self._extensions = extensions
         self._attributes = attributes
 
@@ -202,7 +204,45 @@ def subject_name(self, name: Name) -> CertificateSigningRequestBuilder:
         if self._subject_name is not None:
             raise ValueError("The subject name may only be set once.")
         return CertificateSigningRequestBuilder(
-            name, self._extensions, self._attributes
+            name, self._public_key, self._extensions, self._attributes
+        )
+
+    def public_key(
+        self,
+        public_key: CertificatePublicKeyTypes,
+    ) -> CertificateSigningRequestBuilder:
+        """
+        Sets the requestor's public key.
+        """
+        if not isinstance(
+            public_key,
+            (
+                dsa.DSAPublicKey,
+                rsa.RSAPublicKey,
+                ec.EllipticCurvePublicKey,
+                ed25519.Ed25519PublicKey,
+                ed448.Ed448PublicKey,
+                mldsa.MLDSA44PublicKey,
+                mldsa.MLDSA65PublicKey,
+                mldsa.MLDSA87PublicKey,
+                mlkem.MLKEM768PublicKey,
+                mlkem.MLKEM1024PublicKey,
+                x25519.X25519PublicKey,
+                x448.X448PublicKey,
+            ),
+        ):
+            raise TypeError(
+                "Expecting one of DSAPublicKey, RSAPublicKey,"
+                " EllipticCurvePublicKey, Ed25519PublicKey,"
+                " Ed448PublicKey, MLDSA44PublicKey, MLDSA65PublicKey,"
+                " MLDSA87PublicKey, MLKEM768PublicKey, MLKEM1024PublicKey,"
+                " X25519PublicKey or X448PublicKey."
+            )
+
+        if self._public_key is not None:
+            raise ValueError("The public key may only be set once.")
+        return CertificateSigningRequestBuilder(
+            self._subject_name, public_key, self._extensions, self._attributes
         )
 
     def add_extension(
@@ -219,6 +259,7 @@ def add_extension(
 
         return CertificateSigningRequestBuilder(
             self._subject_name,
+            self._public_key,
             [*self._extensions, extension],
             self._attributes,
         )
@@ -251,6 +292,7 @@ def add_attribute(
 
         return CertificateSigningRequestBuilder(
             self._subject_name,
+            self._public_key,
             self._extensions,
             [*self._attributes, (oid, value, tag)],
         )
@@ -265,7 +307,9 @@ def sign(
         ecdsa_deterministic: bool | None = None,
     ) -> CertificateSigningRequest:
         """
-        Signs the request using the requestor's private key.
+        Signs the request using the requestor's private key. If no public key
+        was indicated, the public key associated with specified private key
+        will be included instead.
         """
         if self._subject_name is None:
             raise ValueError("A CertificateSigningRequest must have a subject")

src/rust/src/x509/csr.rs

@@ -187,14 +187,21 @@ impl CertificateSigningRequest {
     #[getter]
     fn is_signature_valid(&self, py: pyo3::Python<'_>) -> CryptographyResult<bool> {
         let public_key = self.public_key(py)?;
-        Ok(sign::verify_signature_with_signature_algorithm(
+        Ok(self.verify_directly_signed_by(py, public_key).is_ok())
+    }
+
+    fn verify_directly_signed_by<'p>(
+        &self,
+        py: pyo3::Python<'p>,
+        public_key: pyo3::Bound<'p, pyo3::PyAny>,
+    ) -> CryptographyResult<()> {
+        sign::verify_signature_with_signature_algorithm(
             py,
             public_key,
             &self.raw.borrow_dependent().signature_alg,
             self.raw.borrow_dependent().signature.as_bytes(),
             &asn1::write_single(&self.raw.borrow_dependent().csr_info)?,
         )
-        .is_ok())
     }
 }
 
@@ -265,8 +272,12 @@ pub(crate) fn create_x509_csr(
         rsa_padding.clone(),
     )?;
 
-    let spki_bytes = private_key
-        .call_method0(pyo3::intern!(py, "public_key"))?
+    let public_key = match builder.getattr(pyo3::intern!(py, "_public_key"))? {
+        pk if pk.is_none() => private_key.call_method0(pyo3::intern!(py, "public_key"))?,
+        pk => pk,
+    };
+
+    let spki_bytes = public_key
         .call_method1(
             pyo3::intern!(py, "public_bytes"),
             (

tests/x509/test_x509.py

@@ -2037,6 +2037,29 @@ def test_admissions_extension(self):
         assert ext.value == x509.Admissions(authority=None, admissions=[])
 
 
+class TestCertificateRequest:
+    def test_public_key_must_be_public_key(
+        self, rsa_key_2048: rsa.RSAPrivateKey
+    ):
+        private_key = rsa_key_2048
+        builder = x509.CertificateSigningRequestBuilder()
+
+        with pytest.raises(TypeError):
+            builder.public_key(typing.cast(typing.Any, private_key))
+
+    def test_public_key_may_only_be_set_once(
+        self, rsa_key_2048: rsa.RSAPrivateKey
+    ):
+        private_key = rsa_key_2048
+        public_key = private_key.public_key()
+        builder = x509.CertificateSigningRequestBuilder().public_key(
+            public_key
+        )
+
+        with pytest.raises(ValueError):
+            builder.public_key(public_key)
+
+
 class TestRSACertificateRequest:
     @pytest.mark.parametrize(
         ("path", "loader_func"),
@@ -6391,6 +6414,144 @@ def test_tbs_certrequest_bytes(self, backend):
         )
 
 
+class TestMLKEMCertificateRequest:
+    @pytest.mark.supported(
+        only_if=lambda backend: (
+            backend.mldsa_supported() and backend.mlkem_supported()
+        ),
+        skip_message="Does not support ML-DSA and/or ML-KEM",
+    )
+    @pytest.mark.parametrize(
+        (
+            "enclosed_key_path",
+            "enclosed_pub_key_cls",
+            "signing_pub_key_path",
+            "signing_pub_key_cls",
+            "signature_algorithm_oid",
+        ),
+        [
+            (
+                os.path.join("x509", "requests", "mldsa-mlkem768.pem"),
+                mlkem.MLKEM768PublicKey,
+                os.path.join(
+                    "x509", "requests", "mldsa-mlkem768-signing-pubkey.pem"
+                ),
+                mldsa.MLDSA65PublicKey,
+                SignatureAlgorithmOID.ML_DSA_65,
+            ),
+            (
+                os.path.join("x509", "requests", "mldsa-mlkem1024.pem"),
+                mlkem.MLKEM1024PublicKey,
+                os.path.join(
+                    "x509", "requests", "mldsa-mlkem1024-signing-pubkey.pem"
+                ),
+                mldsa.MLDSA65PublicKey,
+                SignatureAlgorithmOID.ML_DSA_65,
+            ),
+        ],
+    )
+    def test_load_request_mldsa_mlkem(
+        self,
+        enclosed_key_path,
+        enclosed_pub_key_cls,
+        signing_pub_key_path,
+        signing_pub_key_cls,
+        signature_algorithm_oid,
+    ):
+        request = _load_cert(enclosed_key_path, x509.load_pem_x509_csr)
+
+        signing_key = _load_cert(
+            signing_pub_key_path, serialization.load_pem_public_key
+        )
+
+        assert isinstance(request.public_key(), enclosed_pub_key_cls)
+        assert isinstance(signing_key, signing_pub_key_cls)
+
+        assert request.signature_algorithm_oid == signature_algorithm_oid
+
+        assert isinstance(request.subject, x509.Name)
+        assert list(request.subject) == [
+            x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
+        ]
+
+        assert not request.is_signature_valid
+        request.verify_directly_signed_by(signing_key)
+
+        with pytest.raises(TypeError):
+            unsupported_algorithm = request.public_key()
+            request.verify_directly_signed_by(unsupported_algorithm)
+
+        with pytest.raises(ValueError):
+            wrong_algorithm = ec.generate_private_key(
+                ec.SECP256R1()
+            ).public_key()
+            request.verify_directly_signed_by(wrong_algorithm)
+
+        with pytest.raises(InvalidSignature):
+            wrong_pub_key = mldsa.MLDSA65PrivateKey.generate().public_key()
+            request.verify_directly_signed_by(wrong_pub_key)
+
+    @pytest.mark.supported(
+        only_if=lambda backend: (
+            backend.mldsa_supported() and backend.mlkem_supported()
+        ),
+        skip_message="Does not support ML-DSA and/or ML-KEM",
+    )
+    @pytest.mark.parametrize(
+        (
+            "enclosed_key_cls",
+            "enclosed_pub_key_cls",
+            "signing_key_cls",
+            "signature_algorithm_oid",
+        ),
+        [
+            (
+                mlkem.MLKEM768PrivateKey,
+                mlkem.MLKEM768PublicKey,
+                mldsa.MLDSA65PrivateKey,
+                SignatureAlgorithmOID.ML_DSA_65,
+            ),
+            (
+                mlkem.MLKEM1024PrivateKey,
+                mlkem.MLKEM1024PublicKey,
+                mldsa.MLDSA65PrivateKey,
+                SignatureAlgorithmOID.ML_DSA_65,
+            ),
+        ],
+    )
+    def test_build_request_mldsa_mlkem(
+        self,
+        enclosed_key_cls,
+        enclosed_pub_key_cls,
+        signing_key_cls,
+        signature_algorithm_oid,
+    ):
+        enclosed_key = enclosed_key_cls.generate()
+        signing_key = signing_key_cls.generate()
+
+        request = (
+            x509.CertificateSigningRequestBuilder()
+            .subject_name(
+                x509.Name([x509.NameAttribute(NameOID.COUNTRY_NAME, "US")])
+            )
+            .public_key(enclosed_key.public_key())
+            .sign(signing_key, None)
+        )
+
+        public_key = request.public_key()
+        assert isinstance(public_key, enclosed_pub_key_cls)
+
+        assert request.signature_algorithm_oid == signature_algorithm_oid
+
+        assert isinstance(request.subject, x509.Name)
+        assert list(request.subject) == [
+            x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
+        ]
+
+        assert not request.is_signature_valid
+        request.verify_directly_signed_by(signing_key.public_key())
+
+
 class TestOtherCertificate:
     def test_unsupported_subject_public_key_info(self):
         cert = _load_cert(