synta-python 0.1.4

//! Python bindings for RFC 4211 Certificate Request Message Format (CRMF).
//!
//! Exposes ``CertReqMessages`` and ``CertReqMsg`` as Python classes and
//! installs registration-control OID constants into the ``synta.crmf``
//! submodule.

use std::sync::OnceLock;

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

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

use crate::error::SyntaErr;

// ── helpers ───────────────────────────────────────────────────────────────────

fn encode_to_der<T: Encode>(v: &T) -> Vec<u8> {
    let mut enc = synta::Encoder::new(Encoding::Der);
    if v.encode(&mut enc).is_err() {
        return Vec::new();
    }
    enc.finish().unwrap_or_default()
}

// ── PyCertReqMsg ──────────────────────────────────────────────────────────────

/// A single certificate request message (RFC 4211 §4).
///
/// Contains a ``CertRequest`` (with an ID and optional ``CertTemplate``) and
/// an optional ``ProofOfPossession``.  Obtained by iterating over a
/// :class:`CertReqMessages` object.
///
/// ``cert_template_der`` contains the re-encoded ``CertTemplate`` SEQUENCE;
/// feed it back to a ``synta.Decoder`` to inspect individual fields.
/// ``popo_der`` (if present) contains the re-encoded ``ProofOfPossession``
/// CHOICE; ``popo_type`` names the active arm.
#[pyclass(frozen, name = "CertReqMsg")]
pub struct PyCertReqMsg {
    cert_req_id: i64,
    cert_template_der: Vec<u8>,
    popo_type: Option<String>,
    popo_der: Option<Vec<u8>>,
    subject_der: Option<Vec<u8>>,
    issuer_der: Option<Vec<u8>>,
}

impl PyCertReqMsg {
    fn from_rust(msg: &synta_certificate::crmf_types::CertReqMsg<'_>) -> Self {
        let cert_req_id = msg.cert_req.cert_req_id.as_i64().unwrap_or(-1);

        let cert_template_der = encode_to_der(&msg.cert_req.cert_template);

        let (popo_type, popo_der) = match &msg.popo {
            None => (None, None),
            Some(pop) => {
                use synta_certificate::crmf_types::ProofOfPossession::*;
                let arm = match pop {
                    RaVerified(_) => "raVerified",
                    Signature(_) => "signature",
                    KeyEncipherment(_) => "keyEncipherment",
                    KeyAgreement(_) => "keyAgreement",
                };
                let der = encode_to_der(pop);
                (Some(arm.to_string()), Some(der))
            }
        };

        let subject_der = msg
            .cert_req
            .cert_template
            .subject
            .as_ref()
            .map(encode_to_der);
        let issuer_der = msg
            .cert_req
            .cert_template
            .issuer
            .as_ref()
            .map(encode_to_der);

        Self {
            cert_req_id,
            cert_template_der,
            popo_type,
            popo_der,
            subject_der,
            issuer_der,
        }
    }
}

#[pymethods]
impl PyCertReqMsg {
    /// The ``certReqId`` integer identifying this request within the batch.
    #[getter]
    fn cert_req_id(&self) -> i64 {
        self.cert_req_id
    }

