_synta/certificate/

pkix.rs

//! Python bindings for PKIX types: [`PyCsr`], [`PyCrl`], [`PyOcspResponse`],
//! and PKCS#7 / PKCS#12 helper functions.

use std::sync::OnceLock;

use pyo3::prelude::*;
use pyo3::types::{PyBytes, PyList, PyString};

use synta::traits::Encode;
use synta::{Decoder, Encoding};
use synta_certificate::Time;

use super::cert::{pem_blocks_to_pyobject, pyobject_to_pem, PyCertificate};
use crate::crypto_keys::PyPrivateKey;
use crate::types::PyObjectIdentifier;

// ── Helpers for CSR / CRL (Name is decoded eagerly, not stored as RawDer) ─────

/// Re-encode a parsed `Name` to DER and format it as an RFC 4514 DN string.
fn name_to_dn_string(name: &synta_certificate::Name<'_>) -> String {
    let mut enc = synta::Encoder::new(synta::Encoding::Der);
    if name.encode(&mut enc).is_err() {
        return String::new();
    }
    synta_certificate::name::format_dn(&enc.finish().unwrap_or_default())
}

/// Re-encode a parsed `Name` to its DER bytes.
fn name_to_der_bytes(name: &synta_certificate::Name<'_>) -> Vec<u8> {
    let mut enc = synta::Encoder::new(synta::Encoding::Der);
    if name.encode(&mut enc).is_err() {
        return Vec::new();
    }
    enc.finish().unwrap_or_default()
}

/// Map an `OCSPResponseStatus` enum value to its RFC 6960 display name.
fn ocsp_status_str(status: synta_certificate::ocsp::OCSPResponseStatus) -> &'static str {
    use synta_certificate::ocsp::OCSPResponseStatus::*;
    match status {
        Successful => "successful",
        MalformedRequest => "malformedRequest",
        InternalError => "internalError",
        TryLater => "tryLater",
        SigRequired => "sigRequired",
        Unauthorized => "unauthorized",
    }
}

// ── PyCsr ─────────────────────────────────────────────────────────────────────

/// PKCS #10 Certificate Signing Request accessible from Python.
///
/// ```python
/// csr = CertificationRequest.from_der(open("req.der", "rb").read())
/// print(csr.subject)
/// print(csr.public_key_algorithm)
/// ```
#[pyclass(frozen, name = "CertificationRequest")]
pub struct PyCsr {
    pub(super) _data: Py<PyBytes>,
    pub(super) raw: &'static [u8],
    inner: OnceLock<Box<synta_certificate::csr::CertificationRequest<'static>>>,
    subject_cache: OnceLock<Py<PyString>>,
    subject_raw_der_cache: OnceLock<Py<PyBytes>>,
    signature_algorithm_cache: OnceLock<Py<PyString>>,
    signature_algorithm_oid_cache: OnceLock<Py<PyObjectIdentifier>>,
    signature_cache: OnceLock<Py<PyBytes>>,
    public_key_algorithm_cache: OnceLock<Py<PyString>>,
    public_key_algorithm_oid_cache: OnceLock<Py<PyObjectIdentifier>>,
    public_key_cache: OnceLock<Py<PyBytes>>,
}

impl PyCsr {
    fn csr(&self) -> PyResult<&synta_certificate::csr::CertificationRequest<'static>> {
        if let Some(v) = self.inner.get() {
            return Ok(v.as_ref());
        }
        let mut decoder = Decoder::new(self.raw, Encoding::Der);
        let decoded = decoder.decode().map_err(|e| {
            pyo3::exceptions::PyValueError::new_err(format!(
                "CertificationRequest DER decode failed: {e}"
            ))
        })?;
        let _ = self.inner.set(Box::new(decoded));
        Ok(self.inner.get().unwrap().as_ref())
    }
}

