Path validation: builder/verifier API skeletons

docs/x509/verification.rst

@@ -19,7 +19,7 @@ chain building, etc.
     OS's root of trust, from a well-known source such as a browser CA bundle,
     or from a small set of manually pre-trusted entities.
 
-    :param certs: A list of one or more :class:`~cryptography.x509.Certificate`
+    :param certs: A list of one or more :class:`cryptography.x509.Certificate`
         instances.
 
 .. class:: Subject
@@ -30,6 +30,31 @@ chain building, etc.
     :class:`cryptography.x509.general_name.DNSName`,
     :class:`cryptography.x509.general_name.IPAddress`.
 
+.. class:: ServerVerifier
+
+    .. versionadded:: 42.0.0
+
+    A ServerVerifier verifies server certificates.
+
+    It contains and describes various pieces of configurable path
+    validation logic, such as which subject to expect, how deep prospective
+    validation chains may go, which signature algorithms are allowed, and
+    so forth.
+
+    ServerVerifier instances cannot be constructed directly;
+    :class:`PolicyBuilder` must be used.
+
+    .. attribute:: subject
+
+        :type: :class:`Subject`
+
+        The verifier's subject.
+
+    .. attribute:: validation_time
+
+        :type: :class:`datetime.datetime`
+
+        The verifier's validation time.
 
 .. class:: PolicyBuilder
 
@@ -40,16 +65,19 @@ chain building, etc.
 
     .. method:: time(new_time)
 
-        Sets the policy's verification time.
+        Sets the verifier's verification time.
+
+        If not called explicitly, this is set to :meth:`datetime.datetime.now`
+        when :meth:`build_server_verifier` is called.
 
-        :param new_time: The :class:`datetime.datetime` to use in the policy
+        :param new_time: The :class:`datetime.datetime` to use in the verifier
 
         :returns: A new instance of :class:`PolicyBuilder`
 
     .. method:: build_server_verifier(subject)
 
         Builds a verifier for verifying server certificates.
 
-        :param subject: A :class:`Subject` to use in the policy
+        :param subject: A :class:`Subject` to use in the verifier
 
-        :raises NotImplementedError: This API is not implemented yet.
+        :returns: An instance of :class:`ServerVerifier`

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

@@ -2,6 +2,8 @@
 # 2.0, and the BSD License. See the LICENSE file in the root of this repository
 # for complete details.
 
+import datetime
+
 from cryptography import x509
 from cryptography.hazmat.primitives import hashes
 from cryptography.hazmat.primitives.asymmetric.padding import PSS, PKCS1v15
@@ -35,12 +37,22 @@ def create_x509_crl(
     private_key: PrivateKeyTypes,
     hash_algorithm: hashes.HashAlgorithm | None,
 ) -> x509.CertificateRevocationList: ...
+def create_server_verifier(
+    name: x509.verification.Subject,
+    time: datetime.datetime | None,
+) -> x509.verification.ServerVerifier: ...
 
 class Sct: ...
 class Certificate: ...
 class RevokedCertificate: ...
 class CertificateRevocationList: ...
 class CertificateSigningRequest: ...
 
+class ServerVerifier:
+    @property
+    def subject(self) -> x509.verification.Subject: ...
+    @property
+    def validation_time(self) -> datetime.datetime: ...
+
 class Store:
     def __init__(self, certs: list[x509.Certificate]) -> None: ...

src/cryptography/x509/__init__.py