    /// Raw DER bytes of the ``CertTemplate`` SEQUENCE.
    ///
    /// All fields in the template are optional.  Use a ``synta.Decoder`` to
    /// inspect individual fields.
    #[getter]
    fn cert_template_der<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
        PyBytes::new(py, &self.cert_template_der)
    }

    /// Name of the active ``ProofOfPossession`` arm, or ``None`` if absent.
    ///
    /// One of: ``"raVerified"``, ``"signature"``, ``"keyEncipherment"``,
    /// ``"keyAgreement"``.
    #[getter]
    fn popo_type<'py>(&self, py: Python<'py>) -> Option<Bound<'py, PyString>> {
        self.popo_type.as_deref().map(|s| PyString::new(py, s))
    }

    /// Raw DER bytes of the ``ProofOfPossession`` CHOICE, or ``None`` if absent.
    #[getter]
    fn popo_der<'py>(&self, py: Python<'py>) -> Option<Bound<'py, PyBytes>> {
        self.popo_der.as_deref().map(|b| PyBytes::new(py, b))
    }

    /// Raw DER bytes of the ``subject`` ``Name`` from the ``CertTemplate``,
    /// or ``None`` if the template does not include a subject.
    #[getter]
    fn subject_der<'py>(&self, py: Python<'py>) -> Option<Bound<'py, PyBytes>> {
        self.subject_der.as_deref().map(|b| PyBytes::new(py, b))
    }

    /// Raw DER bytes of the ``issuer`` ``Name`` from the ``CertTemplate``,
    /// or ``None`` if the template does not include an issuer.
    #[getter]
    fn issuer_der<'py>(&self, py: Python<'py>) -> Option<Bound<'py, PyBytes>> {
        self.issuer_der.as_deref().map(|b| PyBytes::new(py, b))
    }

    fn __repr__(&self) -> String {
        format!(
            "CertReqMsg(cert_req_id={}, popo_type={})",
            self.cert_req_id,
            self.popo_type.as_deref().unwrap_or("None"),
        )
    }
}

// ── PyCertReqMessages ─────────────────────────────────────────────────────────

/// A batch of Certificate Request Messages (RFC 4211 §3).
///
/// ``CertReqMessages`` is a ``SEQUENCE OF CertReqMsg``.  Each element
/// carries a certificate template, an optional proof-of-possession, and an
/// optional list of registration info attributes.
///
/// ```python,ignore
/// import synta.crmf as crmf
/// msgs = crmf.CertReqMessages.from_der(open("req.crmf", "rb").read())
/// for req in msgs:
///     print(req.cert_req_id, req.popo_type)
///     print(req.subject_der.hex() if req.subject_der else "(no subject)")
/// ```
#[pyclass(frozen, name = "CertReqMessages")]
pub struct PyCertReqMessages {
    _data: Py<PyBytes>,
    raw: &'static [u8],
    inner: OnceLock<Vec<synta_certificate::crmf_types::CertReqMsg<'static>>>,
    requests_cache: OnceLock<Py<PyList>>,
}

impl PyCertReqMessages {
    fn msgs(&self) -> PyResult<&Vec<synta_certificate::crmf_types::CertReqMsg<'static>>> {
        if let Some(v) = self.inner.get() {
            return Ok(v);
        }
        let mut dec = Decoder::new(self.raw, Encoding::Der);
        let decoded = dec
            .decode::<synta_certificate::crmf_types::CertReqMessages<'_>>()
            .map_err(SyntaErr)?;
        // SAFETY: raw is pinned by _data for the lifetime of self.
        let decoded: Vec<synta_certificate::crmf_types::CertReqMsg<'static>> =
            unsafe { std::mem::transmute(decoded) };
        let _ = self.inner.set(decoded);
        Ok(self.inner.get().unwrap())
    }

    fn build_requests_list<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
        let msgs = self.msgs()?;
        let list = PyList::empty(py);
        for msg in msgs {
            let obj = Py::new(py, PyCertReqMsg::from_rust(msg))?;
            list.append(obj)?;
        }
        Ok(list)
    }
}

