synta-python 0.3.0

//! Python bindings for CMS digest and authentication types:
//! [`PyDigestedData`] (RFC 5652 §7) and [`PyAuthenticatedData`] (RFC 5652 §9).

use std::sync::OnceLock;

use pyo3::prelude::*;
use pyo3::types::PyBytes;

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

use crate::types::PyObjectIdentifier;

// ── PyDigestedData ────────────────────────────────────────────────────────────

/// CMS ``DigestedData`` (RFC 5652 §7) accessible from Python.
///
/// Obtained by parsing the inner content of a :class:`ContentInfo` whose
/// ``content_type_oid`` is the id-digestedData OID.
#[pyclass(frozen, name = "DigestedData")]
pub struct PyDigestedData {
    _data: Py<PyBytes>,
    raw: &'static [u8],
    inner: OnceLock<Box<synta_certificate::cms_rfc5652_types::DigestedData<'static>>>,
    digest_algorithm_oid_cache: OnceLock<Py<PyObjectIdentifier>>,
    digest_algorithm_params_cache: OnceLock<Option<Py<PyBytes>>>,
    encap_content_type_cache: OnceLock<Py<PyObjectIdentifier>>,
    encap_content_cache: OnceLock<Option<Py<PyBytes>>>,
    digest_cache: OnceLock<Py<PyBytes>>,
}

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

#[pymethods]
impl PyDigestedData {
    /// Parse a DER- or BER-encoded CMS ``DigestedData`` 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(),
            digest_algorithm_oid_cache: OnceLock::new(),
            digest_algorithm_params_cache: OnceLock::new(),
            encap_content_type_cache: OnceLock::new(),
            encap_content_cache: OnceLock::new(),
            digest_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.digested_data()?.version.as_i64().unwrap_or(0))
    }

    /// OID of the 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.digested_data()?.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.digested_data()?.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 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.digested_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
            .digested_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 bytes of the message digest (OCTET STRING value).
    #[getter]
    fn digest<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        if let Some(cached) = self.digest_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let py_bytes = PyBytes::new(py, self.digested_data()?.digest.as_bytes()).unbind();
        let _ = self.digest_cache.set(py_bytes.clone_ref(py));
        Ok(py_bytes.into_bound(py))
    }

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

// ── PyAuthenticatedData ───────────────────────────────────────────────────────

/// CMS ``AuthenticatedData`` (RFC 5652 §9) accessible from Python.
///
/// Obtained by parsing the inner content of a :class:`ContentInfo` whose
/// ``content_type_oid`` is the id-ct-authData OID.
#[pyclass(frozen, name = "AuthenticatedData")]
pub struct PyAuthenticatedData {
    _data: Py<PyBytes>,
    raw: &'static [u8],
    inner: OnceLock<Box<synta_certificate::cms_rfc5652_types::AuthenticatedData<'static>>>,
    originator_info_cache: OnceLock<Option<Py<PyBytes>>>,
    recipient_infos_cache: OnceLock<Py<PyBytes>>,
    mac_algorithm_oid_cache: OnceLock<Py<PyObjectIdentifier>>,
    mac_algorithm_params_cache: OnceLock<Option<Py<PyBytes>>>,
    digest_algorithm_oid_cache: OnceLock<Option<Py<PyObjectIdentifier>>>,
    digest_algorithm_params_cache: OnceLock<Option<Py<PyBytes>>>,
    encap_content_type_cache: OnceLock<Py<PyObjectIdentifier>>,
    encap_content_cache: OnceLock<Option<Py<PyBytes>>>,
    mac_cache: OnceLock<Py<PyBytes>>,
    auth_attrs_cache: OnceLock<Option<Py<PyBytes>>>,
    unauth_attrs_cache: OnceLock<Option<Py<PyBytes>>>,
}

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