@@ -171,6 +171,7 @@
 
 __all__ = [
     "certificate_transparency",
+    "verification",
     "load_pem_x509_certificate",
     "load_pem_x509_certificates",
     "load_der_x509_certificate",

src/cryptography/x509/verification.py

@@ -10,12 +10,14 @@
 from cryptography.hazmat.bindings._rust import x509 as rust_x509
 from cryptography.x509.general_name import DNSName, IPAddress
 
-__all__ = ["Store", "Subject", "PolicyBuilder"]
+__all__ = ["Store", "Subject", "ServerVerifier", "PolicyBuilder"]
 
 Store = rust_x509.Store
 
 Subject = typing.Union[DNSName, IPAddress]
 
+ServerVerifier = rust_x509.ServerVerifier
+
 
 class PolicyBuilder:
     def __init__(
@@ -36,9 +38,9 @@ def time(self, new_time: datetime.datetime) -> PolicyBuilder:
             time=new_time,
         )
 
-    def build_server_verifier(self, subject: Subject) -> typing.NoReturn:
+    def build_server_verifier(self, subject: Subject) -> ServerVerifier:
         """
         Builds a verifier for verifying server certificates.
         """
 
-        raise NotImplementedError
+        return rust_x509.create_server_verifier(subject, self._time)

src/rust/cryptography-x509-validation/src/policy/mod.rs

@@ -12,6 +12,9 @@ use cryptography_x509::common::{
     PSS_SHA512_MASK_GEN_ALG,
 };
 
+use crate::ops::CryptoOps;
+use crate::types::{DNSName, IPAddress};
+
 // RSASSA‐PKCS1‐v1_5 with SHA‐256
 static RSASSA_PKCS1V15_SHA256: AlgorithmIdentifier<'_> = AlgorithmIdentifier {
     oid: asn1::DefinedByMarker::marker(),
@@ -97,6 +100,39 @@ pub static WEBPKI_PERMITTED_ALGORITHMS: Lazy<HashSet<&AlgorithmIdentifier<'_>>>
     ])
 });
 