#[pymethods]
impl PyCsr {
    /// Parse a DER-encoded PKCS #10 Certificate Signing Request.
    #[staticmethod]
    fn from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
        let py_bytes = data.unbind();
        // SAFETY: `py_bytes` holds a strong reference (Py<PyBytes>) that
        // keeps the Python bytes object alive for the lifetime of this struct.
        // CPython's bytes objects have a fixed-address, non-relocating payload
        // buffer (CPython has no moving GC).  The slice lifetime is extended
        // to 'static; the actual safety invariants are:
        //   (1) All reads of `raw` go through `&self`; no borrow of the struct
        //       can outlive the struct, so `raw` is never read after drop begins.
        //   (2) `raw: &'static [u8]` has no destructor (fat pointer, no heap
        //       allocation), so field drop order does not cause use-after-free.
        //   (3) `inner` contains only borrow-typed fields; dropping
        //       Box<CertificationRequest<'static>> does not read through the
        //       contained &'static slices (borrows have no destructors).
        // CPython-only: does not hold for PyPy or GraalPy.
        let raw: &'static [u8] = unsafe {
            let s = py_bytes.bind(py).as_bytes();
            std::slice::from_raw_parts(s.as_ptr(), s.len())
        };
        // Minimal structural scan: outer SEQUENCE tag + length.
        {
            let mut d = Decoder::new(raw, Encoding::Der);
            d.read_tag()
                .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
            d.read_length()
                .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
        }
        Ok(Self {
            _data: py_bytes,
            raw,
            inner: OnceLock::new(),
            subject_cache: OnceLock::new(),
            subject_raw_der_cache: OnceLock::new(),
            signature_algorithm_cache: OnceLock::new(),
            signature_algorithm_oid_cache: OnceLock::new(),
            signature_cache: OnceLock::new(),
            public_key_algorithm_cache: OnceLock::new(),
            public_key_algorithm_oid_cache: OnceLock::new(),
            public_key_cache: OnceLock::new(),
        })
    }

    /// Parse a PEM-encoded PKCS #10 Certificate Signing Request.
    ///
    /// Returns a single :class:`CertificationRequest` for one PEM block, or a
    /// :class:`list` of :class:`CertificationRequest` objects when multiple
    /// blocks are present.  Raises :exc:`ValueError` if no PEM block is found.
    ///
    /// ```python
    /// csr = CertificationRequest.from_pem(open("csr.pem", "rb").read())
    /// print(csr.subject)
    /// ```
    #[staticmethod]
    fn from_pem<'py>(py: Python<'py>, data: Bound<'_, PyBytes>) -> PyResult<Bound<'py, PyAny>> {
        pem_blocks_to_pyobject(py, data.as_bytes(), |py, bytes| {
            let obj = Self::from_der(py, bytes)?;
            Ok(Py::new(py, obj)?.into_bound(py).into_any())
        })
    }

    /// Serialize one CSR or a list of CSRs to PEM format.
    ///
    /// ```python
    /// pem = CertificationRequest.to_pem(csr)
    /// pem = CertificationRequest.to_pem([csr1, csr2])
    /// ```
    #[staticmethod]
    fn to_pem<'py>(
        py: Python<'py>,
        obj_or_list: Bound<'_, PyAny>,
    ) -> PyResult<Bound<'py, PyBytes>> {
        pyobject_to_pem::<Self, _>(py, "CERTIFICATE REQUEST", &obj_or_list, |c| c.raw)
    }

    /// Subject distinguished name as a string.
    #[getter]
    fn subject<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
        if let Some(cached) = self.subject_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let s = name_to_dn_string(&self.csr()?.certification_request_info.subject);
        let py_str = PyString::new(py, &s).unbind();
        let _ = self.subject_cache.set(py_str.clone_ref(py));
        Ok(py_str.into_bound(py))
    }

    /// Raw DER bytes of the subject Name SEQUENCE.
    #[getter]
    fn subject_raw_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        if let Some(cached) = self.subject_raw_der_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let bytes = name_to_der_bytes(&self.csr()?.certification_request_info.subject);
        let py_bytes = PyBytes::new(py, &bytes).unbind();
        let _ = self.subject_raw_der_cache.set(py_bytes.clone_ref(py));
        Ok(py_bytes.into_bound(py))
    }

    /// CSR version (0 = v1 per RFC 2986).
    #[getter]
    fn version(&self) -> PyResult<i64> {
        Ok(self
            .csr()?
            .certification_request_info
            .version
            .as_i64()
            .unwrap_or(0))
    }

    /// Signature algorithm name (e.g. "RSA", "ECDSA", "Ed25519").
    #[getter]
    fn signature_algorithm<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
        if let Some(cached) = self.signature_algorithm_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let oid = &self.csr()?.signature_algorithm.algorithm;
        let name = synta_certificate::identify_signature_algorithm(oid);
        let s = if name != "Other" {
            name.to_string()
        } else {
            oid.to_string()
        };
        let py_str = PyString::new(py, &s).unbind();
        let _ = self.signature_algorithm_cache.set(py_str.clone_ref(py));
        Ok(py_str.into_bound(py))
    }

    /// OID of the signature algorithm.
    #[getter]
    fn signature_algorithm_oid<'py>(
        &self,
        py: Python<'py>,
    ) -> PyResult<Bound<'py, PyObjectIdentifier>> {
        if let Some(cached) = self.signature_algorithm_oid_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let obj = Py::new(
            py,
            PyObjectIdentifier::from_oid(self.csr()?.signature_algorithm.algorithm.clone()),
        )?;
        let _ = self.signature_algorithm_oid_cache.set(obj.clone_ref(py));
        Ok(obj.into_bound(py))
    }

    /// Raw signature bytes.
    #[getter]
    fn signature<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        if let Some(cached) = self.signature_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let py_bytes = PyBytes::new(py, self.csr()?.signature.as_bytes()).unbind();
        let _ = self.signature_cache.set(py_bytes.clone_ref(py));
        Ok(py_bytes.into_bound(py))
    }

    /// Public-key algorithm name (e.g. "RSA", "ECDSA", "Ed25519").
    #[getter]
    fn public_key_algorithm<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
        if let Some(cached) = self.public_key_algorithm_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let oid = &self
            .csr()?
            .certification_request_info
            .subject_pkinfo
            .algorithm
            .algorithm;
        let s = synta_certificate::identify_public_key_algorithm(oid)
            .map(|s| s.to_string())
            .unwrap_or_else(|| oid.to_string());
        let py_str = PyString::new(py, &s).unbind();
        let _ = self.public_key_algorithm_cache.set(py_str.clone_ref(py));
        Ok(py_str.into_bound(py))
    }

    /// OID of the subject public-key algorithm.
    #[getter]
    fn public_key_algorithm_oid<'py>(
        &self,
        py: Python<'py>,
    ) -> PyResult<Bound<'py, PyObjectIdentifier>> {
        if let Some(cached) = self.public_key_algorithm_oid_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let obj = Py::new(
            py,
            PyObjectIdentifier::from_oid(
                self.csr()?
                    .certification_request_info
                    .subject_pkinfo
                    .algorithm
                    .algorithm
                    .clone(),
            ),
        )?;
        let _ = self.public_key_algorithm_oid_cache.set(obj.clone_ref(py));
        Ok(obj.into_bound(py))
    }

    /// Raw subject public-key bytes.
    #[getter]
    fn public_key<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        if let Some(cached) = self.public_key_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let py_bytes = PyBytes::new(
            py,
            self.csr()?
                .certification_request_info
                .subject_pkinfo
                .subject_public_key
                .as_bytes(),
        )
        .unbind();
        let _ = self.public_key_cache.set(py_bytes.clone_ref(py));
        Ok(py_bytes.into_bound(py))
    }

    /// Return the DER value bytes of the named extension from this CSR's
    /// ``extensionRequest`` attribute (OID ``1.2.840.113549.1.9.14``), or
    /// ``None`` if the CSR carries no such extension.
    ///
    /// ``oid`` may be a dotted-decimal string or a
    /// :class:`~synta.ObjectIdentifier`.  The value returned is the content of
    /// the ``extnValue`` OCTET STRING — the DER encoding of the extension value
    /// — without the outer OCTET STRING tag and length.  Returns ``None`` when
    /// the CSR has no ``extensionRequest`` attribute, when the attribute
    /// contains no extensions, or when the requested OID is absent.
    ///
    /// ```python,ignore
    /// SAN_OID = "2.5.29.17"
    /// san_der = csr.get_extension_value_der(SAN_OID)
    /// if san_der:
    ///     names = synta.parse_general_names(san_der)
    /// ```
    fn get_extension_value_der<'py>(
        &self,
        py: Python<'py>,
        oid: &Bound<'_, PyAny>,
    ) -> PyResult<Option<Bound<'py, PyBytes>>> {
        let target = super::oid_from_pyany(oid)?;
        let csr = self.csr()?;
        let attrs = match csr.certification_request_info.attributes.as_ref() {
            Some(a) => a,
            None => return Ok(None),
        };
        for attr in attrs.elements() {
            if attr.attr_type.components() != synta_certificate::oids::PKCS9_EXTENSION_REQUEST {
                continue;
            }
            // attr_values: SetOf<RawDer> — first element is the SEQUENCE OF Extension TLV
            let Some(raw) = attr.attr_values.elements().first() else {
                return Ok(None);
            };
            return Ok(synta_certificate::find_extension_value(
                raw.as_bytes(),
                target.components(),
            )
            .map(|v| PyBytes::new(py, v)));
        }
        Ok(None)
    }

    /// Return the Subject Alternative Names from this CSR's ``extensionRequest``
    /// attribute as a list of ``(tag_number, content)`` tuples.  Returns an
    /// empty list when the CSR carries no SAN extension.
    ///
    /// Tag numbers and content format follow the same convention as
    /// :meth:`~synta.Certificate.subject_alt_names` and
    /// :func:`~synta.parse_general_names`.
    ///
    /// ```python,ignore
    /// import synta.general_name as gn
    /// for tag_num, content in csr.subject_alt_names():
    ///     if tag_num == gn.DNS_NAME:
    ///         print("DNS:", content.decode("ascii"))
    /// ```
    fn subject_alt_names<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
        use pyo3::types::{PyBytes, PyList, PyTuple};
        let list = PyList::empty(py);
        let csr = self.csr()?;
        let attrs = match csr.certification_request_info.attributes.as_ref() {
            Some(a) => a,
            None => return Ok(list),
        };
        for attr in attrs.elements() {
            if attr.attr_type.components() != synta_certificate::oids::PKCS9_EXTENSION_REQUEST {
                continue;
            }
            let Some(raw) = attr.attr_values.elements().first() else {
                return Ok(list);
            };
            if let Some(san_bytes) = synta_certificate::find_extension_value(
                raw.as_bytes(),
                synta_certificate::oids::SUBJECT_ALT_NAME,
            ) {
                for (tag_num, content) in synta_certificate::parse_general_names(san_bytes) {
                    let tuple = PyTuple::new(
                        py,
                        [
                            tag_num.into_pyobject(py)?.into_any(),
                            PyBytes::new(py, &content).into_any(),
                        ],
                    )?;
                    list.append(tuple)?;
                }
            }
            break;
        }
        Ok(list)
    }

    /// Complete DER encoding of this CSR (the original bytes passed to ``from_der``).
    fn to_der<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
        self._data.clone_ref(py).into_bound(py)
    }

    fn __repr__(&self) -> PyResult<String> {
        Ok(format!(
            "CertificationRequest(subject={:?})",
            name_to_dn_string(&self.csr()?.certification_request_info.subject),
        ))
    }
}