#[pymethods]
impl PyAuthenticatedData {
    /// Parse a DER- or BER-encoded CMS ``AuthenticatedData`` 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(),
            originator_info_cache: OnceLock::new(),
            recipient_infos_cache: OnceLock::new(),
            mac_algorithm_oid_cache: OnceLock::new(),
            mac_algorithm_params_cache: OnceLock::new(),
            digest_algorithm_oid_cache: OnceLock::new(),
            digest_algorithm_params_cache: OnceLock::new(),
            encap_content_type_cache: OnceLock::new(),
            encap_content_cache: OnceLock::new(),
            mac_cache: OnceLock::new(),
            auth_attrs_cache: OnceLock::new(),
            unauth_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.authenticated_data()?.version.as_i64().unwrap_or(0))
    }

    /// DER-encoded ``OriginatorInfo`` SEQUENCE, or ``None`` if absent.
    #[getter]
    fn originator_info<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
        if let Some(cached) = self.originator_info_cache.get() {
            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
        }
        let computed = match &self.authenticated_data()?.originator_info {
            None => None,
            Some(oi) => {
                let mut enc = synta::Encoder::new(Encoding::Der);
                oi.encode(&mut enc)
                    .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
                let bytes = enc
                    .finish()
                    .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
                Some(PyBytes::new(py, &bytes))
            }
        };
        let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
        let _ = self.originator_info_cache.set(to_store);
        Ok(computed)
    }

    /// Raw TLV bytes of the ``RecipientInfos`` SET field.
    #[getter]
    fn recipient_infos<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        if let Some(cached) = self.recipient_infos_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let bytes =
            PyBytes::new(py, self.authenticated_data()?.recipient_infos.as_bytes()).unbind();
        let _ = self.recipient_infos_cache.set(bytes.clone_ref(py));
        Ok(bytes.into_bound(py))
    }

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

    /// Raw DER bytes of the MAC algorithm parameters, or ``None``.
    #[getter]
    fn mac_algorithm_params<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
        if let Some(cached) = self.mac_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.authenticated_data()?.mac_algorithm.parameters.as_ref(),
        )?;
        let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
        let _ = self.mac_algorithm_params_cache.set(to_store);
        Ok(computed)
    }

    /// OID of the digest algorithm (``[1] IMPLICIT``), or ``None`` if absent.
    #[getter]
    fn digest_algorithm_oid<'py>(
        &self,
        py: Python<'py>,
    ) -> PyResult<Option<Bound<'py, PyObjectIdentifier>>> {
        if let Some(cached) = self.digest_algorithm_oid_cache.get() {
            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
        }
        let computed: Option<Py<PyObjectIdentifier>> =
            match &self.authenticated_data()?.digest_algorithm {
                None => None,
                Some(da) => Some(Py::new(
                    py,
                    PyObjectIdentifier::from_oid(da.algorithm.clone()),
                )?),
            };
        let to_store = computed.as_ref().map(|b| b.clone_ref(py));
        let _ = self.digest_algorithm_oid_cache.set(to_store);
        Ok(computed.map(|b| b.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.authenticated_data()?
                .digest_algorithm
                .as_ref()
                .and_then(|da| da.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 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.authenticated_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
            .authenticated_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 bytes of the MAC (message authentication code) value.
    #[getter]
    fn mac<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        if let Some(cached) = self.mac_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let bytes = PyBytes::new(py, self.authenticated_data()?.mac.as_bytes()).unbind();
        let _ = self.mac_cache.set(bytes.clone_ref(py));
        Ok(bytes.into_bound(py))
    }

    /// Raw content bytes of the ``authAttrs`` field (IMPLICIT ``[2]``), or ``None``.
    ///
    /// To verify the MAC, replace the leading context-specific tag byte with
    /// the SET tag ``0x31`` before feeding to the MAC function.
    #[getter]
    fn auth_attrs<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
        if let Some(cached) = self.auth_attrs_cache.get() {
            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
        }
        let computed = self
            .authenticated_data()?
            .auth_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.auth_attrs_cache.set(to_store);
        Ok(computed)
    }

    /// Raw content bytes of the ``unauthAttrs`` field (IMPLICIT ``[3]``), or ``None``.
    #[getter]
    fn unauth_attrs<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
        if let Some(cached) = self.unauth_attrs_cache.get() {
            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
        }
        let computed = self
            .authenticated_data()?
            .unauth_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.unauth_attrs_cache.set(to_store);
        Ok(computed)
    }

    fn __repr__(&self) -> PyResult<String> {
        let ad = self.authenticated_data()?;
        Ok(format!(
            "AuthenticatedData(version={})",
            ad.version.as_i64().unwrap_or(0),
        ))
    }
}