Support x509 Certificate, CSR, and CRL types in declarative ASN.1 API

CHANGELOG.rst

@@ -26,6 +26,10 @@ Changelog
   :class:`~cryptography.hazmat.primitives.hpke.KEM` so callers can split the
   encapsulated key from the ciphertext returned by
   :meth:`~cryptography.hazmat.primitives.hpke.Suite.encrypt`.
+* Added support for using :class:`~cryptography.x509.Certificate`,
+  :class:`~cryptography.x509.CertificateSigningRequest`, and
+  :class:`~cryptography.x509.CertificateRevocationList` as field types in
+  :doc:`/hazmat/asn1/index` structures.
 
 .. _v48-0-0:
 

docs/hazmat/asn1/reference.rst

@@ -77,6 +77,24 @@ The following built-in Python types are supported as ASN.1 field types:
 Additionally, :class:`~cryptography.x509.ObjectIdentifier` maps to
 ``OBJECT IDENTIFIER``.
 
+.. versionadded:: 49.0.0
+
+:class:`~cryptography.x509.Certificate`,
+:class:`~cryptography.x509.CertificateSigningRequest`, and
+:class:`~cryptography.x509.CertificateRevocationList` can also be used as
+field types. They are encoded by embedding their DER serialization, and
+decoded by parsing the field as the corresponding X.509 object. These
+fields cannot have :class:`Implicit` annotations.
+
+.. code-block:: python
+
+    from cryptography import x509
+    from cryptography.hazmat import asn1
+
+    @asn1.sequence
+    class Example:
+        cert: x509.Certificate
+
 The following decorators and types are provided for the rest of the ASN.1 types
 that have no direct Python equivalent:
 

src/cryptography/hazmat/asn1/asn1.py

@@ -18,6 +18,7 @@
     LiteralString = typing.LiteralString
 
 from cryptography.hazmat.bindings._rust import declarative_asn1
+from cryptography.hazmat.bindings._rust import x509 as rust_x509
 
 if sys.version_info < (3, 10):
     NoneType = type(None)
@@ -56,6 +57,25 @@ class Variant(typing.Generic[U, Tag]):
 encode_der = declarative_asn1.encode_der
 
 
+_X509_TYPES = (
+    rust_x509.Certificate,
+    rust_x509.CertificateSigningRequest,
+    rust_x509.CertificateRevocationList,
+)
+
+
+def _check_x509_field_annotations(
+    field_type: typing.Any,
+    annotation: declarative_asn1.Annotation,
+    field_name: str,
+) -> None:
+    if field_type in _X509_TYPES and isinstance(annotation.encoding, Implicit):
+        raise TypeError(
+            f"field '{field_name}' has an IMPLICIT annotation, but "
+            "IMPLICIT annotations are not supported for X.509 types."
+        )
+
+
 def _is_union(field_type: type) -> bool:
     # NOTE: types.UnionType for `T | U`, typing.Union for `Union[T, U]`.
     # TODO: Drop the `hasattr()` once the minimum supported Python version
@@ -144,6 +164,8 @@ def _normalize_field_type(
                 "DEFAULT annotations are not supported for TLV types."
             )
 
+    _check_x509_field_annotations(field_type, annotation, field_name)
+
     if hasattr(field_type, "__asn1_root__"):
         root_type = field_type.__asn1_root__
         if not isinstance(
@@ -166,6 +188,11 @@ def _normalize_field_type(
                     "optional TLV types (`TLV | None`) are not "
                     "currently supported"
                 )
+            # For optional types, the annotation is associated with the
+            # union, so we check it against the inner type here.
+            _check_x509_field_annotations(
+                optional_type, annotation, field_name
+            )
             annotated_type = _normalize_field_type(optional_type, field_name)
 
             if not annotated_type.annotation.is_empty():

src/rust/src/declarative_asn1/decode.rs

@@ -192,6 +192,59 @@ fn decode_tlv<'a>(
     )?)
 }
 