impl PyCsr {
    /// Rust-internal constructor: parse DER bytes into a `PyCsr`.
    ///
    /// Mirrors `from_der` but is accessible from sibling Rust modules.
    pub(crate) fn new_from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
        Self::from_der(py, data)
    }
}

// ── PyCrl ─────────────────────────────────────────────────────────────────────

/// X.509 Certificate Revocation List accessible from Python.
///
/// ```python
/// crl = CertificateList.from_der(open("crl.der", "rb").read())
/// print(crl.issuer)
/// print(crl.revoked_count)
/// ```
#[pyclass(frozen, name = "CertificateList")]
pub struct PyCrl {
    pub(super) _data: Py<PyBytes>,
    pub(super) raw: &'static [u8],
    inner: OnceLock<Box<synta_certificate::crl::CertificateList<'static>>>,
    issuer_cache: OnceLock<Py<PyString>>,
    issuer_raw_der_cache: OnceLock<Py<PyBytes>>,
    this_update_cache: OnceLock<Py<PyString>>,
    next_update_cache: OnceLock<Option<Py<PyString>>>,
    signature_algorithm_cache: OnceLock<Py<PyString>>,
    signature_algorithm_oid_cache: OnceLock<Py<PyObjectIdentifier>>,
    signature_value_cache: OnceLock<Py<PyBytes>>,
}

impl PyCrl {
    fn crl(&self) -> PyResult<&synta_certificate::crl::CertificateList<'static>> {
        if let Some(v) = self.inner.get() {
            return Ok(v.as_ref());
        }
        let mut decoder = Decoder::new(self.raw, Encoding::Der);
        let decoded = decoder.decode().map_err(|e| {
            pyo3::exceptions::PyValueError::new_err(format!(
                "CertificateList DER decode failed: {e}"
            ))
        })?;
        let _ = self.inner.set(Box::new(decoded));
        Ok(self.inner.get().unwrap().as_ref())
    }
}