+/// Represents a logical certificate "subject," i.e. a principal matching
+/// one of the names listed in a certificate's `subjectAltNames` extension.
+pub enum Subject<'a> {
+    DNS(DNSName<'a>),
+    IP(IPAddress),
+}
+
+/// A `Policy` describes user-configurable aspects of X.509 path validation.
+pub struct Policy<'a, B: CryptoOps> {
+    _ops: B,
+
+    /// A subject (i.e. DNS name or other name format) that any EE certificates
+    /// validated by this policy must match.
+    /// If `None`, the EE certificate must not contain a SAN.
+    pub subject: Option<Subject<'a>>,
+
+    /// The validation time. All certificates validated by this policy must
+    /// be valid at this time.
+    pub validation_time: asn1::DateTime,
+}
+
+impl<'a, B: CryptoOps> Policy<'a, B> {
+    /// Creates a new policy with the given `CryptoOps`, an optional subject,
+    /// and a validation time.
+    pub fn new(ops: B, subject: Option<Subject<'a>>, time: asn1::DateTime) -> Self {
+        Self {
+            _ops: ops,
+            subject,
+            validation_time: time,
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::ops::Deref;

src/rust/cryptography-x509-validation/src/types.rs

@@ -20,7 +20,7 @@ use std::str::FromStr;
 /// # use cryptography_x509_validation::types::DNSName;
 /// assert_eq!(DNSName::new("foo.com").unwrap(), DNSName::new("FOO.com").unwrap());
 /// ```
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 pub struct DNSName<'a>(asn1::IA5String<'a>);
 
 impl<'a> DNSName<'a> {

src/rust/src/types.rs

@@ -201,7 +201,6 @@ pub static UNRECOGNIZED_EXTENSION: LazyPyImport =
     LazyPyImport::new("cryptography.x509", &["UnrecognizedExtension"]);
 pub static EXTENSION: LazyPyImport = LazyPyImport::new("cryptography.x509", &["Extension"]);
 pub static EXTENSIONS: LazyPyImport = LazyPyImport::new("cryptography.x509", &["Extensions"]);
-pub static IPADDRESS: LazyPyImport = LazyPyImport::new("cryptography.x509", &["IPAddress"]);
 pub static NAME: LazyPyImport = LazyPyImport::new("cryptography.x509", &["Name"]);
 pub static RELATIVE_DISTINGUISHED_NAME: LazyPyImport =
     LazyPyImport::new("cryptography.x509", &["RelativeDistinguishedName"]);
@@ -243,6 +242,7 @@ pub static DIRECTORY_NAME: LazyPyImport =
 pub static UNIFORM_RESOURCE_IDENTIFIER: LazyPyImport =
     LazyPyImport::new("cryptography.x509", &["UniformResourceIdentifier"]);
 pub static DNS_NAME: LazyPyImport = LazyPyImport::new("cryptography.x509", &["DNSName"]);
+pub static IP_ADDRESS: LazyPyImport = LazyPyImport::new("cryptography.x509", &["IPAddress"]);
 pub static RFC822_NAME: LazyPyImport = LazyPyImport::new("cryptography.x509", &["RFC822Name"]);
 pub static OTHER_NAME: LazyPyImport = LazyPyImport::new("cryptography.x509", &["OtherName"]);
 pub static CERTIFICATE_VERSION_V1: LazyPyImport =

src/rust/src/x509/common.rs

@@ -136,7 +136,7 @@ pub(crate) fn encode_general_name<'a>(
         Ok(GeneralName::UniformResourceIdentifier(
             UnvalidatedIA5String(gn_value.extract::<&str>()?),
         ))
-    } else if gn_type.is(types::IPADDRESS.get(py)?) {
+    } else if gn_type.is(types::IP_ADDRESS.get(py)?) {
         Ok(GeneralName::IPAddress(
             gn.call_method0(pyo3::intern!(py, "_packed"))?
                 .extract::<&[u8]>()?,
@@ -272,7 +272,7 @@ pub(crate) fn parse_general_name(
         GeneralName::IPAddress(data) => {
             if data.len() == 4 || data.len() == 16 {
                 let addr = types::IPADDRESS_IPADDRESS.get(py)?.call1((data,))?;
-                types::IPADDRESS.get(py)?.call1((addr,))?.to_object(py)
+                types::IP_ADDRESS.get(py)?.call1((addr,))?.to_object(py)
             } else {
                 // if it's not an IPv4 or IPv6 we assume it's an IPNetwork and
                 // verify length in this function.
@@ -333,7 +333,7 @@ fn create_ip_network(
         prefix?
     );
     let addr = types::IPADDRESS_IPNETWORK.get(py)?.call1((net,))?;
-    Ok(types::IPADDRESS.get(py)?.call1((addr,))?.to_object(py))
+    Ok(types::IP_ADDRESS.get(py)?.call1((addr,))?.to_object(py))
 }
 
 fn ipv4_netmask(num: u32) -> Result<u32, CryptographyError> {

src/rust/src/x509/verify.rs

@@ -3,14 +3,17 @@
 // for complete details.
 
 use cryptography_x509::certificate::Certificate;
-use cryptography_x509_validation::ops::CryptoOps;
+use cryptography_x509_validation::{
+    ops::CryptoOps,
+    policy::{Policy, Subject},
+    types::{DNSName, IPAddress},
+};
 
+use crate::error::{CryptographyError, CryptographyResult};
+use crate::types;
+use crate::x509::certificate::Certificate as PyCertificate;
+use crate::x509::common::{datetime_now, datetime_to_py, py_to_datetime};
 use crate::x509::sign;
-use crate::{
-    error::{CryptographyError, CryptographyResult},
-    types,
-    x509::certificate::Certificate as PyCertificate,
-};
 
 pub(crate) struct PyCryptoOps {}
 
@@ -43,6 +46,125 @@ impl CryptoOps for PyCryptoOps {
     }
 }
 
+struct PyCryptoPolicy<'a>(Policy<'a, PyCryptoOps>);
+
+/// This enum exists solely to provide heterogeneously typed ownership for `OwnedPolicy`.
+enum SubjectOwner {
+    // TODO: Switch this to `Py<PyString>` once Pyo3's `to_str()` preserves a
+    // lifetime relationship between an a `PyString` and its borrowed `&str`
+    // reference in all limited API builds. PyO3 can't currently do that in
+    // older limited API builds because it needs `PyUnicode_AsUTF8AndSize` to do
+    // so, which was only stabilized with 3.10.
+    DNSName(String),
+    IPAddress(pyo3::Py<pyo3::types::PyBytes>),
+}
+
+self_cell::self_cell!(
+    struct OwnedPolicy {
+        owner: SubjectOwner,
+
+        #[covariant]
+        dependent: PyCryptoPolicy,
+    }
+);
+
+#[pyo3::pyclass(
+    name = "ServerVerifier",
+    module = "cryptography.hazmat.bindings._rust.x509"
+)]
+struct PyServerVerifier {
+    #[pyo3(get, name = "subject")]
+    py_subject: pyo3::Py<pyo3::PyAny>,
+    policy: OwnedPolicy,
+}
+
+impl PyServerVerifier {
+    fn as_policy(&self) -> &Policy<'_, PyCryptoOps> {
+        &self.policy.borrow_dependent().0
+    }
+}
+
+#[pyo3::pymethods]
+impl PyServerVerifier {
+    #[getter]
+    fn validation_time<'p>(&self, py: pyo3::Python<'p>) -> pyo3::PyResult<&'p pyo3::PyAny> {
+        datetime_to_py(py, &self.as_policy().validation_time)
+    }
+}
+
+fn build_subject_owner(
+    py: pyo3::Python<'_>,
+    subject: &pyo3::Py<pyo3::PyAny>,
+) -> pyo3::PyResult<SubjectOwner> {
+    let subject = subject.as_ref(py);
+
+    if subject.is_instance(types::DNS_NAME.get(py)?)? {
+        let value = subject
+            .getattr(pyo3::intern!(py, "value"))?
+            .downcast::<pyo3::types::PyString>()?;
+
+        Ok(SubjectOwner::DNSName(value.to_str()?.to_owned()))
+    } else if subject.is_instance(types::IP_ADDRESS.get(py)?)? {
+        let value = subject
+            .getattr(pyo3::intern!(py, "_packed"))?
+            .call0()?
+            .downcast::<pyo3::types::PyBytes>()?;
+
+        Ok(SubjectOwner::IPAddress(value.into()))
+    } else {
+        Err(pyo3::exceptions::PyTypeError::new_err(
+            "unsupported subject type",
+        ))
+    }
+}
+
+fn build_subject<'a>(
+    py: pyo3::Python<'_>,
+    subject: &'a SubjectOwner,
+) -> pyo3::PyResult<Option<Subject<'a>>> {
+    match subject {
+        SubjectOwner::DNSName(dns_name) => {
+            let dns_name = DNSName::new(dns_name)
+                .ok_or_else(|| pyo3::exceptions::PyValueError::new_err("invalid domain name"))?;
+
+            Ok(Some(Subject::DNS(dns_name)))
+        }
+        SubjectOwner::IPAddress(ip_addr) => {
+            let ip_addr = IPAddress::from_bytes(ip_addr.as_bytes(py))
+                .ok_or_else(|| pyo3::exceptions::PyValueError::new_err("invalid IP address"))?;
+
+            Ok(Some(Subject::IP(ip_addr)))
+        }
+    }
+}
+
+#[pyo3::prelude::pyfunction]
+fn create_server_verifier(
+    py: pyo3::Python<'_>,
+    subject: pyo3::Py<pyo3::PyAny>,
+    time: Option<&pyo3::PyAny>,
+) -> pyo3::PyResult<PyServerVerifier> {
+    let time = match time {
+        Some(time) => py_to_datetime(py, time)?,
+        None => datetime_now(py)?,
+    };
+
+    let subject_owner = build_subject_owner(py, &subject)?;
+    let policy = OwnedPolicy::try_new(subject_owner, |subject_owner| {
+        let subject = build_subject(py, subject_owner)?;
+        Ok::<PyCryptoPolicy<'_>, pyo3::PyErr>(PyCryptoPolicy(Policy::new(
+            PyCryptoOps {},
+            subject,
+            time,
+        )))
+    })?;
+
+    Ok(PyServerVerifier {
+        py_subject: subject,
+        policy,
+    })
+}
+
 #[pyo3::pyclass(
     frozen,
     name = "Store",
@@ -64,7 +186,9 @@ impl PyStore {
 }
 
 pub(crate) fn add_to_module(module: &pyo3::prelude::PyModule) -> pyo3::PyResult<()> {
+    module.add_class::<PyServerVerifier>()?;
     module.add_class::<PyStore>()?;
+    module.add_function(pyo3::wrap_pyfunction!(create_server_verifier, module)?)?;
 
     Ok(())
 }