+// Reads a TLV from `parser` and returns its full data as `PyBytes`,
+// suitable for passing to the X.509 loaders.
+fn decode_x509_der_bytes<'a>(
+    py: pyo3::Python<'a>,
+    parser: &mut Parser<'a>,
+    encoding: &Option<pyo3::Py<Encoding>>,
+) -> ParseResult<pyo3::Py<pyo3::types::PyBytes>> {
+    let tlv = match encoding {
+        Some(e) => match e.get() {
+            Encoding::Implicit(_) => Err(CryptographyError::Py(
+                // We don't support IMPLICIT X.509 fields
+                // (they are caught first at the Python level)
+                pyo3::exceptions::PyValueError::new_err(
+                    "invalid type definition: X.509 fields cannot be implicitly encoded",
+                ),
+            ))?,
+            Encoding::Explicit(n) => parser.read_explicit_element::<asn1::Tlv<'_>>(*n),
+        },
+        None => parser.read_element::<asn1::Tlv<'_>>(),
+    }?;
+    Ok(pyo3::types::PyBytes::new(py, tlv.full_data()).unbind())
+}
+
+fn decode_certificate<'a>(
+    py: pyo3::Python<'a>,
+    parser: &mut Parser<'a>,
+    encoding: &Option<pyo3::Py<Encoding>>,
+) -> ParseResult<pyo3::Bound<'a, crate::x509::certificate::Certificate>> {
+    let data = decode_x509_der_bytes(py, parser, encoding)?;
+    let cert = crate::x509::certificate::load_der_x509_certificate(py, data, None)?;
+    Ok(pyo3::Bound::new(py, cert)?)
+}
+
+fn decode_csr<'a>(
+    py: pyo3::Python<'a>,
+    parser: &mut Parser<'a>,
+    encoding: &Option<pyo3::Py<Encoding>>,
+) -> ParseResult<pyo3::Bound<'a, crate::x509::csr::CertificateSigningRequest>> {
+    let data = decode_x509_der_bytes(py, parser, encoding)?;
+    let csr = crate::x509::csr::load_der_x509_csr(py, data, None)?;
+    Ok(pyo3::Bound::new(py, csr)?)
+}
+
+fn decode_crl<'a>(
+    py: pyo3::Python<'a>,
+    parser: &mut Parser<'a>,
+    encoding: &Option<pyo3::Py<Encoding>>,
+) -> ParseResult<pyo3::Bound<'a, crate::x509::crl::CertificateRevocationList>> {
+    let data = decode_x509_der_bytes(py, parser, encoding)?;
+    let crl = crate::x509::crl::load_der_x509_crl(py, data, None)?;
+    Ok(pyo3::Bound::new(py, crl)?)
+}
+
 fn decode_null<'a>(
     py: pyo3::Python<'a>,
     parser: &mut Parser<'a>,
@@ -379,6 +432,9 @@ pub(crate) fn decode_annotated_type<'a>(
         Type::BitString() => decode_bitstring(py, parser, annotation)?.into_any(),
         Type::Tlv() => decode_tlv(py, parser, encoding)?.into_any(),
         Type::Null() => decode_null(py, parser, encoding)?.into_any(),
+        Type::Certificate() => decode_certificate(py, parser, encoding)?.into_any(),
+        Type::CertificateSigningRequest() => decode_csr(py, parser, encoding)?.into_any(),
+        Type::CertificateRevocationList() => decode_crl(py, parser, encoding)?.into_any(),
     };
 
     match &ann_type.annotation.get().default {
@@ -434,4 +490,19 @@ mod tests {
                 .contains("invalid type definition: TLV/ANY fields cannot be implicitly encoded"));
         });
     }
+
+    #[test]
+    fn test_decode_implicit_x509() {
+        pyo3::Python::initialize();
+        pyo3::Python::attach(|py| {
+            let result = asn1::parse(&[], |parser| {
+                let encoding = pyo3::Py::new(py, Encoding::Implicit(0)).ok();
+                super::decode_x509_der_bytes(py, parser, &encoding)
+            });
+            assert!(result.is_err());
+            let error = result.unwrap_err();
+            assert!(format!("{error}")
+                .contains("invalid type definition: X.509 fields cannot be implicitly encoded"));
+        });
+    }
 }

src/rust/src/declarative_asn1/encode.rs

@@ -265,6 +265,30 @@ impl asn1::Asn1Writable for AnnotatedTypeObject<'_> {
                 ),
             )),
             Type::Null() => Ok(write_value(writer, &(), encoding)?),
+            Type::Certificate() => {
+                let val = value.cast::<crate::x509::certificate::Certificate>()?;
+                Ok(write_value(
+                    writer,
+                    val.get().raw.borrow_dependent(),
+                    encoding,
+                )?)
+            }
+            Type::CertificateSigningRequest() => {
+                let val = value.cast::<crate::x509::csr::CertificateSigningRequest>()?;
+                Ok(write_value(
+                    writer,
+                    val.get().raw.borrow_dependent(),
+                    encoding,
+                )?)
+            }
+            Type::CertificateRevocationList() => {
+                let val = value.cast::<crate::x509::crl::CertificateRevocationList>()?;
+                Ok(write_value(
+                    writer,
+                    val.get().owned.borrow_dependent(),
+                    encoding,
+                )?)
+            }
         }
     }
 }