#[pymethods]
impl PyCrl {
    /// Parse a DER-encoded X.509 Certificate Revocation List.
    #[staticmethod]
    fn from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
        let py_bytes = data.unbind();
        // SAFETY: `py_bytes` holds a strong reference (Py<PyBytes>) that
        // keeps the Python bytes object alive for the lifetime of this struct.
        // CPython's bytes objects have a fixed-address, non-relocating payload
        // buffer (CPython has no moving GC).  The slice lifetime is extended
        // to 'static; the actual safety invariants are:
        //   (1) All reads of `raw` go through `&self`; no borrow of the struct
        //       can outlive the struct, so `raw` is never read after drop begins.
        //   (2) `raw: &'static [u8]` has no destructor (fat pointer, no heap
        //       allocation), so field drop order does not cause use-after-free.
        //   (3) `inner` contains only borrow-typed fields; dropping
        //       Box<CertificateList<'static>> does not read through the
        //       contained &'static slices (borrows have no destructors).
        // CPython-only: does not hold for PyPy or GraalPy.
        let raw: &'static [u8] = unsafe {
            let s = py_bytes.bind(py).as_bytes();
            std::slice::from_raw_parts(s.as_ptr(), s.len())
        };
        {
            let mut d = Decoder::new(raw, Encoding::Der);
            d.read_tag()
                .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
            d.read_length()
                .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
        }
        Ok(Self {
            _data: py_bytes,
            raw,
            inner: OnceLock::new(),
            issuer_cache: OnceLock::new(),
            issuer_raw_der_cache: OnceLock::new(),
            this_update_cache: OnceLock::new(),
            next_update_cache: OnceLock::new(),
            signature_algorithm_cache: OnceLock::new(),
            signature_algorithm_oid_cache: OnceLock::new(),
            signature_value_cache: OnceLock::new(),
        })
    }

    /// Parse a PEM-encoded X.509 Certificate Revocation List.
    ///
    /// Returns a single :class:`CertificateList` for one PEM block, or a
    /// :class:`list` of :class:`CertificateList` objects when multiple blocks
    /// are present.  Raises :exc:`ValueError` if no PEM block is found.
    ///
    /// ```python
    /// crl = CertificateList.from_pem(open("crl.pem", "rb").read())
    /// print(crl.issuer)
    /// ```
    #[staticmethod]
    fn from_pem<'py>(py: Python<'py>, data: Bound<'_, PyBytes>) -> PyResult<Bound<'py, PyAny>> {
        pem_blocks_to_pyobject(py, data.as_bytes(), |py, bytes| {
            let obj = Self::from_der(py, bytes)?;
            Ok(Py::new(py, obj)?.into_bound(py).into_any())
        })
    }

    /// Serialize one CRL or a list of CRLs to PEM format.
    ///
    /// ```python
    /// pem = CertificateList.to_pem(crl)
    /// pem = CertificateList.to_pem([crl1, crl2])
    /// ```
    #[staticmethod]
    fn to_pem<'py>(
        py: Python<'py>,
        obj_or_list: Bound<'_, PyAny>,
    ) -> PyResult<Bound<'py, PyBytes>> {
        pyobject_to_pem::<Self, _>(py, "X509 CRL", &obj_or_list, |c| c.raw)
    }

    /// CRL version field (1 = v2), or ``None`` if absent (implies v1).
    ///
    /// RFC 5280 §5.1.2.1: the ``version`` field is present only when the CRL
    /// is version 2.  For RFC 5280-compliant CRLs this value is always ``1``
    /// (the integer 1 encodes version 2).  Returns ``None`` for v1 CRLs.
    #[getter]
    fn version(&self) -> PyResult<Option<i64>> {
        Ok(self
            .crl()?
            .tbs_cert_list
            .version
            .as_ref()
            .and_then(|v| v.as_i64().ok()))
    }

    /// Issuer distinguished name as a string.
    #[getter]
    fn issuer<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
        if let Some(cached) = self.issuer_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let s = name_to_dn_string(&self.crl()?.tbs_cert_list.issuer);
        let py_str = PyString::new(py, &s).unbind();
        let _ = self.issuer_cache.set(py_str.clone_ref(py));
        Ok(py_str.into_bound(py))
    }

    /// Raw DER bytes of the issuer Name SEQUENCE.
    #[getter]
    fn issuer_raw_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        if let Some(cached) = self.issuer_raw_der_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let bytes = name_to_der_bytes(&self.crl()?.tbs_cert_list.issuer);
        let py_bytes = PyBytes::new(py, &bytes).unbind();
        let _ = self.issuer_raw_der_cache.set(py_bytes.clone_ref(py));
        Ok(py_bytes.into_bound(py))
    }

    /// thisUpdate time as a string.
    #[getter]
    fn this_update<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
        if let Some(cached) = self.this_update_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let s = match &self.crl()?.tbs_cert_list.this_update {
            Time::UtcTime(t) => t.to_string(),
            Time::GeneralTime(t) => t.to_string(),
        };
        let py_str = PyString::new(py, &s).unbind();
        let _ = self.this_update_cache.set(py_str.clone_ref(py));
        Ok(py_str.into_bound(py))
    }

    /// nextUpdate time as a string, or ``None`` if absent.
    #[getter]
    fn next_update<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyString>>> {
        if let Some(cached) = self.next_update_cache.get() {
            return Ok(cached.as_ref().map(|s| s.clone_ref(py).into_bound(py)));
        }
        let computed: Option<Bound<'py, PyString>> =
            self.crl()?.tbs_cert_list.next_update.as_ref().map(|t| {
                let s = match t {
                    Time::UtcTime(t) => t.to_string(),
                    Time::GeneralTime(t) => t.to_string(),
                };
                PyString::new(py, &s)
            });
        let to_store = computed.as_ref().map(|s| s.as_unbound().clone_ref(py));
        let _ = self.next_update_cache.set(to_store);
        Ok(computed)
    }

    /// Signature algorithm name (e.g. "RSA", "ECDSA").
    #[getter]
    fn signature_algorithm<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
        if let Some(cached) = self.signature_algorithm_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let oid = &self.crl()?.signature_algorithm.algorithm;
        let name = synta_certificate::identify_signature_algorithm(oid);
        let s = if name != "Other" {
            name.to_string()
        } else {
            oid.to_string()
        };
        let py_str = PyString::new(py, &s).unbind();
        let _ = self.signature_algorithm_cache.set(py_str.clone_ref(py));
        Ok(py_str.into_bound(py))
    }

    /// OID of the signature algorithm.
    #[getter]
    fn signature_algorithm_oid<'py>(
        &self,
        py: Python<'py>,
    ) -> PyResult<Bound<'py, PyObjectIdentifier>> {
        if let Some(cached) = self.signature_algorithm_oid_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let obj = Py::new(
            py,
            PyObjectIdentifier::from_oid(self.crl()?.signature_algorithm.algorithm.clone()),
        )?;
        let _ = self.signature_algorithm_oid_cache.set(obj.clone_ref(py));
        Ok(obj.into_bound(py))
    }

    /// Raw signature bytes.
    #[getter]
    fn signature_value<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        if let Some(cached) = self.signature_value_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let py_bytes = PyBytes::new(py, self.crl()?.signature_value.as_bytes()).unbind();
        let _ = self.signature_value_cache.set(py_bytes.clone_ref(py));
        Ok(py_bytes.into_bound(py))
    }

    /// CRL sequence number from the ``cRLNumber`` extension (OID ``2.5.29.20``),
    /// or ``None`` if the CRL carries no such extension.
    ///
    /// Returns an arbitrary-precision Python :class:`int`.
    ///
    /// ```python,ignore
    /// crl = synta.CertificateList.from_der(data)
    /// print("CRL number:", crl.crl_number)
    /// ```
    #[getter]
    fn crl_number<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyAny>>> {
        let Some(n) = self.crl()?.crl_number() else {
            return Ok(None);
        };
        let py_int = if let Ok(v) = n.as_i64() {
            v.into_pyobject(py)?.into_any()
        } else if let Ok(v) = n.as_i128() {
            v.into_pyobject(py)?.into_any()
        } else {
            // Arbitrarily large CRL number — extremely uncommon but RFC-compliant.
            let bytes_obj = PyBytes::new(py, n.as_bytes());
            let kwargs = pyo3::types::PyDict::new(py);
            kwargs.set_item(pyo3::intern!(py, "signed"), false)?;
            py.get_type::<pyo3::types::PyInt>().call_method(
                pyo3::intern!(py, "from_bytes"),
                (bytes_obj, pyo3::intern!(py, "big")),
                Some(&kwargs),
            )?
        };
        Ok(Some(py_int))
    }

    /// Number of revoked certificates in this CRL (0 if no revoked entries).
    #[getter]
    fn revoked_count(&self) -> PyResult<usize> {
        Ok(self
            .crl()?
            .tbs_cert_list
            .revoked_certificates
            .as_ref()
            .map(|v| v.len())
            .unwrap_or(0))
    }

    /// Return the DER value bytes of the named CRL extension, or ``None`` if
    /// the CRL carries no such extension.
    ///
    /// ``oid`` may be a dotted-decimal string or a
    /// :class:`~synta.ObjectIdentifier`.  The value returned is the content of
    /// the ``extnValue`` OCTET STRING — the DER encoding of the extension value
    /// — without the outer OCTET STRING tag and length.
    ///
    /// ```python,ignore
    /// CRL_NUMBER_OID = "2.5.29.20"
    /// crl_num_der = crl.get_extension_value_der(CRL_NUMBER_OID)
    /// if crl_num_der:
    ///     num = int(synta.Decoder(crl_num_der, synta.Encoding.DER).decode_integer())
    ///     print("CRL Number:", num)
    /// ```
    fn get_extension_value_der<'py>(
        &self,
        py: Python<'py>,
        oid: &Bound<'_, PyAny>,
    ) -> PyResult<Option<Bound<'py, PyBytes>>> {
        let target = super::oid_from_pyany(oid)?;
        let exts = match self.crl()?.tbs_cert_list.crl_extensions.as_ref() {
            Some(e) => e,
            None => return Ok(None),
        };
        for ext in exts {
            if ext.extn_id == target {
                return Ok(Some(PyBytes::new(py, ext.extn_value.as_bytes())));
            }
        }
        Ok(None)
    }

    /// Complete DER encoding of this CRL (the original bytes passed to ``from_der``).
    fn to_der<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
        self._data.clone_ref(py).into_bound(py)
    }

    fn __repr__(&self) -> PyResult<String> {
        let crl = self.crl()?;
        Ok(format!(
            "CertificateList(issuer={:?}, revoked_count={})",
            name_to_dn_string(&crl.tbs_cert_list.issuer),
            crl.tbs_cert_list
                .revoked_certificates
                .as_ref()
                .map(|v| v.len())
                .unwrap_or(0),
        ))
    }
}

