_synta/

x509_verification.rs

//! Python bindings for X.509 certificate chain verification.
//!
//! This module builds the ``synta.x509`` Python submodule, exposing
//! RFC 5280 / CABF certificate path validation powered by
//! [`synta_x509_verification`] with the default signature backend
//! from [`synta_certificate`].
//!
//! # Quick start
//!
//! ```python,ignore
//! import synta
//! import synta.x509 as x509
//!
//! # Load trust anchors (DER bytes).
//! with open("root.der", "rb") as f:
//!     root_der = f.read()
//! store = x509.TrustStore([root_der])
//!
//! # Verify a TLS server certificate chain.
//! with open("leaf.der", "rb") as f:
//!     leaf_der = f.read()
//! policy = x509.VerificationPolicy(server_names=["example.com"])
//! chain = x509.verify_server_certificate(leaf_der, [], store, policy)
//! # chain is list[bytes], root-first: chain[0] is the trust anchor.
//! for cert_der in chain:
//!     cert = synta.Certificate.from_der(cert_der)
//!     print(cert.subject)
//!
//! # CRL-based revocation checking (optional):
//! crl_ders = synta.pem_to_der(open("issuing-ca.crl", "rb").read())
//! crl_store = x509.CrlStore(crl_ders)
//! chain = x509.verify_server_certificate(leaf_der, [], store, policy, crls=crl_store)
//!
//! # OCSP-based revocation checking (optional):
//! ocsp_resp_der = open("ocsp-response.der", "rb").read()
//! ocsp_store = x509.OcspStore([ocsp_resp_der])
//! chain = x509.verify_server_certificate(leaf_der, [], store, policy, ocsp=ocsp_store)
//!
//! # Both CRL and OCSP revocation at once:
//! chain = x509.verify_server_certificate(
//!     leaf_der, [], store, policy, crls=crl_store, ocsp=ocsp_store
//! )
//! ```

use std::net::IpAddr;
use std::time::{SystemTime, UNIX_EPOCH};

use pyo3::exceptions::PyValueError;
use pyo3::prelude::*;
use pyo3::types::PyBytes;

use synta::{Decoder, Encoding};
use synta_certificate::{default_signature_verifier, Certificate, ErasedSignatureVerifier};
use synta_x509_verification::{
    ocsp::OcspStore,
    ops::VerificationCertificate,
    policy::{NameMatchMode, PolicyDefinition, Subject, ValidationProfile, VerificationPolicy},
    revocation::CrlStore,
    trust_store::OwnedStore,
    types::{DNSName, IPAddress},
    verify, RevocationChecks,
};

use crate::error::SyntaErr;
use crate::install_submodule;

// ── Python exception ─────────────────────────────────────────────────────────

pyo3::create_exception!(
    synta.x509,
    X509VerificationError,
    pyo3::exceptions::PyException
);

// ── Python types ─────────────────────────────────────────────────────────────

/// A set of trusted CA certificates for X.509 chain verification.
///
/// Construct with a list of DER-encoded certificate bytes:
///
/// ```python,ignore
/// store = TrustStore([root_ca_der, cross_root_der])
/// ```
///
/// Pass PEM-encoded certificates through :func:`synta.pem_to_der` first:
///
/// ```python,ignore
/// ders = synta.pem_to_der(open("roots.pem", "rb").read())
/// store = TrustStore(ders)
/// ```
#[pyclass(frozen, name = "TrustStore")]
pub struct PyTrustStore {
    store: OwnedStore,
}

#[pymethods]
impl PyTrustStore {
    /// Create a trust store from a list of DER-encoded CA certificates.
    ///
    /// Each entry in ``certs_der`` must be the DER bytes of a single
    /// certificate.  The certificates are parsed and indexed once at
    /// construction time; subsequent verification calls incur no CA parsing
    /// overhead.
    ///
    /// Raises :exc:`synta.SyntaError` if any entry is not a valid DER
    /// certificate.
    #[new]
    fn new(certs_der: Vec<Vec<u8>>) -> PyResult<Self> {
        let store =
            OwnedStore::try_new(certs_der.iter().map(|v| v.as_slice())).map_err(SyntaErr)?;
        Ok(PyTrustStore { store })
    }