src/rust/src/declarative_asn1/types.rs

@@ -61,6 +61,16 @@ pub enum Type {
     Tlv(),
     /// NULL
     Null(),
+
+    // X.509 types that we special-case to allow embedding them
+    // in ASN.1 structures.
+    //
+    /// `x509.Certificate`
+    Certificate(),
+    /// `x509.CertificateSigningRequest`
+    CertificateSigningRequest(),
+    /// `x509.CertificateRevocationList`
+    CertificateRevocationList(),
 }
 
 /// A type that we know how to encode/decode, along with any
@@ -530,6 +540,12 @@ pub fn non_root_python_to_rust<'p>(
         Type::Tlv().into_pyobject(py)
     } else if class.is(Null::type_object(py)) {
         Type::Null().into_pyobject(py)
+    } else if class.is(crate::x509::certificate::Certificate::type_object(py)) {
+        Type::Certificate().into_pyobject(py)
+    } else if class.is(crate::x509::csr::CertificateSigningRequest::type_object(py)) {
+        Type::CertificateSigningRequest().into_pyobject(py)
+    } else if class.is(crate::x509::crl::CertificateRevocationList::type_object(py)) {
+        Type::CertificateRevocationList().into_pyobject(py)
     } else {
         Err(pyo3::exceptions::PyTypeError::new_err(format!(
             "cannot handle type: {class:?}"
@@ -673,6 +689,12 @@ pub(crate) fn is_tag_valid_for_type(
             }
         }
         Type::Null() => check_tag_with_encoding(asn1::Null::TAG, encoding, tag),
+        // Certificates, CSRs, and CRLs are all SEQUENCEs
+        Type::Certificate()
+        | Type::CertificateSigningRequest()
+        | Type::CertificateRevocationList() => {
+            check_tag_with_encoding(asn1::Sequence::TAG, encoding, tag)
+        }
     }
 }
 

src/rust/src/x509/crl.rs

@@ -82,7 +82,7 @@ pub(crate) fn load_pem_x509_crl(
 }
 
 self_cell::self_cell!(
-    struct OwnedCertificateRevocationList {
+    pub(crate) struct OwnedCertificateRevocationList {
         owner: pyo3::Py<pyo3::types::PyBytes>,
         #[covariant]
         dependent: RawCertificateRevocationList,
@@ -91,7 +91,7 @@ self_cell::self_cell!(
 
 #[pyo3::pyclass(frozen, module = "cryptography.hazmat.bindings._rust.x509")]
 pub(crate) struct CertificateRevocationList {
-    owned: OwnedCertificateRevocationList,
+    pub(crate) owned: OwnedCertificateRevocationList,
 
     revoked_certs: pyo3::sync::PyOnceLock<Vec<OwnedRevokedCertificate>>,
     cached_extensions: pyo3::sync::PyOnceLock<pyo3::Py<pyo3::PyAny>>,

src/rust/src/x509/csr.rs

@@ -17,7 +17,7 @@ use crate::x509::{certificate, sign};
 use crate::{exceptions, types, x509};
 
 self_cell::self_cell!(
-    struct OwnedCsr {
+    pub(crate) struct OwnedCsr {
         owner: pyo3::Py<pyo3::types::PyBytes>,
 
         #[covariant]
@@ -27,7 +27,7 @@ self_cell::self_cell!(
 
 #[pyo3::pyclass(frozen, module = "cryptography.hazmat.bindings._rust.x509")]
 pub(crate) struct CertificateSigningRequest {
-    raw: OwnedCsr,
+    pub(crate) raw: OwnedCsr,
     cached_extensions: pyo3::sync::PyOnceLock<pyo3::Py<pyo3::PyAny>>,
     cached_attributes: pyo3::sync::PyOnceLock<pyo3::Py<pyo3::PyAny>>,
 }

tests/hazmat/asn1/__init__.py

@@ -0,0 +1,3 @@
+# This file is dual licensed under the terms of the Apache License, Version
+# 2.0, and the BSD License. See the LICENSE file in the root of this repository
+# for complete details.