// ── PyOcspResponse ────────────────────────────────────────────────────────────

/// OCSP Response (RFC 6960) accessible from Python.
///
/// ```python
/// resp = OCSPResponse.from_der(open("resp.der", "rb").read())
/// print(resp.status)
/// if resp.response_bytes:
///     basic = resp.response_bytes  # DER of BasicOCSPResponse
/// ```
#[pyclass(frozen, name = "OCSPResponse")]
pub struct PyOcspResponse {
    pub(super) _data: Py<PyBytes>,
    pub(super) raw: &'static [u8],
    inner: OnceLock<Box<synta_certificate::ocsp::OCSPResponse<'static>>>,
    status_cache: OnceLock<Py<PyString>>,
    response_type_oid_cache: OnceLock<Option<Py<PyObjectIdentifier>>>,
    response_bytes_cache: OnceLock<Option<Py<PyBytes>>>,
}

impl PyOcspResponse {
    fn ocsp(&self) -> PyResult<&synta_certificate::ocsp::OCSPResponse<'static>> {
        if let Some(v) = self.inner.get() {
            return Ok(v.as_ref());
        }
        let mut decoder = Decoder::new(self.raw, Encoding::Der);
        let decoded = decoder.decode().map_err(|e| {
            pyo3::exceptions::PyValueError::new_err(format!("OCSPResponse DER decode failed: {e}"))
        })?;
        let _ = self.inner.set(Box::new(decoded));
        Ok(self.inner.get().unwrap().as_ref())
    }
}

#[pymethods]
impl PyOcspResponse {
    /// Parse a DER-encoded OCSP Response.
    #[staticmethod]
    fn from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
        let py_bytes = data.unbind();
        // SAFETY: `py_bytes` holds a strong reference (Py<PyBytes>) that
        // keeps the Python bytes object alive for the lifetime of this struct.
        // CPython's bytes objects have a fixed-address, non-relocating payload
        // buffer (CPython has no moving GC).  The slice lifetime is extended
        // to 'static; the actual safety invariants are:
        //   (1) All reads of `raw` go through `&self`; no borrow of the struct
        //       can outlive the struct, so `raw` is never read after drop begins.
        //   (2) `raw: &'static [u8]` has no destructor (fat pointer, no heap
        //       allocation), so field drop order does not cause use-after-free.
        //   (3) `inner` contains only borrow-typed fields; dropping
        //       Box<OCSPResponse<'static>> does not read through the contained
        //       &'static slices (borrows have no destructors).
        // CPython-only: does not hold for PyPy or GraalPy.
        let raw: &'static [u8] = unsafe {
            let s = py_bytes.bind(py).as_bytes();
            std::slice::from_raw_parts(s.as_ptr(), s.len())
        };
        {
            let mut d = Decoder::new(raw, Encoding::Der);
            d.read_tag()
                .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
            d.read_length()
                .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
        }
        Ok(Self {
            _data: py_bytes,
            raw,
            inner: OnceLock::new(),
            status_cache: OnceLock::new(),
            response_type_oid_cache: OnceLock::new(),
            response_bytes_cache: OnceLock::new(),
        })
    }

    /// Parse a PEM-encoded OCSP Response.
    ///
    /// Returns a single :class:`OCSPResponse` for one PEM block, or a
    /// :class:`list` of :class:`OCSPResponse` objects when multiple blocks are
    /// present.  Raises :exc:`ValueError` if no PEM block is found.
    ///
    /// ```python
    /// resp = OCSPResponse.from_pem(open("ocsp.pem", "rb").read())
    /// print(resp.status)
    /// ```
    #[staticmethod]
    fn from_pem<'py>(py: Python<'py>, data: Bound<'_, PyBytes>) -> PyResult<Bound<'py, PyAny>> {
        pem_blocks_to_pyobject(py, data.as_bytes(), |py, bytes| {
            let obj = Self::from_der(py, bytes)?;
            Ok(Py::new(py, obj)?.into_bound(py).into_any())
        })
    }

    /// Serialize one OCSP response or a list of them to PEM format.
    ///
    /// ```python
    /// pem = OCSPResponse.to_pem(resp)
    /// pem = OCSPResponse.to_pem([resp1, resp2])
    /// ```
    #[staticmethod]
    fn to_pem<'py>(
        py: Python<'py>,
        obj_or_list: Bound<'_, PyAny>,
    ) -> PyResult<Bound<'py, PyBytes>> {
        pyobject_to_pem::<Self, _>(py, "OCSP RESPONSE", &obj_or_list, |c| c.raw)
    }

    /// Response status string (e.g. ``"successful"``, ``"tryLater"``).
    #[getter]
    fn status<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
        if let Some(cached) = self.status_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let s = ocsp_status_str(self.ocsp()?.response_status);
        let py_str = PyString::new(py, s).unbind();
        let _ = self.status_cache.set(py_str.clone_ref(py));
        Ok(py_str.into_bound(py))
    }

    /// Dotted-decimal OID of the response type, or ``None`` for error responses
    /// that carry no ``responseBytes`` (e.g. ``tryLater``).
    ///
    /// For successful responses this is typically the id-pkix-ocsp-basic OID
    /// (``"1.3.6.1.5.5.7.48.1.1"``).
    #[getter]
    fn response_type_oid<'py>(
        &self,
        py: Python<'py>,
    ) -> PyResult<Option<Bound<'py, PyObjectIdentifier>>> {
        if let Some(cached) = self.response_type_oid_cache.get() {
            return Ok(cached.as_ref().map(|o| o.clone_ref(py).into_bound(py)));
        }
        let computed: Option<Py<PyObjectIdentifier>> = self
            .ocsp()?
            .response_bytes
            .as_ref()
            .map(|rb| Py::new(py, PyObjectIdentifier::from_oid(rb.response_type.clone())))
            .transpose()?;
        let _ = self
            .response_type_oid_cache
            .set(computed.as_ref().map(|o| o.clone_ref(py)));
        Ok(computed.map(|o| o.into_bound(py)))
    }

    /// Raw DER bytes of the embedded response (the OCTET STRING content of
    /// ``ResponseBytes``), or ``None`` for error responses.
    ///
    /// For id-pkix-ocsp-basic responses, these bytes are a DER-encoded
    /// ``BasicOCSPResponse``.
    #[getter]
    fn response_bytes<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
        if let Some(cached) = self.response_bytes_cache.get() {
            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
        }
        let computed: Option<Bound<'py, PyBytes>> = self
            .ocsp()?
            .response_bytes
            .as_ref()
            .map(|rb| PyBytes::new(py, rb.response.as_bytes()));
        let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
        let _ = self.response_bytes_cache.set(to_store);
        Ok(computed)
    }

    /// Complete DER encoding of this OCSP response (the original bytes passed to ``from_der``).
    fn to_der<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
        self._data.clone_ref(py).into_bound(py)
    }

    fn __repr__(&self) -> PyResult<String> {
        Ok(format!(
            "OCSPResponse(status={})",
            ocsp_status_str(self.ocsp()?.response_status),
        ))
    }
}

