synta-python 0.2.5

//! Python bindings for CMS signature types:
//! [`PySignedData`] (RFC 5652 §5) and [`PySignerInfo`] (RFC 5652 §5.3).

use std::sync::OnceLock;

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

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

use crate::types::PyObjectIdentifier;

// ── PySignedData ──────────────────────────────────────────────────────────────

/// CMS ``SignedData`` (RFC 5652 §5) accessible from Python.
///
/// Obtained by parsing the inner content of a :class:`ContentInfo` whose
/// ``content_type_oid`` equals :data:`~synta.cms.ID_SIGNED_DATA`.  Strip the
/// ``[0] EXPLICIT`` wrapper from ``ci.content`` before calling
/// :meth:`from_der`.
///
/// ```python,ignore
/// ci = ContentInfo.from_der(raw)
/// sd = SignedData.from_der(ci.content[4:])  # strip [0] EXPLICIT tag+len
/// print(sd.version)
/// for si in sd.signer_infos:
///     print(si.digest_algorithm_oid)
/// ```
#[pyclass(frozen, name = "SignedData")]
pub struct PySignedData {
    _data: Py<PyBytes>,
    raw: &'static [u8],
    inner: OnceLock<Box<synta_certificate::cms_rfc5652_types::SignedData<'static>>>,
    encap_content_type_cache: OnceLock<Py<PyObjectIdentifier>>,
    encap_content_cache: OnceLock<Option<Py<PyBytes>>>,
    certificates_cache: OnceLock<Option<Py<PyBytes>>>,
    crls_cache: OnceLock<Option<Py<PyBytes>>>,
    signer_infos_cache: OnceLock<Py<PyList>>,
}

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

#[pymethods]
impl PySignedData {
    /// Parse a DER- or BER-encoded CMS ``SignedData`` SEQUENCE.
    #[staticmethod]
    fn from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
        let py_bytes = data.unbind();
        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::Ber);
            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(),
            encap_content_type_cache: OnceLock::new(),
            encap_content_cache: OnceLock::new(),
            certificates_cache: OnceLock::new(),
            crls_cache: OnceLock::new(),
            signer_infos_cache: OnceLock::new(),
        })
    }

    /// Return the original bytes passed to :meth:`from_der`.
    fn to_der<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
        self._data.clone_ref(py).into_bound(py)
    }

    /// CMS version integer.
    #[getter]
    fn version(&self) -> PyResult<i64> {
        Ok(self.signed_data()?.version.as_i64().unwrap_or(0))
    }

    /// OID identifying the encapsulated content type.
    #[getter]
    fn encap_content_type<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyObjectIdentifier>> {
        if let Some(cached) = self.encap_content_type_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let obj = Py::new(
            py,
            PyObjectIdentifier::from_oid(
                self.signed_data()?
                    .encap_content_info
                    .e_content_type
                    .clone(),
            ),
        )?;
        let _ = self.encap_content_type_cache.set(obj.clone_ref(py));
        Ok(obj.into_bound(py))
    }

    /// Raw bytes of the encapsulated content OCTET STRING value, or ``None``.
    #[getter]
    fn encap_content<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
        if let Some(cached) = self.encap_content_cache.get() {
            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
        }
        let computed = self
            .signed_data()?
            .encap_content_info
            .e_content
            .as_ref()
            .map(|c| PyBytes::new(py, c.as_bytes()));
        let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
        let _ = self.encap_content_cache.set(to_store);
        Ok(computed)
    }

    /// Raw content bytes of the ``certificates`` field (IMPLICIT ``[0]``), or ``None``.
    ///
    /// These are the **value** bytes captured during IMPLICIT decoding (no
    /// context-tag prefix).  Feed them to a :class:`~synta.Decoder` to iterate
    /// the individual certificate TLVs.
    #[getter]
    fn certificates<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
        if let Some(cached) = self.certificates_cache.get() {
            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
        }
        let computed = self
            .signed_data()?
            .certificates
            .as_ref()
            .map(|c| PyBytes::new(py, c.as_bytes()));
        let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
        let _ = self.certificates_cache.set(to_store);
        Ok(computed)
    }

    /// Raw content bytes of the ``crls`` field (IMPLICIT ``[1]``), or ``None``.
    #[getter]
    fn crls<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
        if let Some(cached) = self.crls_cache.get() {
            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
        }
        let computed = self
            .signed_data()?
            .crls
            .as_ref()
            .map(|c| PyBytes::new(py, c.as_bytes()));
        let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
        let _ = self.crls_cache.set(to_store);
        Ok(computed)
    }

    /// List of :class:`SignerInfo` objects (one per signer).
    #[getter]
    fn signer_infos<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
        if let Some(cached) = self.signer_infos_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let list = PyList::empty(py);
        for si in self.signed_data()?.signer_infos.elements() {
            let mut enc = synta::Encoder::new(Encoding::Der);
            si.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}")))?;
            let pybytes = PyBytes::new(py, &der).unbind();
            // SAFETY: `pybytes` is a freshly-allocated PyBytes that owns the
            // `der` bytes.  CPython bytes have a fixed-address payload buffer.
            // `pybytes` is moved into `PySignerInfo::_data`, which holds a
            // strong reference for the lifetime of that struct.  `raw_static`
            // (no destructor) is stored in `PySignerInfo::raw`.  No read of
            // `raw_static` can occur after `_data` is dropped (through `&self`
            // borrows that cannot outlive the struct).
            // CPython-only: does not hold for PyPy or GraalPy.
            let raw_static: &'static [u8] = unsafe {
                let s = pybytes.bind(py).as_bytes();
                std::slice::from_raw_parts(s.as_ptr(), s.len())
            };
            let si_obj = Py::new(
                py,
                PySignerInfo {
                    _data: pybytes,
                    raw: raw_static,
                    inner: OnceLock::new(),
                    sid_cache: OnceLock::new(),
                    digest_algorithm_oid_cache: OnceLock::new(),
                    digest_algorithm_params_cache: OnceLock::new(),
                    signature_algorithm_oid_cache: OnceLock::new(),
                    signature_algorithm_params_cache: OnceLock::new(),
                    signature_cache: OnceLock::new(),
                    signed_attrs_cache: OnceLock::new(),
                    unsigned_attrs_cache: OnceLock::new(),
                },
            )?;
            list.append(si_obj.into_bound(py))?;
        }
        let list_unbound = list.unbind();
        let _ = self.signer_infos_cache.set(list_unbound.clone_ref(py));
        Ok(list_unbound.into_bound(py))
    }

    fn __repr__(&self) -> PyResult<String> {
        let sd = self.signed_data()?;
        Ok(format!(
            "SignedData(version={}, signer_count={})",
            sd.version.as_i64().unwrap_or(0),
            sd.signer_infos.len(),
        ))
    }
}

