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 mut dec = Decoder::new(self.raw, Encoding::Der);
187 let decoded = dec
188 .decode::<synta_certificate::crmf_types::CertReqMessages<'static>>()
189 .map_err(SyntaErr)?;
190 let _ = self.inner.set(decoded);
191 Ok(self.inner.get().unwrap())
192 }
193
194 fn build_requests_list<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
195 let msgs = self.msgs()?;
196 let list = PyList::empty(py);
197 for msg in msgs {
198 let obj = Py::new(py, PyCertReqMsg::from_rust(msg))?;
199 list.append(obj)?;
200 }
201 Ok(list)
202 }
203}
204
205#[pymethods]
206impl PyCertReqMessages {
207 #[staticmethod]
212 fn from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
213 let py_bytes = data.unbind();
214 {
215 let raw = py_bytes.as_bytes(py);
216 Decoder::new(raw, Encoding::Der)
217 .decode::<synta_certificate::crmf_types::CertReqMessages<'_>>()
218 .map_err(SyntaErr)?;
219 }
220 let raw: &'static [u8] = unsafe {
221 let s = py_bytes.bind(py).as_bytes();
222 std::slice::from_raw_parts(s.as_ptr(), s.len())
223 };
224 Ok(Self {
225 _data: py_bytes,
226 raw,
227 inner: OnceLock::new(),
228 requests_cache: OnceLock::new(),
229 })
230 }
231
232 fn to_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
234 Ok(PyBytes::new(py, &self.msgs()?.to_der().map_err(SyntaErr)?))
235 }
236
237 #[getter]
239 fn requests<'py>(&'py self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
240 if let Some(c) = self.requests_cache.get() {
241 return Ok(c.clone_ref(py).into_bound(py));
242 }
243 let list = self.build_requests_list(py)?;
244 let _ = self.requests_cache.set(list.as_unbound().clone_ref(py));
245 Ok(list)
246 }
247
248 fn __len__(&self) -> PyResult<usize> {
249 Ok(self.msgs()?.len())
250 }
251
252 fn __iter__<'py>(&'py self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
253 self.requests(py)?.call_method0("__iter__")
254 }
255
256 fn __getitem__(&self, py: Python<'_>, index: usize) -> PyResult<Py<PyCertReqMsg>> {
257 let msgs = self.msgs()?;
258 if index >= msgs.len() {
259 return Err(pyo3::exceptions::PyIndexError::new_err(format!(
260 "index {index} out of range for CertReqMessages with {} entries",
261 msgs.len()
262 )));
263 }
264 Py::new(py, PyCertReqMsg::from_rust(&msgs[index]))
265 }
266
267 fn __repr__(&self) -> PyResult<String> {
268 Ok(format!("CertReqMessages({} requests)", self.msgs()?.len()))
269 }
270}
271
272enum PubInfoGn {
276 None,
277 Uri(String),
278 Rfc822(String),
279 Dns(String),
280 DirectoryName(Vec<u8>),
281}
282
283#[pyclass(name = "CertReqMsgBuilder")]
301pub struct PyCertReqMsgBuilder {
302 cert_req_id: i64,
303 subject_der: Option<Vec<u8>>,
304 issuer_der: Option<Vec<u8>>,
305 spki_der: Option<Vec<u8>>,
306 popo_ra_verified: bool,
307 pub_info_action: i64,
309 pub_infos: Vec<(i64, PubInfoGn)>,
311}
312
313#[pymethods]
314impl PyCertReqMsgBuilder {
315 #[new]
317 fn new() -> Self {
318 Self {
319 cert_req_id: 0,
320 subject_der: None,
321 issuer_der: None,
322 spki_der: None,
323 popo_ra_verified: false,
324 pub_info_action: 1, pub_infos: Vec::new(),
326 }
327 }
328
329 fn cert_req_id<'py>(slf: Bound<'py, Self>, id: i64) -> Bound<'py, Self> {
331 slf.borrow_mut().cert_req_id = id;
332 slf
333 }
334
335 fn subject_name<'py>(slf: Bound<'py, Self>, name_der: &[u8]) -> Bound<'py, Self> {
340 slf.borrow_mut().subject_der = Some(name_der.to_vec());
341 slf
342 }
343
344 fn issuer_name<'py>(slf: Bound<'py, Self>, name_der: &[u8]) -> Bound<'py, Self> {
346 slf.borrow_mut().issuer_der = Some(name_der.to_vec());
347 slf
348 }
349
350 fn public_key_der<'py>(slf: Bound<'py, Self>, spki_der: &[u8]) -> Bound<'py, Self> {
354 slf.borrow_mut().spki_der = Some(spki_der.to_vec());
355 slf
356 }
357
358 fn popo_ra_verified<'py>(slf: Bound<'py, Self>) -> Bound<'py, Self> {
363 slf.borrow_mut().popo_ra_verified = true;
364 slf
365 }
366
367 fn publication_action<'py>(slf: Bound<'py, Self>, action: i64) -> Bound<'py, Self> {
369 slf.borrow_mut().pub_info_action = action;
370 slf
371 }
372
373 fn add_pub_info<'py>(slf: Bound<'py, Self>, pub_method: i64) -> Bound<'py, Self> {
377 slf.borrow_mut()
378 .pub_infos
379 .push((pub_method, PubInfoGn::None));
380 slf
381 }
382
383 fn pub_location_uri<'py>(
385 slf: Bound<'py, Self>,
386 pub_method: i64,
387 uri: &str,
388 ) -> Bound<'py, Self> {
389 slf.borrow_mut()
390 .pub_infos
391 .push((pub_method, PubInfoGn::Uri(uri.to_string())));
392 slf
393 }
394
395 fn pub_location_rfc822<'py>(
397 slf: Bound<'py, Self>,
398 pub_method: i64,
399 email: &str,
400 ) -> Bound<'py, Self> {
401 slf.borrow_mut()
402 .pub_infos
403 .push((pub_method, PubInfoGn::Rfc822(email.to_string())));
404 slf
405 }
406
407 fn pub_location_dns<'py>(
409 slf: Bound<'py, Self>,
410 pub_method: i64,
411 host: &str,
412 ) -> Bound<'py, Self> {
413 slf.borrow_mut()
414 .pub_infos
415 .push((pub_method, PubInfoGn::Dns(host.to_string())));
416 slf
417 }
418
419 fn pub_location_directory_name<'py>(
422 slf: Bound<'py, Self>,
423 pub_method: i64,
424 name_der: &[u8],
425 ) -> Bound<'py, Self> {
426 slf.borrow_mut()
427 .pub_infos
428 .push((pub_method, PubInfoGn::DirectoryName(name_der.to_vec())));
429 slf
430 }
431
432 fn build<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyCertReqMsg>> {
437 let mut rust_builder =
438 synta_certificate::CertReqMsgBuilder::new().cert_req_id(self.cert_req_id);
439 if let Some(ref b) = self.subject_der {
440 rust_builder = rust_builder.subject_name(b);
441 }
442 if let Some(ref b) = self.issuer_der {
443 rust_builder = rust_builder.issuer_name(b);
444 }
445 if let Some(ref b) = self.spki_der {
446 rust_builder = rust_builder.public_key_der(b);
447 }
448 if self.popo_ra_verified {
449 rust_builder = rust_builder.popo_ra_verified();
450 }
451 rust_builder = rust_builder.publication_action(self.pub_info_action);
452 for (method, gn) in &self.pub_infos {
453 rust_builder = match gn {
454 PubInfoGn::None => rust_builder.add_pub_info(*method),
455 PubInfoGn::Uri(uri) => rust_builder.pub_location_uri(*method, uri),
456 PubInfoGn::Rfc822(email) => rust_builder.pub_location_rfc822(*method, email),
457 PubInfoGn::Dns(host) => rust_builder.pub_location_dns(*method, host),
458 PubInfoGn::DirectoryName(bytes) => {
459 rust_builder.pub_location_directory_name(*method, bytes)
460 }
461 };
462 }
463 let msg_der = rust_builder
464 .build()
465 .map_err(|e| PyValueError::new_err(e.to_string()))?;
466
467 let msg = Decoder::new(&msg_der, Encoding::Der)
469 .decode::<synta_certificate::crmf_types::CertReqMsg<'_>>()
470 .map_err(SyntaErr)?;
471 let mut py_msg = PyCertReqMsg::from_rust(&msg);
472 py_msg.msg_der = msg_der;
474 Py::new(py, py_msg).map(|x| x.into_bound(py))
475 }
476
477 fn __repr__(&self) -> String {
478 format!("CertReqMsgBuilder(cert_req_id={})", self.cert_req_id)
479 }
480}
481
482#[pyclass(name = "CertReqMessagesBuilder")]
501pub struct PyCertReqMessagesBuilder {
502 requests: Vec<Py<PyCertReqMsg>>,
503}
504
505#[pymethods]
506impl PyCertReqMessagesBuilder {
507 #[new]
509 fn new() -> Self {
510 Self {
511 requests: Vec::new(),
512 }
513 }
514
515 fn add_request<'py>(
519 slf: Bound<'py, Self>,
520 req: Bound<'py, PyCertReqMsg>,
521 ) -> PyResult<Bound<'py, Self>> {
522 slf.borrow_mut().requests.push(req.unbind());
523 Ok(slf)
524 }
525
526 fn build<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyCertReqMessages>> {
530 let mut builder = synta_certificate::CertReqMessagesBuilder::new();
531 for req_py in &self.requests {
532 let req = req_py.borrow(py);
533 builder = builder.add_request_der(&req.msg_der);
534 }
535 let der = builder
536 .build()
537 .map_err(|e| PyValueError::new_err(e.to_string()))?;
538 let py_bytes = PyBytes::new(py, &der);
539 let msgs = PyCertReqMessages::from_der(py, py_bytes)?;
540 Py::new(py, msgs).map(|x| x.into_bound(py))
541 }
542
543 fn __repr__(&self) -> String {
544 format!("CertReqMessagesBuilder({} requests)", self.requests.len())
545 }
546}
547
548pub(super) fn register_crmf_submodule(parent: &Bound<'_, PyModule>) -> PyResult<()> {
552 let py = parent.py();
553 let m = PyModule::new(py, "crmf")?;
554
555 m.add_class::<PyCertReqMessages>()?;
556 m.add_class::<PyCertReqMsg>()?;
557 m.add_class::<PyCertReqMessagesBuilder>()?;
558 m.add_class::<PyCertReqMsgBuilder>()?;
559
560 m.add(
562 "PUB_METHOD_DONT_CARE",
563 synta_certificate::PUB_METHOD_DONT_CARE,
564 )?;
565 m.add("PUB_METHOD_X500", synta_certificate::PUB_METHOD_X500)?;
566 m.add("PUB_METHOD_WEB", synta_certificate::PUB_METHOD_WEB)?;
567 m.add("PUB_METHOD_LDAP", synta_certificate::PUB_METHOD_LDAP)?;
568
569 m.add(
571 "ID_REG_CTRL_REG_TOKEN",
572 super::oid_const(py, synta_certificate::crmf_types::ID_REG_CTRL_REG_TOKEN),
573 )?;
574 m.add(
575 "ID_REG_CTRL_AUTHENTICATOR",
576 super::oid_const(py, synta_certificate::crmf_types::ID_REG_CTRL_AUTHENTICATOR),
577 )?;
578 m.add(
579 "ID_REG_CTRL_PKI_PUBLICATION_INFO",
580 super::oid_const(
581 py,
582 synta_certificate::crmf_types::ID_REG_CTRL_PKI_PUBLICATION_INFO,
583 ),
584 )?;
585 m.add(
586 "ID_REG_CTRL_PKI_ARCHIVE_OPTIONS",
587 super::oid_const(
588 py,
589 synta_certificate::crmf_types::ID_REG_CTRL_PKI_ARCHIVE_OPTIONS,
590 ),
591 )?;
592 m.add(
593 "ID_REG_CTRL_OLD_CERT_ID",
594 super::oid_const(py, synta_certificate::crmf_types::ID_REG_CTRL_OLD_CERT_ID),
595 )?;
596 m.add(
597 "ID_REG_CTRL_PROTOCOL_ENCR_KEY",
598 super::oid_const(
599 py,
600 synta_certificate::crmf_types::ID_REG_CTRL_PROTOCOL_ENCR_KEY,
601 ),
602 )?;
603 m.add(
604 "ID_REG_INFO_UTF8_PAIRS",
605 super::oid_const(py, synta_certificate::crmf_types::ID_REG_INFO_UTF8_PAIRS),
606 )?;
607 m.add(
608 "ID_REG_INFO_CERT_REQ",
609 super::oid_const(py, synta_certificate::crmf_types::ID_REG_INFO_CERT_REQ),
610 )?;
611
612 crate::install_submodule(
613 parent,
614 &m,
615 "synta.crmf",
616 Some(concat!(
617 "synta.crmf — RFC 4211 Certificate Request Message Format types.\n\n",
618 "Provides CertReqMessages (SEQUENCE OF CertReqMsg) and CertReqMsg\n",
619 "for decoding CRMF batches used in CMP certificate management,\n",
620 "along with registration-control OID constants.",
621 )),
622 )
623}