// ── PKCS#7 / PKCS#12 free functions ─────────────────────────────────────────

/// Build a `PyList[Certificate]` from a `Vec<Vec<u8>>` of DER-encoded certs.
fn der_certs_to_pylist<'py>(
    py: Python<'py>,
    der_certs: Vec<Vec<u8>>,
) -> PyResult<Bound<'py, PyList>> {
    let list = PyList::empty(py);
    for der in der_certs {
        let py_bytes = PyBytes::new(py, &der);
        let cert = PyCertificate::new_from_der(py, py_bytes)?;
        list.append(Py::new(py, cert)?.into_bound(py))?;
    }
    Ok(list)
}

/// Extract X.509 certificates from a PKCS#7 SignedData blob (DER or BER).
///
/// Accepts raw DER or BER input; BER indefinite-length encodings (common in
/// PKCS#7 files produced by older tools) are handled transparently.  Returns
/// a list of :class:`Certificate` objects in document order.
///
/// Raises :exc:`ValueError` on any ASN.1 structural error or if the input
/// is not id-signedData.
///
/// ```python,ignore
/// data = open("bundle.p7b", "rb").read()
/// certs = synta.load_der_pkcs7_certificates(data)
/// for cert in certs:
///     print(cert.subject)
/// ```
#[pyfunction]
pub fn load_der_pkcs7_certificates<'py>(
    py: Python<'py>,
    data: &[u8],
) -> PyResult<Bound<'py, PyList>> {
    let der_certs = synta_certificate::certs_from_pkcs7(data)
        .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
    der_certs_to_pylist(py, der_certs)
}

/// Extract X.509 certificates from a PEM-encoded PKCS#7 SignedData blob.
///
/// Decodes PEM block(s) in ``data`` (any label is accepted: ``PKCS7``,
/// ``CMS``, ``CERTIFICATE``, etc.) then calls
/// :func:`load_der_pkcs7_certificates` on each DER payload.  All certificates
/// across all blocks are returned as a single flat list.
///
/// Raises :exc:`ValueError` if no PEM block is found or if any block fails
/// to parse as PKCS#7 SignedData.
///
/// ```python,ignore
/// data = open("bundle.pem", "rb").read()
/// certs = synta.load_pem_pkcs7_certificates(data)
/// for cert in certs:
///     print(cert.subject)
/// ```
#[pyfunction]
pub fn load_pem_pkcs7_certificates<'py>(
    py: Python<'py>,
    data: &[u8],
) -> PyResult<Bound<'py, PyList>> {
    let blocks = synta_certificate::pem_blocks(data);
    if blocks.is_empty() {
        return Err(pyo3::exceptions::PyValueError::new_err(
            "no PEM block found in input",
        ));
    }
    let mut all_certs: Vec<Vec<u8>> = Vec::new();
    for (_, der) in &blocks {
        let certs = synta_certificate::certs_from_pkcs7(der)
            .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
        all_certs.extend(certs);
    }
    der_certs_to_pylist(py, all_certs)
}

/// Extract X.509 certificates from a PKCS#12 archive.
///
/// Accepts raw DER or BER input; the encoding is detected automatically.
/// Returns a list of :class:`Certificate` objects in document order.
/// Non-certificate bag types (``keyBag``, ``pkcs8ShroudedKeyBag``,
/// ``crlBag``, ``secretBag``) are silently skipped.
///
/// ``password`` is the archive password as raw bytes (UTF-8 without a NUL
/// terminator).  Pass ``b""`` or omit the argument for password-less archives.
///
/// Encrypted bags are supported when the ``openssl`` Cargo feature is enabled.
/// Without that feature a :exc:`ValueError` is raised on any encrypted bag.
///
/// ```python,ignore
/// data = open("archive.p12", "rb").read()
/// certs = synta.load_pkcs12_certificates(data, b"s3cr3t")
/// for cert in certs:
///     print(cert.subject)
/// ```
#[pyfunction]
#[pyo3(signature = (data, password = None))]
pub fn load_pkcs12_certificates<'py>(
    py: Python<'py>,
    data: &[u8],
    password: Option<&[u8]>,
) -> PyResult<Bound<'py, PyList>> {
    let pw = password.unwrap_or(b"");
    #[cfg(feature = "openssl")]
    let der_certs =
        synta_certificate::certs_from_pkcs12(data, pw, &synta_certificate::OpensslDecryptor)
            .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
    #[cfg(not(feature = "openssl"))]
    let der_certs = synta_certificate::certs_from_pkcs12(data, pw, &synta_certificate::NoCrypto)
        .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
    der_certs_to_pylist(py, der_certs)
}

/// Extract PKCS#8 private keys from a PKCS#12 archive.
///
/// Returns a :class:`list` of :class:`bytes` objects, one per private key
/// found in ``keyBag`` (unencrypted) and ``pkcs8ShroudedKeyBag`` (encrypted,
/// decrypted when the ``openssl`` Cargo feature is enabled and ``password``
/// is given) entries.  Each element is a DER-encoded ``OneAsymmetricKey``
/// (PKCS#8) structure, suitable for passing to a cryptography library.
///
/// ``certBag``, ``crlBag``, ``secretBag`` and any unknown bag types are
/// silently skipped.
///
/// ```python,ignore
/// data = open("archive.p12", "rb").read()
/// keys = synta.load_pkcs12_keys(data, b"s3cr3t")
/// for key_der in keys:
///     from cryptography.hazmat.primitives.serialization import load_der_private_key
///     key = load_der_private_key(key_der, None)
/// ```
#[pyfunction]
#[pyo3(signature = (data, password = None))]
pub fn load_pkcs12_keys<'py>(
    py: Python<'py>,
    data: &[u8],
    password: Option<&[u8]>,
) -> PyResult<Bound<'py, PyList>> {
    let pw = password.unwrap_or(b"");
    #[cfg(feature = "openssl")]
    let der_keys =
        synta_certificate::keys_from_pkcs12(data, pw, &synta_certificate::OpensslDecryptor)
            .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
    #[cfg(not(feature = "openssl"))]
    let der_keys = synta_certificate::keys_from_pkcs12(data, pw, &synta_certificate::NoCrypto)
        .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
    let list = PyList::empty(py);
    for der in der_keys {
        list.append(PyBytes::new(py, &der))?;
    }
    Ok(list)
}

