1use std::sync::OnceLock;
8
9use pyo3::exceptions::PyValueError;
10use pyo3::prelude::*;
11use pyo3::types::{PyBytes, PyList, PyString};
12
13use synta::traits::Encode;
14use synta::{Decoder, Encoding, ToDer};
15
16use crate::error::SyntaErr;
17
18fn 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#[pyclass(frozen, name = "CertReqMsg")]
41pub struct PyCertReqMsg {
42 cert_req_id: i64,
43 cert_template_der: Vec<u8>,
44 popo_type: Option<String>,
45 popo_der: Option<Vec<u8>>,
46 subject_der: Option<Vec<u8>>,
47 issuer_der: Option<Vec<u8>>,
48 msg_der: Vec<u8>,
53}
54
55impl PyCertReqMsg {
56 fn from_rust(msg: &synta_certificate::crmf_types::CertReqMsg<'_>) -> Self {
57 let cert_req_id = msg.cert_req.cert_req_id.as_i64().unwrap_or(-1);
58
59 let cert_template_der = encode_to_der(&msg.cert_req.cert_template);
60
61 let (popo_type, popo_der) = match &msg.popo {
62 None => (None, None),
63 Some(pop) => {
64 use synta_certificate::crmf_types::ProofOfPossession::*;
65 let arm = match pop {
66 RaVerified(_) => "raVerified",
67 Signature(_) => "signature",
68 KeyEncipherment(_) => "keyEncipherment",
69 KeyAgreement(_) => "keyAgreement",
70 };
71 let der = encode_to_der(pop);
72 (Some(arm.to_string()), Some(der))
73 }
74 };
75
76 let subject_der = msg
77 .cert_req
78 .cert_template
79 .subject
80 .as_ref()
81 .map(encode_to_der);
82 let issuer_der = msg
83 .cert_req
84 .cert_template
85 .issuer
86 .as_ref()
87 .map(encode_to_der);
88
89 let msg_der = encode_to_der(msg);
90
91 Self {
92 cert_req_id,
93 cert_template_der,
94 popo_type,
95 popo_der,
96 subject_der,
97 issuer_der,
98 msg_der,
99 }
100 }
101}
102
103#[pymethods]
104impl PyCertReqMsg {
105 #[getter]
107 fn cert_req_id(&self) -> i64 {
108 self.cert_req_id
109 }
110
111 #[getter]
116 fn cert_template_der<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
117 PyBytes::new(py, &self.cert_template_der)
118 }
119
120 #[getter]
125 fn popo_type<'py>(&self, py: Python<'py>) -> Option<Bound<'py, PyString>> {
126 self.popo_type.as_deref().map(|s| PyString::new(py, s))
127 }
128
129 #[getter]
131 fn popo_der<'py>(&self, py: Python<'py>) -> Option<Bound<'py, PyBytes>> {
132 self.popo_der.as_deref().map(|b| PyBytes::new(py, b))
133 }
134
135 #[getter]
138 fn subject_der<'py>(&self, py: Python<'py>) -> Option<Bound<'py, PyBytes>> {
139 self.subject_der.as_deref().map(|b| PyBytes::new(py, b))
140 }
141
142 #[getter]
145 fn issuer_der<'py>(&self, py: Python<'py>) -> Option<Bound<'py, PyBytes>> {
146 self.issuer_der.as_deref().map(|b| PyBytes::new(py, b))
147 }
148
149 fn __repr__(&self) -> String {
150 format!(
151 "CertReqMsg(cert_req_id={}, popo_type={})",
152 self.cert_req_id,
153 self.popo_type.as_deref().unwrap_or("None"),
154 )
155 }
156}
157
158#[pyclass(frozen, name = "CertReqMessages")]
174pub struct PyCertReqMessages {
175 _data: Py<PyBytes>,
176 raw: &'static [u8],
177 inner: OnceLock<Vec<synta_certificate::crmf_types::CertReqMsg<'static>>>,
178 requests_cache: OnceLock<Py<PyList>>,
179}
180
181impl PyCertReqMessages {
182 fn msgs(&self) -> PyResult<&Vec<synta_certificate::crmf_types::CertReqMsg<'static>>> {
183 if let Some(v) = self.inner.get() {
184 return Ok(v);
185 }
186 let decoded =
187 synta_certificate::crmf_types::CertReqMessages::from_der(self.raw).map_err(SyntaErr)?;
188 let _ = self.inner.set(decoded.0);
189 Ok(self.inner.get().unwrap())
190 }
191
192 fn build_requests_list<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
193 let msgs = self.msgs()?;
194 let list = PyList::empty(py);
195 for msg in msgs {
196 let obj = Py::new(py, PyCertReqMsg::from_rust(msg))?;
197 list.append(obj)?;
198 }
199 Ok(list)
200 }
201}
202
203#[pymethods]
204impl PyCertReqMessages {
205 #[staticmethod]
210 fn from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
211 let py_bytes = data.unbind();
212 {
213 let raw = py_bytes.as_bytes(py);
214 Decoder::new(raw, Encoding::Der)
215 .decode::<synta_certificate::crmf_types::CertReqMessages<'_>>()
216 .map_err(SyntaErr)?;
217 }
218 let raw: &'static [u8] = unsafe {
219 let s = py_bytes.bind(py).as_bytes();
220 std::slice::from_raw_parts(s.as_ptr(), s.len())
221 };
222 Ok(Self {
223 _data: py_bytes,
224 raw,
225 inner: OnceLock::new(),
226 requests_cache: OnceLock::new(),
227 })
228 }
229
230 fn to_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
232 Ok(PyBytes::new(py, &self.msgs()?.to_der().map_err(SyntaErr)?))
233 }
234
235 #[getter]
237 fn requests<'py>(&'py self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
238 if let Some(c) = self.requests_cache.get() {
239 return Ok(c.clone_ref(py).into_bound(py));
240 }
241 let list = self.build_requests_list(py)?;
242 let _ = self.requests_cache.set(list.as_unbound().clone_ref(py));
243 Ok(list)
244 }
245
246 fn __len__(&self) -> PyResult<usize> {
247 Ok(self.msgs()?.len())
248 }
249
250 fn __iter__<'py>(&'py self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
251 self.requests(py)?.call_method0("__iter__")
252 }
253
254 fn __getitem__(&self, py: Python<'_>, index: usize) -> PyResult<Py<PyCertReqMsg>> {
255 let msgs = self.msgs()?;
256 if index >= msgs.len() {
257 return Err(pyo3::exceptions::PyIndexError::new_err(format!(
258 "index {index} out of range for CertReqMessages with {} entries",
259 msgs.len()
260 )));
261 }
262 Py::new(py, PyCertReqMsg::from_rust(&msgs[index]))
263 }
264
265 fn __repr__(&self) -> PyResult<String> {
266 Ok(format!("CertReqMessages({} requests)", self.msgs()?.len()))
267 }
268}
269
270enum PubInfoGn {
274 None,
275 Uri(String),
276 Rfc822(String),
277 Dns(String),
278 DirectoryName(Vec<u8>),
279}
280
281#[pyclass(name = "CertReqMsgBuilder")]
299pub struct PyCertReqMsgBuilder {
300 cert_req_id: i64,
301 subject_der: Option<Vec<u8>>,
302 issuer_der: Option<Vec<u8>>,
303 spki_der: Option<Vec<u8>>,
304 popo_ra_verified: bool,
305 pub_info_action: i64,
307 pub_infos: Vec<(i64, PubInfoGn)>,
309}
310
311#[pymethods]
312impl PyCertReqMsgBuilder {
313 #[new]
315 fn new() -> Self {
316 Self {
317 cert_req_id: 0,
318 subject_der: None,
319 issuer_der: None,
320 spki_der: None,
321 popo_ra_verified: false,
322 pub_info_action: 1, pub_infos: Vec::new(),
324 }
325 }
326
327 fn cert_req_id<'py>(slf: Bound<'py, Self>, id: i64) -> Bound<'py, Self> {
329 slf.borrow_mut().cert_req_id = id;
330 slf
331 }
332
333 fn subject_name<'py>(slf: Bound<'py, Self>, name_der: &[u8]) -> Bound<'py, Self> {
338 slf.borrow_mut().subject_der = Some(name_der.to_vec());
339 slf
340 }
341
342 fn issuer_name<'py>(slf: Bound<'py, Self>, name_der: &[u8]) -> Bound<'py, Self> {
344 slf.borrow_mut().issuer_der = Some(name_der.to_vec());
345 slf
346 }
347
348 fn public_key_der<'py>(slf: Bound<'py, Self>, spki_der: &[u8]) -> Bound<'py, Self> {
352 slf.borrow_mut().spki_der = Some(spki_der.to_vec());
353 slf
354 }
355
356 fn popo_ra_verified<'py>(slf: Bound<'py, Self>) -> Bound<'py, Self> {
361 slf.borrow_mut().popo_ra_verified = true;
362 slf
363 }
364
365 fn publication_action<'py>(slf: Bound<'py, Self>, action: i64) -> Bound<'py, Self> {
367 slf.borrow_mut().pub_info_action = action;
368 slf
369 }
370
371 fn add_pub_info<'py>(slf: Bound<'py, Self>, pub_method: i64) -> Bound<'py, Self> {
375 slf.borrow_mut()
376 .pub_infos
377 .push((pub_method, PubInfoGn::None));
378 slf
379 }
380
381 fn pub_location_uri<'py>(
383 slf: Bound<'py, Self>,
384 pub_method: i64,
385 uri: &str,
386 ) -> Bound<'py, Self> {
387 slf.borrow_mut()
388 .pub_infos
389 .push((pub_method, PubInfoGn::Uri(uri.to_string())));
390 slf
391 }
392
393 fn pub_location_rfc822<'py>(
395 slf: Bound<'py, Self>,
396 pub_method: i64,
397 email: &str,
398 ) -> Bound<'py, Self> {
399 slf.borrow_mut()
400 .pub_infos
401 .push((pub_method, PubInfoGn::Rfc822(email.to_string())));
402 slf
403 }
404
405 fn pub_location_dns<'py>(
407 slf: Bound<'py, Self>,
408 pub_method: i64,
409 host: &str,
410 ) -> Bound<'py, Self> {
411 slf.borrow_mut()
412 .pub_infos
413 .push((pub_method, PubInfoGn::Dns(host.to_string())));
414 slf
415 }
416
417 fn pub_location_directory_name<'py>(
420 slf: Bound<'py, Self>,
421 pub_method: i64,
422 name_der: &[u8],
423 ) -> Bound<'py, Self> {
424 slf.borrow_mut()
425 .pub_infos
426 .push((pub_method, PubInfoGn::DirectoryName(name_der.to_vec())));
427 slf
428 }
429
430 fn build<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyCertReqMsg>> {
435 let mut rust_builder =
436 synta_certificate::CertReqMsgBuilder::new().cert_req_id(self.cert_req_id);
437 if let Some(ref b) = self.subject_der {
438 rust_builder = rust_builder.subject_name(b);
439 }
440 if let Some(ref b) = self.issuer_der {
441 rust_builder = rust_builder.issuer_name(b);
442 }
443 if let Some(ref b) = self.spki_der {
444 rust_builder = rust_builder.public_key_der(b);
445 }
446 if self.popo_ra_verified {
447 rust_builder = rust_builder.popo_ra_verified();
448 }
449 rust_builder = rust_builder.publication_action(self.pub_info_action);
450 for (method, gn) in &self.pub_infos {
451 rust_builder = match gn {
452 PubInfoGn::None => rust_builder.add_pub_info(*method),
453 PubInfoGn::Uri(uri) => rust_builder.pub_location_uri(*method, uri),
454 PubInfoGn::Rfc822(email) => rust_builder.pub_location_rfc822(*method, email),
455 PubInfoGn::Dns(host) => rust_builder.pub_location_dns(*method, host),
456 PubInfoGn::DirectoryName(bytes) => {
457 rust_builder.pub_location_directory_name(*method, bytes)
458 }
459 };
460 }
461 let msg_der = rust_builder
462 .build()
463 .map_err(|e| PyValueError::new_err(e.to_string()))?;
464
465 let msg = Decoder::new(&msg_der, Encoding::Der)
467 .decode::<synta_certificate::crmf_types::CertReqMsg<'_>>()
468 .map_err(SyntaErr)?;
469 let mut py_msg = PyCertReqMsg::from_rust(&msg);
470 py_msg.msg_der = msg_der;
472 Py::new(py, py_msg).map(|x| x.into_bound(py))
473 }
474
475 fn __repr__(&self) -> String {
476 format!("CertReqMsgBuilder(cert_req_id={})", self.cert_req_id)
477 }
478}
479
480#[pyclass(name = "CertReqMessagesBuilder")]
499pub struct PyCertReqMessagesBuilder {
500 requests: Vec<Py<PyCertReqMsg>>,
501}
502
503#[pymethods]
504impl PyCertReqMessagesBuilder {
505 #[new]
507 fn new() -> Self {
508 Self {
509 requests: Vec::new(),
510 }
511 }
512
513 fn add_request<'py>(
517 slf: Bound<'py, Self>,
518 req: Bound<'py, PyCertReqMsg>,
519 ) -> PyResult<Bound<'py, Self>> {
520 slf.borrow_mut().requests.push(req.unbind());
521 Ok(slf)
522 }
523
524 fn build<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyCertReqMessages>> {
528 let mut builder = synta_certificate::CertReqMessagesBuilder::new();
529 for req_py in &self.requests {
530 let req = req_py.borrow(py);
531 builder = builder.add_request_der(&req.msg_der);
532 }
533 let der = builder
534 .build()
535 .map_err(|e| PyValueError::new_err(e.to_string()))?;
536 let py_bytes = PyBytes::new(py, &der);
537 let msgs = PyCertReqMessages::from_der(py, py_bytes)?;
538 Py::new(py, msgs).map(|x| x.into_bound(py))
539 }
540
541 fn __repr__(&self) -> String {
542 format!("CertReqMessagesBuilder({} requests)", self.requests.len())
543 }
544}
545
546pub(super) fn register_crmf_submodule(parent: &Bound<'_, PyModule>) -> PyResult<()> {
550 let py = parent.py();
551 let m = PyModule::new(py, "crmf")?;
552
553 m.add_class::<PyCertReqMessages>()?;
554 m.add_class::<PyCertReqMsg>()?;
555 m.add_class::<PyCertReqMessagesBuilder>()?;
556 m.add_class::<PyCertReqMsgBuilder>()?;
557
558 m.add(
560 "PUB_METHOD_DONT_CARE",
561 synta_certificate::PUB_METHOD_DONT_CARE,
562 )?;
563 m.add("PUB_METHOD_X500", synta_certificate::PUB_METHOD_X500)?;
564 m.add("PUB_METHOD_WEB", synta_certificate::PUB_METHOD_WEB)?;
565 m.add("PUB_METHOD_LDAP", synta_certificate::PUB_METHOD_LDAP)?;
566
567 m.add(
569 "ID_REG_CTRL_REG_TOKEN",
570 super::oid_const(py, synta_certificate::crmf_types::ID_REG_CTRL_REG_TOKEN),
571 )?;
572 m.add(
573 "ID_REG_CTRL_AUTHENTICATOR",
574 super::oid_const(py, synta_certificate::crmf_types::ID_REG_CTRL_AUTHENTICATOR),
575 )?;
576 m.add(
577 "ID_REG_CTRL_PKI_PUBLICATION_INFO",
578 super::oid_const(
579 py,
580 synta_certificate::crmf_types::ID_REG_CTRL_PKI_PUBLICATION_INFO,
581 ),
582 )?;
583 m.add(
584 "ID_REG_CTRL_PKI_ARCHIVE_OPTIONS",
585 super::oid_const(
586 py,
587 synta_certificate::crmf_types::ID_REG_CTRL_PKI_ARCHIVE_OPTIONS,
588 ),
589 )?;
590 m.add(
591 "ID_REG_CTRL_OLD_CERT_ID",
592 super::oid_const(py, synta_certificate::crmf_types::ID_REG_CTRL_OLD_CERT_ID),
593 )?;
594 m.add(
595 "ID_REG_CTRL_PROTOCOL_ENCR_KEY",
596 super::oid_const(
597 py,
598 synta_certificate::crmf_types::ID_REG_CTRL_PROTOCOL_ENCR_KEY,
599 ),
600 )?;
601 m.add(
602 "ID_REG_INFO_UTF8_PAIRS",
603 super::oid_const(py, synta_certificate::crmf_types::ID_REG_INFO_UTF8_PAIRS),
604 )?;
605 m.add(
606 "ID_REG_INFO_CERT_REQ",
607 super::oid_const(py, synta_certificate::crmf_types::ID_REG_INFO_CERT_REQ),
608 )?;
609
610 crate::install_submodule(
611 parent,
612 &m,
613 "synta.crmf",
614 Some(concat!(
615 "synta.crmf — RFC 4211 Certificate Request Message Format types.\n\n",
616 "Provides CertReqMessages (SEQUENCE OF CertReqMsg) and CertReqMsg\n",
617 "for decoding CRMF batches used in CMP certificate management,\n",
618 "along with registration-control OID constants.",
619 )),
620 )
621}