_synta/
x509_verification.rs

1//! Python bindings for X.509 certificate chain verification.
2//!
3//! This module builds the ``synta.x509`` Python submodule, exposing
4//! RFC 5280 / CABF certificate path validation powered by
5//! [`synta_x509_verification`] with the OpenSSL signature backend
6//! from [`synta_certificate::OpensslSignatureVerifier`].
7//!
8//! # Quick start
9//!
10//! ```python,ignore
11//! import synta
12//! import synta.x509 as x509
13//!
14//! # Load trust anchors (DER bytes).
15//! with open("root.der", "rb") as f:
16//!     root_der = f.read()
17//! store = x509.TrustStore([root_der])
18//!
19//! # Verify a TLS server certificate chain.
20//! with open("leaf.der", "rb") as f:
21//!     leaf_der = f.read()
22//! policy = x509.VerificationPolicy(server_names=["example.com"])
23//! chain = x509.verify_server_certificate(leaf_der, [], store, policy)
24//! # chain is list[bytes], root-first: chain[0] is the trust anchor.
25//! for cert_der in chain:
26//!     cert = synta.Certificate.from_der(cert_der)
27//!     print(cert.subject)
28//!
29//! # CRL-based revocation checking (optional):
30//! crl_ders = synta.pem_to_der(open("issuing-ca.crl", "rb").read())
31//! crl_store = x509.CrlStore(crl_ders)
32//! chain = x509.verify_server_certificate(leaf_der, [], store, policy, crls=crl_store)
33//!
34//! # OCSP-based revocation checking (optional):
35//! ocsp_resp_der = open("ocsp-response.der", "rb").read()
36//! ocsp_store = x509.OcspStore([ocsp_resp_der])
37//! chain = x509.verify_server_certificate(leaf_der, [], store, policy, ocsp=ocsp_store)
38//!
39//! # Both CRL and OCSP revocation at once:
40//! chain = x509.verify_server_certificate(
41//!     leaf_der, [], store, policy, crls=crl_store, ocsp=ocsp_store
42//! )
43//! ```
44
45use std::net::IpAddr;
46use std::time::{SystemTime, UNIX_EPOCH};
47
48use pyo3::exceptions::PyValueError;
49use pyo3::prelude::*;
50use pyo3::types::PyBytes;
51
52use synta::{Decoder, Encoding};
53use synta_certificate::{Certificate, OpensslSignatureVerifier};
54use synta_x509_verification::{
55    ocsp::OcspStore,
56    ops::VerificationCertificate,
57    policy::{NameMatchMode, PolicyDefinition, Subject, ValidationProfile, VerificationPolicy},
58    revocation::CrlStore,
59    trust_store::Store,
60    types::{DNSName, IPAddress},
61    verify, RevocationChecks,
62};
63
64use crate::error::SyntaErr;
65use crate::install_submodule;
66
67// ── Python exception ─────────────────────────────────────────────────────────
68
69pyo3::create_exception!(
70    synta.x509,
71    X509VerificationError,
72    pyo3::exceptions::PyException
73);
74
75// ── Python types ─────────────────────────────────────────────────────────────
76
77/// A set of trusted CA certificates for X.509 chain verification.
78///
79/// Construct with a list of DER-encoded certificate bytes:
80///
81/// ```python,ignore
82/// store = TrustStore([root_ca_der, cross_root_der])
83/// ```
84///
85/// Pass PEM-encoded certificates through :func:`synta.pem_to_der` first:
86///
87/// ```python,ignore
88/// ders = synta.pem_to_der(open("roots.pem", "rb").read())
89/// store = TrustStore(ders)
90/// ```
91#[pyclass(frozen, name = "TrustStore")]
92pub struct PyTrustStore {
93    certs_der: Vec<Vec<u8>>,
94}
95
96#[pymethods]
97impl PyTrustStore {
98    /// Create a trust store from a list of DER-encoded CA certificates.
99    ///
100    /// Each entry in ``certs_der`` must be the DER bytes of a single
101    /// certificate.
102    #[new]
103    fn new(certs_der: Vec<Vec<u8>>) -> Self {
104        PyTrustStore { certs_der }
105    }
106
107    fn __repr__(&self) -> String {
108        format!("TrustStore(<{} certificate(s)>)", self.certs_der.len())
109    }
110
111    /// Number of trusted certificates in this store.
112    #[getter]
113    fn len(&self) -> usize {
114        self.certs_der.len()
115    }
116}
117
118/// A store of Certificate Revocation Lists (CRLs) for revocation checking.
119///
120/// Construct with a list of DER-encoded CRL byte strings:
121///
122/// ```python,ignore
123/// crl_store = CrlStore([crl_der])
124/// ```
125///
126/// Pass PEM-encoded CRLs through :func:`synta.pem_to_der` first:
127///
128/// ```python,ignore
129/// crl_ders = synta.pem_to_der(open("crl.pem", "rb").read())
130/// crl_store = CrlStore(crl_ders)
131/// ```
132///
133/// Pass a populated :class:`CrlStore` to :func:`verify_server_certificate` or
134/// :func:`verify_client_certificate` via the ``crls`` keyword argument to enable
135/// CRL-based revocation checking.  Certificates whose issuing CA has no matching
136/// CRL in the store are treated as not-revoked (soft-fail).
137#[pyclass(frozen, name = "CrlStore")]
138pub struct PyCrlStore {
139    crl_ders: Vec<Vec<u8>>,
140}
141
142#[pymethods]
143impl PyCrlStore {
144    /// Create a CRL store from a list of DER-encoded CRLs.
145    ///
146    /// Each entry in ``crl_ders`` must be the DER bytes of a single CRL.
147    #[new]
148    fn new(crl_ders: Vec<Vec<u8>>) -> Self {
149        PyCrlStore { crl_ders }
150    }
151
152    fn __repr__(&self) -> String {
153        format!("CrlStore(<{} CRL(s)>)", self.crl_ders.len())
154    }
155
156    /// Number of CRLs in this store.
157    #[getter]
158    fn len(&self) -> usize {
159        self.crl_ders.len()
160    }
161}
162
163/// A store of pre-fetched OCSP responses for revocation checking.
164///
165/// Construct with a list of DER-encoded OCSP response byte strings:
166///
167/// ```python,ignore
168/// ocsp_store = OcspStore([ocsp_resp_der])
169/// ```
170///
171/// Pass a populated :class:`OcspStore` to :func:`verify_server_certificate` or
172/// :func:`verify_client_certificate` via the ``ocsp`` keyword argument to enable
173/// OCSP-based revocation checking.  Certificates for which no valid, matching
174/// OCSP response is found are treated as not-revoked (soft-fail).
175#[pyclass(frozen, name = "OcspStore")]
176pub struct PyOcspStore {
177    ocsp_ders: Vec<Vec<u8>>,
178}
179
180#[pymethods]
181impl PyOcspStore {
182    /// Create an OCSP store from a list of DER-encoded OCSP responses.
183    ///
184    /// Each entry in ``ocsp_ders`` must be the DER bytes of a single OCSP
185    /// response (the outer ``OCSPResponse`` SEQUENCE as defined in RFC 6960).
186    #[new]
187    fn new(ocsp_ders: Vec<Vec<u8>>) -> Self {
188        PyOcspStore { ocsp_ders }
189    }
190
191    fn __repr__(&self) -> String {
192        format!("OcspStore(<{} OCSP response(s)>)", self.ocsp_ders.len())
193    }
194
195    /// Number of OCSP responses in this store.
196    #[getter]
197    fn len(&self) -> usize {
198        self.ocsp_ders.len()
199    }
200}
201
202/// Optional parameters that control certificate chain verification.
203///
204/// All fields have safe defaults so you only need to set what differs from
205/// the standard WebPKI TLS server / client validation.
206///
207/// ``server_names`` accepts a list of DNS hostnames or IP address literals.
208/// When more than one name is given, ``name_match`` controls whether the
209/// certificate must cover **any** (default) or **all** of them:
210///
211/// * ``"any"`` — connection validation: the certificate is accepted if it
212///   matches at least one name in the list (e.g. you are connecting to one
213///   of several possible endpoints).
214/// * ``"all"`` — cert assessment: the certificate must cover every name
215///   (e.g. checking a cert covers your entire domain set before deploying).
216///
217/// ```python,ignore
218/// import synta.x509 as x509
219///
220/// # Single name — classic behaviour:
221/// policy = x509.VerificationPolicy(server_names=["example.com"])
222///
223/// # Any-match: accept the cert if it covers either name (connection validation):
224/// policy = x509.VerificationPolicy(
225///     server_names=["example.com", "www.example.com"],
226///     name_match="any",
227/// )
228///
229/// # All-match: verify the cert covers every name (cert assessment):
230/// policy = x509.VerificationPolicy(
231///     server_names=["example.com", "api.example.com"],
232///     name_match="all",
233/// )
234///
235/// # Strict RFC 5280 profile with a fixed validation time, no SAN check:
236/// policy = x509.VerificationPolicy(
237///     profile="rfc5280",
238///     validation_time=1_700_000_000,
239///     max_chain_depth=4,
240/// )
241/// ```
242#[pyclass(name = "VerificationPolicy")]
243pub struct PyVerificationPolicy {
244    inner: VerificationPolicy,
245}
246
247#[pymethods]
248impl PyVerificationPolicy {
249    #[new]
250    #[pyo3(
251        signature = (*, server_names=None, name_match=None, validation_time=None, max_chain_depth=8, profile=None)
252    )]
253    fn new(
254        server_names: Option<Vec<String>>,
255        name_match: Option<String>,
256        validation_time: Option<i64>,
257        max_chain_depth: u8,
258        profile: Option<String>,
259    ) -> PyResult<Self> {
260        let vprofile = match profile.as_deref() {
261            None | Some("webpki") => ValidationProfile::WebPki,
262            Some("rfc5280") => ValidationProfile::Rfc5280,
263            Some(other) => {
264                return Err(PyValueError::new_err(format!(
265                    "unknown validation profile {other:?}; expected \"webpki\" or \"rfc5280\""
266                )));
267            }
268        };
269        let vmatch = match name_match.as_deref() {
270            None | Some("any") => NameMatchMode::Any,
271            Some("all") => NameMatchMode::All,
272            Some(other) => {
273                return Err(PyValueError::new_err(format!(
274                    "unknown name_match {other:?}; expected \"any\" or \"all\""
275                )));
276            }
277        };
278        Ok(PyVerificationPolicy {
279            inner: VerificationPolicy {
280                server_names: server_names.unwrap_or_default(),
281                name_match: vmatch,
282                validation_time,
283                max_chain_depth,
284                profile: vprofile,
285            },
286        })
287    }
288
289    fn __repr__(&self) -> String {
290        let profile = match self.inner.profile {
291            ValidationProfile::WebPki => "webpki",
292            ValidationProfile::Rfc5280 => "rfc5280",
293        };
294        let name_match = match self.inner.name_match {
295            NameMatchMode::Any => "any",
296            NameMatchMode::All => "all",
297        };
298        format!(
299            "VerificationPolicy(server_names={:?}, name_match={:?}, profile={:?}, \
300             max_chain_depth={}, validation_time={:?})",
301            self.inner.server_names,
302            name_match,
303            profile,
304            self.inner.max_chain_depth,
305            self.inner.validation_time,
306        )
307    }
308}
309
310// ── Helpers ───────────────────────────────────────────────────────────────────
311
312fn now_unix() -> i64 {
313    SystemTime::now()
314        .duration_since(UNIX_EPOCH)
315        .map(|d| d.as_secs() as i64)
316        .unwrap_or(0)
317}
318
319/// Parse a single DER cert.
320fn parse_vcert(der: &[u8]) -> PyResult<Certificate<'_>> {
321    Ok(Decoder::new(der, Encoding::Der)
322        .decode::<Certificate>()
323        .map_err(SyntaErr)?)
324}
325
326/// Core verifier: build `VerificationCertificate`s from borrowed DER slices,
327/// run `verify()` with a [`RevocationChecks`] struct, and return the chain as
328/// owned DER byte vectors.
329///
330/// `intermediate_ders` holds owned byte vecs (kept alive by the caller) so
331/// we can take stable `&[u8]` references into them.
332fn run_verify<'a>(
333    leaf_der: &'a [u8],
334    intermediate_ders: &'a [Vec<u8>],
335    trust_store: &'a PyTrustStore,
336    policy: PolicyDefinition<'a, OpensslSignatureVerifier>,
337    crls: Option<&'a CrlStore>,
338    ocsp: Option<&'a OcspStore>,
339) -> PyResult<Vec<Vec<u8>>> {
340    let leaf_cert = parse_vcert(leaf_der)?;
341    let leaf_vcert = VerificationCertificate::new(leaf_cert, leaf_der);
342
343    let mut intermediate_vcerts = Vec::with_capacity(intermediate_ders.len());
344    for der in intermediate_ders {
345        let cert = parse_vcert(der.as_slice())?;
346        intermediate_vcerts.push(VerificationCertificate::new(cert, der.as_slice()));
347    }
348
349    let mut trusted_vcerts = Vec::with_capacity(trust_store.certs_der.len());
350    for der in &trust_store.certs_der {
351        let cert = parse_vcert(der.as_slice())?;
352        trusted_vcerts.push(VerificationCertificate::new(cert, der.as_slice()));
353    }
354
355    let store = Store::new(trusted_vcerts);
356
357    let revocation = RevocationChecks { crls, ocsp };
358    verify(
359        &leaf_vcert,
360        &intermediate_vcerts,
361        &policy,
362        &store,
363        revocation,
364    )
365    .map(|chain| chain.into_iter().map(|vc| vc.der().to_vec()).collect())
366    .map_err(|e| X509VerificationError::new_err(e.to_string()))
367}
368
369// ── Python functions ──────────────────────────────────────────────────────────
370
371/// Parse a name string (DNS hostname or IP literal) into a [`Subject`].
372fn parse_subject(name: &str) -> PyResult<Subject<'_>> {
373    if let Ok(ip) = name.parse::<IpAddr>() {
374        let addr = match ip {
375            IpAddr::V4(a) => IPAddress::from_bytes(&a.octets()),
376            IpAddr::V6(a) => IPAddress::from_bytes(&a.octets()),
377        };
378        Ok(Subject::Ip(addr.ok_or_else(|| {
379            PyValueError::new_err(format!("invalid IP address for server name: {name}"))
380        })?))
381    } else {
382        Ok(Subject::Dns(DNSName::new(name).ok_or_else(|| {
383            PyValueError::new_err(format!("invalid DNS name for server name: {name:?}"))
384        })?))
385    }
386}
387
388/// Verify a TLS server certificate chain using the WebPKI / CABF profile.
389///
390/// ``leaf_der`` is the DER-encoded end-entity certificate.
391/// ``intermediates_der`` is a list of DER-encoded intermediate CA certificates
392/// (order does not matter; the validator discovers the chain automatically).
393/// ``trust_store`` is a :class:`TrustStore` populated with trusted root CAs.
394/// ``policy`` is an optional :class:`VerificationPolicy` specifying the server
395/// names, name match mode, validation time, chain depth, and compliance profile.
396///
397/// Returns the validated chain as ``list[bytes]``, ordered root-first
398/// (``chain[0]`` is the trust anchor, ``chain[-1]`` is the leaf).
399///
400/// Raises :exc:`X509VerificationError` if verification fails.
401///
402/// ```python,ignore
403/// import synta
404/// import synta.x509 as x509
405///
406/// store = x509.TrustStore([root_der])
407///
408/// # Single name:
409/// policy = x509.VerificationPolicy(server_names=["example.com"])
410/// chain = x509.verify_server_certificate(leaf_der, [intermediate_der], store, policy)
411/// for i, cert_der in enumerate(chain):
412///     cert = synta.Certificate.from_der(cert_der)
413///     print(f"chain[{i}]: {cert.subject}")
414///
415/// # Multi-name, all-match (cert covers every name):
416/// policy = x509.VerificationPolicy(
417///     server_names=["example.com", "api.example.com"],
418///     name_match="all",
419/// )
420/// chain = x509.verify_server_certificate(leaf_der, [], store, policy)
421/// ```
422#[pyfunction]
423#[pyo3(signature = (leaf_der, intermediates_der, trust_store, policy=None, crls=None, ocsp=None))]
424fn verify_server_certificate<'py>(
425    py: Python<'py>,
426    leaf_der: Vec<u8>,
427    intermediates_der: Vec<Vec<u8>>,
428    trust_store: &PyTrustStore,
429    policy: Option<&PyVerificationPolicy>,
430    crls: Option<&PyCrlStore>,
431    ocsp: Option<&PyOcspStore>,
432) -> PyResult<Vec<Bound<'py, PyBytes>>> {
433    let default_policy;
434    let vp: &VerificationPolicy = match policy {
435        Some(p) => &p.inner,
436        None => {
437            default_policy = VerificationPolicy::new_client();
438            &default_policy
439        }
440    };
441
442    let now = vp.validation_time.unwrap_or_else(now_unix);
443
444    let subjects = vp
445        .server_names
446        .iter()
447        .map(|name| parse_subject(name))
448        .collect::<PyResult<Vec<_>>>()?;
449
450    let mut pd = PolicyDefinition::new_server(OpensslSignatureVerifier, subjects, now);
451    pd.profile = vp.profile;
452    pd.max_chain_depth = vp.max_chain_depth;
453    pd.name_match = vp.name_match;
454
455    // Build a CrlStore from the Python CRL store if provided.
456    let mut crl_store = CrlStore::new();
457    if let Some(py_crls) = crls {
458        for der in &py_crls.crl_ders {
459            crl_store.add_der(der.clone());
460        }
461    }
462    let crl_opt = crls.map(|_| &crl_store);
463
464    // Build an OcspStore from the Python OCSP store if provided.
465    let mut ocsp_store = OcspStore::new();
466    if let Some(py_ocsp) = ocsp {
467        for der in &py_ocsp.ocsp_ders {
468            ocsp_store.add_der(der.clone());
469        }
470    }
471    let ocsp_opt = ocsp.map(|_| &ocsp_store);
472
473    run_verify(
474        leaf_der.as_slice(),
475        &intermediates_der,
476        trust_store,
477        pd,
478        crl_opt,
479        ocsp_opt,
480    )?
481    .into_iter()
482    .map(|der| Ok(PyBytes::new(py, &der)))
483    .collect()
484}
485
486/// Verify a TLS client certificate chain.
487///
488/// Like :func:`verify_server_certificate` but uses the ``clientAuth`` EKU and
489/// skips SAN / server-name matching.
490///
491/// ``policy`` may set ``validation_time``, ``max_chain_depth``, and
492/// ``profile``; its ``server_names`` field is ignored.
493///
494/// Returns the validated chain as ``list[bytes]``, root-first.
495///
496/// Raises :exc:`X509VerificationError` if verification fails.
497///
498/// ```python,ignore
499/// import synta.x509 as x509
500///
501/// store = x509.TrustStore([root_der])
502/// chain = x509.verify_client_certificate(leaf_der, [], store)
503/// ```
504#[pyfunction]
505#[pyo3(signature = (leaf_der, intermediates_der, trust_store, policy=None, crls=None, ocsp=None))]
506fn verify_client_certificate<'py>(
507    py: Python<'py>,
508    leaf_der: Vec<u8>,
509    intermediates_der: Vec<Vec<u8>>,
510    trust_store: &PyTrustStore,
511    policy: Option<&PyVerificationPolicy>,
512    crls: Option<&PyCrlStore>,
513    ocsp: Option<&PyOcspStore>,
514) -> PyResult<Vec<Bound<'py, PyBytes>>> {
515    let default_policy;
516    let vp: &VerificationPolicy = match policy {
517        Some(p) => &p.inner,
518        None => {
519            default_policy = VerificationPolicy::new_client();
520            &default_policy
521        }
522    };
523
524    let now = vp.validation_time.unwrap_or_else(now_unix);
525
526    let mut pd = PolicyDefinition::new_client(OpensslSignatureVerifier, now);
527    pd.profile = vp.profile;
528    pd.max_chain_depth = vp.max_chain_depth;
529
530    // Build a CrlStore from the Python CRL store if provided.
531    let mut crl_store = CrlStore::new();
532    if let Some(py_crls) = crls {
533        for der in &py_crls.crl_ders {
534            crl_store.add_der(der.clone());
535        }
536    }
537    let crl_opt = crls.map(|_| &crl_store);
538
539    // Build an OcspStore from the Python OCSP store if provided.
540    let mut ocsp_store = OcspStore::new();
541    if let Some(py_ocsp) = ocsp {
542        for der in &py_ocsp.ocsp_ders {
543            ocsp_store.add_der(der.clone());
544        }
545    }
546    let ocsp_opt = ocsp.map(|_| &ocsp_store);
547
548    run_verify(
549        leaf_der.as_slice(),
550        &intermediates_der,
551        trust_store,
552        pd,
553        crl_opt,
554        ocsp_opt,
555    )?
556    .into_iter()
557    .map(|der| Ok(PyBytes::new(py, &der)))
558    .collect()
559}
560
561// ── Module registration ───────────────────────────────────────────────────────
562
563/// Register the ``synta.x509`` submodule onto ``parent``.
564pub fn register_x509_module(parent: &Bound<'_, PyModule>) -> PyResult<()> {
565    let py = parent.py();
566    let m = PyModule::new(py, "x509")?;
567
568    m.add(
569        "X509VerificationError",
570        py.get_type::<X509VerificationError>(),
571    )?;
572    m.add_class::<PyTrustStore>()?;
573    m.add_class::<PyCrlStore>()?;
574    m.add_class::<PyOcspStore>()?;
575    m.add_class::<PyVerificationPolicy>()?;
576    m.add_function(wrap_pyfunction!(verify_server_certificate, &m)?)?;
577    m.add_function(wrap_pyfunction!(verify_client_certificate, &m)?)?;
578
579    install_submodule(
580        parent,
581        &m,
582        "synta.x509",
583        Some("RFC 5280 X.509 certificate chain verification with OpenSSL backend."),
584    )?;
585    Ok(())
586}
_synta/x509_verification.rs

_synta/
x509_verification.rs