/// Extract both certificates and private keys from a PKCS#12 archive.
///
/// Returns a 2-tuple ``(certs, keys)`` where:
///
/// - ``certs`` is a :class:`list` of :class:`Certificate` objects.
/// - ``keys`` is a :class:`list` of :class:`bytes`, each a DER-encoded
///   PKCS#8 ``OneAsymmetricKey`` structure.
///
/// Encrypted bags are decrypted with ``password`` when the ``openssl``
/// Cargo feature is enabled; without it, encrypted bags are silently
/// skipped and a :exc:`ValueError` is raised only on structural errors.
/// Pass ``b""`` or omit ``password`` for password-less archives.
///
/// Use :func:`load_pkcs12_certificates` if you only need certificates, or
/// :func:`load_pkcs12_keys` if you only need keys.
///
/// ```python,ignore
/// data = open("archive.p12", "rb").read()
/// certs, keys = synta.load_pkcs12(data, b"s3cr3t")
/// for cert in certs:
///     print(cert.subject)
/// for key_der in keys:
///     from cryptography.hazmat.primitives.serialization import load_der_private_key
///     key = load_der_private_key(key_der, None)
/// ```
#[pyfunction]
#[pyo3(signature = (data, password = None))]
pub fn load_pkcs12<'py>(
    py: Python<'py>,
    data: &[u8],
    password: Option<&[u8]>,
) -> PyResult<Bound<'py, pyo3::types::PyTuple>> {
    let pw = password.unwrap_or(b"");
    #[cfg(feature = "openssl")]
    let pki = synta_certificate::pki_from_pkcs12(data, pw, &synta_certificate::OpensslDecryptor)
        .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
    #[cfg(not(feature = "openssl"))]
    let pki = synta_certificate::pki_from_pkcs12(data, pw, &synta_certificate::NoCrypto)
        .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
    let certs = der_certs_to_pylist(py, pki.certs)?;
    let keys = PyList::empty(py);
    for der in pki.keys {
        keys.append(PyBytes::new(py, &der))?;
    }
    pyo3::types::PyTuple::new(py, [certs.into_any(), keys.into_any()])
}

/// Auto-detect the encoding of ``data`` and return every PKI object as a
/// ``(label, der_bytes)`` tuple, in document order.
///
/// Supported formats: PEM (any label), PKCS#7 / CMS SignedData (DER or BER),
/// PKCS#12 PFX (DER or BER), and raw DER.
///
/// **Labels:**
///
/// - PEM blocks whose payload is a PKCS#7 ``ContentInfo`` (detected by
///   structure, not label) are expanded: the embedded certificates are
///   extracted and each is returned as ``("CERTIFICATE", der)``.
/// - All other PEM block types pass through with their original label
///   (e.g. ``"CERTIFICATE"``, ``"PRIVATE KEY"``).  Malformed PEM blocks
///   (bad base64, truncated header) are silently skipped.
/// - Binary PKCS#7 and raw DER yield ``"CERTIFICATE"`` labels.
/// - Binary PKCS#12 yields ``"CERTIFICATE"`` for ``certBag`` entries and
///   ``"PRIVATE KEY"`` for ``keyBag`` / ``pkcs8ShroudedKeyBag`` entries.
///
/// **Decryption:**
///
/// ``password`` applies only to PKCS#12 archives.  When ``password`` is
/// given **and** the library was built with the ``openssl`` Cargo feature,
/// encrypted SafeBags are decrypted.  Otherwise encrypted bags are silently
/// skipped; unencrypted certificates in the same archive are still returned.
///
/// :raises ValueError: For any structural failure:
///
///   - A PKCS#7 block (binary or PEM-wrapped) has malformed DER, or its
///     ``contentType`` OID is not ``id-signedData``.
///   - The PKCS#12 input has a structural ASN.1 error.
///   - ``password`` was supplied with the ``openssl`` feature enabled and
///     decryption of an encrypted SafeBag failed (e.g. wrong password,
///     unsupported cipher).
///   - The PKCS#12 authSafe uses an unsupported ``ContentInfo`` type.
///
/// :raises TypeError: If ``data`` is not :class:`bytes`, or ``password``
///   is not :class:`bytes` or ``None``.
///
/// ```python,ignore
/// data = open("bundle.p7b", "rb").read()
/// blocks = synta.read_pki_blocks(data)
/// for label, der in blocks:
///     print(f"{label}: {len(der)} bytes")
///
/// # PKCS#12 with encryption
/// data = open("archive.p12", "rb").read()
/// blocks = synta.read_pki_blocks(data, b"s3cr3t")
/// ```
#[pyfunction]
#[pyo3(signature = (data, password = None))]
pub fn read_pki_blocks<'py>(
    py: Python<'py>,
    data: &[u8],
    password: Option<&[u8]>,
) -> PyResult<Bound<'py, PyList>> {
    let pw = password.unwrap_or(b"");
    #[cfg(feature = "openssl")]
    let blocks = if password.is_some() {
        synta_certificate::read_pki_blocks(
            data,
            pw,
            Some(&synta_certificate::OpensslDecryptor as &dyn synta_certificate::PkiDecryptor),
        )
    } else {
        synta_certificate::read_pki_blocks(data, pw, None::<&dyn synta_certificate::PkiDecryptor>)
    };
    #[cfg(not(feature = "openssl"))]
    let blocks =
        synta_certificate::read_pki_blocks(data, pw, None::<&dyn synta_certificate::PkiDecryptor>);
    let blocks = blocks.map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
    let list = PyList::empty(py);
    for (label, der) in blocks {
        let tuple = pyo3::types::PyTuple::new(
            py,
            [
                PyString::new(py, &label).into_any(),
                PyBytes::new(py, &der).into_any(),
            ],
        )?;
        list.append(tuple)?;
    }
    Ok(list)
}

// ── X.509 encoding utilities ──────────────────────────────────────────────────

/// Encode a list of OIDs as a DER ``SEQUENCE OF OID`` (ExtendedKeyUsage value).
///
/// ``oids`` must be a :class:`list` of :class:`~synta.ObjectIdentifier` or
/// dotted-decimal ``str`` values.  Returns the complete DER bytes of the
/// ``SEQUENCE``, suitable for use as an X.509v3 Extended Key Usage extension
/// value.
///
/// Delegates to synta's ``Vec<ObjectIdentifier>`` encoder which matches the
/// generated ``ExtendedKeyUsage = Vec<KeyPurposeId>`` type from the X.509 schema.
///
/// ```python,ignore
/// eku_der = synta.encode_extended_key_usage([
///     "1.3.6.1.5.5.7.3.1",   # id-kp-serverAuth
///     "1.3.6.1.5.5.7.3.2",   # id-kp-clientAuth
/// ])
/// ```
#[pyfunction]
pub fn encode_extended_key_usage<'py>(
    py: Python<'py>,
    oids: &Bound<'_, PyList>,
) -> PyResult<Bound<'py, PyBytes>> {
    use synta::traits::Encode;
    let mut eku: Vec<synta::ObjectIdentifier> = Vec::with_capacity(oids.len());
    for item in oids.iter() {
        eku.push(super::oid_from_pyany(&item)?);
    }
    let mut enc = synta::Encoder::new(synta::Encoding::Der);
    eku.encode(&mut enc)
        .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
    let der = enc
        .finish()
        .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
    Ok(PyBytes::new(py, &der))
}