// ── PySignerInfo ──────────────────────────────────────────────────────────────

/// CMS ``SignerInfo`` (RFC 5652 §5.3) accessible from Python.
///
/// Represents one signer's information within a :class:`SignedData`.
/// Retrieve instances via :attr:`SignedData.signer_infos` or construct
/// directly with :meth:`from_der`.
///
/// ```python,ignore
/// for si in sd.signer_infos:
///     print(si.digest_algorithm_oid)
///     print(si.signature_algorithm_oid)
/// ```
#[pyclass(frozen, name = "SignerInfo")]
pub struct PySignerInfo {
    pub(super) _data: Py<PyBytes>,
    pub(super) raw: &'static [u8],
    pub(super) inner: OnceLock<Box<synta_certificate::cms_rfc5652_types::SignerInfo<'static>>>,
    pub(super) sid_cache: OnceLock<Py<PyBytes>>,
    pub(super) digest_algorithm_oid_cache: OnceLock<Py<PyObjectIdentifier>>,
    pub(super) digest_algorithm_params_cache: OnceLock<Option<Py<PyBytes>>>,
    pub(super) signature_algorithm_oid_cache: OnceLock<Py<PyObjectIdentifier>>,
    pub(super) signature_algorithm_params_cache: OnceLock<Option<Py<PyBytes>>>,
    pub(super) signature_cache: OnceLock<Py<PyBytes>>,
    pub(super) signed_attrs_cache: OnceLock<Option<Py<PyBytes>>>,
    pub(super) unsigned_attrs_cache: OnceLock<Option<Py<PyBytes>>>,
}

impl PySignerInfo {
    fn signer_info(&self) -> PyResult<&synta_certificate::cms_rfc5652_types::SignerInfo<'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!("SignerInfo DER decode failed: {e}"))
        })?;
        let _ = self.inner.set(Box::new(decoded));
        Ok(self.inner.get().unwrap().as_ref())
    }
}

