synta-python 0.2.5

//! 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::exceptions::PyValueError;
use pyo3::prelude::*;
use pyo3::types::{PyBytes, PyList, PyString};

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

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>>,
    /// Full DER encoding of the `CertReqMsg` SEQUENCE.
    ///
    /// Populated by both `from_rust` and `CertReqMsgBuilder::build` so that
    /// `CertReqMessagesBuilder::add_request` can collect raw message bytes.
    msg_der: 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);

        let msg_der = encode_to_der(msg);

        Self {
            cert_req_id,
            cert_template_der,
            popo_type,
            popo_der,
            subject_der,
            issuer_der,
            msg_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>> {
        Ok(PyBytes::new(py, &self.msgs()?.to_der().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()))
    }
}

// ── PyCertReqMsgBuilder ───────────────────────────────────────────────────────

/// Pub-location discriminant for `PyCertReqMsgBuilder::pub_infos`.
enum PubInfoGn {
    None,
    Uri(String),
    Rfc822(String),
    Dns(String),
    DirectoryName(Vec<u8>),
}

/// Builder for a single CRMF certificate request message (RFC 4211 §4).
///
/// Construct with ``CertReqMsgBuilder()``, chain setters, and call
/// ``.build()`` to obtain a :class:`CertReqMsg`.
///
/// ```python,ignore
/// import synta, synta.crmf as crmf
///
/// name_der = synta.NameBuilder().common_name("host.example.com").build()
/// req = (
///     crmf.CertReqMsgBuilder()
///     .cert_req_id(0)
///     .subject_name(name_der)
///     .popo_ra_verified()
///     .build()
/// )
/// ```
#[pyclass(name = "CertReqMsgBuilder")]
pub struct PyCertReqMsgBuilder {
    cert_req_id: i64,
    subject_der: Option<Vec<u8>>,
    issuer_der: Option<Vec<u8>>,
    spki_der: Option<Vec<u8>>,
    popo_ra_verified: bool,
    /// `PKIPublicationInfo.action` (0 = dontPublish, 1 = pleasePublish).
    pub_info_action: i64,
    /// Accumulated `SinglePubInfo` entries.
    pub_infos: Vec<(i64, PubInfoGn)>,
}

#[pymethods]
impl PyCertReqMsgBuilder {
    /// Create a new, empty builder with ``certReqId`` = 0.
    #[new]
    fn new() -> Self {
        Self {
            cert_req_id: 0,
            subject_der: None,
            issuer_der: None,
            spki_der: None,
            popo_ra_verified: false,
            pub_info_action: 1, // pleasePublish
            pub_infos: Vec::new(),
        }
    }

