add Statement of Possession attribute support to CSRs

src/cryptography/hazmat/_oid.py

@@ -240,6 +240,7 @@ class CertificatePoliciesOID:
 class AttributeOID:
     CHALLENGE_PASSWORD = ObjectIdentifier("1.2.840.113549.1.9.7")
     UNSTRUCTURED_NAME = ObjectIdentifier("1.2.840.113549.1.9.2")
+    STATEMENT_OF_POSSESSION = ObjectIdentifier("1.3.6.1.5.5.7.11.2")
 
 
 _OID_NAMES = {

src/cryptography/x509/__init__.py

@@ -18,6 +18,7 @@
     InvalidVersion,
     RevokedCertificate,
     RevokedCertificateBuilder,
+    StatementOfPossession,
     Version,
     load_der_x509_certificate,
     load_der_x509_crl,
@@ -246,6 +247,7 @@
     "RevokedCertificateBuilder",
     "SignatureAlgorithmOID",
     "SignedCertificateTimestamps",
+    "StatementOfPossession",
     "SubjectAlternativeName",
     "SubjectInformationAccess",
     "SubjectKeyIdentifier",

src/cryptography/x509/base.py

@@ -34,7 +34,15 @@
     _make_sequence_methods,
 )
 from cryptography.x509.name import Name, _ASN1Type
-from cryptography.x509.oid import ObjectIdentifier
+from cryptography.x509.oid import AttributeOID, ObjectIdentifier
+
+if typing.TYPE_CHECKING:
+    from cryptography.hazmat.primitives.asymmetric.types import (
+        CertificateIssuerPrivateKeyTypes,
+    )
+    from cryptography.hazmat.primitives.asymmetric.types import (
+        CertificatePublicKeyTypes as PublicKeyTypes,
+    )
 
 _EARLIEST_UTC_TIME = datetime.datetime(1950, 1, 1)
 
@@ -159,6 +167,20 @@ def __init__(self, msg: str, parsed_version: int) -> None:
         self.parsed_version = parsed_version
 
 
+class StatementOfPossession(Attribute):
+    def __init__(self, value: bytes) -> None:
+        if not isinstance(value, bytes):
+            raise TypeError("value must be bytes")
+        super().__init__(
+            AttributeOID.STATEMENT_OF_POSSESSION,
+            value,
+            _type=0x30,  # 0x30 = ASN.1 SEQUENCE tag
+        )
+
+    def __repr__(self) -> str:
+        return f"<StatementOfPossession(value={self.value!r})>"
+
+
 Certificate = rust_x509.Certificate
 RevokedCertificate = rust_x509.RevokedCertificate
 
@@ -185,13 +207,15 @@ def __init__(
         subject_name: Name | None = None,
         extensions: list[Extension[ExtensionType]] = [],
         attributes: list[tuple[ObjectIdentifier, bytes, int | None]] = [],
+        public_key: PublicKeyTypes | None = None,
     ):
         """
         Creates an empty X.509 certificate request (v1).
         """
         self._subject_name = subject_name
         self._extensions = extensions
         self._attributes = attributes