    fn __repr__(&self) -> String {
        format!("TrustStore(<{} certificate(s)>)", self.store.len())
    }

    /// Number of trusted certificates in this store.
    #[getter]
    fn len(&self) -> usize {
        self.store.len()
    }
}

/// A store of Certificate Revocation Lists (CRLs) for revocation checking.
///
/// Construct with a list of DER-encoded CRL byte strings:
///
/// ```python,ignore
/// crl_store = CrlStore([crl_der])
/// ```
///
/// Pass PEM-encoded CRLs through :func:`synta.pem_to_der` first:
///
/// ```python,ignore
/// crl_ders = synta.pem_to_der(open("crl.pem", "rb").read())
/// crl_store = CrlStore(crl_ders)
/// ```
///
/// Pass a populated :class:`CrlStore` to :func:`verify_server_certificate` or
/// :func:`verify_client_certificate` via the ``crls`` keyword argument to enable
/// CRL-based revocation checking.  Certificates whose issuing CA has no matching
/// CRL in the store are treated as not-revoked (soft-fail).
#[pyclass(frozen, name = "CrlStore")]
pub struct PyCrlStore {
    crl_ders: Vec<Vec<u8>>,
}

#[pymethods]
impl PyCrlStore {
    /// Create a CRL store from a list of DER-encoded CRLs.
    ///
    /// Each entry in ``crl_ders`` must be the DER bytes of a single CRL.
    #[new]
    fn new(crl_ders: Vec<Vec<u8>>) -> Self {
        PyCrlStore { crl_ders }
    }

    fn __repr__(&self) -> String {
        format!("CrlStore(<{} CRL(s)>)", self.crl_ders.len())
    }

    /// Number of CRLs in this store.
    #[getter]
    fn len(&self) -> usize {
        self.crl_ders.len()
    }
}

/// A store of pre-fetched OCSP responses for revocation checking.
///
/// Construct with a list of DER-encoded OCSP response byte strings:
///
/// ```python,ignore
/// ocsp_store = OcspStore([ocsp_resp_der])
/// ```
///
/// Pass a populated :class:`OcspStore` to :func:`verify_server_certificate` or
/// :func:`verify_client_certificate` via the ``ocsp`` keyword argument to enable
/// OCSP-based revocation checking.  Certificates for which no valid, matching
/// OCSP response is found are treated as not-revoked (soft-fail).
#[pyclass(frozen, name = "OcspStore")]
pub struct PyOcspStore {
    ocsp_ders: Vec<Vec<u8>>,
}

#[pymethods]
impl PyOcspStore {
    /// Create an OCSP store from a list of DER-encoded OCSP responses.
    ///
    /// Each entry in ``ocsp_ders`` must be the DER bytes of a single OCSP
    /// response (the outer ``OCSPResponse`` SEQUENCE as defined in RFC 6960).
    #[new]
    fn new(ocsp_ders: Vec<Vec<u8>>) -> Self {
        PyOcspStore { ocsp_ders }
    }

    fn __repr__(&self) -> String {
        format!("OcspStore(<{} OCSP response(s)>)", self.ocsp_ders.len())
    }

    /// Number of OCSP responses in this store.
    #[getter]
    fn len(&self) -> usize {
        self.ocsp_ders.len()
    }
}

/// Optional parameters that control certificate chain verification.
///
/// All fields have safe defaults so you only need to set what differs from
/// the standard WebPKI TLS server / client validation.
///
/// ``server_names`` accepts a list of DNS hostnames or IP address literals.
/// When more than one name is given, ``name_match`` controls whether the
/// certificate must cover **any** (default) or **all** of them:
///
/// * ``"any"`` — connection validation: the certificate is accepted if it
///   matches at least one name in the list (e.g. you are connecting to one
///   of several possible endpoints).
/// * ``"all"`` — cert assessment: the certificate must cover every name
///   (e.g. checking a cert covers your entire domain set before deploying).
///
/// ```python,ignore
/// import synta.x509 as x509
///
/// # Single name — classic behaviour:
/// policy = x509.VerificationPolicy(server_names=["example.com"])
///
/// # Any-match: accept the cert if it covers either name (connection validation):
/// policy = x509.VerificationPolicy(
///     server_names=["example.com", "www.example.com"],
///     name_match="any",
/// )
///
/// # All-match: verify the cert covers every name (cert assessment):
/// policy = x509.VerificationPolicy(
///     server_names=["example.com", "api.example.com"],
///     name_match="all",
/// )
///
/// # Strict RFC 5280 profile with a fixed validation time, no SAN check:
/// policy = x509.VerificationPolicy(
///     profile="rfc5280",
///     validation_time=1_700_000_000,
///     max_chain_depth=4,
/// )
/// ```
#[pyclass(name = "VerificationPolicy")]
pub struct PyVerificationPolicy {
    inner: VerificationPolicy,
}