/// Encode a list of ``(tag_number, content_bytes)`` pairs as a DER
/// ``SEQUENCE OF GeneralName``.
///
/// This is the inverse of :func:`~synta.parse_general_names`: the input
/// format is identical to what that function returns.  Tag numbers follow
/// RFC 5280 §4.2.1.6; use the :mod:`synta.general_name` constants.
///
/// ``content_bytes`` must be the raw **value** bytes — without the outer
/// context-specific TLV — exactly as returned by
/// :func:`~synta.parse_general_names`.
///
/// Delegates to :func:`synta_certificate::encode_general_names` which uses the
/// generated :class:`GeneralName` enum's ``Encode`` implementation.
///
/// ```python,ignore
/// import synta.general_name as gn
///
/// san_der = synta.encode_subject_alt_names([
///     (gn.DNS_NAME,   b"example.com"),
///     (gn.IP_ADDRESS, b"\x7f\x00\x00\x01"),   # 127.0.0.1
/// ])
/// ```
#[pyfunction]
pub fn encode_subject_alt_names<'py>(
    py: Python<'py>,
    names: &Bound<'_, PyList>,
) -> PyResult<Bound<'py, PyBytes>> {
    // Collect owned (tag, bytes) first so we can create &[u8] references that
    // all share the same lifetime, satisfying Vec<GeneralName<'_>> constraints.
    let mut owned: Vec<(u32, Vec<u8>)> = Vec::with_capacity(names.len());
    for item in names.iter() {
        let tuple = item.cast::<pyo3::types::PyTuple>().map_err(|_| {
            pyo3::exceptions::PyTypeError::new_err(
                "each element must be a (tag_number, bytes) 2-tuple",
            )
        })?;
        if tuple.len() != 2 {
            return Err(pyo3::exceptions::PyTypeError::new_err(
                "each element must be a (tag_number, bytes) 2-tuple",
            ));
        }
        let tag_num: u32 = tuple.get_item(0)?.extract()?;
        let content: Vec<u8> = tuple.get_item(1)?.extract()?;
        owned.push((tag_num, content));
    }
    let entries: Vec<(u32, &[u8])> = owned.iter().map(|(t, v)| (*t, v.as_slice())).collect();
    let der = synta_certificate::encode_general_names(&entries).ok_or_else(|| {
        pyo3::exceptions::PyValueError::new_err(
            "failed to encode GeneralName entries (invalid content bytes or unsupported tag)",
        )
    })?;
    Ok(PyBytes::new(py, &der))
}

/// Compare two DER-encoded X.500 Names for byte-for-byte equality.
///
/// Returns ``True`` if and only if the two byte sequences are identical.
/// This is a strict comparison; RFC 5280 name matching rules (case-insensitive
/// string comparison, etc.) are **not** applied.
///
/// Useful for quickly checking whether issuer and subject names match without
/// decoding the full structure:
///
/// ```python,ignore
/// same = synta.name_der_equal(cert.issuer_raw_der, ca_cert.subject_raw_der)
/// ```
#[pyfunction]
pub fn name_der_equal(a: &[u8], b: &[u8]) -> bool {
    a == b
}

/// Build a DER-encoded PKCS#12 ``PFX`` archive.
///
/// Assembles a ``PFX`` from one or more certificates and an optional
/// DER-encoded PKCS#8 private key.  The archive is protected by ``password``
/// using PBES2 (PBKDF2-SHA256 + AES-256-CBC) for key encryption and
/// HMAC-SHA256 (PBKDF2-SHA256 derived key) for the integrity MAC.
///
/// :param certificates: sequence of :class:`~synta.Certificate` objects or
///     raw DER ``bytes``.  Both types may appear in the same list.
/// :param private_key: a :class:`~synta.PrivateKey` object or a DER-encoded
///     PKCS#8 ``OneAsymmetricKey`` ``bytes``, or ``None`` to omit.
/// :param password: archive password (``bytes``), or ``None`` for a
///     password-less archive.
/// :returns: DER-encoded ``PFX`` archive bytes.
/// :raises TypeError: if any element of ``certificates`` is neither a
///     :class:`~synta.Certificate` nor ``bytes``, or if ``private_key`` is
///     not a :class:`~synta.PrivateKey` nor ``bytes``.
/// :raises ValueError: if no certificates or key are supplied, or if
///     encryption fails (e.g. OpenSSL not available).
///
/// ```python,ignore
/// # Certificate and PrivateKey objects — no .to_der() call needed
/// pfx = synta.create_pkcs12(
///     [cert, ca_cert],
///     private_key=key,
///     password=b"s3cr3t",
/// )
///
/// # Raw DER bytes also accepted for both
/// pfx = synta.create_pkcs12([cert.to_der()], private_key=key.to_der())
///
/// with open("out.p12", "wb") as f:
///     f.write(pfx)
/// ```
#[pyfunction]
#[pyo3(signature = (certificates, private_key = None, password = None))]
pub fn create_pkcs12<'py>(
    py: Python<'py>,
    certificates: &Bound<'_, PyList>,
    private_key: Option<Bound<'_, PyAny>>,
    password: Option<&[u8]>,
) -> PyResult<Bound<'py, PyBytes>> {
    use synta_certificate::Pkcs12Builder;

    let password = password.unwrap_or(b"");
    let mut builder = Pkcs12Builder::new();
    for item in certificates.iter() {
        if let Ok(py_cert) = item.cast::<PyCertificate>() {
            // Zero-copy: raw is &'static [u8] backed by the PyBytes in _data.
            // Pkcs12Builder::certificate() copies into Vec<u8> immediately, so
            // the borrow does not outlive this loop iteration.
            builder = builder.certificate(py_cert.get().raw);
        } else if let Ok(py_bytes) = item.cast::<PyBytes>() {
            builder = builder.certificate(py_bytes.as_bytes());
        } else {
            return Err(pyo3::exceptions::PyTypeError::new_err(
                "certificates must be synta.Certificate or bytes",
            ));
        }
    }
    if let Some(key_arg) = private_key {
        if let Ok(py_key) = key_arg.cast::<PyPrivateKey>() {
            // PyPrivateKey stores PKCS#8 DER internally; retrieve it directly.
            let key_der = py_key
                .get()
                .inner
                .to_der()
                .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
            builder = builder.private_key(&key_der);
        } else if let Ok(py_bytes) = key_arg.cast::<PyBytes>() {
            builder = builder.private_key(py_bytes.as_bytes());
        } else {
            return Err(pyo3::exceptions::PyTypeError::new_err(
                "private_key must be synta.PrivateKey or bytes",
            ));
        }
    }

    #[cfg(feature = "openssl")]
    let pfx_der = builder
        .build(password, &synta_certificate::OpensslPkcs12Encryptor::new())
        .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;

    #[cfg(not(feature = "openssl"))]
    let pfx_der = {
        use synta_certificate::NoPkcs12Encryptor;
        builder
            .build(password, &NoPkcs12Encryptor)
            .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?
    };

    Ok(PyBytes::new(py, &pfx_der))
}