+        self._public_key = public_key
 
     def subject_name(self, name: Name) -> CertificateSigningRequestBuilder:
         """
@@ -202,7 +226,10 @@ 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._extensions,
+            self._attributes,
+            public_key=self._public_key,
         )
 
     def add_extension(
@@ -221,6 +248,7 @@ def add_extension(
             self._subject_name,
             [*self._extensions, extension],
             self._attributes,
+            public_key=self._public_key,
         )
 
     def add_attribute(
@@ -253,6 +281,7 @@ def add_attribute(
             self._subject_name,
             self._extensions,
             [*self._attributes, (oid, value, tag)],
+            public_key=self._public_key,
         )
 
     def sign(
@@ -290,6 +319,18 @@ def sign(
             ecdsa_deterministic,
         )
 
+    def public_key(
+        self, public_key: PublicKeyTypes
+    ) -> CertificateSigningRequestBuilder:
+        if self._public_key is not None:
+            raise ValueError("The public key has already been set")
+        return CertificateSigningRequestBuilder(
+            subject_name=self._subject_name,
+            extensions=self._extensions,
+            attributes=self._attributes,
+            public_key=public_key,
+        )
+
 
 class CertificateBuilder:
     _extensions: list[Extension[ExtensionType]]

src/rust/src/x509/csr.rs

@@ -265,8 +265,15 @@ pub(crate) fn create_x509_csr(
         rsa_padding.clone(),
     )?;
 
-    let spki_bytes = private_key
-        .call_method0(pyo3::intern!(py, "public_key"))?
+    let py_public_key = builder.getattr(pyo3::intern!(py, "_public_key"))?;
+
+    let public_key = if py_public_key.is_none() {
+        private_key.call_method0(pyo3::intern!(py, "public_key"))?
+    } else {
+        py_public_key
+    };
+
+    let spki_bytes = public_key
         .call_method1(
             pyo3::intern!(py, "public_bytes"),
             (
@@ -308,8 +315,8 @@ pub(crate) fn create_x509_csr(
             Option<u8>,
         ) = py_attr?.extract()?;
         let oid = py_oid_to_oid(py_oid)?;
-        let tag = if let Some(tag) = tag {
-            asn1::Tag::from_bytes(&[tag])?.0
+        let tag = if let Some(tag_num) = tag {
+            asn1::Tag::from_bytes(&[tag_num])?.0
         } else {
             if std::str::from_utf8(&value).is_err() {
                 return Err(CryptographyError::from(

tests/x509/test_x509.py

@@ -7448,3 +7448,86 @@ def test_load_pem_x509_certificates():
     assert len(certs) == 2
     assert certs[0].serial_number == 16160
     assert certs[1].serial_number == 146039
+
+
+def test_csr_decoupled_signing_key():
+    subject_private_key = rsa.generate_private_key(
+        public_exponent=65537, key_size=2048
+    )
+    signing_private_key = ec.generate_private_key(ec.SECP256R1())
+
+    builder = (
+        x509.CertificateSigningRequestBuilder()
+        .subject_name(
+            x509.Name(
+                [x509.NameAttribute(NameOID.COMMON_NAME, "RFC 9883 Test")]
+            )
+        )
+        .public_key(subject_private_key.public_key())
+    )
+
+    csr = builder.sign(signing_private_key, hashes.SHA256())
+
+    # 1. Subject public key in CSR matches subject_private_key
+    assert csr.public_key().public_bytes(
+        serialization.Encoding.DER,
+        serialization.PublicFormat.SubjectPublicKeyInfo,
+    ) == subject_private_key.public_key().public_bytes(
+        serialization.Encoding.DER,
+        serialization.PublicFormat.SubjectPublicKeyInfo,
+    )
+
+    # 2. Self-signature validation using embedded SPKI returns False
+    # (expected for decoupled signing key)
+    assert not csr.is_signature_valid
+
+
+def test_csr_statement_of_possession_attribute():
+    raw_pop_bytes = b"sample_pop_structure"
+
+    builder = (
+        x509.CertificateSigningRequestBuilder()
+        .subject_name(
+            x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "SoP Test")])
+        )
+        .add_attribute(
+            x509.oid.AttributeOID.STATEMENT_OF_POSSESSION,
+            raw_pop_bytes,
+        )
+    )
+
+    csr = builder.sign(
+        rsa.generate_private_key(public_exponent=65537, key_size=2048),
+        hashes.SHA256(),
+    )
+    der = csr.public_bytes(serialization.Encoding.DER)
+
+    parsed_csr = x509.load_der_x509_csr(der)
+
+    attr = parsed_csr.attributes.get_attribute_for_oid(
+        x509.oid.AttributeOID.STATEMENT_OF_POSSESSION
+    )
+    assert attr.value == raw_pop_bytes
+
+
+def test_statement_of_possession_invalid_type_and_repr():
+    # (TypeError guard)
+    with pytest.raises(TypeError, match="value must be bytes"):
+        x509.StatementOfPossession("not_bytes")  # type: ignore[arg-type]
+
+    # (__repr__)
+    raw_val = b"\x30\x00"
+    sop = x509.StatementOfPossession(raw_val)
+    assert repr(sop) == f"<StatementOfPossession(value={raw_val!r})>"
+
+
+def test_csr_builder_public_key_already_set():
+    # (ValueError on duplicate public key assignment)
+    builder = x509.CertificateSigningRequestBuilder()
+    key = ec.generate_private_key(ec.SECP256R1()).public_key()
+
+    builder = builder.public_key(key)
+    with pytest.raises(
+        ValueError, match="The public key has already been set"
+    ):
+        builder.public_key(key)