Skip to main content

_synta/certificate/
cert.rs

1//! Python binding for [`PyCertificate`] (X.509 Certificate).
2
3use std::sync::OnceLock;
4
5use pyo3::prelude::*;
6use pyo3::types::{PyBytes, PyList, PyString};
7use pyo3::PyClass;
8
9use synta::traits::Encode;
10use synta::{Decoder, Encoding};
11use synta_certificate::{Certificate, PolicyQualifierInfo, Time};
12
13use crate::types::PyObjectIdentifier;
14
15use super::oid_from_pyany;
16
17/// Re-encode an ASN.1 `Element` to DER bytes, returning `None` for a missing
18/// optional field.  Used by the algorithm-parameter getters.
19fn encode_element_opt<'py>(
20    py: Python<'py>,
21    elem: Option<&synta::Element<'_>>,
22) -> PyResult<Option<Bound<'py, PyBytes>>> {
23    match elem {
24        None => Ok(None),
25        Some(e) => {
26            let mut encoder = synta::Encoder::new(Encoding::Der);
27            e.encode(&mut encoder)
28                .map_err(|err| pyo3::exceptions::PyValueError::new_err(format!("{err}")))?;
29            let bytes = encoder
30                .finish()
31                .map_err(|err| pyo3::exceptions::PyValueError::new_err(format!("{err}")))?;
32            Ok(Some(PyBytes::new(py, &bytes)))
33        }
34    }
35}
36
37/// Decode a raw DER Name SEQUENCE and format it as an RFC 4514-style string.
38fn decode_name_debug(raw: &[u8]) -> String {
39    synta_certificate::name::format_dn(raw)
40}
41
42/// Shared implementation for every `from_pem` static method.
43///
44/// Decodes all PEM blocks in `data` and constructs a Python object for each
45/// by calling `make_obj(py, der_bytes)`.  The closure is responsible for
46/// parsing the DER bytes and wrapping the result as a `Bound<'py, PyAny>`;
47/// this lets the caller use concrete types (with their full `Py::new` bounds)
48/// without any generic `PyClass` constraint here.
49///
50/// Returns a single object for one block, a `list` for multiple blocks, and
51/// raises `ValueError` when no block is found.
52pub(super) fn pem_blocks_to_pyobject<'py, F>(
53    py: Python<'py>,
54    data: &[u8],
55    make_obj: F,
56) -> PyResult<Bound<'py, PyAny>>
57where
58    F: for<'a> Fn(Python<'py>, Bound<'a, PyBytes>) -> PyResult<Bound<'py, PyAny>>,
59{
60    let blocks = synta_certificate::pem_blocks(data);
61    match blocks.len() {
62        0 => Err(pyo3::exceptions::PyValueError::new_err(
63            "no PEM block found in input",
64        )),
65        1 => make_obj(py, PyBytes::new(py, &blocks[0].1)),
66        _ => {
67            let list = PyList::empty(py);
68            for (_, der) in &blocks {
69                list.append(make_obj(py, PyBytes::new(py, der))?)?;
70            }
71            Ok(list.into_any())
72        }
73    }
74}
75
76/// Shared implementation for every `to_pem` static method.
77///
78/// Accepts either a single Python object of type `T` or a `list` of them,
79/// serialises each to a PEM block labelled `label`, and returns the
80/// concatenated bytes.  The `get_der` closure extracts the raw DER bytes from
81/// a Rust reference to `T`.
82///
83/// Using a closure (rather than a trait method) avoids requiring `T: PyClass`
84/// with internal PyO3 initialiser bounds while still letting the caller use
85/// concrete types.
86pub(super) fn pyobject_to_pem<'py, T, D>(
87    py: Python<'py>,
88    label: &str,
89    obj_or_list: &Bound<'_, PyAny>,
90    get_der: D,
91) -> PyResult<Bound<'py, PyBytes>>
92where
93    T: PyClass,
94    D: for<'r> Fn(&'r T) -> &'r [u8],
95{
96    let mut pem: Vec<u8> = Vec::new();
97    if let Ok(list) = obj_or_list.cast::<PyList>() {
98        for item in list.iter() {
99            let bound_t = item.cast::<T>().map_err(|_| {
100                pyo3::exceptions::PyTypeError::new_err(format!(
101                    "list items must be {label} objects"
102                ))
103            })?;
104            let borrow = bound_t.borrow();
105            pem.extend_from_slice(&synta_certificate::der_to_pem(label, get_der(&borrow)));
106        }
107    } else {
108        let bound_t = obj_or_list.cast::<T>().map_err(|_| {
109            pyo3::exceptions::PyTypeError::new_err(format!(
110                "expected a {label} object or list[{label}]"
111            ))
112        })?;
113        let borrow = bound_t.borrow();
114        pem.extend_from_slice(&synta_certificate::der_to_pem(label, get_der(&borrow)));
115    }
116    Ok(PyBytes::new(py, &pem))
117}
118
119/// A single policy qualifier from the CertificatePolicies extension.
120///
121/// Corresponds to `PolicyQualifierInfo` in RFC 5280:
122///
123/// ```asn1
124/// PolicyQualifierInfo ::= SEQUENCE {
125///     policyQualifierId  PolicyQualifierId,
126///     qualifier          ANY DEFINED BY policyQualifierId }
127/// ```
128///
129/// The ``qualifier_value`` bytes are the complete DER TLV of the ``qualifier``
130/// field.  For ``id-qt-cps`` (OID ``1.3.6.1.5.5.7.2.1``) this encodes an
131/// IA5String containing the CPS URI.  For ``id-qt-unotice``
132/// (``1.3.6.1.5.5.7.2.2``) it encodes a ``UserNotice`` SEQUENCE.
133#[pyclass(frozen, name = "PolicyQualifier", module = "synta")]
134pub struct PyPolicyQualifier {
135    pub qualifier_oid: synta::ObjectIdentifier,
136    pub qualifier_value: Vec<u8>,
137}
138
139#[pymethods]
140impl PyPolicyQualifier {
141    /// OID identifying the qualifier type.
142    ///
143    /// Common values:
144    ///
145    /// - ``1.3.6.1.5.5.7.2.1`` — CPS pointer (URI string)
146    /// - ``1.3.6.1.5.5.7.2.2`` — user notice
147    #[getter]
148    fn qualifier_oid<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyObjectIdentifier>> {
149        Py::new(py, PyObjectIdentifier::from_oid(self.qualifier_oid.clone()))
150            .map(|p| p.into_bound(py))
151    }
152
153    /// Raw DER bytes of the qualifier value (tag + length + value).
154    ///
155    /// For a CPS qualifier (OID ``1.3.6.1.5.5.7.2.1``) this is an IA5String
156    /// encoding of the CPS URI.  Decode with:
157    ///
158    /// ```python,ignore
159    /// synta.Decoder(data, synta.Encoding.DER).decode_ia5_string()
160    /// ```
161    #[getter]
162    fn qualifier_value<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
163        PyBytes::new(py, &self.qualifier_value)
164    }
165
166    fn __repr__(&self) -> String {
167        format!("PolicyQualifier(qualifier_oid={})", self.qualifier_oid)
168    }
169}
170
171/// A single ``PolicyInformation`` entry from the CertificatePolicies extension.
172///
173/// Corresponds to ``PolicyInformation`` in RFC 5280:
174///
175/// ```asn1
176/// PolicyInformation ::= SEQUENCE {
177///     policyIdentifier   CertPolicyId,
178///     policyQualifiers   PolicyQualifiers OPTIONAL }
179/// ```
180#[pyclass(frozen, name = "PolicyInformation", module = "synta")]
181pub struct PyPolicyInformation {
182    pub policy_oid: synta::ObjectIdentifier,
183    pub qualifiers: Vec<(synta::ObjectIdentifier, Vec<u8>)>,
184}
185
186#[pymethods]
187impl PyPolicyInformation {
188    /// The certificate policy OID.
189    #[getter]
190    fn policy_oid<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyObjectIdentifier>> {
191        Py::new(py, PyObjectIdentifier::from_oid(self.policy_oid.clone())).map(|p| p.into_bound(py))
192    }
193
194    /// List of :class:`PolicyQualifier` entries (empty when ``policyQualifiers`` is absent).
195    #[getter]
196    fn qualifiers<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
197        let list = PyList::empty(py);
198        for (qoid, qval) in &self.qualifiers {
199            let pq = Py::new(
200                py,
201                PyPolicyQualifier {
202                    qualifier_oid: qoid.clone(),
203                    qualifier_value: qval.clone(),
204                },
205            )?;
206            list.append(pq.into_bound(py))?;
207        }
208        Ok(list)
209    }
210
211    fn __repr__(&self) -> String {
212        format!("PolicyInformation(policy_oid={})", self.policy_oid)
213    }
214}
215
216/// X.509 Certificate accessible from Python.
217///
218/// Example (inside a Python extension that called `register_module`):
219///
220/// ```python
221/// cert = Certificate.from_der(open("cert.der", "rb").read())
222/// print(cert.subject)
223/// ```
224#[pyclass(frozen, name = "Certificate")]
225pub struct PyCertificate {
226    // Holds a strong reference to the Python bytes object that backs the
227    // DER data.  The CPython refcount prevents the underlying bytes buffer
228    // from being freed for the full lifetime of this struct.  After the
229    // struct is collected by Python's GC, `_data` drops (in Rust's
230    // declaration order) and decrements the refcount; by that point no
231    // Rust code can hold a borrow of this struct, so no read of `raw`
232    // can occur through `&self` after drop begins.
233    pub(super) _data: Py<PyBytes>,
234    // Raw slice pointing into `_data`'s buffer.  SAFETY: valid as long as
235    // `_data` is alive; CPython bytes objects have a fixed-address buffer
236    // that is never relocated.  Stored here so the OnceLock init closure in
237    // `cert()` can access the bytes without requiring a GIL token.
238    pub(super) raw: &'static [u8],
239    // Full decoded certificate, heap-allocated and initialised lazily on first
240    // field access.  `Box` keeps the 568-byte `Certificate` off the struct so
241    // that `PyCertificate` stays within CPython's 512-byte `pymalloc`
242    // threshold (after adding the Python object header).  Allocating through
243    // pymalloc instead of the system allocator is roughly 3× faster, which is
244    // measurable at parse-only speeds.  The full recursive decode is deferred
245    // so that parse-only workloads pay only the shallow envelope-scan cost.
246    inner: OnceLock<Box<Certificate<'static>>>,
247    // Byte range within `_data` covering the complete TBSCertificate TLV.
248    tbs_range: std::ops::Range<usize>,
249    // Lazily-computed Python objects.  Each field is initialised on first
250    // Python access and returned via clone_ref on every subsequent call,
251    // avoiding repeated allocation and PyO3 string construction.
252    issuer_cache: OnceLock<Py<PyString>>,
253    subject_cache: OnceLock<Py<PyString>>,
254    signature_algorithm_cache: OnceLock<Py<PyString>>,
255    not_before_cache: OnceLock<Py<PyString>>,
256    not_after_cache: OnceLock<Py<PyString>>,
257    public_key_algorithm_cache: OnceLock<Py<PyString>>,
258    serial_number_cache: OnceLock<Py<PyAny>>,
259    signature_value_cache: OnceLock<Py<PyBytes>>,
260    public_key_cache: OnceLock<Py<PyBytes>>,
261    tbs_bytes_cache: OnceLock<Py<PyBytes>>,
262    signature_algorithm_oid_cache: OnceLock<Py<PyObjectIdentifier>>,
263    public_key_algorithm_oid_cache: OnceLock<Py<PyObjectIdentifier>>,
264    signature_algorithm_der_cache: OnceLock<Py<PyBytes>>,
265    signature_algorithm_params_cache: OnceLock<Option<Py<PyBytes>>>,
266    public_key_algorithm_params_cache: OnceLock<Option<Py<PyBytes>>>,
267    extensions_der_cache: OnceLock<Option<Py<PyBytes>>>,
268    issuer_raw_der_cache: OnceLock<Py<PyBytes>>,
269    subject_raw_der_cache: OnceLock<Py<PyBytes>>,
270    not_before_utc_cache: OnceLock<Py<PyAny>>,
271    not_after_utc_cache: OnceLock<Py<PyAny>>,
272    spki_der_cache: OnceLock<Py<PyBytes>>,
273}
274
275impl PyCertificate {
276    /// Return the fully-decoded certificate, triggering a full recursive
277    /// `Certificate::decode()` on the first call.
278    ///
279    /// The result is cached in `inner` so subsequent calls are a single
280    /// atomic load.  Returns `Err(PyValueError)` if the full decode fails
281    /// (e.g. malformed inner content that passed the shallow envelope scan
282    /// in `from_der`).  Unlike a panic, this surfaces as a catchable Python
283    /// `ValueError` rather than an uncatchable `PanicException`.
284    pub(super) fn cert(&self) -> PyResult<&Certificate<'static>> {
285        if let Some(v) = self.inner.get() {
286            return Ok(v.as_ref());
287        }
288        let mut decoder = Decoder::new(self.raw, Encoding::Der);
289        let decoded = decoder.decode().map_err(|e| {
290            pyo3::exceptions::PyValueError::new_err(format!("Certificate DER decode failed: {e}"))
291        })?;
292        let _ = self.inner.set(Box::new(decoded));
293        Ok(self.inner.get().unwrap().as_ref())
294    }
295}
296
297#[pymethods]
298impl PyCertificate {
299    /// Parse a DER-encoded X.509 certificate.
300    #[staticmethod]
301    fn from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
302        // Store the Python bytes object directly — no copy of the DER data.
303        let py_bytes = data.unbind();
304
305        // SAFETY: `py_bytes` holds a strong reference (Py<PyBytes>) that
306        // keeps the Python bytes object alive for the lifetime of this struct.
307        // CPython's bytes objects have a fixed-address, non-relocating payload
308        // buffer (CPython has no moving GC for bytes objects).  The slice
309        // lifetime is extended to 'static; the actual safety invariants are:
310        //   (1) All reads of `raw` go through `&self` (a borrow of the struct).
311        //       No borrow of the struct can outlive the struct, so `raw` is
312        //       never read after the struct begins dropping.
313        //   (2) `raw: &'static [u8]` has no destructor (it is a fat pointer
314        //       with no heap allocation), so Rust dropping `_data` before `raw`
315        //       (fields drop in declaration order) does not cause a
316        //       use-after-free during the drop sequence itself.
317        //   (3) `inner` contains `Certificate<'static>` references into the
318        //       buffer; dropping `Box<Certificate<'static>>` frees the box
319        //       allocation but does not read through the contained &'static
320        //       slices (borrows have no destructors in Rust).
321        // CPython-only: this pattern does not hold for PyPy or GraalPy,
322        // which may relocate or compact heap objects.
323        let raw: &'static [u8] = unsafe {
324            let s = py_bytes.bind(py).as_bytes();
325            std::slice::from_raw_parts(s.as_ptr(), s.len())
326        };
327
328        // Shallow structural scan: validate the Certificate SEQUENCE envelope
329        // and locate the TBSCertificate TLV range.  This is ~4 decoder
330        // operations — roughly 10× faster than a full Certificate::decode().
331        // The full decode is deferred to the first field access via `cert()`.
332        //
333        // Certificate ::= SEQUENCE {
334        //     tbsCertificate   TBSCertificate,   -- child 0: SEQUENCE
335        //     signatureAlgorithm AlgorithmIdentifier, -- child 1: SEQUENCE
336        //     signature        BIT STRING            -- child 2
337        // }
338        let tbs_range = {
339            let mut d = Decoder::new(raw, Encoding::Der);
340            // Outer Certificate SEQUENCE tag + length.
341            d.read_tag()
342                .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
343            d.read_length()
344                .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
345            // TBSCertificate: record the full TLV range (tag + length + content).
346            let tbs_start = d.position();
347            d.read_tag()
348                .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
349            let tbs_len = d
350                .read_length()
351                .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
352            let tbs_content_len = tbs_len
353                .definite()
354                .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
355            tbs_start..(d.position() + tbs_content_len)
356        };
357
358        Ok(Self {
359            _data: py_bytes,
360            raw,
361            inner: OnceLock::new(),
362            tbs_range,
363            issuer_cache: OnceLock::new(),
364            subject_cache: OnceLock::new(),
365            signature_algorithm_cache: OnceLock::new(),
366            not_before_cache: OnceLock::new(),
367            not_after_cache: OnceLock::new(),
368            public_key_algorithm_cache: OnceLock::new(),
369            serial_number_cache: OnceLock::new(),
370            signature_value_cache: OnceLock::new(),
371            public_key_cache: OnceLock::new(),
372            tbs_bytes_cache: OnceLock::new(),
373            signature_algorithm_oid_cache: OnceLock::new(),
374            public_key_algorithm_oid_cache: OnceLock::new(),
375            signature_algorithm_der_cache: OnceLock::new(),
376            signature_algorithm_params_cache: OnceLock::new(),
377            public_key_algorithm_params_cache: OnceLock::new(),
378            extensions_der_cache: OnceLock::new(),
379            issuer_raw_der_cache: OnceLock::new(),
380            subject_raw_der_cache: OnceLock::new(),
381            not_before_utc_cache: OnceLock::new(),
382            not_after_utc_cache: OnceLock::new(),
383            spki_der_cache: OnceLock::new(),
384        })
385    }
386
387    /// Parse a DER-encoded X.509 certificate and perform a full RFC 5280 decode immediately.
388    ///
389    /// Unlike :meth:`from_der`, which performs only a shallow 4-operation
390    /// envelope scan and defers the full :class:`Certificate` decode to the
391    /// first field access, this method triggers the complete recursive
392    /// ``Certificate::decode()`` at construction time.
393    ///
394    /// Use this when you need all fields to be available without any
395    /// lazy-decode latency on the first getter call, or when benchmarking the
396    /// true full-parse cost that is comparable to Criterion's ``rust_typed``
397    /// numbers.
398    ///
399    /// ```python
400    /// # Full parse happens here — no deferred work on first field access.
401    /// cert = Certificate.full_from_der(der)
402    /// print(cert.issuer)   # warm path only
403    /// ```
404    #[staticmethod]
405    fn full_from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
406        let cert = Self::from_der(py, data)?;
407        // Prime the OnceLock: runs the full Certificate::decode() now.
408        cert.cert()?;
409        Ok(cert)
410    }
411
412    /// Parse a PEM-encoded X.509 certificate.
413    ///
414    /// Strips the ``-----BEGIN CERTIFICATE-----`` / ``-----END CERTIFICATE-----``
415    /// boundary lines, decodes the base64 body, and calls :meth:`from_der`.
416    /// No external dependencies are required — the decoder is implemented in
417    /// pure Rust inside ``synta-certificate``.
418    ///
419    /// Returns a single :class:`Certificate` when the input contains exactly
420    /// one PEM block.  When multiple blocks are present (e.g. a certificate
421    /// chain file), returns a :class:`list` of :class:`Certificate` objects in
422    /// the order they appear.  Raises :exc:`ValueError` if no PEM block is
423    /// found.
424    ///
425    /// ```python
426    /// # Single certificate — returns Certificate directly.
427    /// cert = Certificate.from_pem(open("cert.pem", "rb").read())
428    /// print(cert.subject)
429    ///
430    /// # Certificate chain — returns list[Certificate].
431    /// chain = Certificate.from_pem(open("chain.pem", "rb").read())
432    /// for cert in chain:
433    ///     print(cert.subject)
434    /// ```
435    #[staticmethod]
436    fn from_pem<'py>(py: Python<'py>, data: Bound<'_, PyBytes>) -> PyResult<Bound<'py, PyAny>> {
437        pem_blocks_to_pyobject(py, data.as_bytes(), |py, bytes| {
438            let obj = Self::from_der(py, bytes)?;
439            Ok(Py::new(py, obj)?.into_bound(py).into_any())
440        })
441    }
442
443    /// Convert to a ``cryptography.x509.Certificate`` (PyCA) object.
444    ///
445    /// Passes the original DER bytes directly to
446    /// ``cryptography.x509.load_der_x509_certificate``; no re-encoding is
447    /// performed.  Requires the ``cryptography`` package to be installed.
448    ///
449    /// ```python
450    /// synta_cert = synta.Certificate.from_der(der)
451    /// pyca_cert  = synta_cert.to_pyca()
452    /// # Full cryptographic operations are now available via PyCA:
453    /// pyca_cert.public_key().verify(signature, message, ...)
454    /// ```
455    fn to_pyca<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
456        let m = py.import("cryptography.x509").map_err(|_| {
457            pyo3::exceptions::PyImportError::new_err(
458                "the 'cryptography' package is required; install it with: pip install cryptography",
459            )
460        })?;
461        m.call_method1(
462            "load_der_x509_certificate",
463            (self._data.clone_ref(py).into_bound(py),),
464        )
465    }
466
467    /// Construct a ``Certificate`` from a ``cryptography.x509.Certificate``.
468    ///
469    /// Serialises the PyCA certificate to DER via
470    /// ``cert.public_bytes(Encoding.DER)`` and parses the result with
471    /// :meth:`from_der`.  Requires the ``cryptography`` package to be
472    /// installed.
473    ///
474    /// **Fast path** — if the object exposes a ``_synta_der_bytes`` attribute
475    /// whose value is a ``bytes`` object, that buffer is used directly and
476    /// the ``public_bytes()`` call is skipped entirely.  Wrapper classes that
477    /// already hold the raw DER (e.g. an ``IPACertificate`` loaded from LDAP)
478    /// can opt in by setting the attribute at construction time:
479    ///
480    /// ```python
481    /// class IPACertificate:
482    ///     def __init__(self, der: bytes):
483    ///         self._synta_der_bytes = der          # enables fast path
484    ///         self._pyca = x509.load_der_x509_certificate(der)
485    ///
486    /// # Zero re-encoding cost — DER is read from _synta_der_bytes directly:
487    /// synta_cert = synta.Certificate.from_pyca(ipa_cert)
488    /// ```
489    ///
490    /// Without the attribute the standard path is used:
491    ///
492    /// ```python
493    /// pyca_cert  = cryptography.x509.load_pem_x509_certificate(pem)
494    /// synta_cert = synta.Certificate.from_pyca(pyca_cert)
495    /// ```
496    #[staticmethod]
497    fn from_pyca(py: Python<'_>, pyca_cert: Bound<'_, PyAny>) -> PyResult<Self> {
498        // Optional fast path: if the caller has cached raw DER bytes on the
499        // object under the `_synta_der_bytes` attribute, use that directly to
500        // avoid the re-encoding cost of `public_bytes(Encoding.DER)`.  This
501        // lets wrappers like FreeIPA's `IPACertificate` opt in by setting
502        // `self._synta_der_bytes = der` at construction time.
503        if let Ok(attr) = pyca_cert.getattr("_synta_der_bytes") {
504            if let Ok(der_bytes) = attr.cast_into::<PyBytes>() {
505                return Self::from_der(py, der_bytes);
506            }
507        }
508
509        let ser = py
510            .import("cryptography.hazmat.primitives.serialization")
511            .map_err(|_| {
512                pyo3::exceptions::PyImportError::new_err(
513                    "the 'cryptography' package is required; install it with: pip install cryptography",
514                )
515            })?;
516        let der_bytes: Bound<'_, PyBytes> = pyca_cert
517            .call_method1("public_bytes", (ser.getattr("Encoding")?.getattr("DER")?,))?
518            .cast_into()?;
519        Self::from_der(py, der_bytes)
520    }
521
522    /// Serialize one certificate or a list of certificates to PEM format.
523    ///
524    /// The mirror of :meth:`from_pem`: accepts either a single
525    /// :class:`Certificate` or a :class:`list` of them, and returns a
526    /// :class:`bytes` value containing one or more
527    /// ``-----BEGIN CERTIFICATE-----`` blocks concatenated in order.
528    ///
529    /// ```python
530    /// # Single certificate:
531    /// pem = Certificate.to_pem(cert)
532    /// open("cert.pem", "wb").write(pem)
533    ///
534    /// # Certificate chain:
535    /// pem = Certificate.to_pem([leaf, intermediate, root])
536    /// open("chain.pem", "wb").write(pem)
537    /// ```
538    #[staticmethod]
539    fn to_pem<'py>(
540        py: Python<'py>,
541        obj_or_list: Bound<'_, PyAny>,
542    ) -> PyResult<Bound<'py, PyBytes>> {
543        pyobject_to_pem::<Self, _>(py, "CERTIFICATE", &obj_or_list, |c| c.raw)
544    }
545
546    /// Serial number as a Python int.
547    ///
548    /// Returns a native Python `int` regardless of size (X.509 serials can be
549    /// up to 20 bytes / 160 bits per RFC 5280).  The Python object is created
550    /// once and cached; subsequent accesses return a reference to the same object.
551    #[getter]
552    fn serial_number<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
553        // `get_or_try_init` is not stable on `OnceLock`, so build the value
554        // outside the lock if the cache is empty, then store it.
555        if let Some(cached) = self.serial_number_cache.get() {
556            return Ok(cached.clone_ref(py).into_bound(py));
557        }
558        let serial = &self.cert()?.tbs_certificate.serial_number;
559        let py_int: Py<PyAny> = if let Ok(v) = serial.as_i64() {
560            // Fast path: fits in i64 (covers traditional short serials)
561            v.into_pyobject(py)?.into_any().unbind()
562        } else if let Ok(v) = serial.as_i128() {
563            // Fits in i128 (covers up to 16-byte random serials)
564            v.into_pyobject(py)?.into_any().unbind()
565        } else {
566            // Large serial (17–20 bytes): call int.from_bytes() directly.
567            // Using call_method avoids the parse+compile overhead of py.eval()
568            // on every cold-path invocation.  `signed` is keyword-only in
569            // Python's int.from_bytes signature, so it must go in kwargs.
570            let bytes_obj = PyBytes::new(py, serial.as_bytes());
571            let kwargs = pyo3::types::PyDict::new(py);
572            kwargs.set_item(pyo3::intern!(py, "signed"), true)?;
573            py.get_type::<pyo3::types::PyInt>()
574                .call_method(
575                    pyo3::intern!(py, "from_bytes"),
576                    (bytes_obj, pyo3::intern!(py, "big")),
577                    Some(&kwargs),
578                )
579                .map(|r| r.unbind())?
580        };
581        // Ignore a racing writer; both produce the same value.
582        let _ = self.serial_number_cache.set(py_int.clone_ref(py));
583        Ok(py_int.into_bound(py))
584    }
585
586    /// Issuer distinguished name as a string.
587    #[getter]
588    fn issuer<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
589        if let Some(cached) = self.issuer_cache.get() {
590            return Ok(cached.clone_ref(py).into_bound(py));
591        }
592        let s = decode_name_debug(self.cert()?.tbs_certificate.issuer.as_bytes());
593        let py_str = PyString::new(py, &s).unbind();
594        let _ = self.issuer_cache.set(py_str.clone_ref(py));
595        Ok(py_str.into_bound(py))
596    }
597
598    /// Subject distinguished name as a string.
599    #[getter]
600    fn subject<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
601        if let Some(cached) = self.subject_cache.get() {
602            return Ok(cached.clone_ref(py).into_bound(py));
603        }
604        let s = decode_name_debug(self.cert()?.tbs_certificate.subject.as_bytes());
605        let py_str = PyString::new(py, &s).unbind();
606        let _ = self.subject_cache.set(py_str.clone_ref(py));
607        Ok(py_str.into_bound(py))
608    }
609
610    /// Signature algorithm name (e.g. "RSA", "ECDSA", "Ed25519", "ML-DSA-65"),
611    /// or the dotted OID notation for unrecognized algorithms.
612    #[getter]
613    fn signature_algorithm<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
614        if let Some(cached) = self.signature_algorithm_cache.get() {
615            return Ok(cached.clone_ref(py).into_bound(py));
616        }
617        let oid = &self.cert()?.signature_algorithm.algorithm;
618        let name = synta_certificate::identify_signature_algorithm(oid);
619        let s = if name != "Other" {
620            name.to_string()
621        } else {
622            oid.to_string()
623        };
624        let py_str = PyString::new(py, &s).unbind();
625        let _ = self.signature_algorithm_cache.set(py_str.clone_ref(py));
626        Ok(py_str.into_bound(py))
627    }
628
629    /// Raw signature bytes.
630    ///
631    /// The `bytes` object is created once on first access and the same Python
632    /// object is returned (by reference) on every subsequent call,
633    /// avoiding repeated allocation and PyO3 string construction.
634    #[getter]
635    fn signature_value<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
636        if let Some(cached) = self.signature_value_cache.get() {
637            return Ok(cached.clone_ref(py).into_bound(py));
638        }
639        let py_bytes = PyBytes::new(py, self.cert()?.signature_value.as_bytes()).unbind();
640        let _ = self.signature_value_cache.set(py_bytes.clone_ref(py));
641        Ok(py_bytes.into_bound(py))
642    }
643
644    /// notBefore time as a string.
645    #[getter]
646    fn not_before<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
647        if let Some(cached) = self.not_before_cache.get() {
648            return Ok(cached.clone_ref(py).into_bound(py));
649        }
650        let s = match &self.cert()?.tbs_certificate.validity.not_before {
651            Time::UtcTime(t) => t.to_string(),
652            Time::GeneralTime(t) => t.to_string(),
653        };
654        let py_str = PyString::new(py, &s).unbind();
655        let _ = self.not_before_cache.set(py_str.clone_ref(py));
656        Ok(py_str.into_bound(py))
657    }
658
659    /// notAfter time as a string.
660    #[getter]
661    fn not_after<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
662        if let Some(cached) = self.not_after_cache.get() {
663            return Ok(cached.clone_ref(py).into_bound(py));
664        }
665        let s = match &self.cert()?.tbs_certificate.validity.not_after {
666            Time::UtcTime(t) => t.to_string(),
667            Time::GeneralTime(t) => t.to_string(),
668        };
669        let py_str = PyString::new(py, &s).unbind();
670        let _ = self.not_after_cache.set(py_str.clone_ref(py));
671        Ok(py_str.into_bound(py))
672    }
673
674    /// Subject public-key algorithm name (e.g. "RSA", "ECDSA", "Ed25519", "ML-DSA-65"),
675    /// or the dotted OID notation for unrecognized algorithms.
676    #[getter]
677    fn public_key_algorithm<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
678        if let Some(cached) = self.public_key_algorithm_cache.get() {
679            return Ok(cached.clone_ref(py).into_bound(py));
680        }
681        let oid = &self
682            .cert()?
683            .tbs_certificate
684            .subject_public_key_info
685            .algorithm
686            .algorithm;
687        let s = synta_certificate::identify_public_key_algorithm(oid)
688            .map(|s| s.to_string())
689            .unwrap_or_else(|| oid.to_string());
690        let py_str = PyString::new(py, &s).unbind();
691        let _ = self.public_key_algorithm_cache.set(py_str.clone_ref(py));
692        Ok(py_str.into_bound(py))
693    }
694
695    /// Raw subject public-key bytes.
696    ///
697    /// The `bytes` object is created once on first access and the same Python
698    /// object is returned (by reference) on every subsequent call,
699    /// avoiding repeated allocation and PyO3 string construction.
700    #[getter]
701    fn public_key<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
702        if let Some(cached) = self.public_key_cache.get() {
703            return Ok(cached.clone_ref(py).into_bound(py));
704        }
705        let py_bytes = PyBytes::new(
706            py,
707            self.cert()?
708                .tbs_certificate
709                .subject_public_key_info
710                .subject_public_key
711                .as_bytes(),
712        )
713        .unbind();
714        let _ = self.public_key_cache.set(py_bytes.clone_ref(py));
715        Ok(py_bytes.into_bound(py))
716    }
717
718    /// Version field (0 = v1, 1 = v2, 2 = v3), or None if absent.
719    #[getter]
720    fn version(&self) -> PyResult<Option<i64>> {
721        Ok(self
722            .cert()?
723            .tbs_certificate
724            .version
725            .as_ref()
726            .and_then(|v| v.as_i64().ok()))
727    }
728
729    /// Complete DER encoding of this certificate (the original bytes passed to
730    /// ``from_der``).
731    fn to_der<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
732        self._data.clone_ref(py).into_bound(py)
733    }
734
735    /// Raw DER bytes of the TBSCertificate structure (the bytes that were
736    /// signed by the issuer's key).
737    #[getter]
738    fn tbs_bytes<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
739        self.tbs_bytes_cache
740            .get_or_init(|| {
741                let data = self._data.bind(py).as_bytes();
742                PyBytes::new(py, &data[self.tbs_range.clone()]).unbind()
743            })
744            .clone_ref(py)
745            .into_bound(py)
746    }
747
748    /// OID of the signature algorithm
749    /// (e.g. ``ObjectIdentifier("1.2.840.113549.1.1.11")`` for sha256WithRSAEncryption).
750    ///
751    /// Unlike ``signature_algorithm``, this always returns the machine-readable
752    /// OID, even for algorithms that synta does not recognise by name.
753    #[getter]
754    fn signature_algorithm_oid<'py>(
755        &self,
756        py: Python<'py>,
757    ) -> PyResult<Bound<'py, PyObjectIdentifier>> {
758        if let Some(cached) = self.signature_algorithm_oid_cache.get() {
759            return Ok(cached.clone_ref(py).into_bound(py));
760        }
761        let obj = Py::new(
762            py,
763            PyObjectIdentifier::from_oid(self.cert()?.signature_algorithm.algorithm.clone()),
764        )?;
765        let _ = self.signature_algorithm_oid_cache.set(obj.clone_ref(py));
766        Ok(obj.into_bound(py))
767    }
768
769    /// Raw DER bytes of the signature algorithm parameters, or ``None`` if
770    /// the AlgorithmIdentifier has no parameters field (e.g. Ed25519).
771    #[getter]
772    fn signature_algorithm_params<'py>(
773        &self,
774        py: Python<'py>,
775    ) -> PyResult<Option<Bound<'py, PyBytes>>> {
776        if let Some(cached) = self.signature_algorithm_params_cache.get() {
777            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
778        }
779        let computed =
780            encode_element_opt(py, self.cert()?.signature_algorithm.parameters.as_ref())?;
781        let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
782        let _ = self.signature_algorithm_params_cache.set(to_store);
783        Ok(computed)
784    }
785
786    /// DER-encoded ``AlgorithmIdentifier`` SEQUENCE from the certificate's outer
787    /// ``signatureAlgorithm`` field.
788    ///
789    /// This is the raw DER form of the field — suitable for passing directly to
790    /// :meth:`PublicKey.verify_certificate_signature` without any re-encoding.
791    ///
792    /// ```python,ignore
793    /// pub = synta.PublicKey.from_der(issuer_cert.public_key)
794    /// pub.verify_certificate_signature(
795    ///     cert.tbs_certificate_der,
796    ///     cert.signature_algorithm_der,
797    ///     cert.signature_value,
798    /// )
799    /// ```
800    #[getter]
801    fn signature_algorithm_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
802        if let Some(cached) = self.signature_algorithm_der_cache.get() {
803            return Ok(cached.clone_ref(py).into_bound(py));
804        }
805        let ranges = synta_certificate::cert_byte_ranges(self.raw).ok_or_else(|| {
806            pyo3::exceptions::PyValueError::new_err("certificate DER structure is invalid")
807        })?;
808        let py_bytes = PyBytes::new(py, &self.raw[ranges.signature_algorithm]).unbind();
809        let _ = self
810            .signature_algorithm_der_cache
811            .set(py_bytes.clone_ref(py));
812        Ok(py_bytes.into_bound(py))
813    }
814
815    /// OID of the subject public-key algorithm
816    /// (e.g. ``ObjectIdentifier("1.2.840.10045.2.1")`` for id-ecPublicKey).
817    #[getter]
818    fn public_key_algorithm_oid<'py>(
819        &self,
820        py: Python<'py>,
821    ) -> PyResult<Bound<'py, PyObjectIdentifier>> {
822        if let Some(cached) = self.public_key_algorithm_oid_cache.get() {
823            return Ok(cached.clone_ref(py).into_bound(py));
824        }
825        let obj = Py::new(
826            py,
827            PyObjectIdentifier::from_oid(
828                self.cert()?
829                    .tbs_certificate
830                    .subject_public_key_info
831                    .algorithm
832                    .algorithm
833                    .clone(),
834            ),
835        )?;
836        let _ = self.public_key_algorithm_oid_cache.set(obj.clone_ref(py));
837        Ok(obj.into_bound(py))
838    }
839
840    /// Raw DER bytes of the public-key algorithm parameters, or ``None``.
841    ///
842    /// For EC keys this is an OID naming the curve (e.g. secp256r1).
843    /// For RSA and Ed25519 this is typically ``None`` or a NULL element.
844    #[getter]
845    fn public_key_algorithm_params<'py>(
846        &self,
847        py: Python<'py>,
848    ) -> PyResult<Option<Bound<'py, PyBytes>>> {
849        if let Some(cached) = self.public_key_algorithm_params_cache.get() {
850            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
851        }
852        let computed = encode_element_opt(
853            py,
854            self.cert()?
855                .tbs_certificate
856                .subject_public_key_info
857                .algorithm
858                .parameters
859                .as_ref(),
860        )?;
861        let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
862        let _ = self.public_key_algorithm_params_cache.set(to_store);
863        Ok(computed)
864    }
865
866    /// Raw DER bytes of the extensions SEQUENCE OF, or ``None`` for v1/v2
867    /// certificates that carry no extensions.
868    ///
869    /// The returned bytes begin with the SEQUENCE tag (``0x30``) and contain
870    /// the full SEQUENCE OF Extension encoding.  Pass them to a ``Decoder``
871    /// to iterate the individual extensions:
872    ///
873    /// ```python
874    /// ext_der = cert.extensions_der
875    /// if ext_der:
876    ///     dec = synta.Decoder(ext_der, synta.Encoding.DER)
877    ///     exts_dec = dec.decode_sequence()
878    ///     while not exts_dec.is_empty():
879    ///         ext_tlv = exts_dec.decode_raw_tlv()
880    /// ```
881    #[getter]
882    fn extensions_der<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
883        if let Some(cached) = self.extensions_der_cache.get() {
884            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
885        }
886        let computed = self
887            .cert()?
888            .tbs_certificate
889            .extensions
890            .as_ref()
891            .map(|raw: &synta::RawDer<'_>| PyBytes::new(py, raw.as_bytes()));
892        let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
893        let _ = self.extensions_der_cache.set(to_store);
894        Ok(computed)
895    }
896
897    /// Return the DER content of a named extension's value, or ``None``.
898    ///
899    /// Searches the certificate's extension list for an extension whose
900    /// ``extnID`` equals *oid* (dotted-decimal notation, e.g.
901    /// ``"2.5.29.17"`` for SubjectAltName).  When found, the bytes inside
902    /// the ``extnValue`` OCTET STRING — the DER-encoded extension-specific
903    /// structure — are returned.  Returns ``None`` when the certificate
904    /// carries no extensions or the named OID is absent.
905    ///
906    /// Example — parse SubjectAltName:
907    ///
908    /// ```python
909    ///     san_der = cert.get_extension_value_der("2.5.29.17")
910    ///     if san_der:
911    ///         dec = synta.Decoder(san_der, synta.Encoding.DER)
912    ///         san_seq = dec.decode_sequence()
913    ///         while not san_seq.is_empty():
914    ///             tag_num, tag_class, _ = san_seq.peek_tag()
915    ///             child = san_seq.decode_implicit_tag(tag_num, tag_class)
916    ///             if tag_num == 2:  # dNSName
917    ///                 print(child.remaining_bytes().decode("ascii"))
918    /// ```
919    fn get_extension_value_der<'py>(
920        &self,
921        py: Python<'py>,
922        oid: &Bound<'_, PyAny>,
923    ) -> PyResult<Option<Bound<'py, PyBytes>>> {
924        let target = oid_from_pyany(oid)?;
925        let raw = match self.cert()?.tbs_certificate.extensions.as_ref() {
926            Some(r) => r,
927            None => return Ok(None),
928        };
929        for ext in &synta_certificate::decode_extensions(raw.as_bytes()) {
930            if ext.extn_id == target {
931                return Ok(Some(PyBytes::new(py, ext.extn_value.as_bytes())));
932            }
933        }
934        Ok(None)
935    }
936
937    /// Return the Subject Alternative Names of this certificate as typed objects.
938    ///
939    /// Combines looking up the SAN extension (OID ``2.5.29.17``) and parsing
940    /// its ``GeneralName`` entries into a single call.  Returns a
941    /// :class:`list` of typed :mod:`synta.general_name` objects in document
942    /// order; returns an empty list when the certificate carries no SAN
943    /// extension.
944    ///
945    /// Each element is one of:
946    ///
947    /// - :class:`~synta.general_name.OtherName` — tag 0
948    /// - :class:`~synta.general_name.RFC822Name` — tag 1 (e-mail address)
949    /// - :class:`~synta.general_name.DNSName` — tag 2
950    /// - :class:`~synta.general_name.X400Address` — tag 3 (raw DER)
951    /// - :class:`~synta.general_name.DirectoryName` — tag 4
952    /// - :class:`~synta.general_name.EDIPartyName` — tag 5 (raw DER)
953    /// - :class:`~synta.general_name.UniformResourceIdentifier` — tag 6
954    /// - :class:`~synta.general_name.IPAddress` — tag 7
955    /// - :class:`~synta.general_name.RegisteredID` — tag 8
956    ///
957    /// ```python,ignore
958    /// import ipaddress
959    /// import synta.general_name as gn
960    ///
961    /// for name in cert.subject_alt_names():
962    ///     if isinstance(name, gn.DNSName):
963    ///         print("DNS:", name.value)
964    ///     elif isinstance(name, gn.IPAddress):
965    ///         print("IP:", ipaddress.ip_address(name.address))
966    ///     elif isinstance(name, gn.RFC822Name):
967    ///         print("email:", name.value)
968    ///     elif isinstance(name, gn.DirectoryName):
969    ///         attrs = synta.parse_name_attrs(name.name_der)
970    ///         print("DirName:", attrs)
971    ///     elif isinstance(name, gn.UniformResourceIdentifier):
972    ///         print("URI:", name.value)
973    /// ```
974    fn subject_alt_names<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
975        use super::general_name::decode_general_names_to_py;
976        use synta_certificate::find_extension_value;
977
978        let cert = self.cert()?;
979        let raw = match cert.tbs_certificate.extensions.as_ref() {
980            Some(r) => r,
981            None => return Ok(PyList::empty(py)),
982        };
983        let san_bytes =
984            match find_extension_value(raw.as_bytes(), synta_certificate::oids::SUBJECT_ALT_NAME) {
985                Some(b) => b,
986                None => return Ok(PyList::empty(py)),
987            };
988        decode_general_names_to_py(py, san_bytes)
989    }
990
991    /// Return the GeneralNames from any extension that encodes a
992    /// ``SEQUENCE OF GeneralName``, identified by OID.
993    ///
994    /// Looks up the extension with the given *oid* and decodes its value as a
995    /// ``SEQUENCE OF GeneralName``.  Suitable for SAN (``2.5.29.17``), IAN
996    /// (``2.5.29.18``), or any private extension that uses the same format.
997    ///
998    /// Returns a :class:`list` of typed :mod:`synta.general_name` objects, or
999    /// an empty list when the extension is absent or cannot be decoded.
1000    ///
1001    /// ```python,ignore
1002    /// import synta
1003    /// import synta.general_name as gn
1004    ///
1005    /// cert = synta.Certificate.from_der(der)
1006    /// # Decode the Issuer Alternative Names extension:
1007    /// for name in cert.general_names("2.5.29.18"):
1008    ///     if isinstance(name, gn.DirectoryName):
1009    ///         print(synta.parse_name_attrs(name.name_der))
1010    /// ```
1011    fn general_names<'py>(
1012        &self,
1013        py: Python<'py>,
1014        oid: &Bound<'_, PyAny>,
1015    ) -> PyResult<Bound<'py, PyList>> {
1016        use super::general_name::decode_general_names_to_py;
1017        use super::oid_from_pyany;
1018        use synta_certificate::find_extension_value;
1019
1020        let target = oid_from_pyany(oid)?;
1021        let cert = self.cert()?;
1022        let raw = match cert.tbs_certificate.extensions.as_ref() {
1023            Some(r) => r,
1024            None => return Ok(PyList::empty(py)),
1025        };
1026        let ext_bytes = match find_extension_value(raw.as_bytes(), target.components()) {
1027            Some(b) => b,
1028            None => return Ok(PyList::empty(py)),
1029        };
1030        decode_general_names_to_py(py, ext_bytes)
1031    }
1032
1033    /// Return the CertificatePolicies entries from the extension (OID ``2.5.29.32``).
1034    ///
1035    /// Decodes the ``CertificatePolicies`` extension value
1036    /// (``SEQUENCE OF PolicyInformation``) and returns a list of
1037    /// :class:`PolicyInformation` objects, each containing the policy OID and
1038    /// any ``policyQualifiers``.
1039    ///
1040    /// Returns an empty list when the extension is absent or cannot be parsed.
1041    ///
1042    /// ```python,ignore
1043    /// import synta
1044    /// cert = synta.Certificate.from_der(der)
1045    /// for pi in cert.certificate_policies():
1046    ///     print(str(pi.policy_oid))   # e.g. "1.3.6.1.4.1.311.21.8...."
1047    ///     for q in pi.qualifiers:
1048    ///         print(str(q.qualifier_oid))
1049    /// ```
1050    fn certificate_policies<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
1051        let list = PyList::empty(py);
1052        let cert = self.cert()?;
1053        let ext_raw = match cert.tbs_certificate.extensions.as_ref() {
1054            Some(r) => r,
1055            None => return Ok(list),
1056        };
1057        let ext_value_bytes = synta_certificate::find_extension_value(
1058            ext_raw.as_bytes(),
1059            synta_certificate::oids::CERTIFICATE_POLICIES,
1060        );
1061        let ext_bytes = match ext_value_bytes {
1062            Some(b) => b,
1063            None => return Ok(list),
1064        };
1065        // Decode using the code-generated PolicyInformation<'a> type.
1066        let mut decoder = synta::Decoder::new(ext_bytes, synta::Encoding::Der);
1067        let infos: Vec<synta_certificate::PolicyInformation<'_>> = match decoder.decode() {
1068            Ok(v) => v,
1069            Err(_) => return Ok(list),
1070        };
1071        for pi in infos {
1072            let qualifiers = pi
1073                .policy_qualifiers
1074                .unwrap_or_default()
1075                .into_iter()
1076                .filter_map(|qi: PolicyQualifierInfo<'_>| {
1077                    let mut enc = synta::Encoder::new(synta::Encoding::Der);
1078                    qi.qualifier.encode(&mut enc).ok()?;
1079                    let qval = enc.finish().ok()?;
1080                    Some((qi.policy_qualifier_id, qval))
1081                })
1082                .collect();
1083            let py_pi = Py::new(
1084                py,
1085                PyPolicyInformation {
1086                    policy_oid: pi.policy_identifier,
1087                    qualifiers,
1088                },
1089            )?;
1090            list.append(py_pi.into_bound(py))?;
1091        }
1092        Ok(list)
1093    }
1094
1095    /// Raw DER bytes of the issuer Name SEQUENCE.
1096    #[getter]
1097    fn issuer_raw_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
1098        if let Some(cached) = self.issuer_raw_der_cache.get() {
1099            return Ok(cached.clone_ref(py).into_bound(py));
1100        }
1101        let py_bytes = PyBytes::new(py, self.cert()?.tbs_certificate.issuer.as_bytes()).unbind();
1102        let _ = self.issuer_raw_der_cache.set(py_bytes.clone_ref(py));
1103        Ok(py_bytes.into_bound(py))
1104    }
1105
1106    /// Raw DER bytes of the subject Name SEQUENCE.
1107    #[getter]
1108    fn subject_raw_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
1109        if let Some(cached) = self.subject_raw_der_cache.get() {
1110            return Ok(cached.clone_ref(py).into_bound(py));
1111        }
1112        let py_bytes = PyBytes::new(py, self.cert()?.tbs_certificate.subject.as_bytes()).unbind();
1113        let _ = self.subject_raw_der_cache.set(py_bytes.clone_ref(py));
1114        Ok(py_bytes.into_bound(py))
1115    }
1116
1117    /// notBefore time as a UTC-aware ``datetime.datetime``.
1118    ///
1119    /// Equivalent to ``cert.not_valid_before.replace(tzinfo=datetime.timezone.utc)``
1120    /// from the ``cryptography`` library.  Derives the value from the parsed
1121    /// ASN.1 ``Time`` field (``UTCTime`` or ``GeneralizedTime``) without calling
1122    /// any Python date-parsing code.
1123    ///
1124    /// ```python,ignore
1125    /// import datetime
1126    /// cert = synta.Certificate.from_der(der)
1127    /// dt = cert.not_before_utc
1128    /// print(dt.isoformat())       # e.g. "2023-01-01T00:00:00+00:00"
1129    /// print(dt.tzinfo)            # datetime.timezone.utc
1130    /// ```
1131    #[getter]
1132    fn not_before_utc<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
1133        if let Some(cached) = self.not_before_utc_cache.get() {
1134            return Ok(cached.clone_ref(py).into_bound(py));
1135        }
1136        let unix_secs = synta_x509_verification::certificate::time_to_unix(
1137            &self.cert()?.tbs_certificate.validity.not_before,
1138        )
1139        .ok_or_else(|| {
1140            pyo3::exceptions::PyValueError::new_err("notBefore date is structurally invalid")
1141        })?;
1142        let datetime_mod = py.import("datetime")?;
1143        let utc = datetime_mod.getattr("timezone")?.getattr("utc")?;
1144        let dt = datetime_mod
1145            .getattr("datetime")?
1146            .call_method1("fromtimestamp", (unix_secs, &utc))?
1147            .unbind();
1148        let _ = self.not_before_utc_cache.set(dt.clone_ref(py));
1149        Ok(dt.into_bound(py))
1150    }
1151
1152    /// notAfter time as a UTC-aware ``datetime.datetime``.
1153    ///
1154    /// Equivalent to ``cert.not_valid_after.replace(tzinfo=datetime.timezone.utc)``
1155    /// from the ``cryptography`` library.  Derives the value from the parsed
1156    /// ASN.1 ``Time`` field (``UTCTime`` or ``GeneralizedTime``) without calling
1157    /// any Python date-parsing code.
1158    ///
1159    /// ```python,ignore
1160    /// import datetime
1161    /// cert = synta.Certificate.from_der(der)
1162    /// dt = cert.not_after_utc
1163    /// print(dt.isoformat())       # e.g. "2033-01-01T00:00:00+00:00"
1164    /// print(dt.tzinfo)            # datetime.timezone.utc
1165    /// ```
1166    #[getter]
1167    fn not_after_utc<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
1168        if let Some(cached) = self.not_after_utc_cache.get() {
1169            return Ok(cached.clone_ref(py).into_bound(py));
1170        }
1171        let unix_secs = synta_x509_verification::certificate::time_to_unix(
1172            &self.cert()?.tbs_certificate.validity.not_after,
1173        )
1174        .ok_or_else(|| {
1175            pyo3::exceptions::PyValueError::new_err("notAfter date is structurally invalid")
1176        })?;
1177        let datetime_mod = py.import("datetime")?;
1178        let utc = datetime_mod.getattr("timezone")?.getattr("utc")?;
1179        let dt = datetime_mod
1180            .getattr("datetime")?
1181            .call_method1("fromtimestamp", (unix_secs, &utc))?
1182            .unbind();
1183        let _ = self.not_after_utc_cache.set(dt.clone_ref(py));
1184        Ok(dt.into_bound(py))
1185    }
1186
1187    /// Hash algorithm name component of the certificate's outer signature algorithm.
1188    ///
1189    /// Returns the lowercase hash algorithm name (e.g. ``"sha256"``, ``"sha1"``,
1190    /// ``"sha384"``) implied by the signature algorithm OID.  Returns ``None``
1191    /// for algorithms that do not use a traditional hash (Ed25519, Ed448,
1192    /// ML-DSA, etc.).
1193    ///
1194    /// For RSASSA-PSS the hash is encoded in the parameters field, which is not
1195    /// inspected here; ``"sha256"`` is returned as a conservative default.
1196    ///
1197    /// ```python,ignore
1198    /// cert = synta.Certificate.from_der(der)
1199    /// hash_name = cert.signature_hash_algorithm_name
1200    /// if hash_name:
1201    ///     print(hash_name)    # e.g. "sha256"
1202    /// else:
1203    ///     print("no hash (e.g. Ed25519)")
1204    /// ```
1205    #[getter]
1206    fn signature_hash_algorithm_name(&self) -> PyResult<Option<&'static str>> {
1207        use synta_certificate::oids;
1208        let oid = &self.cert()?.signature_algorithm.algorithm;
1209        let c = oid.components();
1210        // RSA PKCS#1 v1.5 variants (1.2.840.113549.1.1.*)
1211        if c == oids::SHA1_WITH_RSA {
1212            return Ok(Some("sha1"));
1213        }
1214        if c == oids::SHA256_WITH_RSA {
1215            return Ok(Some("sha256"));
1216        }
1217        if c == oids::SHA384_WITH_RSA {
1218            return Ok(Some("sha384"));
1219        }
1220        if c == oids::SHA512_WITH_RSA {
1221            return Ok(Some("sha512"));
1222        }
1223        // RSA-PSS (hash is in params; return sha256 as safe default)
1224        if c == oids::RSASSA_PSS {
1225            return Ok(Some("sha256"));
1226        }
1227        // ECDSA variants
1228        if c == oids::ECDSA_WITH_SHA1 {
1229            return Ok(Some("sha1"));
1230        }
1231        if c == oids::ECDSA_WITH_SHA256 {
1232            return Ok(Some("sha256"));
1233        }
1234        if c == oids::ECDSA_WITH_SHA384 {
1235            return Ok(Some("sha384"));
1236        }
1237        if c == oids::ECDSA_WITH_SHA512 {
1238            return Ok(Some("sha512"));
1239        }
1240        // Ed25519, Ed448, ML-DSA-*, SLH-DSA-* and everything else: no hash
1241        Ok(None)
1242    }
1243
1244    /// Complete DER encoding of the SubjectPublicKeyInfo SEQUENCE.
1245    ///
1246    /// Returns the full ``SubjectPublicKeyInfo`` SEQUENCE TLV as ``bytes``,
1247    /// including the algorithm identifier and the BIT STRING carrying the
1248    /// actual public key.  This is the DER encoding that
1249    /// ``cryptography``'s ``public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo)``
1250    /// produces.
1251    ///
1252    /// ```python,ignore
1253    /// cert = synta.Certificate.from_der(der)
1254    /// spki_der = cert.subject_public_key_info_der
1255    /// # Load the public key with cryptography:
1256    /// from cryptography.hazmat.primitives.serialization import load_der_public_key
1257    /// pub_key = load_der_public_key(spki_der)
1258    /// ```
1259    #[getter]
1260    fn subject_public_key_info_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
1261        if let Some(cached) = self.spki_der_cache.get() {
1262            return Ok(cached.clone_ref(py).into_bound(py));
1263        }
1264        let spki = &self.cert()?.tbs_certificate.subject_public_key_info;
1265        let mut enc = synta::Encoder::new(synta::Encoding::Der);
1266        spki.encode(&mut enc)
1267            .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
1268        let bytes = enc
1269            .finish()
1270            .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
1271        let py_bytes = PyBytes::new(py, &bytes).unbind();
1272        let _ = self.spki_der_cache.set(py_bytes.clone_ref(py));
1273        Ok(py_bytes.into_bound(py))
1274    }
1275
1276    /// Raw DER bytes of the ``TBSCertificate`` structure — the bytes that
1277    /// were signed by the issuer's key.
1278    ///
1279    /// This is the DER encoding of the full ``TBSCertificate`` SEQUENCE TLV,
1280    /// suitable for signature verification or as the input to a hash function.
1281    /// Equivalent to ``cryptography``'s ``cert.tbs_certificate_bytes``.
1282    ///
1283    /// ```python,ignore
1284    /// cert = synta.Certificate.from_der(der)
1285    /// tbs = cert.tbs_certificate_der
1286    /// # Verify signature manually:
1287    /// import synta.crypto
1288    /// digest = synta.digest("sha256", tbs)
1289    /// ```
1290    #[getter]
1291    fn tbs_certificate_der<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
1292        self.tbs_bytes(py)
1293    }
1294
1295    /// Verify that this certificate was directly issued by ``issuer``.
1296    ///
1297    /// Checks two things:
1298    ///
1299    /// 1. The ``issuer`` field of this certificate's TBS matches the
1300    ///    ``subject`` field of the provided issuer certificate (byte-exact
1301    ///    DER comparison).
1302    /// 2. The certificate's signature is valid under the issuer's public key.
1303    ///
1304    /// Raises :exc:`ValueError` if either check fails.  This is the synta
1305    /// equivalent of ``cryptography``'s
1306    /// ``cert.verify_directly_issued_by(issuer)``.
1307    ///
1308    /// ```python,ignore
1309    /// root = synta.Certificate.from_pem(open("root.pem", "rb").read())
1310    /// leaf = synta.Certificate.from_pem(open("leaf.pem", "rb").read())
1311    /// leaf.verify_issued_by(root)   # raises ValueError if not valid
1312    /// ```
1313    fn verify_issued_by(&self, issuer: &PyCertificate) -> PyResult<()> {
1314        use synta_certificate::{cert_byte_ranges, default_signature_verifier, SignatureVerifier};
1315
1316        // 1. Name check: our issuer Name must byte-equal their subject Name.
1317        let cert = self.cert()?;
1318        let issuer_cert = issuer.cert()?;
1319        if cert.tbs_certificate.issuer.as_bytes() != issuer_cert.tbs_certificate.subject.as_bytes()
1320        {
1321            return Err(pyo3::exceptions::PyValueError::new_err(
1322                "issuer name does not match subject of provided certificate",
1323            ));
1324        }
1325
1326        // 2. Locate byte ranges within each cert's DER without re-encoding.
1327        let my_ranges = cert_byte_ranges(self.raw).ok_or_else(|| {
1328            pyo3::exceptions::PyValueError::new_err(
1329                "failed to locate byte ranges in certificate DER",
1330            )
1331        })?;
1332        let issuer_ranges = cert_byte_ranges(issuer.raw).ok_or_else(|| {
1333            pyo3::exceptions::PyValueError::new_err(
1334                "failed to locate byte ranges in issuer certificate DER",
1335            )
1336        })?;
1337
1338        // 3. Signature bits (raw bytes, no unused-bits prefix).
1339        let signature_bits = cert.signature_value.as_bytes();
1340
1341        // 4. Verify using the backend-agnostic verifier.
1342        default_signature_verifier()
1343            .verify_certificate_signature(
1344                &self.raw[my_ranges.tbs],
1345                &self.raw[my_ranges.signature_algorithm],
1346                signature_bits,
1347                &issuer.raw[issuer_ranges.subject_public_key_info],
1348            )
1349            .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
1350
1351        Ok(())
1352    }
1353
1354    /// Compute a hash fingerprint of the complete certificate DER.
1355    ///
1356    /// ``algorithm`` must be one of ``"sha1"``, ``"sha224"``, ``"sha256"``,
1357    /// ``"sha384"``, ``"sha512"``, or ``"md5"``.  Raises :exc:`ValueError` for
1358    /// unknown algorithm names.
1359    ///
1360    /// ```python,ignore
1361    /// cert = synta.Certificate.from_der(der)
1362    /// fp = cert.fingerprint("sha256")
1363    /// print(fp.hex())     # e.g. "3a2b1c..."
1364    /// ```
1365    fn fingerprint<'py>(&self, py: Python<'py>, algorithm: &str) -> PyResult<Bound<'py, PyBytes>> {
1366        use synta_certificate::{default_data_hasher, DataHasher};
1367        let digest = default_data_hasher()
1368            .hash_data(algorithm, self.raw)
1369            .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
1370        Ok(PyBytes::new(py, &digest))
1371    }
1372
1373    fn __repr__(&self) -> PyResult<String> {
1374        let cert = self.cert()?;
1375        let serial = &cert.tbs_certificate.serial_number;
1376        let serial_str = if let Ok(v) = serial.as_i64() {
1377            v.to_string()
1378        } else {
1379            format!("<{} bytes>", serial.as_bytes().len())
1380        };
1381        Ok(format!(
1382            "Certificate(subject={:?}, serial={})",
1383            decode_name_debug(cert.tbs_certificate.subject.as_bytes()),
1384            serial_str,
1385        ))
1386    }
1387}
1388
1389impl PyCertificate {
1390    /// Parse a DER-encoded X.509 certificate (internal helper visible to sibling modules).
1391    ///
1392    /// This is the same logic as the `#[staticmethod] from_der` exposed to Python but
1393    /// accessible from Rust sibling modules via `pub(super)` visibility.
1394    pub(crate) fn new_from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
1395        Self::from_der(py, data)
1396    }
1397}