    /// Set the ``certReqId`` integer identifying this request in the batch.
    fn cert_req_id<'py>(slf: Bound<'py, Self>, id: i64) -> Bound<'py, Self> {
        slf.borrow_mut().cert_req_id = id;
        slf
    }

    /// Set the subject ``Name`` from pre-encoded DER bytes.
    ///
    /// Pass the return value of ``synta.NameBuilder().build()`` or
    /// ``Certificate.subject_raw_der`` directly.
    fn subject_name<'py>(slf: Bound<'py, Self>, name_der: &[u8]) -> Bound<'py, Self> {
        slf.borrow_mut().subject_der = Some(name_der.to_vec());
        slf
    }

    /// Set the issuer ``Name`` from pre-encoded DER bytes.
    fn issuer_name<'py>(slf: Bound<'py, Self>, name_der: &[u8]) -> Bound<'py, Self> {
        slf.borrow_mut().issuer_der = Some(name_der.to_vec());
        slf
    }

    /// Set the ``SubjectPublicKeyInfo`` from pre-encoded DER bytes.
    ///
    /// Pass ``CertificationRequest.subject_public_key_info_der`` directly.
    fn public_key_der<'py>(slf: Bound<'py, Self>, spki_der: &[u8]) -> Bound<'py, Self> {
        slf.borrow_mut().spki_der = Some(spki_der.to_vec());
        slf
    }

    /// Set the proof-of-possession to ``raVerified`` (``[0] IMPLICIT NULL``).
    ///
    /// This is the simplest POPO type — no signing required; the RA asserts
    /// that private-key possession has already been verified out of band.
    fn popo_ra_verified<'py>(slf: Bound<'py, Self>) -> Bound<'py, Self> {
        slf.borrow_mut().popo_ra_verified = true;
        slf
    }

    /// Set the ``PKIPublicationInfo`` action (0 = dontPublish, 1 = pleasePublish).
    fn publication_action<'py>(slf: Bound<'py, Self>, action: i64) -> Bound<'py, Self> {
        slf.borrow_mut().pub_info_action = action;
        slf
    }

    /// Add a ``SinglePubInfo`` with no location.
    ///
    /// ``pub_method`` should be one of the ``PUB_METHOD_*`` constants.
    fn add_pub_info<'py>(slf: Bound<'py, Self>, pub_method: i64) -> Bound<'py, Self> {
        slf.borrow_mut()
            .pub_infos
            .push((pub_method, PubInfoGn::None));
        slf
    }

    /// Add a ``SinglePubInfo`` with a URI location.
    fn pub_location_uri<'py>(
        slf: Bound<'py, Self>,
        pub_method: i64,
        uri: &str,
    ) -> Bound<'py, Self> {
        slf.borrow_mut()
            .pub_infos
            .push((pub_method, PubInfoGn::Uri(uri.to_string())));
        slf
    }

    /// Add a ``SinglePubInfo`` with an RFC 822 (email) location.
    fn pub_location_rfc822<'py>(
        slf: Bound<'py, Self>,
        pub_method: i64,
        email: &str,
    ) -> Bound<'py, Self> {
        slf.borrow_mut()
            .pub_infos
            .push((pub_method, PubInfoGn::Rfc822(email.to_string())));
        slf
    }

    /// Add a ``SinglePubInfo`` with a DNS name location.
    fn pub_location_dns<'py>(
        slf: Bound<'py, Self>,
        pub_method: i64,
        host: &str,
    ) -> Bound<'py, Self> {
        slf.borrow_mut()
            .pub_infos
            .push((pub_method, PubInfoGn::Dns(host.to_string())));
        slf
    }

    /// Add a ``SinglePubInfo`` with a directory name location
    /// (pre-encoded DER ``Name`` bytes).
    fn pub_location_directory_name<'py>(
        slf: Bound<'py, Self>,
        pub_method: i64,
        name_der: &[u8],
    ) -> Bound<'py, Self> {
        slf.borrow_mut()
            .pub_infos
            .push((pub_method, PubInfoGn::DirectoryName(name_der.to_vec())));
        slf
    }

    /// Build and return the encoded :class:`CertReqMsg`.
    ///
    /// :raises ValueError: if any stored Name/SPKI bytes are invalid or if
    ///     ASN.1 encoding fails.
    fn build<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyCertReqMsg>> {
        let mut rust_builder =
            synta_certificate::CertReqMsgBuilder::new().cert_req_id(self.cert_req_id);
        if let Some(ref b) = self.subject_der {
            rust_builder = rust_builder.subject_name(b);
        }
        if let Some(ref b) = self.issuer_der {
            rust_builder = rust_builder.issuer_name(b);
        }
        if let Some(ref b) = self.spki_der {
            rust_builder = rust_builder.public_key_der(b);
        }
        if self.popo_ra_verified {
            rust_builder = rust_builder.popo_ra_verified();
        }
        rust_builder = rust_builder.publication_action(self.pub_info_action);
        for (method, gn) in &self.pub_infos {
            rust_builder = match gn {
                PubInfoGn::None => rust_builder.add_pub_info(*method),
                PubInfoGn::Uri(uri) => rust_builder.pub_location_uri(*method, uri),
                PubInfoGn::Rfc822(email) => rust_builder.pub_location_rfc822(*method, email),
                PubInfoGn::Dns(host) => rust_builder.pub_location_dns(*method, host),
                PubInfoGn::DirectoryName(bytes) => {
                    rust_builder.pub_location_directory_name(*method, bytes)
                }
            };
        }
        let msg_der = rust_builder
            .build()
            .map_err(|e| PyValueError::new_err(e.to_string()))?;

        // Parse the encoded msg to populate the PyCertReqMsg fields.
        let msg = Decoder::new(&msg_der, Encoding::Der)
            .decode::<synta_certificate::crmf_types::CertReqMsg<'_>>()
            .map_err(SyntaErr)?;
        let mut py_msg = PyCertReqMsg::from_rust(&msg);
        // Overwrite msg_der with the one we already have (avoids a second encode).
        py_msg.msg_der = msg_der;
        Py::new(py, py_msg).map(|x| x.into_bound(py))
    }

    fn __repr__(&self) -> String {
        format!("CertReqMsgBuilder(cert_req_id={})", self.cert_req_id)
    }
}

