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_u64() {
560            // Fast path: fits in u64 (covers traditional short serials).
561            // Use unsigned to match RFC 5280 §4.1.2.2 (serials are positive).
562            v.into_pyobject(py)?.into_any().unbind()
563        } else if let Ok(v) = serial.as_u128() {
564            // Fits in u128 after stripping any DER leading 0x00 byte.
565            // Covers 128-bit RSNv3 random serials correctly encoded with a
566            // leading 0x00 (17 bytes total) and 127-bit serials (16 bytes).
567            v.into_pyobject(py)?.into_any().unbind()
568        } else if let Ok(v) = serial.as_i64() {
569            // Fallback for small negative serials (technically invalid per
570            // RFC 5280 but be lenient when decoding).
571            v.into_pyobject(py)?.into_any().unbind()
572        } else if let Ok(v) = serial.as_i128() {
573            // Fallback: 16-byte signed value.
574            v.into_pyobject(py)?.into_any().unbind()
575        } else {
576            // Large serial (17–20 bytes): call int.from_bytes() directly.
577            // Using call_method avoids the parse+compile overhead of py.eval()
578            // on every cold-path invocation.  `signed` is keyword-only in
579            // Python's int.from_bytes signature, so it must go in kwargs.
580            // Use signed=False: RFC 5280 §4.1.2.2 requires positive serials,
581            // and a leading 0x00 in correctly-encoded values is already
582            // stripped before reaching this branch.
583            let bytes_obj = PyBytes::new(py, serial.as_bytes());
584            let kwargs = pyo3::types::PyDict::new(py);
585            kwargs.set_item(pyo3::intern!(py, "signed"), false)?;
586            py.get_type::<pyo3::types::PyInt>()
587                .call_method(
588                    pyo3::intern!(py, "from_bytes"),
589                    (bytes_obj, pyo3::intern!(py, "big")),
590                    Some(&kwargs),
591                )
592                .map(|r| r.unbind())?
593        };
594        // Ignore a racing writer; both produce the same value.
595        let _ = self.serial_number_cache.set(py_int.clone_ref(py));
596        Ok(py_int.into_bound(py))
597    }
598
599    /// Issuer distinguished name as a string.
600    #[getter]
601    fn issuer<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
602        if let Some(cached) = self.issuer_cache.get() {
603            return Ok(cached.clone_ref(py).into_bound(py));
604        }
605        let s = decode_name_debug(self.cert()?.tbs_certificate.issuer.as_bytes());
606        let py_str = PyString::new(py, &s).unbind();
607        let _ = self.issuer_cache.set(py_str.clone_ref(py));
608        Ok(py_str.into_bound(py))
609    }
610
611    /// Subject distinguished name as a string.
612    #[getter]
613    fn subject<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
614        if let Some(cached) = self.subject_cache.get() {
615            return Ok(cached.clone_ref(py).into_bound(py));
616        }
617        let s = decode_name_debug(self.cert()?.tbs_certificate.subject.as_bytes());
618        let py_str = PyString::new(py, &s).unbind();
619        let _ = self.subject_cache.set(py_str.clone_ref(py));
620        Ok(py_str.into_bound(py))
621    }
622
623    /// Signature algorithm name (e.g. "RSA", "ECDSA", "Ed25519", "ML-DSA-65"),
624    /// or the dotted OID notation for unrecognized algorithms.
625    #[getter]
626    fn signature_algorithm<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
627        if let Some(cached) = self.signature_algorithm_cache.get() {
628            return Ok(cached.clone_ref(py).into_bound(py));
629        }
630        let oid = &self.cert()?.signature_algorithm.algorithm;
631        let name = synta_certificate::identify_signature_algorithm(oid);
632        let s = if name != "Other" {
633            name.to_string()
634        } else {
635            oid.to_string()
636        };
637        let py_str = PyString::new(py, &s).unbind();
638        let _ = self.signature_algorithm_cache.set(py_str.clone_ref(py));
639        Ok(py_str.into_bound(py))
640    }
641
642    /// Raw signature bytes.
643    ///
644    /// The `bytes` object is created once on first access and the same Python
645    /// object is returned (by reference) on every subsequent call,
646    /// avoiding repeated allocation and PyO3 string construction.
647    #[getter]
648    fn signature_value<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
649        if let Some(cached) = self.signature_value_cache.get() {
650            return Ok(cached.clone_ref(py).into_bound(py));
651        }
652        let py_bytes = PyBytes::new(py, self.cert()?.signature_value.as_bytes()).unbind();
653        let _ = self.signature_value_cache.set(py_bytes.clone_ref(py));
654        Ok(py_bytes.into_bound(py))
655    }
656
657    /// notBefore time as a string.
658    #[getter]
659    fn not_before<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
660        if let Some(cached) = self.not_before_cache.get() {
661            return Ok(cached.clone_ref(py).into_bound(py));
662        }
663        let s = match &self.cert()?.tbs_certificate.validity.not_before {
664            Time::UtcTime(t) => t.to_string(),
665            Time::GeneralTime(t) => t.to_string(),
666        };
667        let py_str = PyString::new(py, &s).unbind();
668        let _ = self.not_before_cache.set(py_str.clone_ref(py));
669        Ok(py_str.into_bound(py))
670    }
671
672    /// notAfter time as a string.
673    #[getter]
674    fn not_after<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
675        if let Some(cached) = self.not_after_cache.get() {
676            return Ok(cached.clone_ref(py).into_bound(py));
677        }
678        let s = match &self.cert()?.tbs_certificate.validity.not_after {
679            Time::UtcTime(t) => t.to_string(),
680            Time::GeneralTime(t) => t.to_string(),
681        };
682        let py_str = PyString::new(py, &s).unbind();
683        let _ = self.not_after_cache.set(py_str.clone_ref(py));
684        Ok(py_str.into_bound(py))
685    }
686
687    /// Subject public-key algorithm name (e.g. "RSA", "ECDSA", "Ed25519", "ML-DSA-65"),
688    /// or the dotted OID notation for unrecognized algorithms.
689    #[getter]
690    fn public_key_algorithm<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
691        if let Some(cached) = self.public_key_algorithm_cache.get() {
692            return Ok(cached.clone_ref(py).into_bound(py));
693        }
694        let oid = &self
695            .cert()?
696            .tbs_certificate
697            .subject_public_key_info
698            .algorithm
699            .algorithm;
700        let s = synta_certificate::identify_public_key_algorithm(oid)
701            .map(|s| s.to_string())
702            .unwrap_or_else(|| oid.to_string());
703        let py_str = PyString::new(py, &s).unbind();
704        let _ = self.public_key_algorithm_cache.set(py_str.clone_ref(py));
705        Ok(py_str.into_bound(py))
706    }
707
708    /// Raw subject public-key bytes.
709    ///
710    /// The `bytes` object is created once on first access and the same Python
711    /// object is returned (by reference) on every subsequent call,
712    /// avoiding repeated allocation and PyO3 string construction.
713    #[getter]
714    fn public_key<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
715        if let Some(cached) = self.public_key_cache.get() {
716            return Ok(cached.clone_ref(py).into_bound(py));
717        }
718        let py_bytes = PyBytes::new(
719            py,
720            self.cert()?
721                .tbs_certificate
722                .subject_public_key_info
723                .subject_public_key
724                .as_bytes(),
725        )
726        .unbind();
727        let _ = self.public_key_cache.set(py_bytes.clone_ref(py));
728        Ok(py_bytes.into_bound(py))
729    }
730
731    /// Version field (0 = v1, 1 = v2, 2 = v3), or None if absent.
732    #[getter]
733    fn version(&self) -> PyResult<Option<i64>> {
734        Ok(self
735            .cert()?
736            .tbs_certificate
737            .version
738            .as_ref()
739            .and_then(|v| v.as_i64().ok()))
740    }
741
742    /// Complete DER encoding of this certificate (the original bytes passed to
743    /// ``from_der``).
744    fn to_der<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
745        self._data.clone_ref(py).into_bound(py)
746    }
747
748    /// Raw DER bytes of the TBSCertificate structure (the bytes that were
749    /// signed by the issuer's key).
750    #[getter]
751    fn tbs_bytes<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
752        self.tbs_bytes_cache
753            .get_or_init(|| {
754                let data = self._data.bind(py).as_bytes();
755                PyBytes::new(py, &data[self.tbs_range.clone()]).unbind()
756            })
757            .clone_ref(py)
758            .into_bound(py)
759    }
760
761    /// OID of the signature algorithm
762    /// (e.g. ``ObjectIdentifier("1.2.840.113549.1.1.11")`` for sha256WithRSAEncryption).
763    ///
764    /// Unlike ``signature_algorithm``, this always returns the machine-readable
765    /// OID, even for algorithms that synta does not recognise by name.
766    #[getter]
767    fn signature_algorithm_oid<'py>(
768        &self,
769        py: Python<'py>,
770    ) -> PyResult<Bound<'py, PyObjectIdentifier>> {
771        if let Some(cached) = self.signature_algorithm_oid_cache.get() {
772            return Ok(cached.clone_ref(py).into_bound(py));
773        }
774        let obj = Py::new(
775            py,
776            PyObjectIdentifier::from_oid(self.cert()?.signature_algorithm.algorithm.clone()),
777        )?;
778        let _ = self.signature_algorithm_oid_cache.set(obj.clone_ref(py));
779        Ok(obj.into_bound(py))
780    }
781
782    /// Raw DER bytes of the signature algorithm parameters, or ``None`` if
783    /// the AlgorithmIdentifier has no parameters field (e.g. Ed25519).
784    #[getter]
785    fn signature_algorithm_params<'py>(
786        &self,
787        py: Python<'py>,
788    ) -> PyResult<Option<Bound<'py, PyBytes>>> {
789        if let Some(cached) = self.signature_algorithm_params_cache.get() {
790            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
791        }
792        let computed =
793            encode_element_opt(py, self.cert()?.signature_algorithm.parameters.as_ref())?;
794        let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
795        let _ = self.signature_algorithm_params_cache.set(to_store);
796        Ok(computed)
797    }
798
799    /// DER-encoded ``AlgorithmIdentifier`` SEQUENCE from the certificate's outer
800    /// ``signatureAlgorithm`` field.
801    ///
802    /// This is the raw DER form of the field — suitable for passing directly to
803    /// :meth:`PublicKey.verify_certificate_signature` without any re-encoding.
804    ///
805    /// ```python,ignore
806    /// pub = synta.PublicKey.from_der(issuer_cert.public_key)
807    /// pub.verify_certificate_signature(
808    ///     cert.tbs_certificate_der,
809    ///     cert.signature_algorithm_der,
810    ///     cert.signature_value,
811    /// )
812    /// ```
813    #[getter]
814    fn signature_algorithm_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
815        if let Some(cached) = self.signature_algorithm_der_cache.get() {
816            return Ok(cached.clone_ref(py).into_bound(py));
817        }
818        let ranges = synta_certificate::cert_byte_ranges(self.raw).ok_or_else(|| {
819            pyo3::exceptions::PyValueError::new_err("certificate DER structure is invalid")
820        })?;
821        let py_bytes = PyBytes::new(py, &self.raw[ranges.signature_algorithm]).unbind();
822        let _ = self
823            .signature_algorithm_der_cache
824            .set(py_bytes.clone_ref(py));
825        Ok(py_bytes.into_bound(py))
826    }
827
828    /// OID of the subject public-key algorithm
829    /// (e.g. ``ObjectIdentifier("1.2.840.10045.2.1")`` for id-ecPublicKey).
830    #[getter]
831    fn public_key_algorithm_oid<'py>(
832        &self,
833        py: Python<'py>,
834    ) -> PyResult<Bound<'py, PyObjectIdentifier>> {
835        if let Some(cached) = self.public_key_algorithm_oid_cache.get() {
836            return Ok(cached.clone_ref(py).into_bound(py));
837        }
838        let obj = Py::new(
839            py,
840            PyObjectIdentifier::from_oid(
841                self.cert()?
842                    .tbs_certificate
843                    .subject_public_key_info
844                    .algorithm
845                    .algorithm
846                    .clone(),
847            ),
848        )?;
849        let _ = self.public_key_algorithm_oid_cache.set(obj.clone_ref(py));
850        Ok(obj.into_bound(py))
851    }
852
853    /// Raw DER bytes of the public-key algorithm parameters, or ``None``.
854    ///
855    /// For EC keys this is an OID naming the curve (e.g. secp256r1).
856    /// For RSA and Ed25519 this is typically ``None`` or a NULL element.
857    #[getter]
858    fn public_key_algorithm_params<'py>(
859        &self,
860        py: Python<'py>,
861    ) -> PyResult<Option<Bound<'py, PyBytes>>> {
862        if let Some(cached) = self.public_key_algorithm_params_cache.get() {
863            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
864        }
865        let computed = encode_element_opt(
866            py,
867            self.cert()?
868                .tbs_certificate
869                .subject_public_key_info
870                .algorithm
871                .parameters
872                .as_ref(),
873        )?;
874        let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
875        let _ = self.public_key_algorithm_params_cache.set(to_store);
876        Ok(computed)
877    }
878
879    /// Raw DER bytes of the extensions SEQUENCE OF, or ``None`` for v1/v2
880    /// certificates that carry no extensions.
881    ///
882    /// The returned bytes begin with the SEQUENCE tag (``0x30``) and contain
883    /// the full SEQUENCE OF Extension encoding.  Pass them to a ``Decoder``
884    /// to iterate the individual extensions:
885    ///
886    /// ```python
887    /// ext_der = cert.extensions_der
888    /// if ext_der:
889    ///     dec = synta.Decoder(ext_der, synta.Encoding.DER)
890    ///     exts_dec = dec.decode_sequence()
891    ///     while not exts_dec.is_empty():
892    ///         ext_tlv = exts_dec.decode_raw_tlv()
893    /// ```
894    #[getter]
895    fn extensions_der<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
896        if let Some(cached) = self.extensions_der_cache.get() {
897            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
898        }
899        let computed = self
900            .cert()?
901            .tbs_certificate
902            .extensions
903            .as_ref()
904            .map(|raw: &synta::RawDer<'_>| PyBytes::new(py, raw.as_bytes()));
905        let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
906        let _ = self.extensions_der_cache.set(to_store);
907        Ok(computed)
908    }
909
910    /// Return the DER content of a named extension's value, or ``None``.
911    ///
912    /// Searches the certificate's extension list for an extension whose
913    /// ``extnID`` equals *oid* (dotted-decimal notation, e.g.
914    /// ``"2.5.29.17"`` for SubjectAltName).  When found, the bytes inside
915    /// the ``extnValue`` OCTET STRING — the DER-encoded extension-specific
916    /// structure — are returned.  Returns ``None`` when the certificate
917    /// carries no extensions or the named OID is absent.
918    ///
919    /// Example — parse SubjectAltName:
920    ///
921    /// ```python
922    ///     san_der = cert.get_extension_value_der("2.5.29.17")
923    ///     if san_der:
924    ///         dec = synta.Decoder(san_der, synta.Encoding.DER)
925    ///         san_seq = dec.decode_sequence()
926    ///         while not san_seq.is_empty():
927    ///             tag_num, tag_class, _ = san_seq.peek_tag()
928    ///             child = san_seq.decode_implicit_tag(tag_num, tag_class)
929    ///             if tag_num == 2:  # dNSName
930    ///                 print(child.remaining_bytes().decode("ascii"))
931    /// ```
932    fn get_extension_value_der<'py>(
933        &self,
934        py: Python<'py>,
935        oid: &Bound<'_, PyAny>,
936    ) -> PyResult<Option<Bound<'py, PyBytes>>> {
937        let target = oid_from_pyany(oid)?;
938        let raw = match self.cert()?.tbs_certificate.extensions.as_ref() {
939            Some(r) => r,
940            None => return Ok(None),
941        };
942        for ext in &synta_certificate::decode_extensions(raw.as_bytes()) {
943            if ext.extn_id == target {
944                return Ok(Some(PyBytes::new(py, ext.extn_value.as_bytes())));
945            }
946        }
947        Ok(None)
948    }
949
950    /// Return the Subject Alternative Names of this certificate as typed objects.
951    ///
952    /// Combines looking up the SAN extension (OID ``2.5.29.17``) and parsing
953    /// its ``GeneralName`` entries into a single call.  Returns a
954    /// :class:`list` of typed :mod:`synta.general_name` objects in document
955    /// order; returns an empty list when the certificate carries no SAN
956    /// extension.
957    ///
958    /// Each element is one of:
959    ///
960    /// - :class:`~synta.general_name.OtherName` — tag 0
961    /// - :class:`~synta.general_name.RFC822Name` — tag 1 (e-mail address)
962    /// - :class:`~synta.general_name.DNSName` — tag 2
963    /// - :class:`~synta.general_name.X400Address` — tag 3 (raw DER)
964    /// - :class:`~synta.general_name.DirectoryName` — tag 4
965    /// - :class:`~synta.general_name.EDIPartyName` — tag 5 (raw DER)
966    /// - :class:`~synta.general_name.UniformResourceIdentifier` — tag 6
967    /// - :class:`~synta.general_name.IPAddress` — tag 7
968    /// - :class:`~synta.general_name.RegisteredID` — tag 8
969    ///
970    /// ```python,ignore
971    /// import ipaddress
972    /// import synta.general_name as gn
973    ///
974    /// for name in cert.subject_alt_names():
975    ///     if isinstance(name, gn.DNSName):
976    ///         print("DNS:", name.value)
977    ///     elif isinstance(name, gn.IPAddress):
978    ///         print("IP:", ipaddress.ip_address(name.address))
979    ///     elif isinstance(name, gn.RFC822Name):
980    ///         print("email:", name.value)
981    ///     elif isinstance(name, gn.DirectoryName):
982    ///         attrs = synta.parse_name_attrs(name.name_der)
983    ///         print("DirName:", attrs)
984    ///     elif isinstance(name, gn.UniformResourceIdentifier):
985    ///         print("URI:", name.value)
986    /// ```
987    fn subject_alt_names<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
988        use super::general_name::decode_general_names_to_py;
989        use synta_certificate::find_extension_value;
990
991        let cert = self.cert()?;
992        let raw = match cert.tbs_certificate.extensions.as_ref() {
993            Some(r) => r,
994            None => return Ok(PyList::empty(py)),
995        };
996        let san_bytes =
997            match find_extension_value(raw.as_bytes(), synta_certificate::oids::SUBJECT_ALT_NAME) {
998                Some(b) => b,
999                None => return Ok(PyList::empty(py)),
1000            };
1001        decode_general_names_to_py(py, san_bytes)
1002    }
1003
1004    /// Return the GeneralNames from any extension that encodes a
1005    /// ``SEQUENCE OF GeneralName``, identified by OID.
1006    ///
1007    /// Looks up the extension with the given *oid* and decodes its value as a
1008    /// ``SEQUENCE OF GeneralName``.  Suitable for SAN (``2.5.29.17``), IAN
1009    /// (``2.5.29.18``), or any private extension that uses the same format.
1010    ///
1011    /// Returns a :class:`list` of typed :mod:`synta.general_name` objects, or
1012    /// an empty list when the extension is absent or cannot be decoded.
1013    ///
1014    /// ```python,ignore
1015    /// import synta
1016    /// import synta.general_name as gn
1017    ///
1018    /// cert = synta.Certificate.from_der(der)
1019    /// # Decode the Issuer Alternative Names extension:
1020    /// for name in cert.general_names("2.5.29.18"):
1021    ///     if isinstance(name, gn.DirectoryName):
1022    ///         print(synta.parse_name_attrs(name.name_der))
1023    /// ```
1024    fn general_names<'py>(
1025        &self,
1026        py: Python<'py>,
1027        oid: &Bound<'_, PyAny>,
1028    ) -> PyResult<Bound<'py, PyList>> {
1029        use super::general_name::decode_general_names_to_py;
1030        use super::oid_from_pyany;
1031        use synta_certificate::find_extension_value;
1032
1033        let target = oid_from_pyany(oid)?;
1034        let cert = self.cert()?;
1035        let raw = match cert.tbs_certificate.extensions.as_ref() {
1036            Some(r) => r,
1037            None => return Ok(PyList::empty(py)),
1038        };
1039        let ext_bytes = match find_extension_value(raw.as_bytes(), target.components()) {
1040            Some(b) => b,
1041            None => return Ok(PyList::empty(py)),
1042        };
1043        decode_general_names_to_py(py, ext_bytes)
1044    }
1045
1046    /// Return the CertificatePolicies entries from the extension (OID ``2.5.29.32``).
1047    ///
1048    /// Decodes the ``CertificatePolicies`` extension value
1049    /// (``SEQUENCE OF PolicyInformation``) and returns a list of
1050    /// :class:`PolicyInformation` objects, each containing the policy OID and
1051    /// any ``policyQualifiers``.
1052    ///
1053    /// Returns an empty list when the extension is absent or cannot be parsed.
1054    ///
1055    /// ```python,ignore
1056    /// import synta
1057    /// cert = synta.Certificate.from_der(der)
1058    /// for pi in cert.certificate_policies():
1059    ///     print(str(pi.policy_oid))   # e.g. "1.3.6.1.4.1.311.21.8...."
1060    ///     for q in pi.qualifiers:
1061    ///         print(str(q.qualifier_oid))
1062    /// ```
1063    fn certificate_policies<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
1064        let list = PyList::empty(py);
1065        let cert = self.cert()?;
1066        let ext_raw = match cert.tbs_certificate.extensions.as_ref() {
1067            Some(r) => r,
1068            None => return Ok(list),
1069        };
1070        let ext_value_bytes = synta_certificate::find_extension_value(
1071            ext_raw.as_bytes(),
1072            synta_certificate::oids::CERTIFICATE_POLICIES,
1073        );
1074        let ext_bytes = match ext_value_bytes {
1075            Some(b) => b,
1076            None => return Ok(list),
1077        };
1078        // Decode using the code-generated PolicyInformation<'a> type.
1079        let mut decoder = synta::Decoder::new(ext_bytes, synta::Encoding::Der);
1080        let infos: Vec<synta_certificate::PolicyInformation<'_>> = match decoder.decode() {
1081            Ok(v) => v,
1082            Err(_) => return Ok(list),
1083        };
1084        for pi in infos {
1085            let qualifiers = pi
1086                .policy_qualifiers
1087                .unwrap_or_default()
1088                .into_iter()
1089                .filter_map(|qi: PolicyQualifierInfo<'_>| {
1090                    let mut enc = synta::Encoder::new(synta::Encoding::Der);
1091                    qi.qualifier.encode(&mut enc).ok()?;
1092                    let qval = enc.finish().ok()?;
1093                    Some((qi.policy_qualifier_id, qval))
1094                })
1095                .collect();
1096            let py_pi = Py::new(
1097                py,
1098                PyPolicyInformation {
1099                    policy_oid: pi.policy_identifier,
1100                    qualifiers,
1101                },
1102            )?;
1103            list.append(py_pi.into_bound(py))?;
1104        }
1105        Ok(list)
1106    }
1107
1108    /// Raw DER bytes of the issuer Name SEQUENCE.
1109    #[getter]
1110    fn issuer_raw_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
1111        if let Some(cached) = self.issuer_raw_der_cache.get() {
1112            return Ok(cached.clone_ref(py).into_bound(py));
1113        }
1114        let py_bytes = PyBytes::new(py, self.cert()?.tbs_certificate.issuer.as_bytes()).unbind();
1115        let _ = self.issuer_raw_der_cache.set(py_bytes.clone_ref(py));
1116        Ok(py_bytes.into_bound(py))
1117    }
1118
1119    /// Raw DER bytes of the subject Name SEQUENCE.
1120    #[getter]
1121    fn subject_raw_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
1122        if let Some(cached) = self.subject_raw_der_cache.get() {
1123            return Ok(cached.clone_ref(py).into_bound(py));
1124        }
1125        let py_bytes = PyBytes::new(py, self.cert()?.tbs_certificate.subject.as_bytes()).unbind();
1126        let _ = self.subject_raw_der_cache.set(py_bytes.clone_ref(py));
1127        Ok(py_bytes.into_bound(py))
1128    }
1129
1130    /// notBefore time as a UTC-aware ``datetime.datetime``.
1131    ///
1132    /// Equivalent to ``cert.not_valid_before.replace(tzinfo=datetime.timezone.utc)``
1133    /// from the ``cryptography`` library.  Derives the value from the parsed
1134    /// ASN.1 ``Time`` field (``UTCTime`` or ``GeneralizedTime``) without calling
1135    /// any Python date-parsing code.
1136    ///
1137    /// ```python,ignore
1138    /// import datetime
1139    /// cert = synta.Certificate.from_der(der)
1140    /// dt = cert.not_before_utc
1141    /// print(dt.isoformat())       # e.g. "2023-01-01T00:00:00+00:00"
1142    /// print(dt.tzinfo)            # datetime.timezone.utc
1143    /// ```
1144    #[getter]
1145    fn not_before_utc<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
1146        if let Some(cached) = self.not_before_utc_cache.get() {
1147            return Ok(cached.clone_ref(py).into_bound(py));
1148        }
1149        let unix_secs = synta_x509_verification::certificate::time_to_unix(
1150            &self.cert()?.tbs_certificate.validity.not_before,
1151        )
1152        .ok_or_else(|| {
1153            pyo3::exceptions::PyValueError::new_err("notBefore date is structurally invalid")
1154        })?;
1155        let datetime_mod = py.import("datetime")?;
1156        let utc = datetime_mod.getattr("timezone")?.getattr("utc")?;
1157        let dt = datetime_mod
1158            .getattr("datetime")?
1159            .call_method1("fromtimestamp", (unix_secs, &utc))?
1160            .unbind();
1161        let _ = self.not_before_utc_cache.set(dt.clone_ref(py));
1162        Ok(dt.into_bound(py))
1163    }
1164
1165    /// notAfter time as a UTC-aware ``datetime.datetime``.
1166    ///
1167    /// Equivalent to ``cert.not_valid_after.replace(tzinfo=datetime.timezone.utc)``
1168    /// from the ``cryptography`` library.  Derives the value from the parsed
1169    /// ASN.1 ``Time`` field (``UTCTime`` or ``GeneralizedTime``) without calling
1170    /// any Python date-parsing code.
1171    ///
1172    /// ```python,ignore
1173    /// import datetime
1174    /// cert = synta.Certificate.from_der(der)
1175    /// dt = cert.not_after_utc
1176    /// print(dt.isoformat())       # e.g. "2033-01-01T00:00:00+00:00"
1177    /// print(dt.tzinfo)            # datetime.timezone.utc
1178    /// ```
1179    #[getter]
1180    fn not_after_utc<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
1181        if let Some(cached) = self.not_after_utc_cache.get() {
1182            return Ok(cached.clone_ref(py).into_bound(py));
1183        }
1184        let unix_secs = synta_x509_verification::certificate::time_to_unix(
1185            &self.cert()?.tbs_certificate.validity.not_after,
1186        )
1187        .ok_or_else(|| {
1188            pyo3::exceptions::PyValueError::new_err("notAfter date is structurally invalid")
1189        })?;
1190        let datetime_mod = py.import("datetime")?;
1191        let utc = datetime_mod.getattr("timezone")?.getattr("utc")?;
1192        let dt = datetime_mod
1193            .getattr("datetime")?
1194            .call_method1("fromtimestamp", (unix_secs, &utc))?
1195            .unbind();
1196        let _ = self.not_after_utc_cache.set(dt.clone_ref(py));
1197        Ok(dt.into_bound(py))
1198    }
1199
1200    /// Hash algorithm name component of the certificate's outer signature algorithm.
1201    ///
1202    /// Returns the lowercase hash algorithm name (e.g. ``"sha256"``, ``"sha1"``,
1203    /// ``"sha384"``) implied by the signature algorithm OID.  Returns ``None``
1204    /// for algorithms that do not use a traditional hash (Ed25519, Ed448,
1205    /// ML-DSA, etc.).
1206    ///
1207    /// For RSASSA-PSS the hash is encoded in the parameters field, which is not
1208    /// inspected here; ``"sha256"`` is returned as a conservative default.
1209    ///
1210    /// ```python,ignore
1211    /// cert = synta.Certificate.from_der(der)
1212    /// hash_name = cert.signature_hash_algorithm_name
1213    /// if hash_name:
1214    ///     print(hash_name)    # e.g. "sha256"
1215    /// else:
1216    ///     print("no hash (e.g. Ed25519)")
1217    /// ```
1218    #[getter]
1219    fn signature_hash_algorithm_name(&self) -> PyResult<Option<&'static str>> {
1220        use synta_certificate::oids;
1221        let oid = &self.cert()?.signature_algorithm.algorithm;
1222        let c = oid.components();
1223        // RSA PKCS#1 v1.5 variants (1.2.840.113549.1.1.*)
1224        if c == oids::SHA1_WITH_RSA {
1225            return Ok(Some("sha1"));
1226        }
1227        if c == oids::SHA256_WITH_RSA {
1228            return Ok(Some("sha256"));
1229        }
1230        if c == oids::SHA384_WITH_RSA {
1231            return Ok(Some("sha384"));
1232        }
1233        if c == oids::SHA512_WITH_RSA {
1234            return Ok(Some("sha512"));
1235        }
1236        // RSA-PSS (hash is in params; return sha256 as safe default)
1237        if c == oids::RSASSA_PSS {
1238            return Ok(Some("sha256"));
1239        }
1240        // ECDSA variants
1241        if c == oids::ECDSA_WITH_SHA1 {
1242            return Ok(Some("sha1"));
1243        }
1244        if c == oids::ECDSA_WITH_SHA256 {
1245            return Ok(Some("sha256"));
1246        }
1247        if c == oids::ECDSA_WITH_SHA384 {
1248            return Ok(Some("sha384"));
1249        }
1250        if c == oids::ECDSA_WITH_SHA512 {
1251            return Ok(Some("sha512"));
1252        }
1253        // Ed25519, Ed448, ML-DSA-*, SLH-DSA-* and everything else: no hash
1254        Ok(None)
1255    }
1256
1257    /// Complete DER encoding of the SubjectPublicKeyInfo SEQUENCE.
1258    ///
1259    /// Returns the full ``SubjectPublicKeyInfo`` SEQUENCE TLV as ``bytes``,
1260    /// including the algorithm identifier and the BIT STRING carrying the
1261    /// actual public key.  This is the DER encoding that
1262    /// ``cryptography``'s ``public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo)``
1263    /// produces.
1264    ///
1265    /// ```python,ignore
1266    /// cert = synta.Certificate.from_der(der)
1267    /// spki_der = cert.subject_public_key_info_der
1268    /// # Load the public key with cryptography:
1269    /// from cryptography.hazmat.primitives.serialization import load_der_public_key
1270    /// pub_key = load_der_public_key(spki_der)
1271    /// ```
1272    #[getter]
1273    fn subject_public_key_info_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
1274        if let Some(cached) = self.spki_der_cache.get() {
1275            return Ok(cached.clone_ref(py).into_bound(py));
1276        }
1277        let spki = &self.cert()?.tbs_certificate.subject_public_key_info;
1278        let mut enc = synta::Encoder::new(synta::Encoding::Der);
1279        spki.encode(&mut enc)
1280            .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
1281        let bytes = enc
1282            .finish()
1283            .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
1284        let py_bytes = PyBytes::new(py, &bytes).unbind();
1285        let _ = self.spki_der_cache.set(py_bytes.clone_ref(py));
1286        Ok(py_bytes.into_bound(py))
1287    }
1288
1289    /// Raw DER bytes of the ``TBSCertificate`` structure — the bytes that
1290    /// were signed by the issuer's key.
1291    ///
1292    /// This is the DER encoding of the full ``TBSCertificate`` SEQUENCE TLV,
1293    /// suitable for signature verification or as the input to a hash function.
1294    /// Equivalent to ``cryptography``'s ``cert.tbs_certificate_bytes``.
1295    ///
1296    /// ```python,ignore
1297    /// cert = synta.Certificate.from_der(der)
1298    /// tbs = cert.tbs_certificate_der
1299    /// # Verify signature manually:
1300    /// import synta.crypto
1301    /// digest = synta.digest("sha256", tbs)
1302    /// ```
1303    #[getter]
1304    fn tbs_certificate_der<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
1305        self.tbs_bytes(py)
1306    }
1307
1308    /// Verify that this certificate was directly issued by ``issuer``.
1309    ///
1310    /// Checks two things:
1311    ///
1312    /// 1. The ``issuer`` field of this certificate's TBS matches the
1313    ///    ``subject`` field of the provided issuer certificate (byte-exact
1314    ///    DER comparison).
1315    /// 2. The certificate's signature is valid under the issuer's public key.
1316    ///
1317    /// Raises :exc:`ValueError` if either check fails.  This is the synta
1318    /// equivalent of ``cryptography``'s
1319    /// ``cert.verify_directly_issued_by(issuer)``.
1320    ///
1321    /// ```python,ignore
1322    /// root = synta.Certificate.from_pem(open("root.pem", "rb").read())
1323    /// leaf = synta.Certificate.from_pem(open("leaf.pem", "rb").read())
1324    /// leaf.verify_issued_by(root)   # raises ValueError if not valid
1325    /// ```
1326    fn verify_issued_by(&self, issuer: &PyCertificate) -> PyResult<()> {
1327        use synta_certificate::{cert_byte_ranges, default_signature_verifier, SignatureVerifier};
1328
1329        // 1. Name check: our issuer Name must byte-equal their subject Name.
1330        let cert = self.cert()?;
1331        let issuer_cert = issuer.cert()?;
1332        if cert.tbs_certificate.issuer.as_bytes() != issuer_cert.tbs_certificate.subject.as_bytes()
1333        {
1334            return Err(pyo3::exceptions::PyValueError::new_err(
1335                "issuer name does not match subject of provided certificate",
1336            ));
1337        }
1338
1339        // 2. Locate byte ranges within each cert's DER without re-encoding.
1340        let my_ranges = cert_byte_ranges(self.raw).ok_or_else(|| {
1341            pyo3::exceptions::PyValueError::new_err(
1342                "failed to locate byte ranges in certificate DER",
1343            )
1344        })?;
1345        let issuer_ranges = cert_byte_ranges(issuer.raw).ok_or_else(|| {
1346            pyo3::exceptions::PyValueError::new_err(
1347                "failed to locate byte ranges in issuer certificate DER",
1348            )
1349        })?;
1350
1351        // 3. Signature bits (raw bytes, no unused-bits prefix).
1352        let signature_bits = cert.signature_value.as_bytes();
1353
1354        // 4. Verify using the backend-agnostic verifier.
1355        default_signature_verifier()
1356            .verify_certificate_signature(
1357                &self.raw[my_ranges.tbs],
1358                &self.raw[my_ranges.signature_algorithm],
1359                signature_bits,
1360                &issuer.raw[issuer_ranges.subject_public_key_info],
1361            )
1362            .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
1363
1364        Ok(())
1365    }
1366
1367    /// Compute a hash fingerprint of the complete certificate DER.
1368    ///
1369    /// ``algorithm`` must be one of ``"sha1"``, ``"sha224"``, ``"sha256"``,
1370    /// ``"sha384"``, ``"sha512"``, or ``"md5"``.  Raises :exc:`ValueError` for
1371    /// unknown algorithm names.
1372    ///
1373    /// ```python,ignore
1374    /// cert = synta.Certificate.from_der(der)
1375    /// fp = cert.fingerprint("sha256")
1376    /// print(fp.hex())     # e.g. "3a2b1c..."
1377    /// ```
1378    fn fingerprint<'py>(&self, py: Python<'py>, algorithm: &str) -> PyResult<Bound<'py, PyBytes>> {
1379        use synta_certificate::{default_data_hasher, DataHasher};
1380        let digest = default_data_hasher()
1381            .hash_data(algorithm, self.raw)
1382            .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
1383        Ok(PyBytes::new(py, &digest))
1384    }
1385
1386    fn __repr__(&self) -> PyResult<String> {
1387        let cert = self.cert()?;
1388        let serial = &cert.tbs_certificate.serial_number;
1389        let serial_str = if let Ok(v) = serial.as_i64() {
1390            v.to_string()
1391        } else {
1392            format!("<{} bytes>", serial.as_bytes().len())
1393        };
1394        Ok(format!(
1395            "Certificate(subject={:?}, serial={})",
1396            decode_name_debug(cert.tbs_certificate.subject.as_bytes()),
1397            serial_str,
1398        ))
1399    }
1400}
1401
1402impl PyCertificate {
1403    /// Parse a DER-encoded X.509 certificate (internal helper visible to sibling modules).
1404    ///
1405    /// This is the same logic as the `#[staticmethod] from_der` exposed to Python but
1406    /// accessible from Rust sibling modules via `pub(super)` visibility.
1407    pub(crate) fn new_from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
1408        Self::from_der(py, data)
1409    }
1410}