_synta/certificate/
crmf.rs

1//! Python bindings for RFC 4211 Certificate Request Message Format (CRMF).
2//!
3//! Exposes ``CertReqMessages`` and ``CertReqMsg`` as Python classes and
4//! installs registration-control OID constants into the ``synta.crmf``
5//! submodule.
6
7use std::sync::OnceLock;
8
9use pyo3::prelude::*;
10use pyo3::types::{PyBytes, PyList, PyString};
11
12use synta::traits::Encode;
13use synta::{Decoder, Encoding};
14
15use crate::error::SyntaErr;
16
17// ── helpers ───────────────────────────────────────────────────────────────────
18
19fn encode_to_der<T: Encode>(v: &T) -> Vec<u8> {
20    let mut enc = synta::Encoder::new(Encoding::Der);
21    if v.encode(&mut enc).is_err() {
22        return Vec::new();
23    }
24    enc.finish().unwrap_or_default()
25}
26
27// ── PyCertReqMsg ──────────────────────────────────────────────────────────────
28
29/// A single certificate request message (RFC 4211 §4).
30///
31/// Contains a ``CertRequest`` (with an ID and optional ``CertTemplate``) and
32/// an optional ``ProofOfPossession``.  Obtained by iterating over a
33/// :class:`CertReqMessages` object.
34///
35/// ``cert_template_der`` contains the re-encoded ``CertTemplate`` SEQUENCE;
36/// feed it back to a ``synta.Decoder`` to inspect individual fields.
37/// ``popo_der`` (if present) contains the re-encoded ``ProofOfPossession``
38/// CHOICE; ``popo_type`` names the active arm.
39#[pyclass(frozen, name = "CertReqMsg")]
40pub struct PyCertReqMsg {
41    cert_req_id: i64,
42    cert_template_der: Vec<u8>,
43    popo_type: Option<String>,
44    popo_der: Option<Vec<u8>>,
45    subject_der: Option<Vec<u8>>,
46    issuer_der: Option<Vec<u8>>,
47}
48
49impl PyCertReqMsg {
50    fn from_rust(msg: &synta_certificate::crmf_types::CertReqMsg<'_>) -> Self {
51        let cert_req_id = msg.cert_req.cert_req_id.as_i64().unwrap_or(-1);
52
53        let cert_template_der = encode_to_der(&msg.cert_req.cert_template);
54
55        let (popo_type, popo_der) = match &msg.popo {
56            None => (None, None),
57            Some(pop) => {
58                use synta_certificate::crmf_types::ProofOfPossession::*;
59                let arm = match pop {
60                    RaVerified(_) => "raVerified",
61                    Signature(_) => "signature",
62                    KeyEncipherment(_) => "keyEncipherment",
63                    KeyAgreement(_) => "keyAgreement",
64                };
65                let der = encode_to_der(pop);
66                (Some(arm.to_string()), Some(der))
67            }
68        };
69
70        let subject_der = msg
71            .cert_req
72            .cert_template
73            .subject
74            .as_ref()
75            .map(encode_to_der);
76        let issuer_der = msg
77            .cert_req
78            .cert_template
79            .issuer
80            .as_ref()
81            .map(encode_to_der);
82
83        Self {
84            cert_req_id,
85            cert_template_der,
86            popo_type,
87            popo_der,
88            subject_der,
89            issuer_der,
90        }
91    }
92}
93
94#[pymethods]
95impl PyCertReqMsg {
96    /// The ``certReqId`` integer identifying this request within the batch.
97    #[getter]
98    fn cert_req_id(&self) -> i64 {
99        self.cert_req_id
100    }
101
102    /// Raw DER bytes of the ``CertTemplate`` SEQUENCE.
103    ///
104    /// All fields in the template are optional.  Use a ``synta.Decoder`` to
105    /// inspect individual fields.
106    #[getter]
107    fn cert_template_der<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
108        PyBytes::new(py, &self.cert_template_der)
109    }
110
111    /// Name of the active ``ProofOfPossession`` arm, or ``None`` if absent.
112    ///
113    /// One of: ``"raVerified"``, ``"signature"``, ``"keyEncipherment"``,
114    /// ``"keyAgreement"``.
115    #[getter]
116    fn popo_type<'py>(&self, py: Python<'py>) -> Option<Bound<'py, PyString>> {
117        self.popo_type.as_deref().map(|s| PyString::new(py, s))
118    }
119
120    /// Raw DER bytes of the ``ProofOfPossession`` CHOICE, or ``None`` if absent.
121    #[getter]
122    fn popo_der<'py>(&self, py: Python<'py>) -> Option<Bound<'py, PyBytes>> {
123        self.popo_der.as_deref().map(|b| PyBytes::new(py, b))
124    }
125
126    /// Raw DER bytes of the ``subject`` ``Name`` from the ``CertTemplate``,
127    /// or ``None`` if the template does not include a subject.
128    #[getter]
129    fn subject_der<'py>(&self, py: Python<'py>) -> Option<Bound<'py, PyBytes>> {
130        self.subject_der.as_deref().map(|b| PyBytes::new(py, b))
131    }
132
133    /// Raw DER bytes of the ``issuer`` ``Name`` from the ``CertTemplate``,
134    /// or ``None`` if the template does not include an issuer.
135    #[getter]
136    fn issuer_der<'py>(&self, py: Python<'py>) -> Option<Bound<'py, PyBytes>> {
137        self.issuer_der.as_deref().map(|b| PyBytes::new(py, b))
138    }
139
140    fn __repr__(&self) -> String {
141        format!(
142            "CertReqMsg(cert_req_id={}, popo_type={})",
143            self.cert_req_id,
144            self.popo_type.as_deref().unwrap_or("None"),
145        )
146    }
147}
148
149// ── PyCertReqMessages ─────────────────────────────────────────────────────────
150
151/// A batch of Certificate Request Messages (RFC 4211 §3).
152///
153/// ``CertReqMessages`` is a ``SEQUENCE OF CertReqMsg``.  Each element
154/// carries a certificate template, an optional proof-of-possession, and an
155/// optional list of registration info attributes.
156///
157/// ```python,ignore
158/// import synta.crmf as crmf
159/// msgs = crmf.CertReqMessages.from_der(open("req.crmf", "rb").read())
160/// for req in msgs:
161///     print(req.cert_req_id, req.popo_type)
162///     print(req.subject_der.hex() if req.subject_der else "(no subject)")
163/// ```
164#[pyclass(frozen, name = "CertReqMessages")]
165pub struct PyCertReqMessages {
166    _data: Py<PyBytes>,
167    raw: &'static [u8],
168    inner: OnceLock<Vec<synta_certificate::crmf_types::CertReqMsg<'static>>>,
169    requests_cache: OnceLock<Py<PyList>>,
170}
171
172impl PyCertReqMessages {
173    fn msgs(&self) -> PyResult<&Vec<synta_certificate::crmf_types::CertReqMsg<'static>>> {
174        if let Some(v) = self.inner.get() {
175            return Ok(v);
176        }
177        let mut dec = Decoder::new(self.raw, Encoding::Der);
178        let decoded = dec
179            .decode::<synta_certificate::crmf_types::CertReqMessages<'_>>()
180            .map_err(SyntaErr)?;
181        // SAFETY: raw is pinned by _data for the lifetime of self.
182        let decoded: Vec<synta_certificate::crmf_types::CertReqMsg<'static>> =
183            unsafe { std::mem::transmute(decoded) };
184        let _ = self.inner.set(decoded);
185        Ok(self.inner.get().unwrap())
186    }
187
188    fn build_requests_list<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
189        let msgs = self.msgs()?;
190        let list = PyList::empty(py);
191        for msg in msgs {
192            let obj = Py::new(py, PyCertReqMsg::from_rust(msg))?;
193            list.append(obj)?;
194        }
195        Ok(list)
196    }
197}
198
199#[pymethods]
200impl PyCertReqMessages {
201    /// Parse a DER-encoded ``CertReqMessages`` SEQUENCE OF.
202    ///
203    /// :param data: DER bytes of the ``CertReqMessages`` structure.
204    /// :raises ValueError: if the bytes cannot be decoded.
205    #[staticmethod]
206    fn from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
207        let py_bytes = data.unbind();
208        {
209            let raw = py_bytes.as_bytes(py);
210            Decoder::new(raw, Encoding::Der)
211                .decode::<synta_certificate::crmf_types::CertReqMessages<'_>>()
212                .map_err(SyntaErr)?;
213        }
214        let raw: &'static [u8] = unsafe { std::mem::transmute(py_bytes.as_bytes(py)) };
215        Ok(Self {
216            _data: py_bytes,
217            raw,
218            inner: OnceLock::new(),
219            requests_cache: OnceLock::new(),
220        })
221    }
222
223    /// Return the DER encoding of this ``CertReqMessages`` SEQUENCE OF.
224    fn to_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
225        let msgs = self.msgs()?;
226        let mut enc = synta::Encoder::new(Encoding::Der);
227        msgs.encode(&mut enc).map_err(SyntaErr)?;
228        Ok(PyBytes::new(py, &enc.finish().map_err(SyntaErr)?))
229    }
230
231    /// List of :class:`CertReqMsg` objects, one per request in the batch.
232    #[getter]
233    fn requests<'py>(&'py self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
234        if let Some(c) = self.requests_cache.get() {
235            return Ok(c.clone_ref(py).into_bound(py));
236        }
237        let list = self.build_requests_list(py)?;
238        let _ = self.requests_cache.set(list.as_unbound().clone_ref(py));
239        Ok(list)
240    }
241
242    fn __len__(&self) -> PyResult<usize> {
243        Ok(self.msgs()?.len())
244    }
245
246    fn __iter__<'py>(&'py self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
247        self.requests(py)?.call_method0("__iter__")
248    }
249
250    fn __getitem__(&self, py: Python<'_>, index: usize) -> PyResult<Py<PyCertReqMsg>> {
251        let msgs = self.msgs()?;
252        if index >= msgs.len() {
253            return Err(pyo3::exceptions::PyIndexError::new_err(format!(
254                "index {index} out of range for CertReqMessages with {} entries",
255                msgs.len()
256            )));
257        }
258        Py::new(py, PyCertReqMsg::from_rust(&msgs[index]))
259    }
260
261    fn __repr__(&self) -> PyResult<String> {
262        Ok(format!("CertReqMessages({} requests)", self.msgs()?.len()))
263    }
264}
265
266// ── register_crmf_submodule ───────────────────────────────────────────────────
267
268/// Build and register the ``synta.crmf`` submodule.
269pub(super) fn register_crmf_submodule(parent: &Bound<'_, PyModule>) -> PyResult<()> {
270    let py = parent.py();
271    let m = PyModule::new(py, "crmf")?;
272
273    m.add_class::<PyCertReqMessages>()?;
274    m.add_class::<PyCertReqMsg>()?;
275
276    // ── Registration-control OIDs (RFC 4211 §6 / RFC 9810 Appendix) ──────────
277    m.add(
278        "ID_REG_CTRL_REG_TOKEN",
279        super::oid_const(py, synta_certificate::crmf_types::ID_REG_CTRL_REG_TOKEN),
280    )?;
281    m.add(
282        "ID_REG_CTRL_AUTHENTICATOR",
283        super::oid_const(py, synta_certificate::crmf_types::ID_REG_CTRL_AUTHENTICATOR),
284    )?;
285    m.add(
286        "ID_REG_CTRL_PKI_PUBLICATION_INFO",
287        super::oid_const(
288            py,
289            synta_certificate::crmf_types::ID_REG_CTRL_PKI_PUBLICATION_INFO,
290        ),
291    )?;
292    m.add(
293        "ID_REG_CTRL_PKI_ARCHIVE_OPTIONS",
294        super::oid_const(
295            py,
296            synta_certificate::crmf_types::ID_REG_CTRL_PKI_ARCHIVE_OPTIONS,
297        ),
298    )?;
299    m.add(
300        "ID_REG_CTRL_OLD_CERT_ID",
301        super::oid_const(py, synta_certificate::crmf_types::ID_REG_CTRL_OLD_CERT_ID),
302    )?;
303    m.add(
304        "ID_REG_CTRL_PROTOCOL_ENCR_KEY",
305        super::oid_const(
306            py,
307            synta_certificate::crmf_types::ID_REG_CTRL_PROTOCOL_ENCR_KEY,
308        ),
309    )?;
310    m.add(
311        "ID_REG_INFO_UTF8_PAIRS",
312        super::oid_const(py, synta_certificate::crmf_types::ID_REG_INFO_UTF8_PAIRS),
313    )?;
314    m.add(
315        "ID_REG_INFO_CERT_REQ",
316        super::oid_const(py, synta_certificate::crmf_types::ID_REG_INFO_CERT_REQ),
317    )?;
318
319    crate::install_submodule(
320        parent,
321        &m,
322        "synta.crmf",
323        Some(concat!(
324            "synta.crmf — RFC 4211 Certificate Request Message Format types.\n\n",
325            "Provides CertReqMessages (SEQUENCE OF CertReqMsg) and CertReqMsg\n",
326            "for decoding CRMF batches used in CMP certificate management,\n",
327            "along with registration-control OID constants.",
328        )),
329    )
330}
_synta/certificate/crmf.rs

_synta/certificate/
crmf.rs
