synta-python 0.2.5

//! Python bindings for basic CMS container types:
//! [`PyContentInfo`] (RFC 5652 §3) and [`PyIssuerAndSerialNumber`] (RFC 5652 §10.2.4).

use std::sync::OnceLock;

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

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

use crate::types::PyObjectIdentifier;

// ── PyContentInfo ─────────────────────────────────────────────────────────────

/// CMS ContentInfo (RFC 5652 §3) accessible from Python.
///
/// ``ContentInfo`` is the outermost CMS envelope.  It names the content type
/// via an OID and carries the content itself (e.g. a ``SignedData`` or
/// ``EnvelopedData`` SEQUENCE) as an opaque DER blob.
///
/// ```python,ignore
/// ci = ContentInfo.from_der(raw)
/// if ci.content_type_oid == synta.cms.ID_SIGNED_DATA:
///     signed_data_der = ci.content   # raw DER of the [0] EXPLICIT wrapper
/// ```
#[pyclass(frozen, name = "ContentInfo")]
pub struct PyContentInfo {
    _data: Py<PyBytes>,
    raw: &'static [u8],
    inner: OnceLock<Box<synta_certificate::pkcs7_types::ContentInfo<'static>>>,
    content_type_oid_cache: OnceLock<Py<PyObjectIdentifier>>,
    content_cache: OnceLock<Py<PyBytes>>,
}

impl PyContentInfo {
    fn content_info(&self) -> PyResult<&synta_certificate::pkcs7_types::ContentInfo<'static>> {
        if let Some(v) = self.inner.get() {
            return Ok(v.as_ref());
        }
        // Use BER so that indefinite-length PKCS#7 wrappers are accepted.
        let mut decoder = Decoder::new(self.raw, Encoding::Ber);
        let decoded = decoder.decode().map_err(|e| {
            pyo3::exceptions::PyValueError::new_err(format!("ContentInfo BER decode failed: {e}"))
        })?;
        let _ = self.inner.set(Box::new(decoded));
        Ok(self.inner.get().unwrap().as_ref())
    }
}

#[pymethods]
impl PyContentInfo {
    /// Parse a DER- or BER-encoded CMS ``ContentInfo`` SEQUENCE.
    ///
    /// BER indefinite-length encodings (common in PKCS#7 files produced by
    /// older tools and OpenSSL) are accepted in addition to strict DER.
    /// Raises :exc:`ValueError` if the outer SEQUENCE envelope is malformed.
    #[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(),
            content_type_oid_cache: OnceLock::new(),
            content_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)
    }

    /// OID identifying the content type
    /// (e.g. :data:`ID_SIGNED_DATA`, :data:`ID_ENVELOPED_DATA`).
    #[getter]
    fn content_type_oid<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyObjectIdentifier>> {
        if let Some(cached) = self.content_type_oid_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let obj = Py::new(
            py,
            PyObjectIdentifier::from_oid(self.content_info()?.content_type.clone()),
        )?;
        let _ = self.content_type_oid_cache.set(obj.clone_ref(py));
        Ok(obj.into_bound(py))
    }

    /// Raw DER bytes of the ``content`` field.
    ///
    /// This is the entire ``[0] EXPLICIT`` TLV; callers that need the inner
    /// value (e.g. the ``SignedData`` SEQUENCE) should strip the context tag
    /// and length with a :class:`~synta.Decoder`.
    #[getter]
    fn content<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        if let Some(cached) = self.content_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let py_bytes = PyBytes::new(py, self.content_info()?.content.as_bytes()).unbind();
        let _ = self.content_cache.set(py_bytes.clone_ref(py));
        Ok(py_bytes.into_bound(py))
    }

    fn __repr__(&self) -> PyResult<String> {
        Ok(format!(
            "ContentInfo(content_type={})",
            self.content_info()?.content_type,
        ))
    }
}

// ── PyIssuerAndSerialNumber ───────────────────────────────────────────────────

/// CMS IssuerAndSerialNumber (RFC 5652 §10.2.4) accessible from Python.
///
/// Identifies a recipient or signer by the issuer distinguished name and
/// certificate serial number of their certificate.  Used as one alternative
/// of the ``RecipientIdentifier`` CHOICE inside :class:`KEMRecipientInfo`.
///
/// ```python,ignore
/// ias = IssuerAndSerialNumber.from_der(raw)
/// print(ias.issuer)          # "CN=My CA,O=Example,C=US"
/// print(ias.serial_number)   # 12345678
/// ```
#[pyclass(frozen, name = "IssuerAndSerialNumber")]
pub struct PyIssuerAndSerialNumber {
    _data: Py<PyBytes>,
    raw: &'static [u8],
    inner: OnceLock<Box<synta_certificate::cms_2010_types::IssuerAndSerialNumber<'static>>>,
    issuer_cache: OnceLock<Py<PyString>>,
    issuer_raw_der_cache: OnceLock<Py<PyBytes>>,
    serial_number_cache: OnceLock<Py<PyAny>>,
}

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

/// 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()
}

#[pymethods]
impl PyIssuerAndSerialNumber {
    /// Parse a DER-encoded ``IssuerAndSerialNumber`` SEQUENCE.
    /// Raises :exc:`ValueError` if the outer SEQUENCE envelope is malformed.
    #[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(),
            issuer_cache: OnceLock::new(),
            issuer_raw_der_cache: OnceLock::new(),
            serial_number_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)
    }

    /// Issuer distinguished name as an RFC 4514-style 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.ias()?.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.ias()?.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))
    }

    /// Certificate serial number as a Python ``int``.
    ///
    /// Returns a Python ``int`` for serials that fit in 128 bits; for larger
    /// values calls ``int.from_bytes(raw, "big", signed=True)``.
    #[getter]
    fn serial_number<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
        if let Some(cached) = self.serial_number_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let serial = &self.ias()?.serial_number;
        let py_int: Py<PyAny> = if let Ok(v) = serial.as_i64() {
            v.into_pyobject(py)?.into_any().unbind()
        } else if let Ok(v) = serial.as_i128() {
            v.into_pyobject(py)?.into_any().unbind()
        } else {
            let bytes_obj = PyBytes::new(py, serial.as_bytes());
            let kwargs = pyo3::types::PyDict::new(py);
            kwargs.set_item(pyo3::intern!(py, "signed"), true)?;
            py.get_type::<pyo3::types::PyInt>()
                .call_method(
                    pyo3::intern!(py, "from_bytes"),
                    (bytes_obj, pyo3::intern!(py, "big")),
                    Some(&kwargs),
                )
                .map(|r| r.unbind())?
        };
        let _ = self.serial_number_cache.set(py_int.clone_ref(py));
        Ok(py_int.into_bound(py))
    }

    fn __repr__(&self) -> PyResult<String> {
        let ias = self.ias()?;
        let serial = &ias.serial_number;
        Ok(format!(
            "IssuerAndSerialNumber(issuer={:?}, serial={})",
            name_to_dn_string(&ias.issuer),
            serial
                .as_i64()
                .map(|v| v.to_string())
                .unwrap_or_else(|_| format!("<{} bytes>", serial.as_bytes().len())),
        ))
    }
}