_synta/certificate/cert.rs
1//! Python binding for [`PyCertificate`] (X.509 Certificate).
2
3use std::sync::OnceLock;
4
5use pyo3::prelude::*;
6use pyo3::types::{PyBytes, PyList, PyString};
7use pyo3::PyClass;
8
9use synta::traits::Encode;
10use synta::{Decoder, Encoding};
11use synta_certificate::{Certificate, PolicyQualifierInfo, Time};
12
13use crate::types::PyObjectIdentifier;
14
15use super::oid_from_pyany;
16
17/// Re-encode an ASN.1 `Element` to DER bytes, returning `None` for a missing
18/// optional field. Used by the algorithm-parameter getters.
19fn encode_element_opt<'py>(
20 py: Python<'py>,
21 elem: Option<&synta::Element<'_>>,
22) -> PyResult<Option<Bound<'py, PyBytes>>> {
23 match elem {
24 None => Ok(None),
25 Some(e) => {
26 let mut encoder = synta::Encoder::new(Encoding::Der);
27 e.encode(&mut encoder)
28 .map_err(|err| pyo3::exceptions::PyValueError::new_err(format!("{err}")))?;
29 let bytes = encoder
30 .finish()
31 .map_err(|err| pyo3::exceptions::PyValueError::new_err(format!("{err}")))?;
32 Ok(Some(PyBytes::new(py, &bytes)))
33 }
34 }
35}
36
37/// Decode a raw DER Name SEQUENCE and format it as an RFC 4514-style string.
38fn decode_name_debug(raw: &[u8]) -> String {
39 synta_certificate::name::format_dn(raw)
40}
41
42/// Shared implementation for every `from_pem` static method.
43///
44/// Decodes all PEM blocks in `data` and constructs a Python object for each
45/// by calling `make_obj(py, der_bytes)`. The closure is responsible for
46/// parsing the DER bytes and wrapping the result as a `Bound<'py, PyAny>`;
47/// this lets the caller use concrete types (with their full `Py::new` bounds)
48/// without any generic `PyClass` constraint here.
49///
50/// Returns a single object for one block, a `list` for multiple blocks, and
51/// raises `ValueError` when no block is found.
52pub(super) fn pem_blocks_to_pyobject<'py, F>(
53 py: Python<'py>,
54 data: &[u8],
55 make_obj: F,
56) -> PyResult<Bound<'py, PyAny>>
57where
58 F: for<'a> Fn(Python<'py>, Bound<'a, PyBytes>) -> PyResult<Bound<'py, PyAny>>,
59{
60 let blocks = synta_certificate::pem_blocks(data);
61 match blocks.len() {
62 0 => Err(pyo3::exceptions::PyValueError::new_err(
63 "no PEM block found in input",
64 )),
65 1 => make_obj(py, PyBytes::new(py, &blocks[0].1)),
66 _ => {
67 let list = PyList::empty(py);
68 for (_, der) in &blocks {
69 list.append(make_obj(py, PyBytes::new(py, der))?)?;
70 }
71 Ok(list.into_any())
72 }
73 }
74}
75
76/// Shared implementation for every `to_pem` static method.
77///
78/// Accepts either a single Python object of type `T` or a `list` of them,
79/// serialises each to a PEM block labelled `label`, and returns the
80/// concatenated bytes. The `get_der` closure extracts the raw DER bytes from
81/// a Rust reference to `T`.
82///
83/// Using a closure (rather than a trait method) avoids requiring `T: PyClass`
84/// with internal PyO3 initialiser bounds while still letting the caller use
85/// concrete types.
86pub(super) fn pyobject_to_pem<'py, T, D>(
87 py: Python<'py>,
88 label: &str,
89 obj_or_list: &Bound<'_, PyAny>,
90 get_der: D,
91) -> PyResult<Bound<'py, PyBytes>>
92where
93 T: PyClass,
94 D: for<'r> Fn(&'r T) -> &'r [u8],
95{
96 let mut pem: Vec<u8> = Vec::new();
97 if let Ok(list) = obj_or_list.cast::<PyList>() {
98 for item in list.iter() {
99 let bound_t = item.cast::<T>().map_err(|_| {
100 pyo3::exceptions::PyTypeError::new_err(format!(
101 "list items must be {label} objects"
102 ))
103 })?;
104 let borrow = bound_t.borrow();
105 pem.extend_from_slice(&synta_certificate::der_to_pem(label, get_der(&borrow)));
106 }
107 } else {
108 let bound_t = obj_or_list.cast::<T>().map_err(|_| {
109 pyo3::exceptions::PyTypeError::new_err(format!(
110 "expected a {label} object or list[{label}]"
111 ))
112 })?;
113 let borrow = bound_t.borrow();
114 pem.extend_from_slice(&synta_certificate::der_to_pem(label, get_der(&borrow)));
115 }
116 Ok(PyBytes::new(py, &pem))
117}
118
119/// A single policy qualifier from the CertificatePolicies extension.
120///
121/// Corresponds to `PolicyQualifierInfo` in RFC 5280:
122///
123/// ```asn1
124/// PolicyQualifierInfo ::= SEQUENCE {
125/// policyQualifierId PolicyQualifierId,
126/// qualifier ANY DEFINED BY policyQualifierId }
127/// ```
128///
129/// The ``qualifier_value`` bytes are the complete DER TLV of the ``qualifier``
130/// field. For ``id-qt-cps`` (OID ``1.3.6.1.5.5.7.2.1``) this encodes an
131/// IA5String containing the CPS URI. For ``id-qt-unotice``
132/// (``1.3.6.1.5.5.7.2.2``) it encodes a ``UserNotice`` SEQUENCE.
133#[pyclass(frozen, name = "PolicyQualifier", module = "synta")]
134pub struct PyPolicyQualifier {
135 pub qualifier_oid: synta::ObjectIdentifier,
136 pub qualifier_value: Vec<u8>,
137}
138
139#[pymethods]
140impl PyPolicyQualifier {
141 /// OID identifying the qualifier type.
142 ///
143 /// Common values:
144 ///
145 /// - ``1.3.6.1.5.5.7.2.1`` — CPS pointer (URI string)
146 /// - ``1.3.6.1.5.5.7.2.2`` — user notice
147 #[getter]
148 fn qualifier_oid<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyObjectIdentifier>> {
149 Py::new(py, PyObjectIdentifier::from_oid(self.qualifier_oid.clone()))
150 .map(|p| p.into_bound(py))
151 }
152
153 /// Raw DER bytes of the qualifier value (tag + length + value).
154 ///
155 /// For a CPS qualifier (OID ``1.3.6.1.5.5.7.2.1``) this is an IA5String
156 /// encoding of the CPS URI. Decode with:
157 ///
158 /// ```python,ignore
159 /// synta.Decoder(data, synta.Encoding.DER).decode_ia5_string()
160 /// ```
161 #[getter]
162 fn qualifier_value<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
163 PyBytes::new(py, &self.qualifier_value)
164 }
165
166 fn __repr__(&self) -> String {
167 format!("PolicyQualifier(qualifier_oid={})", self.qualifier_oid)
168 }
169}
170
171/// A single ``PolicyInformation`` entry from the CertificatePolicies extension.
172///
173/// Corresponds to ``PolicyInformation`` in RFC 5280:
174///
175/// ```asn1
176/// PolicyInformation ::= SEQUENCE {
177/// policyIdentifier CertPolicyId,
178/// policyQualifiers PolicyQualifiers OPTIONAL }
179/// ```
180#[pyclass(frozen, name = "PolicyInformation", module = "synta")]
181pub struct PyPolicyInformation {
182 pub policy_oid: synta::ObjectIdentifier,
183 pub qualifiers: Vec<(synta::ObjectIdentifier, Vec<u8>)>,
184}
185
186#[pymethods]
187impl PyPolicyInformation {
188 /// The certificate policy OID.
189 #[getter]
190 fn policy_oid<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyObjectIdentifier>> {
191 Py::new(py, PyObjectIdentifier::from_oid(self.policy_oid.clone())).map(|p| p.into_bound(py))
192 }
193
194 /// List of :class:`PolicyQualifier` entries (empty when ``policyQualifiers`` is absent).
195 #[getter]
196 fn qualifiers<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
197 let list = PyList::empty(py);
198 for (qoid, qval) in &self.qualifiers {
199 let pq = Py::new(
200 py,
201 PyPolicyQualifier {
202 qualifier_oid: qoid.clone(),
203 qualifier_value: qval.clone(),
204 },
205 )?;
206 list.append(pq.into_bound(py))?;
207 }
208 Ok(list)
209 }
210
211 fn __repr__(&self) -> String {
212 format!("PolicyInformation(policy_oid={})", self.policy_oid)
213 }
214}
215
216/// X.509 Certificate accessible from Python.
217///
218/// Example (inside a Python extension that called `register_module`):
219///
220/// ```python
221/// cert = Certificate.from_der(open("cert.der", "rb").read())
222/// print(cert.subject)
223/// ```
224#[pyclass(frozen, name = "Certificate")]
225pub struct PyCertificate {
226 // Holds a strong reference to the Python bytes object that backs the
227 // DER data. The CPython refcount prevents the underlying bytes buffer
228 // from being freed for the full lifetime of this struct. After the
229 // struct is collected by Python's GC, `_data` drops (in Rust's
230 // declaration order) and decrements the refcount; by that point no
231 // Rust code can hold a borrow of this struct, so no read of `raw`
232 // can occur through `&self` after drop begins.
233 pub(super) _data: Py<PyBytes>,
234 // Raw slice pointing into `_data`'s buffer. SAFETY: valid as long as
235 // `_data` is alive; CPython bytes objects have a fixed-address buffer
236 // that is never relocated. Stored here so the OnceLock init closure in
237 // `cert()` can access the bytes without requiring a GIL token.
238 pub(super) raw: &'static [u8],
239 // Full decoded certificate, heap-allocated and initialised lazily on first
240 // field access. `Box` keeps the 568-byte `Certificate` off the struct so
241 // that `PyCertificate` stays within CPython's 512-byte `pymalloc`
242 // threshold (after adding the Python object header). Allocating through
243 // pymalloc instead of the system allocator is roughly 3× faster, which is
244 // measurable at parse-only speeds. The full recursive decode is deferred
245 // so that parse-only workloads pay only the shallow envelope-scan cost.
246 inner: OnceLock<Box<Certificate<'static>>>,
247 // Byte range within `_data` covering the complete TBSCertificate TLV.
248 tbs_range: std::ops::Range<usize>,
249 // Lazily-computed Python objects. Each field is initialised on first
250 // Python access and returned via clone_ref on every subsequent call,
251 // avoiding repeated allocation and PyO3 string construction.
252 issuer_cache: OnceLock<Py<PyString>>,
253 subject_cache: OnceLock<Py<PyString>>,
254 signature_algorithm_cache: OnceLock<Py<PyString>>,
255 not_before_cache: OnceLock<Py<PyString>>,
256 not_after_cache: OnceLock<Py<PyString>>,
257 public_key_algorithm_cache: OnceLock<Py<PyString>>,
258 serial_number_cache: OnceLock<Py<PyAny>>,
259 signature_value_cache: OnceLock<Py<PyBytes>>,
260 public_key_cache: OnceLock<Py<PyBytes>>,
261 tbs_bytes_cache: OnceLock<Py<PyBytes>>,
262 signature_algorithm_oid_cache: OnceLock<Py<PyObjectIdentifier>>,
263 public_key_algorithm_oid_cache: OnceLock<Py<PyObjectIdentifier>>,
264 signature_algorithm_der_cache: OnceLock<Py<PyBytes>>,
265 signature_algorithm_params_cache: OnceLock<Option<Py<PyBytes>>>,
266 public_key_algorithm_params_cache: OnceLock<Option<Py<PyBytes>>>,
267 extensions_der_cache: OnceLock<Option<Py<PyBytes>>>,
268 issuer_raw_der_cache: OnceLock<Py<PyBytes>>,
269 subject_raw_der_cache: OnceLock<Py<PyBytes>>,
270 not_before_utc_cache: OnceLock<Py<PyAny>>,
271 not_after_utc_cache: OnceLock<Py<PyAny>>,
272 spki_der_cache: OnceLock<Py<PyBytes>>,
273}
274
275impl PyCertificate {
276 /// Return the fully-decoded certificate, triggering a full recursive
277 /// `Certificate::decode()` on the first call.
278 ///
279 /// The result is cached in `inner` so subsequent calls are a single
280 /// atomic load. Returns `Err(PyValueError)` if the full decode fails
281 /// (e.g. malformed inner content that passed the shallow envelope scan
282 /// in `from_der`). Unlike a panic, this surfaces as a catchable Python
283 /// `ValueError` rather than an uncatchable `PanicException`.
284 pub(super) fn cert(&self) -> PyResult<&Certificate<'static>> {
285 if let Some(v) = self.inner.get() {
286 return Ok(v.as_ref());
287 }
288 let mut decoder = Decoder::new(self.raw, Encoding::Der);
289 let decoded = decoder.decode().map_err(|e| {
290 pyo3::exceptions::PyValueError::new_err(format!("Certificate DER decode failed: {e}"))
291 })?;
292 let _ = self.inner.set(Box::new(decoded));
293 Ok(self.inner.get().unwrap().as_ref())
294 }
295}
296
297#[pymethods]
298impl PyCertificate {
299 /// Parse a DER-encoded X.509 certificate.
300 #[staticmethod]
301 fn from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
302 // Store the Python bytes object directly — no copy of the DER data.
303 let py_bytes = data.unbind();
304
305 // SAFETY: `py_bytes` holds a strong reference (Py<PyBytes>) that
306 // keeps the Python bytes object alive for the lifetime of this struct.
307 // CPython's bytes objects have a fixed-address, non-relocating payload
308 // buffer (CPython has no moving GC for bytes objects). The slice
309 // lifetime is extended to 'static; the actual safety invariants are:
310 // (1) All reads of `raw` go through `&self` (a borrow of the struct).
311 // No borrow of the struct can outlive the struct, so `raw` is
312 // never read after the struct begins dropping.
313 // (2) `raw: &'static [u8]` has no destructor (it is a fat pointer
314 // with no heap allocation), so Rust dropping `_data` before `raw`
315 // (fields drop in declaration order) does not cause a
316 // use-after-free during the drop sequence itself.
317 // (3) `inner` contains `Certificate<'static>` references into the
318 // buffer; dropping `Box<Certificate<'static>>` frees the box
319 // allocation but does not read through the contained &'static
320 // slices (borrows have no destructors in Rust).
321 // CPython-only: this pattern does not hold for PyPy or GraalPy,
322 // which may relocate or compact heap objects.
323 let raw: &'static [u8] = unsafe {
324 let s = py_bytes.bind(py).as_bytes();
325 std::slice::from_raw_parts(s.as_ptr(), s.len())
326 };
327
328 // Shallow structural scan: validate the Certificate SEQUENCE envelope
329 // and locate the TBSCertificate TLV range. This is ~4 decoder
330 // operations — roughly 10× faster than a full Certificate::decode().
331 // The full decode is deferred to the first field access via `cert()`.
332 //
333 // Certificate ::= SEQUENCE {
334 // tbsCertificate TBSCertificate, -- child 0: SEQUENCE
335 // signatureAlgorithm AlgorithmIdentifier, -- child 1: SEQUENCE
336 // signature BIT STRING -- child 2
337 // }
338 let tbs_range = {
339 let mut d = Decoder::new(raw, Encoding::Der);
340 // Outer Certificate SEQUENCE tag + length.
341 d.read_tag()
342 .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
343 d.read_length()
344 .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
345 // TBSCertificate: record the full TLV range (tag + length + content).
346 let tbs_start = d.position();
347 d.read_tag()
348 .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
349 let tbs_len = d
350 .read_length()
351 .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
352 let tbs_content_len = tbs_len
353 .definite()
354 .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
355 tbs_start..(d.position() + tbs_content_len)
356 };
357
358 Ok(Self {
359 _data: py_bytes,
360 raw,
361 inner: OnceLock::new(),
362 tbs_range,
363 issuer_cache: OnceLock::new(),
364 subject_cache: OnceLock::new(),
365 signature_algorithm_cache: OnceLock::new(),
366 not_before_cache: OnceLock::new(),
367 not_after_cache: OnceLock::new(),
368 public_key_algorithm_cache: OnceLock::new(),
369 serial_number_cache: OnceLock::new(),
370 signature_value_cache: OnceLock::new(),
371 public_key_cache: OnceLock::new(),
372 tbs_bytes_cache: OnceLock::new(),
373 signature_algorithm_oid_cache: OnceLock::new(),
374 public_key_algorithm_oid_cache: OnceLock::new(),
375 signature_algorithm_der_cache: OnceLock::new(),
376 signature_algorithm_params_cache: OnceLock::new(),
377 public_key_algorithm_params_cache: OnceLock::new(),
378 extensions_der_cache: OnceLock::new(),
379 issuer_raw_der_cache: OnceLock::new(),
380 subject_raw_der_cache: OnceLock::new(),
381 not_before_utc_cache: OnceLock::new(),
382 not_after_utc_cache: OnceLock::new(),
383 spki_der_cache: OnceLock::new(),
384 })
385 }
386
387 /// Parse a DER-encoded X.509 certificate and perform a full RFC 5280 decode immediately.
388 ///
389 /// Unlike :meth:`from_der`, which performs only a shallow 4-operation
390 /// envelope scan and defers the full :class:`Certificate` decode to the
391 /// first field access, this method triggers the complete recursive
392 /// ``Certificate::decode()`` at construction time.
393 ///
394 /// Use this when you need all fields to be available without any
395 /// lazy-decode latency on the first getter call, or when benchmarking the
396 /// true full-parse cost that is comparable to Criterion's ``rust_typed``
397 /// numbers.
398 ///
399 /// ```python
400 /// # Full parse happens here — no deferred work on first field access.
401 /// cert = Certificate.full_from_der(der)
402 /// print(cert.issuer) # warm path only
403 /// ```
404 #[staticmethod]
405 fn full_from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
406 let cert = Self::from_der(py, data)?;
407 // Prime the OnceLock: runs the full Certificate::decode() now.
408 cert.cert()?;
409 Ok(cert)
410 }
411
412 /// Parse a PEM-encoded X.509 certificate.
413 ///
414 /// Strips the ``-----BEGIN CERTIFICATE-----`` / ``-----END CERTIFICATE-----``
415 /// boundary lines, decodes the base64 body, and calls :meth:`from_der`.
416 /// No external dependencies are required — the decoder is implemented in
417 /// pure Rust inside ``synta-certificate``.
418 ///
419 /// Returns a single :class:`Certificate` when the input contains exactly
420 /// one PEM block. When multiple blocks are present (e.g. a certificate
421 /// chain file), returns a :class:`list` of :class:`Certificate` objects in
422 /// the order they appear. Raises :exc:`ValueError` if no PEM block is
423 /// found.
424 ///
425 /// ```python
426 /// # Single certificate — returns Certificate directly.
427 /// cert = Certificate.from_pem(open("cert.pem", "rb").read())
428 /// print(cert.subject)
429 ///
430 /// # Certificate chain — returns list[Certificate].
431 /// chain = Certificate.from_pem(open("chain.pem", "rb").read())
432 /// for cert in chain:
433 /// print(cert.subject)
434 /// ```
435 #[staticmethod]
436 fn from_pem<'py>(py: Python<'py>, data: Bound<'_, PyBytes>) -> PyResult<Bound<'py, PyAny>> {
437 pem_blocks_to_pyobject(py, data.as_bytes(), |py, bytes| {
438 let obj = Self::from_der(py, bytes)?;
439 Ok(Py::new(py, obj)?.into_bound(py).into_any())
440 })
441 }
442
443 /// Convert to a ``cryptography.x509.Certificate`` (PyCA) object.
444 ///
445 /// Passes the original DER bytes directly to
446 /// ``cryptography.x509.load_der_x509_certificate``; no re-encoding is
447 /// performed. Requires the ``cryptography`` package to be installed.
448 ///
449 /// ```python
450 /// synta_cert = synta.Certificate.from_der(der)
451 /// pyca_cert = synta_cert.to_pyca()
452 /// # Full cryptographic operations are now available via PyCA:
453 /// pyca_cert.public_key().verify(signature, message, ...)
454 /// ```
455 fn to_pyca<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
456 let m = py.import("cryptography.x509").map_err(|_| {
457 pyo3::exceptions::PyImportError::new_err(
458 "the 'cryptography' package is required; install it with: pip install cryptography",
459 )
460 })?;
461 m.call_method1(
462 "load_der_x509_certificate",
463 (self._data.clone_ref(py).into_bound(py),),
464 )
465 }
466
467 /// Construct a ``Certificate`` from a ``cryptography.x509.Certificate``.
468 ///
469 /// Serialises the PyCA certificate to DER via
470 /// ``cert.public_bytes(Encoding.DER)`` and parses the result with
471 /// :meth:`from_der`. Requires the ``cryptography`` package to be
472 /// installed.
473 ///
474 /// **Fast path** — if the object exposes a ``_synta_der_bytes`` attribute
475 /// whose value is a ``bytes`` object, that buffer is used directly and
476 /// the ``public_bytes()`` call is skipped entirely. Wrapper classes that
477 /// already hold the raw DER (e.g. an ``IPACertificate`` loaded from LDAP)
478 /// can opt in by setting the attribute at construction time:
479 ///
480 /// ```python
481 /// class IPACertificate:
482 /// def __init__(self, der: bytes):
483 /// self._synta_der_bytes = der # enables fast path
484 /// self._pyca = x509.load_der_x509_certificate(der)
485 ///
486 /// # Zero re-encoding cost — DER is read from _synta_der_bytes directly:
487 /// synta_cert = synta.Certificate.from_pyca(ipa_cert)
488 /// ```
489 ///
490 /// Without the attribute the standard path is used:
491 ///
492 /// ```python
493 /// pyca_cert = cryptography.x509.load_pem_x509_certificate(pem)
494 /// synta_cert = synta.Certificate.from_pyca(pyca_cert)
495 /// ```
496 #[staticmethod]
497 fn from_pyca(py: Python<'_>, pyca_cert: Bound<'_, PyAny>) -> PyResult<Self> {
498 // Optional fast path: if the caller has cached raw DER bytes on the
499 // object under the `_synta_der_bytes` attribute, use that directly to
500 // avoid the re-encoding cost of `public_bytes(Encoding.DER)`. This
501 // lets wrappers like FreeIPA's `IPACertificate` opt in by setting
502 // `self._synta_der_bytes = der` at construction time.
503 if let Ok(attr) = pyca_cert.getattr("_synta_der_bytes") {
504 if let Ok(der_bytes) = attr.cast_into::<PyBytes>() {
505 return Self::from_der(py, der_bytes);
506 }
507 }
508
509 let ser = py
510 .import("cryptography.hazmat.primitives.serialization")
511 .map_err(|_| {
512 pyo3::exceptions::PyImportError::new_err(
513 "the 'cryptography' package is required; install it with: pip install cryptography",
514 )
515 })?;
516 let der_bytes: Bound<'_, PyBytes> = pyca_cert
517 .call_method1("public_bytes", (ser.getattr("Encoding")?.getattr("DER")?,))?
518 .cast_into()?;
519 Self::from_der(py, der_bytes)
520 }
521
522 /// Serialize one certificate or a list of certificates to PEM format.
523 ///
524 /// The mirror of :meth:`from_pem`: accepts either a single
525 /// :class:`Certificate` or a :class:`list` of them, and returns a
526 /// :class:`bytes` value containing one or more
527 /// ``-----BEGIN CERTIFICATE-----`` blocks concatenated in order.
528 ///
529 /// ```python
530 /// # Single certificate:
531 /// pem = Certificate.to_pem(cert)
532 /// open("cert.pem", "wb").write(pem)
533 ///
534 /// # Certificate chain:
535 /// pem = Certificate.to_pem([leaf, intermediate, root])
536 /// open("chain.pem", "wb").write(pem)
537 /// ```
538 #[staticmethod]
539 fn to_pem<'py>(
540 py: Python<'py>,
541 obj_or_list: Bound<'_, PyAny>,
542 ) -> PyResult<Bound<'py, PyBytes>> {
543 pyobject_to_pem::<Self, _>(py, "CERTIFICATE", &obj_or_list, |c| c.raw)
544 }
545
546 /// Serial number as a Python int.
547 ///
548 /// Returns a native Python `int` regardless of size (X.509 serials can be
549 /// up to 20 bytes / 160 bits per RFC 5280). The Python object is created
550 /// once and cached; subsequent accesses return a reference to the same object.
551 #[getter]
552 fn serial_number<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
553 // `get_or_try_init` is not stable on `OnceLock`, so build the value
554 // outside the lock if the cache is empty, then store it.
555 if let Some(cached) = self.serial_number_cache.get() {
556 return Ok(cached.clone_ref(py).into_bound(py));
557 }
558 let serial = &self.cert()?.tbs_certificate.serial_number;
559 let py_int: Py<PyAny> = if let Ok(v) = serial.as_u64() {
560 // Fast path: fits in u64 (covers traditional short serials).
561 // Use unsigned to match RFC 5280 §4.1.2.2 (serials are positive).
562 v.into_pyobject(py)?.into_any().unbind()
563 } else if let Ok(v) = serial.as_u128() {
564 // Fits in u128 after stripping any DER leading 0x00 byte.
565 // Covers 128-bit RSNv3 random serials correctly encoded with a
566 // leading 0x00 (17 bytes total) and 127-bit serials (16 bytes).
567 v.into_pyobject(py)?.into_any().unbind()
568 } else if let Ok(v) = serial.as_i64() {
569 // Fallback for small negative serials (technically invalid per
570 // RFC 5280 but be lenient when decoding).
571 v.into_pyobject(py)?.into_any().unbind()
572 } else if let Ok(v) = serial.as_i128() {
573 // Fallback: 16-byte signed value.
574 v.into_pyobject(py)?.into_any().unbind()
575 } else {
576 // Large serial (17–20 bytes): call int.from_bytes() directly.
577 // Using call_method avoids the parse+compile overhead of py.eval()
578 // on every cold-path invocation. `signed` is keyword-only in
579 // Python's int.from_bytes signature, so it must go in kwargs.
580 // Use signed=False: RFC 5280 §4.1.2.2 requires positive serials,
581 // and a leading 0x00 in correctly-encoded values is already
582 // stripped before reaching this branch.
583 let bytes_obj = PyBytes::new(py, serial.as_bytes());
584 let kwargs = pyo3::types::PyDict::new(py);
585 kwargs.set_item(pyo3::intern!(py, "signed"), false)?;
586 py.get_type::<pyo3::types::PyInt>()
587 .call_method(
588 pyo3::intern!(py, "from_bytes"),
589 (bytes_obj, pyo3::intern!(py, "big")),
590 Some(&kwargs),
591 )
592 .map(|r| r.unbind())?
593 };
594 // Ignore a racing writer; both produce the same value.
595 let _ = self.serial_number_cache.set(py_int.clone_ref(py));
596 Ok(py_int.into_bound(py))
597 }
598
599 /// Issuer distinguished name as a string.
600 #[getter]
601 fn issuer<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
602 if let Some(cached) = self.issuer_cache.get() {
603 return Ok(cached.clone_ref(py).into_bound(py));
604 }
605 let s = decode_name_debug(self.cert()?.tbs_certificate.issuer.as_bytes());
606 let py_str = PyString::new(py, &s).unbind();
607 let _ = self.issuer_cache.set(py_str.clone_ref(py));
608 Ok(py_str.into_bound(py))
609 }
610
611 /// Subject distinguished name as a string.
612 #[getter]
613 fn subject<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
614 if let Some(cached) = self.subject_cache.get() {
615 return Ok(cached.clone_ref(py).into_bound(py));
616 }
617 let s = decode_name_debug(self.cert()?.tbs_certificate.subject.as_bytes());
618 let py_str = PyString::new(py, &s).unbind();
619 let _ = self.subject_cache.set(py_str.clone_ref(py));
620 Ok(py_str.into_bound(py))
621 }
622
623 /// Signature algorithm name (e.g. "RSA", "ECDSA", "Ed25519", "ML-DSA-65"),
624 /// or the dotted OID notation for unrecognized algorithms.
625 #[getter]
626 fn signature_algorithm<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
627 if let Some(cached) = self.signature_algorithm_cache.get() {
628 return Ok(cached.clone_ref(py).into_bound(py));
629 }
630 let oid = &self.cert()?.signature_algorithm.algorithm;
631 let name = synta_certificate::identify_signature_algorithm(oid);
632 let s = if name != "Other" {
633 name.to_string()
634 } else {
635 oid.to_string()
636 };
637 let py_str = PyString::new(py, &s).unbind();
638 let _ = self.signature_algorithm_cache.set(py_str.clone_ref(py));
639 Ok(py_str.into_bound(py))
640 }
641
642 /// Raw signature bytes.
643 ///
644 /// The `bytes` object is created once on first access and the same Python
645 /// object is returned (by reference) on every subsequent call,
646 /// avoiding repeated allocation and PyO3 string construction.
647 #[getter]
648 fn signature_value<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
649 if let Some(cached) = self.signature_value_cache.get() {
650 return Ok(cached.clone_ref(py).into_bound(py));
651 }
652 let py_bytes = PyBytes::new(py, self.cert()?.signature_value.as_bytes()).unbind();
653 let _ = self.signature_value_cache.set(py_bytes.clone_ref(py));
654 Ok(py_bytes.into_bound(py))
655 }
656
657 /// notBefore time as a string.
658 #[getter]
659 fn not_before<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
660 if let Some(cached) = self.not_before_cache.get() {
661 return Ok(cached.clone_ref(py).into_bound(py));
662 }
663 let s = match &self.cert()?.tbs_certificate.validity.not_before {
664 Time::UtcTime(t) => t.to_string(),
665 Time::GeneralTime(t) => t.to_string(),
666 };
667 let py_str = PyString::new(py, &s).unbind();
668 let _ = self.not_before_cache.set(py_str.clone_ref(py));
669 Ok(py_str.into_bound(py))
670 }
671
672 /// notAfter time as a string.
673 #[getter]
674 fn not_after<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
675 if let Some(cached) = self.not_after_cache.get() {
676 return Ok(cached.clone_ref(py).into_bound(py));
677 }
678 let s = match &self.cert()?.tbs_certificate.validity.not_after {
679 Time::UtcTime(t) => t.to_string(),
680 Time::GeneralTime(t) => t.to_string(),
681 };
682 let py_str = PyString::new(py, &s).unbind();
683 let _ = self.not_after_cache.set(py_str.clone_ref(py));
684 Ok(py_str.into_bound(py))
685 }
686
687 /// Subject public-key algorithm name (e.g. "RSA", "ECDSA", "Ed25519", "ML-DSA-65"),
688 /// or the dotted OID notation for unrecognized algorithms.
689 #[getter]
690 fn public_key_algorithm<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
691 if let Some(cached) = self.public_key_algorithm_cache.get() {
692 return Ok(cached.clone_ref(py).into_bound(py));
693 }
694 let oid = &self
695 .cert()?
696 .tbs_certificate
697 .subject_public_key_info
698 .algorithm
699 .algorithm;
700 let s = synta_certificate::identify_public_key_algorithm(oid)
701 .map(|s| s.to_string())
702 .unwrap_or_else(|| oid.to_string());
703 let py_str = PyString::new(py, &s).unbind();
704 let _ = self.public_key_algorithm_cache.set(py_str.clone_ref(py));
705 Ok(py_str.into_bound(py))
706 }
707
708 /// Raw subject public-key bytes.
709 ///
710 /// The `bytes` object is created once on first access and the same Python
711 /// object is returned (by reference) on every subsequent call,
712 /// avoiding repeated allocation and PyO3 string construction.
713 #[getter]
714 fn public_key<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
715 if let Some(cached) = self.public_key_cache.get() {
716 return Ok(cached.clone_ref(py).into_bound(py));
717 }
718 let py_bytes = PyBytes::new(
719 py,
720 self.cert()?
721 .tbs_certificate
722 .subject_public_key_info
723 .subject_public_key
724 .as_bytes(),
725 )
726 .unbind();
727 let _ = self.public_key_cache.set(py_bytes.clone_ref(py));
728 Ok(py_bytes.into_bound(py))
729 }
730
731 /// Version field (0 = v1, 1 = v2, 2 = v3), or None if absent.
732 #[getter]
733 fn version(&self) -> PyResult<Option<i64>> {
734 Ok(self
735 .cert()?
736 .tbs_certificate
737 .version
738 .as_ref()
739 .and_then(|v| v.as_i64().ok()))
740 }
741
742 /// Complete DER encoding of this certificate (the original bytes passed to
743 /// ``from_der``).
744 fn to_der<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
745 self._data.clone_ref(py).into_bound(py)
746 }
747
748 /// Raw DER bytes of the TBSCertificate structure (the bytes that were
749 /// signed by the issuer's key).
750 #[getter]
751 fn tbs_bytes<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
752 self.tbs_bytes_cache
753 .get_or_init(|| {
754 let data = self._data.bind(py).as_bytes();
755 PyBytes::new(py, &data[self.tbs_range.clone()]).unbind()
756 })
757 .clone_ref(py)
758 .into_bound(py)
759 }
760
761 /// OID of the signature algorithm
762 /// (e.g. ``ObjectIdentifier("1.2.840.113549.1.1.11")`` for sha256WithRSAEncryption).
763 ///
764 /// Unlike ``signature_algorithm``, this always returns the machine-readable
765 /// OID, even for algorithms that synta does not recognise by name.
766 #[getter]
767 fn signature_algorithm_oid<'py>(
768 &self,
769 py: Python<'py>,
770 ) -> PyResult<Bound<'py, PyObjectIdentifier>> {
771 if let Some(cached) = self.signature_algorithm_oid_cache.get() {
772 return Ok(cached.clone_ref(py).into_bound(py));
773 }
774 let obj = Py::new(
775 py,
776 PyObjectIdentifier::from_oid(self.cert()?.signature_algorithm.algorithm.clone()),
777 )?;
778 let _ = self.signature_algorithm_oid_cache.set(obj.clone_ref(py));
779 Ok(obj.into_bound(py))
780 }
781
782 /// Raw DER bytes of the signature algorithm parameters, or ``None`` if
783 /// the AlgorithmIdentifier has no parameters field (e.g. Ed25519).
784 #[getter]
785 fn signature_algorithm_params<'py>(
786 &self,
787 py: Python<'py>,
788 ) -> PyResult<Option<Bound<'py, PyBytes>>> {
789 if let Some(cached) = self.signature_algorithm_params_cache.get() {
790 return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
791 }
792 let computed =
793 encode_element_opt(py, self.cert()?.signature_algorithm.parameters.as_ref())?;
794 let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
795 let _ = self.signature_algorithm_params_cache.set(to_store);
796 Ok(computed)
797 }
798
799 /// DER-encoded ``AlgorithmIdentifier`` SEQUENCE from the certificate's outer
800 /// ``signatureAlgorithm`` field.
801 ///
802 /// This is the raw DER form of the field — suitable for passing directly to
803 /// :meth:`PublicKey.verify_certificate_signature` without any re-encoding.
804 ///
805 /// ```python,ignore
806 /// pub = synta.PublicKey.from_der(issuer_cert.public_key)
807 /// pub.verify_certificate_signature(
808 /// cert.tbs_certificate_der,
809 /// cert.signature_algorithm_der,
810 /// cert.signature_value,
811 /// )
812 /// ```
813 #[getter]
814 fn signature_algorithm_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
815 if let Some(cached) = self.signature_algorithm_der_cache.get() {
816 return Ok(cached.clone_ref(py).into_bound(py));
817 }
818 let ranges = synta_certificate::cert_byte_ranges(self.raw).ok_or_else(|| {
819 pyo3::exceptions::PyValueError::new_err("certificate DER structure is invalid")
820 })?;
821 let py_bytes = PyBytes::new(py, &self.raw[ranges.signature_algorithm]).unbind();
822 let _ = self
823 .signature_algorithm_der_cache
824 .set(py_bytes.clone_ref(py));
825 Ok(py_bytes.into_bound(py))
826 }
827
828 /// OID of the subject public-key algorithm
829 /// (e.g. ``ObjectIdentifier("1.2.840.10045.2.1")`` for id-ecPublicKey).
830 #[getter]
831 fn public_key_algorithm_oid<'py>(
832 &self,
833 py: Python<'py>,
834 ) -> PyResult<Bound<'py, PyObjectIdentifier>> {
835 if let Some(cached) = self.public_key_algorithm_oid_cache.get() {
836 return Ok(cached.clone_ref(py).into_bound(py));
837 }
838 let obj = Py::new(
839 py,
840 PyObjectIdentifier::from_oid(
841 self.cert()?
842 .tbs_certificate
843 .subject_public_key_info
844 .algorithm
845 .algorithm
846 .clone(),
847 ),
848 )?;
849 let _ = self.public_key_algorithm_oid_cache.set(obj.clone_ref(py));
850 Ok(obj.into_bound(py))
851 }
852
853 /// Raw DER bytes of the public-key algorithm parameters, or ``None``.
854 ///
855 /// For EC keys this is an OID naming the curve (e.g. secp256r1).
856 /// For RSA and Ed25519 this is typically ``None`` or a NULL element.
857 #[getter]
858 fn public_key_algorithm_params<'py>(
859 &self,
860 py: Python<'py>,
861 ) -> PyResult<Option<Bound<'py, PyBytes>>> {
862 if let Some(cached) = self.public_key_algorithm_params_cache.get() {
863 return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
864 }
865 let computed = encode_element_opt(
866 py,
867 self.cert()?
868 .tbs_certificate
869 .subject_public_key_info
870 .algorithm
871 .parameters
872 .as_ref(),
873 )?;
874 let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
875 let _ = self.public_key_algorithm_params_cache.set(to_store);
876 Ok(computed)
877 }
878
879 /// Raw DER bytes of the extensions SEQUENCE OF, or ``None`` for v1/v2
880 /// certificates that carry no extensions.
881 ///
882 /// The returned bytes begin with the SEQUENCE tag (``0x30``) and contain
883 /// the full SEQUENCE OF Extension encoding. Pass them to a ``Decoder``
884 /// to iterate the individual extensions:
885 ///
886 /// ```python
887 /// ext_der = cert.extensions_der
888 /// if ext_der:
889 /// dec = synta.Decoder(ext_der, synta.Encoding.DER)
890 /// exts_dec = dec.decode_sequence()
891 /// while not exts_dec.is_empty():
892 /// ext_tlv = exts_dec.decode_raw_tlv()
893 /// ```
894 #[getter]
895 fn extensions_der<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
896 if let Some(cached) = self.extensions_der_cache.get() {
897 return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
898 }
899 let computed = self
900 .cert()?
901 .tbs_certificate
902 .extensions
903 .as_ref()
904 .map(|raw: &synta::RawDer<'_>| PyBytes::new(py, raw.as_bytes()));
905 let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
906 let _ = self.extensions_der_cache.set(to_store);
907 Ok(computed)
908 }
909
910 /// Return the DER content of a named extension's value, or ``None``.
911 ///
912 /// Searches the certificate's extension list for an extension whose
913 /// ``extnID`` equals *oid* (dotted-decimal notation, e.g.
914 /// ``"2.5.29.17"`` for SubjectAltName). When found, the bytes inside
915 /// the ``extnValue`` OCTET STRING — the DER-encoded extension-specific
916 /// structure — are returned. Returns ``None`` when the certificate
917 /// carries no extensions or the named OID is absent.
918 ///
919 /// Example — parse SubjectAltName:
920 ///
921 /// ```python
922 /// san_der = cert.get_extension_value_der("2.5.29.17")
923 /// if san_der:
924 /// dec = synta.Decoder(san_der, synta.Encoding.DER)
925 /// san_seq = dec.decode_sequence()
926 /// while not san_seq.is_empty():
927 /// tag_num, tag_class, _ = san_seq.peek_tag()
928 /// child = san_seq.decode_implicit_tag(tag_num, tag_class)
929 /// if tag_num == 2: # dNSName
930 /// print(child.remaining_bytes().decode("ascii"))
931 /// ```
932 fn get_extension_value_der<'py>(
933 &self,
934 py: Python<'py>,
935 oid: &Bound<'_, PyAny>,
936 ) -> PyResult<Option<Bound<'py, PyBytes>>> {
937 let target = oid_from_pyany(oid)?;
938 let raw = match self.cert()?.tbs_certificate.extensions.as_ref() {
939 Some(r) => r,
940 None => return Ok(None),
941 };
942 for ext in &synta_certificate::decode_extensions(raw.as_bytes()) {
943 if ext.extn_id == target {
944 return Ok(Some(PyBytes::new(py, ext.extn_value.as_bytes())));
945 }
946 }
947 Ok(None)
948 }
949
950 /// Return the Subject Alternative Names of this certificate as typed objects.
951 ///
952 /// Combines looking up the SAN extension (OID ``2.5.29.17``) and parsing
953 /// its ``GeneralName`` entries into a single call. Returns a
954 /// :class:`list` of typed :mod:`synta.general_name` objects in document
955 /// order; returns an empty list when the certificate carries no SAN
956 /// extension.
957 ///
958 /// Each element is one of:
959 ///
960 /// - :class:`~synta.general_name.OtherName` — tag 0
961 /// - :class:`~synta.general_name.RFC822Name` — tag 1 (e-mail address)
962 /// - :class:`~synta.general_name.DNSName` — tag 2
963 /// - :class:`~synta.general_name.X400Address` — tag 3 (raw DER)
964 /// - :class:`~synta.general_name.DirectoryName` — tag 4
965 /// - :class:`~synta.general_name.EDIPartyName` — tag 5 (raw DER)
966 /// - :class:`~synta.general_name.UniformResourceIdentifier` — tag 6
967 /// - :class:`~synta.general_name.IPAddress` — tag 7
968 /// - :class:`~synta.general_name.RegisteredID` — tag 8
969 ///
970 /// ```python,ignore
971 /// import ipaddress
972 /// import synta.general_name as gn
973 ///
974 /// for name in cert.subject_alt_names():
975 /// if isinstance(name, gn.DNSName):
976 /// print("DNS:", name.value)
977 /// elif isinstance(name, gn.IPAddress):
978 /// print("IP:", ipaddress.ip_address(name.address))
979 /// elif isinstance(name, gn.RFC822Name):
980 /// print("email:", name.value)
981 /// elif isinstance(name, gn.DirectoryName):
982 /// attrs = synta.parse_name_attrs(name.name_der)
983 /// print("DirName:", attrs)
984 /// elif isinstance(name, gn.UniformResourceIdentifier):
985 /// print("URI:", name.value)
986 /// ```
987 fn subject_alt_names<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
988 use super::general_name::decode_general_names_to_py;
989 use synta_certificate::find_extension_value;
990
991 let cert = self.cert()?;
992 let raw = match cert.tbs_certificate.extensions.as_ref() {
993 Some(r) => r,
994 None => return Ok(PyList::empty(py)),
995 };
996 let san_bytes =
997 match find_extension_value(raw.as_bytes(), synta_certificate::oids::SUBJECT_ALT_NAME) {
998 Some(b) => b,
999 None => return Ok(PyList::empty(py)),
1000 };
1001 decode_general_names_to_py(py, san_bytes)
1002 }
1003
1004 /// Return the GeneralNames from any extension that encodes a
1005 /// ``SEQUENCE OF GeneralName``, identified by OID.
1006 ///
1007 /// Looks up the extension with the given *oid* and decodes its value as a
1008 /// ``SEQUENCE OF GeneralName``. Suitable for SAN (``2.5.29.17``), IAN
1009 /// (``2.5.29.18``), or any private extension that uses the same format.
1010 ///
1011 /// Returns a :class:`list` of typed :mod:`synta.general_name` objects, or
1012 /// an empty list when the extension is absent or cannot be decoded.
1013 ///
1014 /// ```python,ignore
1015 /// import synta
1016 /// import synta.general_name as gn
1017 ///
1018 /// cert = synta.Certificate.from_der(der)
1019 /// # Decode the Issuer Alternative Names extension:
1020 /// for name in cert.general_names("2.5.29.18"):
1021 /// if isinstance(name, gn.DirectoryName):
1022 /// print(synta.parse_name_attrs(name.name_der))
1023 /// ```
1024 fn general_names<'py>(
1025 &self,
1026 py: Python<'py>,
1027 oid: &Bound<'_, PyAny>,
1028 ) -> PyResult<Bound<'py, PyList>> {
1029 use super::general_name::decode_general_names_to_py;
1030 use super::oid_from_pyany;
1031 use synta_certificate::find_extension_value;
1032
1033 let target = oid_from_pyany(oid)?;
1034 let cert = self.cert()?;
1035 let raw = match cert.tbs_certificate.extensions.as_ref() {
1036 Some(r) => r,
1037 None => return Ok(PyList::empty(py)),
1038 };
1039 let ext_bytes = match find_extension_value(raw.as_bytes(), target.components()) {
1040 Some(b) => b,
1041 None => return Ok(PyList::empty(py)),
1042 };
1043 decode_general_names_to_py(py, ext_bytes)
1044 }
1045
1046 /// Return the CertificatePolicies entries from the extension (OID ``2.5.29.32``).
1047 ///
1048 /// Decodes the ``CertificatePolicies`` extension value
1049 /// (``SEQUENCE OF PolicyInformation``) and returns a list of
1050 /// :class:`PolicyInformation` objects, each containing the policy OID and
1051 /// any ``policyQualifiers``.
1052 ///
1053 /// Returns an empty list when the extension is absent or cannot be parsed.
1054 ///
1055 /// ```python,ignore
1056 /// import synta
1057 /// cert = synta.Certificate.from_der(der)
1058 /// for pi in cert.certificate_policies():
1059 /// print(str(pi.policy_oid)) # e.g. "1.3.6.1.4.1.311.21.8...."
1060 /// for q in pi.qualifiers:
1061 /// print(str(q.qualifier_oid))
1062 /// ```
1063 fn certificate_policies<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
1064 let list = PyList::empty(py);
1065 let cert = self.cert()?;
1066 let ext_raw = match cert.tbs_certificate.extensions.as_ref() {
1067 Some(r) => r,
1068 None => return Ok(list),
1069 };
1070 let ext_value_bytes = synta_certificate::find_extension_value(
1071 ext_raw.as_bytes(),
1072 synta_certificate::oids::CERTIFICATE_POLICIES,
1073 );
1074 let ext_bytes = match ext_value_bytes {
1075 Some(b) => b,
1076 None => return Ok(list),
1077 };
1078 // Decode using the code-generated PolicyInformation<'a> type.
1079 let mut decoder = synta::Decoder::new(ext_bytes, synta::Encoding::Der);
1080 let infos: Vec<synta_certificate::PolicyInformation<'_>> = match decoder.decode() {
1081 Ok(v) => v,
1082 Err(_) => return Ok(list),
1083 };
1084 for pi in infos {
1085 let qualifiers = pi
1086 .policy_qualifiers
1087 .unwrap_or_default()
1088 .into_iter()
1089 .filter_map(|qi: PolicyQualifierInfo<'_>| {
1090 let mut enc = synta::Encoder::new(synta::Encoding::Der);
1091 qi.qualifier.encode(&mut enc).ok()?;
1092 let qval = enc.finish().ok()?;
1093 Some((qi.policy_qualifier_id, qval))
1094 })
1095 .collect();
1096 let py_pi = Py::new(
1097 py,
1098 PyPolicyInformation {
1099 policy_oid: pi.policy_identifier,
1100 qualifiers,
1101 },
1102 )?;
1103 list.append(py_pi.into_bound(py))?;
1104 }
1105 Ok(list)
1106 }
1107
1108 /// Raw DER bytes of the issuer Name SEQUENCE.
1109 #[getter]
1110 fn issuer_raw_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
1111 if let Some(cached) = self.issuer_raw_der_cache.get() {
1112 return Ok(cached.clone_ref(py).into_bound(py));
1113 }
1114 let py_bytes = PyBytes::new(py, self.cert()?.tbs_certificate.issuer.as_bytes()).unbind();
1115 let _ = self.issuer_raw_der_cache.set(py_bytes.clone_ref(py));
1116 Ok(py_bytes.into_bound(py))
1117 }
1118
1119 /// Raw DER bytes of the subject Name SEQUENCE.
1120 #[getter]
1121 fn subject_raw_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
1122 if let Some(cached) = self.subject_raw_der_cache.get() {
1123 return Ok(cached.clone_ref(py).into_bound(py));
1124 }
1125 let py_bytes = PyBytes::new(py, self.cert()?.tbs_certificate.subject.as_bytes()).unbind();
1126 let _ = self.subject_raw_der_cache.set(py_bytes.clone_ref(py));
1127 Ok(py_bytes.into_bound(py))
1128 }
1129
1130 /// notBefore time as a UTC-aware ``datetime.datetime``.
1131 ///
1132 /// Equivalent to ``cert.not_valid_before.replace(tzinfo=datetime.timezone.utc)``
1133 /// from the ``cryptography`` library. Derives the value from the parsed
1134 /// ASN.1 ``Time`` field (``UTCTime`` or ``GeneralizedTime``) without calling
1135 /// any Python date-parsing code.
1136 ///
1137 /// ```python,ignore
1138 /// import datetime
1139 /// cert = synta.Certificate.from_der(der)
1140 /// dt = cert.not_before_utc
1141 /// print(dt.isoformat()) # e.g. "2023-01-01T00:00:00+00:00"
1142 /// print(dt.tzinfo) # datetime.timezone.utc
1143 /// ```
1144 #[getter]
1145 fn not_before_utc<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
1146 if let Some(cached) = self.not_before_utc_cache.get() {
1147 return Ok(cached.clone_ref(py).into_bound(py));
1148 }
1149 let unix_secs = synta_x509_verification::certificate::time_to_unix(
1150 &self.cert()?.tbs_certificate.validity.not_before,
1151 )
1152 .ok_or_else(|| {
1153 pyo3::exceptions::PyValueError::new_err("notBefore date is structurally invalid")
1154 })?;
1155 let datetime_mod = py.import("datetime")?;
1156 let utc = datetime_mod.getattr("timezone")?.getattr("utc")?;
1157 let dt = datetime_mod
1158 .getattr("datetime")?
1159 .call_method1("fromtimestamp", (unix_secs, &utc))?
1160 .unbind();
1161 let _ = self.not_before_utc_cache.set(dt.clone_ref(py));
1162 Ok(dt.into_bound(py))
1163 }
1164
1165 /// notAfter time as a UTC-aware ``datetime.datetime``.
1166 ///
1167 /// Equivalent to ``cert.not_valid_after.replace(tzinfo=datetime.timezone.utc)``
1168 /// from the ``cryptography`` library. Derives the value from the parsed
1169 /// ASN.1 ``Time`` field (``UTCTime`` or ``GeneralizedTime``) without calling
1170 /// any Python date-parsing code.
1171 ///
1172 /// ```python,ignore
1173 /// import datetime
1174 /// cert = synta.Certificate.from_der(der)
1175 /// dt = cert.not_after_utc
1176 /// print(dt.isoformat()) # e.g. "2033-01-01T00:00:00+00:00"
1177 /// print(dt.tzinfo) # datetime.timezone.utc
1178 /// ```
1179 #[getter]
1180 fn not_after_utc<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
1181 if let Some(cached) = self.not_after_utc_cache.get() {
1182 return Ok(cached.clone_ref(py).into_bound(py));
1183 }
1184 let unix_secs = synta_x509_verification::certificate::time_to_unix(
1185 &self.cert()?.tbs_certificate.validity.not_after,
1186 )
1187 .ok_or_else(|| {
1188 pyo3::exceptions::PyValueError::new_err("notAfter date is structurally invalid")
1189 })?;
1190 let datetime_mod = py.import("datetime")?;
1191 let utc = datetime_mod.getattr("timezone")?.getattr("utc")?;
1192 let dt = datetime_mod
1193 .getattr("datetime")?
1194 .call_method1("fromtimestamp", (unix_secs, &utc))?
1195 .unbind();
1196 let _ = self.not_after_utc_cache.set(dt.clone_ref(py));
1197 Ok(dt.into_bound(py))
1198 }
1199
1200 /// Hash algorithm name component of the certificate's outer signature algorithm.
1201 ///
1202 /// Returns the lowercase hash algorithm name (e.g. ``"sha256"``, ``"sha1"``,
1203 /// ``"sha384"``) implied by the signature algorithm OID. Returns ``None``
1204 /// for algorithms that do not use a traditional hash (Ed25519, Ed448,
1205 /// ML-DSA, etc.).
1206 ///
1207 /// For RSASSA-PSS the hash is encoded in the parameters field, which is not
1208 /// inspected here; ``"sha256"`` is returned as a conservative default.
1209 ///
1210 /// ```python,ignore
1211 /// cert = synta.Certificate.from_der(der)
1212 /// hash_name = cert.signature_hash_algorithm_name
1213 /// if hash_name:
1214 /// print(hash_name) # e.g. "sha256"
1215 /// else:
1216 /// print("no hash (e.g. Ed25519)")
1217 /// ```
1218 #[getter]
1219 fn signature_hash_algorithm_name(&self) -> PyResult<Option<&'static str>> {
1220 use synta_certificate::oids;
1221 let oid = &self.cert()?.signature_algorithm.algorithm;
1222 let c = oid.components();
1223 // RSA PKCS#1 v1.5 variants (1.2.840.113549.1.1.*)
1224 if c == oids::SHA1_WITH_RSA {
1225 return Ok(Some("sha1"));
1226 }
1227 if c == oids::SHA256_WITH_RSA {
1228 return Ok(Some("sha256"));
1229 }
1230 if c == oids::SHA384_WITH_RSA {
1231 return Ok(Some("sha384"));
1232 }
1233 if c == oids::SHA512_WITH_RSA {
1234 return Ok(Some("sha512"));
1235 }
1236 // RSA-PSS (hash is in params; return sha256 as safe default)
1237 if c == oids::RSASSA_PSS {
1238 return Ok(Some("sha256"));
1239 }
1240 // ECDSA variants
1241 if c == oids::ECDSA_WITH_SHA1 {
1242 return Ok(Some("sha1"));
1243 }
1244 if c == oids::ECDSA_WITH_SHA256 {
1245 return Ok(Some("sha256"));
1246 }
1247 if c == oids::ECDSA_WITH_SHA384 {
1248 return Ok(Some("sha384"));
1249 }
1250 if c == oids::ECDSA_WITH_SHA512 {
1251 return Ok(Some("sha512"));
1252 }
1253 // Ed25519, Ed448, ML-DSA-*, SLH-DSA-* and everything else: no hash
1254 Ok(None)
1255 }
1256
1257 /// Complete DER encoding of the SubjectPublicKeyInfo SEQUENCE.
1258 ///
1259 /// Returns the full ``SubjectPublicKeyInfo`` SEQUENCE TLV as ``bytes``,
1260 /// including the algorithm identifier and the BIT STRING carrying the
1261 /// actual public key. This is the DER encoding that
1262 /// ``cryptography``'s ``public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo)``
1263 /// produces.
1264 ///
1265 /// ```python,ignore
1266 /// cert = synta.Certificate.from_der(der)
1267 /// spki_der = cert.subject_public_key_info_der
1268 /// # Load the public key with cryptography:
1269 /// from cryptography.hazmat.primitives.serialization import load_der_public_key
1270 /// pub_key = load_der_public_key(spki_der)
1271 /// ```
1272 #[getter]
1273 fn subject_public_key_info_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
1274 if let Some(cached) = self.spki_der_cache.get() {
1275 return Ok(cached.clone_ref(py).into_bound(py));
1276 }
1277 let spki = &self.cert()?.tbs_certificate.subject_public_key_info;
1278 let mut enc = synta::Encoder::new(synta::Encoding::Der);
1279 spki.encode(&mut enc)
1280 .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
1281 let bytes = enc
1282 .finish()
1283 .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
1284 let py_bytes = PyBytes::new(py, &bytes).unbind();
1285 let _ = self.spki_der_cache.set(py_bytes.clone_ref(py));
1286 Ok(py_bytes.into_bound(py))
1287 }
1288
1289 /// Raw DER bytes of the ``TBSCertificate`` structure — the bytes that
1290 /// were signed by the issuer's key.
1291 ///
1292 /// This is the DER encoding of the full ``TBSCertificate`` SEQUENCE TLV,
1293 /// suitable for signature verification or as the input to a hash function.
1294 /// Equivalent to ``cryptography``'s ``cert.tbs_certificate_bytes``.
1295 ///
1296 /// ```python,ignore
1297 /// cert = synta.Certificate.from_der(der)
1298 /// tbs = cert.tbs_certificate_der
1299 /// # Verify signature manually:
1300 /// import synta.crypto
1301 /// digest = synta.digest("sha256", tbs)
1302 /// ```
1303 #[getter]
1304 fn tbs_certificate_der<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
1305 self.tbs_bytes(py)
1306 }
1307
1308 /// Verify that this certificate was directly issued by ``issuer``.
1309 ///
1310 /// Checks two things:
1311 ///
1312 /// 1. The ``issuer`` field of this certificate's TBS matches the
1313 /// ``subject`` field of the provided issuer certificate (byte-exact
1314 /// DER comparison).
1315 /// 2. The certificate's signature is valid under the issuer's public key.
1316 ///
1317 /// Raises :exc:`ValueError` if either check fails. This is the synta
1318 /// equivalent of ``cryptography``'s
1319 /// ``cert.verify_directly_issued_by(issuer)``.
1320 ///
1321 /// ```python,ignore
1322 /// root = synta.Certificate.from_pem(open("root.pem", "rb").read())
1323 /// leaf = synta.Certificate.from_pem(open("leaf.pem", "rb").read())
1324 /// leaf.verify_issued_by(root) # raises ValueError if not valid
1325 /// ```
1326 fn verify_issued_by(&self, issuer: &PyCertificate) -> PyResult<()> {
1327 use synta_certificate::{cert_byte_ranges, default_signature_verifier, SignatureVerifier};
1328
1329 // 1. Name check: our issuer Name must byte-equal their subject Name.
1330 let cert = self.cert()?;
1331 let issuer_cert = issuer.cert()?;
1332 if cert.tbs_certificate.issuer.as_bytes() != issuer_cert.tbs_certificate.subject.as_bytes()
1333 {
1334 return Err(pyo3::exceptions::PyValueError::new_err(
1335 "issuer name does not match subject of provided certificate",
1336 ));
1337 }
1338
1339 // 2. Locate byte ranges within each cert's DER without re-encoding.
1340 let my_ranges = cert_byte_ranges(self.raw).ok_or_else(|| {
1341 pyo3::exceptions::PyValueError::new_err(
1342 "failed to locate byte ranges in certificate DER",
1343 )
1344 })?;
1345 let issuer_ranges = cert_byte_ranges(issuer.raw).ok_or_else(|| {
1346 pyo3::exceptions::PyValueError::new_err(
1347 "failed to locate byte ranges in issuer certificate DER",
1348 )
1349 })?;
1350
1351 // 3. Signature bits (raw bytes, no unused-bits prefix).
1352 let signature_bits = cert.signature_value.as_bytes();
1353
1354 // 4. Verify using the backend-agnostic verifier.
1355 default_signature_verifier()
1356 .verify_certificate_signature(
1357 &self.raw[my_ranges.tbs],
1358 &self.raw[my_ranges.signature_algorithm],
1359 signature_bits,
1360 &issuer.raw[issuer_ranges.subject_public_key_info],
1361 )
1362 .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
1363
1364 Ok(())
1365 }
1366
1367 /// Compute a hash fingerprint of the complete certificate DER.
1368 ///
1369 /// ``algorithm`` must be one of ``"sha1"``, ``"sha224"``, ``"sha256"``,
1370 /// ``"sha384"``, ``"sha512"``, or ``"md5"``. Raises :exc:`ValueError` for
1371 /// unknown algorithm names.
1372 ///
1373 /// ```python,ignore
1374 /// cert = synta.Certificate.from_der(der)
1375 /// fp = cert.fingerprint("sha256")
1376 /// print(fp.hex()) # e.g. "3a2b1c..."
1377 /// ```
1378 fn fingerprint<'py>(&self, py: Python<'py>, algorithm: &str) -> PyResult<Bound<'py, PyBytes>> {
1379 use synta_certificate::{default_data_hasher, DataHasher};
1380 let digest = default_data_hasher()
1381 .hash_data(algorithm, self.raw)
1382 .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
1383 Ok(PyBytes::new(py, &digest))
1384 }
1385
1386 fn __repr__(&self) -> PyResult<String> {
1387 let cert = self.cert()?;
1388 let serial = &cert.tbs_certificate.serial_number;
1389 let serial_str = if let Ok(v) = serial.as_i64() {
1390 v.to_string()
1391 } else {
1392 format!("<{} bytes>", serial.as_bytes().len())
1393 };
1394 Ok(format!(
1395 "Certificate(subject={:?}, serial={})",
1396 decode_name_debug(cert.tbs_certificate.subject.as_bytes()),
1397 serial_str,
1398 ))
1399 }
1400}
1401
1402impl PyCertificate {
1403 /// Parse a DER-encoded X.509 certificate (internal helper visible to sibling modules).
1404 ///
1405 /// This is the same logic as the `#[staticmethod] from_der` exposed to Python but
1406 /// accessible from Rust sibling modules via `pub(super)` visibility.
1407 pub(crate) fn new_from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
1408 Self::from_der(py, data)
1409 }
1410}