// ── PyCertReqMessagesBuilder ──────────────────────────────────────────────────

/// Builder for a CRMF batch of certificate request messages (RFC 4211 §3).
///
/// ```python,ignore
/// import synta, synta.crmf as crmf
///
/// name_der = synta.NameBuilder().common_name("host.example.com").build()
/// req = (
///     crmf.CertReqMsgBuilder()
///     .cert_req_id(0)
///     .subject_name(name_der)
///     .popo_ra_verified()
///     .build()
/// )
/// msgs = crmf.CertReqMessagesBuilder().add_request(req).build()
/// raw = msgs.to_der()
/// ```
#[pyclass(name = "CertReqMessagesBuilder")]
pub struct PyCertReqMessagesBuilder {
    requests: Vec<Py<PyCertReqMsg>>,
}

#[pymethods]
impl PyCertReqMessagesBuilder {
    /// Create a new, empty builder.
    #[new]
    fn new() -> Self {
        Self {
            requests: Vec::new(),
        }
    }

    /// Add a :class:`CertReqMsg` to the batch.
    ///
    /// Returns the same builder for chaining.
    fn add_request<'py>(
        slf: Bound<'py, Self>,
        req: Bound<'py, PyCertReqMsg>,
    ) -> PyResult<Bound<'py, Self>> {
        slf.borrow_mut().requests.push(req.unbind());
        Ok(slf)
    }

    /// Build and return the encoded :class:`CertReqMessages` SEQUENCE OF.
    ///
    /// :raises ValueError: if ASN.1 encoding fails.
    fn build<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyCertReqMessages>> {
        let mut builder = synta_certificate::CertReqMessagesBuilder::new();
        for req_py in &self.requests {
            let req = req_py.borrow(py);
            builder = builder.add_request_der(&req.msg_der);
        }
        let der = builder
            .build()
            .map_err(|e| PyValueError::new_err(e.to_string()))?;
        let py_bytes = PyBytes::new(py, &der);
        let msgs = PyCertReqMessages::from_der(py, py_bytes)?;
        Py::new(py, msgs).map(|x| x.into_bound(py))
    }

    fn __repr__(&self) -> String {
        format!("CertReqMessagesBuilder({} requests)", self.requests.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>()?;
    m.add_class::<PyCertReqMessagesBuilder>()?;
    m.add_class::<PyCertReqMsgBuilder>()?;

    // ── pubMethod integer constants (RFC 4211 §11) ────────────────────────────
    m.add(
        "PUB_METHOD_DONT_CARE",
        synta_certificate::PUB_METHOD_DONT_CARE,
    )?;
    m.add("PUB_METHOD_X500", synta_certificate::PUB_METHOD_X500)?;
    m.add("PUB_METHOD_WEB", synta_certificate::PUB_METHOD_WEB)?;
    m.add("PUB_METHOD_LDAP", synta_certificate::PUB_METHOD_LDAP)?;

    // ── 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.",
        )),
    )
}