#[pymethods]
impl PyVerificationPolicy {
    #[new]
    #[pyo3(
        signature = (*, server_names=None, name_match=None, validation_time=None, max_chain_depth=8, profile=None)
    )]
    fn new(
        server_names: Option<Vec<String>>,
        name_match: Option<String>,
        validation_time: Option<i64>,
        max_chain_depth: u8,
        profile: Option<String>,
    ) -> PyResult<Self> {
        let vprofile = match profile.as_deref() {
            None | Some("webpki") => ValidationProfile::WebPki,
            Some("rfc5280") => ValidationProfile::Rfc5280,
            Some(other) => {
                return Err(PyValueError::new_err(format!(
                    "unknown validation profile {other:?}; expected \"webpki\" or \"rfc5280\""
                )));
            }
        };
        let vmatch = match name_match.as_deref() {
            None | Some("any") => NameMatchMode::Any,
            Some("all") => NameMatchMode::All,
            Some(other) => {
                return Err(PyValueError::new_err(format!(
                    "unknown name_match {other:?}; expected \"any\" or \"all\""
                )));
            }
        };
        Ok(PyVerificationPolicy {
            inner: VerificationPolicy {
                server_names: server_names.unwrap_or_default(),
                name_match: vmatch,
                validation_time,
                max_chain_depth,
                profile: vprofile,
            },
        })
    }

    fn __repr__(&self) -> String {
        let profile = match self.inner.profile {
            ValidationProfile::WebPki => "webpki",
            ValidationProfile::Rfc5280 => "rfc5280",
        };
        let name_match = match self.inner.name_match {
            NameMatchMode::Any => "any",
            NameMatchMode::All => "all",
        };
        format!(
            "VerificationPolicy(server_names={:?}, name_match={:?}, profile={:?}, \
             max_chain_depth={}, validation_time={:?})",
            self.inner.server_names,
            name_match,
            profile,
            self.inner.max_chain_depth,
            self.inner.validation_time,
        )
    }
}

// ── Helpers ───────────────────────────────────────────────────────────────────

fn now_unix() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0)
}

/// Parse a single DER cert.
fn parse_vcert(der: &[u8]) -> PyResult<Certificate<'_>> {
    Ok(Decoder::new(der, Encoding::Der)
        .decode::<Certificate>()
        .map_err(SyntaErr)?)
}

/// Core verifier: build `VerificationCertificate`s from borrowed DER slices,
/// run `verify()` with a [`RevocationChecks`] struct, and return the chain as
/// owned DER byte vectors.
///
/// `intermediate_ders` holds owned byte vecs (kept alive by the caller) so
/// we can take stable `&[u8]` references into them.
fn run_verify<'a>(
    leaf_der: &'a [u8],
    intermediate_ders: &'a [Vec<u8>],
    trust_store: &'a PyTrustStore,
    policy: PolicyDefinition<'a, Box<dyn ErasedSignatureVerifier>>,
    crls: Option<&'a CrlStore>,
    ocsp: Option<&'a OcspStore>,
) -> PyResult<Vec<Vec<u8>>> {
    let leaf_cert = parse_vcert(leaf_der)?;
    let leaf_vcert = VerificationCertificate::new(leaf_cert, leaf_der);

    let mut intermediate_vcerts = Vec::with_capacity(intermediate_ders.len());
    for der in intermediate_ders {
        let cert = parse_vcert(der.as_slice())?;
        intermediate_vcerts.push(VerificationCertificate::new(cert, der.as_slice()));
    }

    // Trust anchors are pre-parsed in OwnedStore — no per-call CA parsing.
    let store = trust_store.store.as_store();

    verify(
        &leaf_vcert,
        &intermediate_vcerts,
        &policy,
        store,
        RevocationChecks { crls, ocsp },
    )
    .map(|chain| chain.into_iter().map(|vc| vc.der().to_vec()).collect())
    .map_err(|e| X509VerificationError::new_err(e.to_string()))
}

