_synta/certificate/
cmp.rs

1//! Python bindings for RFC 9810 Certificate Management Protocol (CMP) types.
2//!
3//! Exposes ``CMPMessage`` (wrapping ``PKIMessage``) as a Python class and
4//! installs MAC algorithm and key-purpose OID constants into the ``synta.cmp``
5//! submodule.
6
7use std::sync::OnceLock;
8
9use pyo3::prelude::*;
10use pyo3::types::{PyBytes, PyString};
11
12use synta::traits::Encode;
13use synta::{Decoder, Encoding};
14
15use crate::error::SyntaErr;
16use crate::types::PyObjectIdentifier;
17
18// ── helpers ───────────────────────────────────────────────────────────────────
19
20fn encode_to_der<T: Encode>(v: &T) -> Vec<u8> {
21    let mut enc = synta::Encoder::new(Encoding::Der);
22    if v.encode(&mut enc).is_err() {
23        return Vec::new();
24    }
25    enc.finish().unwrap_or_default()
26}
27
28/// Map a `PKIBody` variant to its CMP message-type name (lowercase, per RFC 9810).
29fn body_type_name(body: &synta_certificate::cmp_types::PKIBody<'_>) -> &'static str {
30    use synta_certificate::cmp_types::PKIBody::*;
31    match body {
32        Ir(_) => "ir",
33        Ip(_) => "ip",
34        Cr(_) => "cr",
35        Cp(_) => "cp",
36        P10cr(_) => "p10cr",
37        Popdecc(_) => "popdecc",
38        Popdecr(_) => "popdecr",
39        Kur(_) => "kur",
40        Kup(_) => "kup",
41        Krr(_) => "krr",
42        Krp(_) => "krp",
43        Rr(_) => "rr",
44        Rp(_) => "rp",
45        Ccr(_) => "ccr",
46        Ccp(_) => "ccp",
47        Ckuann(_) => "ckuann",
48        Cann(_) => "cann",
49        Rann(_) => "rann",
50        Crlann(_) => "crlann",
51        Pkiconf(_) => "pkiconf",
52        Nested(_) => "nested",
53        Genm(_) => "genm",
54        Genp(_) => "genp",
55        Error(_) => "error",
56        CertConf(_) => "certConf",
57        PollReq(_) => "pollReq",
58        PollRep(_) => "pollRep",
59    }
60}
61
62/// Extract raw body DER from a `PKIBody` variant.  Returns `None` for
63/// the `pkiconf` (NULL) arm; returns the `RawDer` bytes for all others.
64fn body_raw_der(body: &synta_certificate::cmp_types::PKIBody<'_>) -> Option<Vec<u8>> {
65    use synta_certificate::cmp_types::PKIBody::*;
66    match body {
67        Pkiconf(_) => None,
68        Ir(r) | Ip(r) | Cr(r) | Cp(r) | P10cr(r) | Popdecc(r) | Popdecr(r) | Kur(r) | Kup(r)
69        | Krr(r) | Krp(r) | Rr(r) | Rp(r) | Ccr(r) | Ccp(r) | Ckuann(r) | Cann(r) | Rann(r)
70        | Crlann(r) | Nested(r) | Genm(r) | Genp(r) | Error(r) | CertConf(r) | PollReq(r)
71        | PollRep(r) => Some(r.as_bytes().to_vec()),
72    }
73}
74
75// ── PyCMPMessage ──────────────────────────────────────────────────────────────
76
77/// A CMP ``PKIMessage`` (RFC 9810 §5.1).
78///
79/// ``PKIMessage`` is the top-level CMP envelope, carrying a ``PKIHeader``
80/// (metadata: pvno, sender, recipient, timestamps, nonces, …) and a
81/// ``PKIBody`` (the message payload: IR, IP, CR, CP, RR, RP, …).
82///
83/// The ``body_type`` property names the active ``PKIBody`` arm in lowercase
84/// (``"ir"``, ``"ip"``, ``"cr"``, ``"cp"``, ``"pkiconf"``, …).  For all arms
85/// except ``pkiconf``, ``body_der`` returns the raw DER bytes of the body
86/// content for further decoding with a ``synta.Decoder``.
87///
88/// ```python,ignore
89/// import synta.cmp as cmp
90///
91/// msg = cmp.CMPMessage.from_der(open("cmp.der", "rb").read())
92/// print(msg.pvno, msg.body_type)
93/// if msg.body_type == "ir":
94///     # body_der contains the DER of CertReqMessages
95///     import synta.crmf as crmf
96///     reqs = crmf.CertReqMessages.from_der(msg.body_der)
97///     for req in reqs:
98///         print(req.cert_req_id, req.popo_type)
99/// ```
100#[pyclass(frozen, name = "CMPMessage")]
101pub struct PyCMPMessage {
102    _data: Py<PyBytes>,
103    raw: &'static [u8],
104    inner: OnceLock<Box<synta_certificate::cmp_types::PKIMessage<'static>>>,
105    // Field caches
106    pvno_cache: OnceLock<i64>,
107    body_type_cache: OnceLock<&'static str>,
108    body_der_cache: OnceLock<Option<Py<PyBytes>>>,
109    sender_der_cache: OnceLock<Py<PyBytes>>,
110    recipient_der_cache: OnceLock<Py<PyBytes>>,
111    transaction_id_cache: OnceLock<Option<Py<PyBytes>>>,
112    sender_nonce_cache: OnceLock<Option<Py<PyBytes>>>,
113    recip_nonce_cache: OnceLock<Option<Py<PyBytes>>>,
114    protection_alg_oid_cache: OnceLock<Option<Py<PyObjectIdentifier>>>,
115    message_time_cache: OnceLock<Option<Py<PyString>>>,
116}
117
118impl PyCMPMessage {
119    fn msg(&self) -> PyResult<&synta_certificate::cmp_types::PKIMessage<'static>> {
120        if let Some(v) = self.inner.get() {
121            return Ok(v.as_ref());
122        }
123        let mut dec = Decoder::new(self.raw, Encoding::Der);
124        let decoded = dec
125            .decode::<synta_certificate::cmp_types::PKIMessage<'_>>()
126            .map_err(SyntaErr)?;
127        // SAFETY: raw is pinned by _data for the lifetime of self.
128        let decoded: synta_certificate::cmp_types::PKIMessage<'static> =
129            unsafe { std::mem::transmute(decoded) };
130        let _ = self.inner.set(Box::new(decoded));
131        Ok(self.inner.get().unwrap().as_ref())
132    }
133}
134
135#[pymethods]
136impl PyCMPMessage {
137    /// Parse a DER-encoded ``PKIMessage`` SEQUENCE.
138    ///
139    /// :param data: DER bytes of the ``PKIMessage``.
140    /// :raises ValueError: if the bytes cannot be decoded.
141    #[staticmethod]
142    fn from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
143        let py_bytes = data.unbind();
144        {
145            let raw = py_bytes.as_bytes(py);
146            Decoder::new(raw, Encoding::Der)
147                .decode::<synta_certificate::cmp_types::PKIMessage<'_>>()
148                .map_err(SyntaErr)?;
149        }
150        let raw: &'static [u8] = unsafe { std::mem::transmute(py_bytes.as_bytes(py)) };
151        Ok(Self {
152            _data: py_bytes,
153            raw,
154            inner: OnceLock::new(),
155            pvno_cache: OnceLock::new(),
156            body_type_cache: OnceLock::new(),
157            body_der_cache: OnceLock::new(),
158            sender_der_cache: OnceLock::new(),
159            recipient_der_cache: OnceLock::new(),
160            transaction_id_cache: OnceLock::new(),
161            sender_nonce_cache: OnceLock::new(),
162            recip_nonce_cache: OnceLock::new(),
163            protection_alg_oid_cache: OnceLock::new(),
164            message_time_cache: OnceLock::new(),
165        })
166    }
167
168    /// Return the DER encoding of this ``PKIMessage``.
169    fn to_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
170        let mut enc = synta::Encoder::new(Encoding::Der);
171        self.msg()?.encode(&mut enc).map_err(SyntaErr)?;
172        Ok(PyBytes::new(py, &enc.finish().map_err(SyntaErr)?))
173    }
174
175    /// CMP protocol version number (``pvno`` field; ``2`` for CMP v2/v3).
176    #[getter]
177    fn pvno(&self) -> PyResult<i64> {
178        if let Some(v) = self.pvno_cache.get() {
179            return Ok(*v);
180        }
181        let v = self.msg()?.header.pvno.as_i64().unwrap_or(2);
182        let _ = self.pvno_cache.set(v);
183        Ok(v)
184    }
185
186    /// Name of the active ``PKIBody`` arm in lowercase (e.g. ``"ir"``, ``"ip"``,
187    /// ``"cr"``, ``"cp"``, ``"rr"``, ``"rp"``, ``"pkiconf"``).
188    #[getter]
189    fn body_type(&self) -> PyResult<&'static str> {
190        if let Some(v) = self.body_type_cache.get() {
191            return Ok(v);
192        }
193        let name = body_type_name(&self.msg()?.body);
194        let _ = self.body_type_cache.set(name);
195        Ok(name)
196    }
197
198    /// Raw DER bytes of the ``PKIBody`` content, or ``None`` for ``pkiconf``
199    /// (which carries a NULL value and has no further content to decode).
200    ///
201    /// For ``"ir"``, ``"cr"`` etc., these bytes are a ``CertReqMessages``
202    /// SEQUENCE OF that can be decoded with ``synta.crmf.CertReqMessages.from_der``.
203    #[getter]
204    fn body_der<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
205        if let Some(cached) = self.body_der_cache.get() {
206            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
207        }
208        let der = body_raw_der(&self.msg()?.body);
209        let py_opt = der
210            .as_deref()
211            .map(|b| PyBytes::new(py, b).as_unbound().clone_ref(py));
212        let _ = self.body_der_cache.set(py_opt);
213        Ok(self
214            .body_der_cache
215            .get()
216            .unwrap()
217            .as_ref()
218            .map(|b| b.clone_ref(py).into_bound(py)))
219    }
220
221    /// Raw DER bytes of the ``sender`` ``GeneralName`` field.
222    #[getter]
223    fn sender_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
224        if let Some(c) = self.sender_der_cache.get() {
225            return Ok(c.clone_ref(py).into_bound(py));
226        }
227        let der = encode_to_der(&self.msg()?.header.sender);
228        let b = PyBytes::new(py, &der);
229        let _ = self.sender_der_cache.set(b.as_unbound().clone_ref(py));
230        Ok(b)
231    }
232
233    /// Raw DER bytes of the ``recipient`` ``GeneralName`` field.
234    #[getter]
235    fn recipient_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
236        if let Some(c) = self.recipient_der_cache.get() {
237            return Ok(c.clone_ref(py).into_bound(py));
238        }
239        let der = encode_to_der(&self.msg()?.header.recipient);
240        let b = PyBytes::new(py, &der);
241        let _ = self.recipient_der_cache.set(b.as_unbound().clone_ref(py));
242        Ok(b)
243    }
244
245    /// ``transactionID`` bytes, or ``None`` if not present.
246    #[getter]
247    fn transaction_id<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
248        if let Some(cached) = self.transaction_id_cache.get() {
249            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
250        }
251        let opt = self
252            .msg()?
253            .header
254            .transaction_id
255            .as_ref()
256            .map(|o| PyBytes::new(py, o.as_bytes()).as_unbound().clone_ref(py));
257        let _ = self.transaction_id_cache.set(opt);
258        Ok(self
259            .transaction_id_cache
260            .get()
261            .unwrap()
262            .as_ref()
263            .map(|b| b.clone_ref(py).into_bound(py)))
264    }
265
266    /// ``senderNonce`` bytes, or ``None`` if not present.
267    #[getter]
268    fn sender_nonce<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
269        if let Some(cached) = self.sender_nonce_cache.get() {
270            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
271        }
272        let opt = self
273            .msg()?
274            .header
275            .sender_nonce
276            .as_ref()
277            .map(|o| PyBytes::new(py, o.as_bytes()).as_unbound().clone_ref(py));
278        let _ = self.sender_nonce_cache.set(opt);
279        Ok(self
280            .sender_nonce_cache
281            .get()
282            .unwrap()
283            .as_ref()
284            .map(|b| b.clone_ref(py).into_bound(py)))
285    }
286
287    /// ``recipNonce`` bytes, or ``None`` if not present.
288    #[getter]
289    fn recip_nonce<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
290        if let Some(cached) = self.recip_nonce_cache.get() {
291            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
292        }
293        let opt = self
294            .msg()?
295            .header
296            .recip_nonce
297            .as_ref()
298            .map(|o| PyBytes::new(py, o.as_bytes()).as_unbound().clone_ref(py));
299        let _ = self.recip_nonce_cache.set(opt);
300        Ok(self
301            .recip_nonce_cache
302            .get()
303            .unwrap()
304            .as_ref()
305            .map(|b| b.clone_ref(py).into_bound(py)))
306    }
307
308    /// OID of the ``protectionAlg`` ``AlgorithmIdentifier``, or ``None``.
309    #[getter]
310    fn protection_alg_oid(&self, py: Python<'_>) -> PyResult<Option<Py<PyObjectIdentifier>>> {
311        if let Some(cached) = self.protection_alg_oid_cache.get() {
312            return Ok(cached.as_ref().map(|o| o.clone_ref(py)));
313        }
314        let opt = self
315            .msg()?
316            .header
317            .protection_alg
318            .as_ref()
319            .map(|alg| Py::new(py, PyObjectIdentifier::from_oid(alg.algorithm.clone())))
320            .transpose()?;
321        let _ = self
322            .protection_alg_oid_cache
323            .set(opt.as_ref().map(|o| o.clone_ref(py)));
324        Ok(opt)
325    }
326
327    /// ``messageTime`` as a GeneralizedTime string, or ``None`` if not present.
328    #[getter]
329    fn message_time<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyString>>> {
330        if let Some(cached) = self.message_time_cache.get() {
331            return Ok(cached.as_ref().map(|s| s.clone_ref(py).into_bound(py)));
332        }
333        let opt = self
334            .msg()?
335            .header
336            .message_time
337            .as_ref()
338            .map(|t| PyString::new(py, &t.to_string()).as_unbound().clone_ref(py));
339        let _ = self.message_time_cache.set(opt);
340        Ok(self
341            .message_time_cache
342            .get()
343            .unwrap()
344            .as_ref()
345            .map(|s| s.clone_ref(py).into_bound(py)))
346    }
347
348    fn __repr__(&self) -> PyResult<String> {
349        Ok(format!(
350            "CMPMessage(pvno={}, body_type={})",
351            self.pvno()?,
352            self.body_type()?,
353        ))
354    }
355}
356
357// ── register_cmp_submodule ────────────────────────────────────────────────────
358
359/// Build and register the ``synta.cmp`` submodule.
360pub(super) fn register_cmp_submodule(parent: &Bound<'_, PyModule>) -> PyResult<()> {
361    let py = parent.py();
362    let m = PyModule::new(py, "cmp")?;
363
364    m.add_class::<PyCMPMessage>()?;
365
366    // ── MAC algorithm OIDs (RFC 9810 Appendix D) ─────────────────────────────
367    m.add(
368        "ID_PASSWORD_BASED_MAC",
369        super::oid_const(py, synta_certificate::cmp_types::ID_PASSWORD_BASED_MAC),
370    )?;
371    m.add(
372        "ID_DHBASED_MAC",
373        super::oid_const(py, synta_certificate::cmp_types::ID_DHBASED_MAC),
374    )?;
375    m.add(
376        "ID_KEM_BASED_MAC",
377        super::oid_const(py, synta_certificate::cmp_types::ID_KEM_BASED_MAC),
378    )?;
379
380    // ── CMP key-purpose OID (RFC 9810 §4) ────────────────────────────────────
381    m.add(
382        "ID_KP_CM_KGA",
383        super::oid_const(py, synta_certificate::cmp_types::ID_KP_CM_KGA),
384    )?;
385
386    // ── CMP registration-control OIDs shared with CRMF (RFC 9810 Appendix) ──
387    m.add(
388        "ID_REG_CTRL_ALT_CERT_TEMPLATE",
389        super::oid_const(
390            py,
391            synta_certificate::cmp_types::ID_REG_CTRL_ALT_CERT_TEMPLATE,
392        ),
393    )?;
394    m.add(
395        "ID_REG_CTRL_ALG_ID",
396        super::oid_const(py, synta_certificate::cmp_types::ID_REG_CTRL_ALG_ID),
397    )?;
398    m.add(
399        "ID_REG_CTRL_RSA_KEY_LEN",
400        super::oid_const(py, synta_certificate::cmp_types::ID_REG_CTRL_RSA_KEY_LEN),
401    )?;
402
403    crate::install_submodule(
404        parent,
405        &m,
406        "synta.cmp",
407        Some(concat!(
408            "synta.cmp — RFC 9810 Certificate Management Protocol v3 types.\n\n",
409            "Provides CMPMessage (wrapping PKIMessage) for decoding CMP\n",
410            "envelopes.  body_type names the active PKIBody arm; body_der\n",
411            "returns the raw bytes for further decoding with synta.crmf or\n",
412            "a synta.Decoder.  Also exposes MAC algorithm and key-purpose\n",
413            "OID constants.",
414        )),
415    )
416}
_synta/certificate/cmp.rs

_synta/certificate/
cmp.rs