#[pymethods]
impl PyCertReqMessages {
    /// Parse a DER-encoded ``CertReqMessages`` SEQUENCE OF.
    ///
    /// :param data: DER bytes of the ``CertReqMessages`` structure.
    /// :raises ValueError: if the bytes cannot be decoded.
    #[staticmethod]
    fn from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
        let py_bytes = data.unbind();
        {
            let raw = py_bytes.as_bytes(py);
            Decoder::new(raw, Encoding::Der)
                .decode::<synta_certificate::crmf_types::CertReqMessages<'_>>()
                .map_err(SyntaErr)?;
        }
        let raw: &'static [u8] = unsafe { std::mem::transmute(py_bytes.as_bytes(py)) };
        Ok(Self {
            _data: py_bytes,
            raw,
            inner: OnceLock::new(),
            requests_cache: OnceLock::new(),
        })
    }

    /// Return the DER encoding of this ``CertReqMessages`` SEQUENCE OF.
    fn to_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        let msgs = self.msgs()?;
        let mut enc = synta::Encoder::new(Encoding::Der);
        msgs.encode(&mut enc).map_err(SyntaErr)?;
        Ok(PyBytes::new(py, &enc.finish().map_err(SyntaErr)?))
    }

    /// List of :class:`CertReqMsg` objects, one per request in the batch.
    #[getter]
    fn requests<'py>(&'py self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
        if let Some(c) = self.requests_cache.get() {
            return Ok(c.clone_ref(py).into_bound(py));
        }
        let list = self.build_requests_list(py)?;
        let _ = self.requests_cache.set(list.as_unbound().clone_ref(py));
        Ok(list)
    }

    fn __len__(&self) -> PyResult<usize> {
        Ok(self.msgs()?.len())
    }

    fn __iter__<'py>(&'py self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
        self.requests(py)?.call_method0("__iter__")
    }

    fn __getitem__(&self, py: Python<'_>, index: usize) -> PyResult<Py<PyCertReqMsg>> {
        let msgs = self.msgs()?;
        if index >= msgs.len() {
            return Err(pyo3::exceptions::PyIndexError::new_err(format!(
                "index {index} out of range for CertReqMessages with {} entries",
                msgs.len()
            )));
        }
        Py::new(py, PyCertReqMsg::from_rust(&msgs[index]))
    }

    fn __repr__(&self) -> PyResult<String> {
        Ok(format!("CertReqMessages({} requests)", self.msgs()?.len()))
    }
}

// ── register_crmf_submodule ───────────────────────────────────────────────────

/// Build and register the ``synta.crmf`` submodule.
pub(super) fn register_crmf_submodule(parent: &Bound<'_, PyModule>) -> PyResult<()> {
    let py = parent.py();
    let m = PyModule::new(py, "crmf")?;

    m.add_class::<PyCertReqMessages>()?;
    m.add_class::<PyCertReqMsg>()?;

    // ── Registration-control OIDs (RFC 4211 §6 / RFC 9810 Appendix) ──────────
    m.add(
        "ID_REG_CTRL_REG_TOKEN",
        super::oid_const(py, synta_certificate::crmf_types::ID_REG_CTRL_REG_TOKEN),
    )?;
    m.add(
        "ID_REG_CTRL_AUTHENTICATOR",
        super::oid_const(py, synta_certificate::crmf_types::ID_REG_CTRL_AUTHENTICATOR),
    )?;
    m.add(
        "ID_REG_CTRL_PKI_PUBLICATION_INFO",
        super::oid_const(
            py,
            synta_certificate::crmf_types::ID_REG_CTRL_PKI_PUBLICATION_INFO,
        ),
    )?;
    m.add(
        "ID_REG_CTRL_PKI_ARCHIVE_OPTIONS",
        super::oid_const(
            py,
            synta_certificate::crmf_types::ID_REG_CTRL_PKI_ARCHIVE_OPTIONS,
        ),
    )?;
    m.add(
        "ID_REG_CTRL_OLD_CERT_ID",
        super::oid_const(py, synta_certificate::crmf_types::ID_REG_CTRL_OLD_CERT_ID),
    )?;
    m.add(
        "ID_REG_CTRL_PROTOCOL_ENCR_KEY",
        super::oid_const(
            py,
            synta_certificate::crmf_types::ID_REG_CTRL_PROTOCOL_ENCR_KEY,
        ),
    )?;
    m.add(
        "ID_REG_INFO_UTF8_PAIRS",
        super::oid_const(py, synta_certificate::crmf_types::ID_REG_INFO_UTF8_PAIRS),
    )?;
    m.add(
        "ID_REG_INFO_CERT_REQ",
        super::oid_const(py, synta_certificate::crmf_types::ID_REG_INFO_CERT_REQ),
    )?;

    crate::install_submodule(
        parent,
        &m,
        "synta.crmf",
        Some(concat!(
            "synta.crmf — RFC 4211 Certificate Request Message Format types.\n\n",
            "Provides CertReqMessages (SEQUENCE OF CertReqMsg) and CertReqMsg\n",
            "for decoding CRMF batches used in CMP certificate management,\n",
            "along with registration-control OID constants.",
        )),
    )
}