// ── Python functions ──────────────────────────────────────────────────────────

/// Parse a name string (DNS hostname or IP literal) into a [`Subject`].
fn parse_subject(name: &str) -> PyResult<Subject<'_>> {
    if let Ok(ip) = name.parse::<IpAddr>() {
        let addr = match ip {
            IpAddr::V4(a) => IPAddress::from_bytes(&a.octets()),
            IpAddr::V6(a) => IPAddress::from_bytes(&a.octets()),
        };
        Ok(Subject::Ip(addr.ok_or_else(|| {
            PyValueError::new_err(format!("invalid IP address for server name: {name}"))
        })?))
    } else {
        Ok(Subject::Dns(DNSName::new(name).ok_or_else(|| {
            PyValueError::new_err(format!("invalid DNS name for server name: {name:?}"))
        })?))
    }
}

/// Verify a TLS server certificate chain using the WebPKI / CABF profile.
///
/// ``leaf_der`` is the DER-encoded end-entity certificate.
/// ``intermediates_der`` is a list of DER-encoded intermediate CA certificates
/// (order does not matter; the validator discovers the chain automatically).
/// ``trust_store`` is a :class:`TrustStore` populated with trusted root CAs.
/// ``policy`` is an optional :class:`VerificationPolicy` specifying the server
/// names, name match mode, validation time, chain depth, and compliance profile.
///
/// Returns the validated chain as ``list[bytes]``, ordered root-first
/// (``chain[0]`` is the trust anchor, ``chain[-1]`` is the leaf).
///
/// Raises :exc:`X509VerificationError` if verification fails.
///
/// ```python,ignore
/// import synta
/// import synta.x509 as x509
///
/// store = x509.TrustStore([root_der])
///
/// # Single name:
/// policy = x509.VerificationPolicy(server_names=["example.com"])
/// chain = x509.verify_server_certificate(leaf_der, [intermediate_der], store, policy)
/// for i, cert_der in enumerate(chain):
///     cert = synta.Certificate.from_der(cert_der)
///     print(f"chain[{i}]: {cert.subject}")
///
/// # Multi-name, all-match (cert covers every name):
/// policy = x509.VerificationPolicy(
///     server_names=["example.com", "api.example.com"],
///     name_match="all",
/// )
/// chain = x509.verify_server_certificate(leaf_der, [], store, policy)
/// ```
#[pyfunction]
#[pyo3(signature = (leaf_der, intermediates_der, trust_store, policy=None, crls=None, ocsp=None))]
fn verify_server_certificate<'py>(
    py: Python<'py>,
    leaf_der: Vec<u8>,
    intermediates_der: Vec<Vec<u8>>,
    trust_store: &PyTrustStore,
    policy: Option<&PyVerificationPolicy>,
    crls: Option<&PyCrlStore>,
    ocsp: Option<&PyOcspStore>,
) -> PyResult<Vec<Bound<'py, PyBytes>>> {
    let default_policy;
    let vp: &VerificationPolicy = match policy {
        Some(p) => &p.inner,
        None => {
            default_policy = VerificationPolicy::new_client();
            &default_policy
        }
    };

    let now = vp.validation_time.unwrap_or_else(now_unix);

    let subjects = vp
        .server_names
        .iter()
        .map(|name| parse_subject(name))
        .collect::<PyResult<Vec<_>>>()?;

    let mut pd = PolicyDefinition::new_server(default_signature_verifier(), subjects, now);
    pd.profile = vp.profile;
    pd.max_chain_depth = vp.max_chain_depth;
    pd.name_match = vp.name_match;

    // Build a CrlStore from the Python CRL store if provided.
    let mut crl_store = CrlStore::new();
    if let Some(py_crls) = crls {
        for der in &py_crls.crl_ders {
            crl_store.add_der(der.clone());
        }
    }
    let crl_opt = crls.map(|_| &crl_store);

    // Build an OcspStore from the Python OCSP store if provided.
    let mut ocsp_store = OcspStore::new();
    if let Some(py_ocsp) = ocsp {
        for der in &py_ocsp.ocsp_ders {
            ocsp_store.add_der(der.clone());
        }
    }
    let ocsp_opt = ocsp.map(|_| &ocsp_store);

    run_verify(
        leaf_der.as_slice(),
        &intermediates_der,
        trust_store,
        pd,
        crl_opt,
        ocsp_opt,
    )?
    .into_iter()
    .map(|der| Ok(PyBytes::new(py, &der)))
    .collect()
}