#[pymethods]
impl PySignerInfo {
    /// Parse a DER-encoded CMS ``SignerInfo`` SEQUENCE.
    #[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 in the decoded type;
        //       dropping the Box does not read through the contained &'static
        //       slices (borrows have no destructors in Rust).
        // 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(),
            sid_cache: OnceLock::new(),
            digest_algorithm_oid_cache: OnceLock::new(),
            digest_algorithm_params_cache: OnceLock::new(),
            signature_algorithm_oid_cache: OnceLock::new(),
            signature_algorithm_params_cache: OnceLock::new(),
            signature_cache: OnceLock::new(),
            signed_attrs_cache: OnceLock::new(),
            unsigned_attrs_cache: OnceLock::new(),
        })
    }

    /// Return the original bytes passed to :meth:`from_der`.
    fn to_der<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
        self._data.clone_ref(py).into_bound(py)
    }

    /// CMS version integer.
    #[getter]
    fn version(&self) -> PyResult<i64> {
        Ok(self.signer_info()?.version.as_i64().unwrap_or(0))
    }

    /// Raw TLV bytes of the ``SignerIdentifier`` CHOICE field.
    ///
    /// Peek the tag byte to distinguish ``issuerAndSerialNumber`` (SEQUENCE,
    /// tag ``0x30``) from ``subjectKeyIdentifier`` (context ``[0]``, tag
    /// ``0x80``).
    #[getter]
    fn sid<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        if let Some(cached) = self.sid_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let py_bytes = PyBytes::new(py, self.signer_info()?.sid.as_bytes()).unbind();
        let _ = self.sid_cache.set(py_bytes.clone_ref(py));
        Ok(py_bytes.into_bound(py))
    }

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

    /// Raw DER bytes of the digest-algorithm parameters, or ``None``.
    #[getter]
    fn digest_algorithm_params<'py>(
        &self,
        py: Python<'py>,
    ) -> PyResult<Option<Bound<'py, PyBytes>>> {
        if let Some(cached) = self.digest_algorithm_params_cache.get() {
            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
        }
        let computed = super::encode_element_opt(
            py,
            self.signer_info()?.digest_algorithm.parameters.as_ref(),
        )?;
        let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
        let _ = self.digest_algorithm_params_cache.set(to_store);
        Ok(computed)
    }

    /// 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.signer_info()?.signature_algorithm.algorithm.clone()),
        )?;
        let _ = self.signature_algorithm_oid_cache.set(obj.clone_ref(py));
        Ok(obj.into_bound(py))
    }

    /// Raw DER bytes of the signature-algorithm parameters, or ``None``.
    #[getter]
    fn signature_algorithm_params<'py>(
        &self,
        py: Python<'py>,
    ) -> PyResult<Option<Bound<'py, PyBytes>>> {
        if let Some(cached) = self.signature_algorithm_params_cache.get() {
            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
        }
        let computed = super::encode_element_opt(
            py,
            self.signer_info()?.signature_algorithm.parameters.as_ref(),
        )?;
        let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
        let _ = self.signature_algorithm_params_cache.set(to_store);
        Ok(computed)
    }

    /// Raw bytes of the signature value.
    #[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.signer_info()?.signature.as_bytes()).unbind();
        let _ = self.signature_cache.set(py_bytes.clone_ref(py));
        Ok(py_bytes.into_bound(py))
    }

    /// Raw content bytes of the ``signedAttrs`` field (IMPLICIT ``[0]``), or ``None``.
    ///
    /// To verify the signature, replace the leading context-specific tag byte
    /// with the SET tag ``0x31`` before hashing.
    #[getter]
    fn signed_attrs<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
        if let Some(cached) = self.signed_attrs_cache.get() {
            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
        }
        let computed = self
            .signer_info()?
            .signed_attrs
            .as_ref()
            .map(|a| PyBytes::new(py, a.as_bytes()));
        let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
        let _ = self.signed_attrs_cache.set(to_store);
        Ok(computed)
    }

    /// Raw content bytes of the ``unsignedAttrs`` field (IMPLICIT ``[1]``), or ``None``.
    #[getter]
    fn unsigned_attrs<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
        if let Some(cached) = self.unsigned_attrs_cache.get() {
            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
        }
        let computed = self
            .signer_info()?
            .unsigned_attrs
            .as_ref()
            .map(|a| PyBytes::new(py, a.as_bytes()));
        let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
        let _ = self.unsigned_attrs_cache.set(to_store);
        Ok(computed)
    }

    fn __repr__(&self) -> PyResult<String> {
        let si = self.signer_info()?;
        Ok(format!(
            "SignerInfo(version={}, digest_algorithm={})",
            si.version.as_i64().unwrap_or(0),
            si.digest_algorithm.algorithm,
        ))
    }
}