/// Verify a TLS client certificate chain.
///
/// Like :func:`verify_server_certificate` but uses the ``clientAuth`` EKU and
/// skips SAN / server-name matching.
///
/// ``policy`` may set ``validation_time``, ``max_chain_depth``, and
/// ``profile``; its ``server_names`` field is ignored.
///
/// Returns the validated chain as ``list[bytes]``, root-first.
///
/// Raises :exc:`X509VerificationError` if verification fails.
///
/// ```python,ignore
/// import synta.x509 as x509
///
/// store = x509.TrustStore([root_der])
/// chain = x509.verify_client_certificate(leaf_der, [], store)
/// ```
#[pyfunction]
#[pyo3(signature = (leaf_der, intermediates_der, trust_store, policy=None, crls=None, ocsp=None))]
fn verify_client_certificate<'py>(
    py: Python<'py>,
    leaf_der: Vec<u8>,
    intermediates_der: Vec<Vec<u8>>,
    trust_store: &PyTrustStore,
    policy: Option<&PyVerificationPolicy>,
    crls: Option<&PyCrlStore>,
    ocsp: Option<&PyOcspStore>,
) -> PyResult<Vec<Bound<'py, PyBytes>>> {
    let default_policy;
    let vp: &VerificationPolicy = match policy {
        Some(p) => &p.inner,
        None => {
            default_policy = VerificationPolicy::new_client();
            &default_policy
        }
    };

    let now = vp.validation_time.unwrap_or_else(now_unix);

    let mut pd = PolicyDefinition::new_client(default_signature_verifier(), now);
    pd.profile = vp.profile;
    pd.max_chain_depth = vp.max_chain_depth;

    // Build a CrlStore from the Python CRL store if provided.
    let mut crl_store = CrlStore::new();
    if let Some(py_crls) = crls {
        for der in &py_crls.crl_ders {
            crl_store.add_der(der.clone());
        }
    }
    let crl_opt = crls.map(|_| &crl_store);

    // Build an OcspStore from the Python OCSP store if provided.
    let mut ocsp_store = OcspStore::new();
    if let Some(py_ocsp) = ocsp {
        for der in &py_ocsp.ocsp_ders {
            ocsp_store.add_der(der.clone());
        }
    }
    let ocsp_opt = ocsp.map(|_| &ocsp_store);

    run_verify(
        leaf_der.as_slice(),
        &intermediates_der,
        trust_store,
        pd,
        crl_opt,
        ocsp_opt,
    )?
    .into_iter()
    .map(|der| Ok(PyBytes::new(py, &der)))
    .collect()
}

// ── Module registration ───────────────────────────────────────────────────────

/// Register the ``synta.x509`` submodule onto ``parent``.
pub fn register_x509_module(parent: &Bound<'_, PyModule>) -> PyResult<()> {
    let py = parent.py();
    let m = PyModule::new(py, "x509")?;

    m.add(
        "X509VerificationError",
        py.get_type::<X509VerificationError>(),
    )?;
    m.add_class::<PyTrustStore>()?;
    m.add_class::<PyCrlStore>()?;
    m.add_class::<PyOcspStore>()?;
    m.add_class::<PyVerificationPolicy>()?;
    m.add_function(wrap_pyfunction!(verify_server_certificate, &m)?)?;
    m.add_function(wrap_pyfunction!(verify_client_certificate, &m)?)?;

    install_submodule(
        parent,
        &m,
        "synta.x509",
        Some("RFC 5280 X.509 certificate chain verification."),
    )?;
    Ok(())
}