_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_i64() {
560 // Fast path: fits in i64 (covers traditional short serials)
561 v.into_pyobject(py)?.into_any().unbind()
562 } else if let Ok(v) = serial.as_i128() {
563 // Fits in i128 (covers up to 16-byte random serials)
564 v.into_pyobject(py)?.into_any().unbind()
565 } else {
566 // Large serial (17–20 bytes): call int.from_bytes() directly.
567 // Using call_method avoids the parse+compile overhead of py.eval()
568 // on every cold-path invocation. `signed` is keyword-only in
569 // Python's int.from_bytes signature, so it must go in kwargs.
570 let bytes_obj = PyBytes::new(py, serial.as_bytes());
571 let kwargs = pyo3::types::PyDict::new(py);
572 kwargs.set_item(pyo3::intern!(py, "signed"), true)?;
573 py.get_type::<pyo3::types::PyInt>()
574 .call_method(
575 pyo3::intern!(py, "from_bytes"),
576 (bytes_obj, pyo3::intern!(py, "big")),
577 Some(&kwargs),
578 )
579 .map(|r| r.unbind())?
580 };
581 // Ignore a racing writer; both produce the same value.
582 let _ = self.serial_number_cache.set(py_int.clone_ref(py));
583 Ok(py_int.into_bound(py))
584 }
585
586 /// Issuer distinguished name as a string.
587 #[getter]
588 fn issuer<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
589 if let Some(cached) = self.issuer_cache.get() {
590 return Ok(cached.clone_ref(py).into_bound(py));
591 }
592 let s = decode_name_debug(self.cert()?.tbs_certificate.issuer.as_bytes());
593 let py_str = PyString::new(py, &s).unbind();
594 let _ = self.issuer_cache.set(py_str.clone_ref(py));
595 Ok(py_str.into_bound(py))
596 }
597
598 /// Subject distinguished name as a string.
599 #[getter]
600 fn subject<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
601 if let Some(cached) = self.subject_cache.get() {
602 return Ok(cached.clone_ref(py).into_bound(py));
603 }
604 let s = decode_name_debug(self.cert()?.tbs_certificate.subject.as_bytes());
605 let py_str = PyString::new(py, &s).unbind();
606 let _ = self.subject_cache.set(py_str.clone_ref(py));
607 Ok(py_str.into_bound(py))
608 }
609
610 /// Signature algorithm name (e.g. "RSA", "ECDSA", "Ed25519", "ML-DSA-65"),
611 /// or the dotted OID notation for unrecognized algorithms.
612 #[getter]
613 fn signature_algorithm<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
614 if let Some(cached) = self.signature_algorithm_cache.get() {
615 return Ok(cached.clone_ref(py).into_bound(py));
616 }
617 let oid = &self.cert()?.signature_algorithm.algorithm;
618 let name = synta_certificate::identify_signature_algorithm(oid);
619 let s = if name != "Other" {
620 name.to_string()
621 } else {
622 oid.to_string()
623 };
624 let py_str = PyString::new(py, &s).unbind();
625 let _ = self.signature_algorithm_cache.set(py_str.clone_ref(py));
626 Ok(py_str.into_bound(py))
627 }
628
629 /// Raw signature bytes.
630 ///
631 /// The `bytes` object is created once on first access and the same Python
632 /// object is returned (by reference) on every subsequent call,
633 /// avoiding repeated allocation and PyO3 string construction.
634 #[getter]
635 fn signature_value<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
636 if let Some(cached) = self.signature_value_cache.get() {
637 return Ok(cached.clone_ref(py).into_bound(py));
638 }
639 let py_bytes = PyBytes::new(py, self.cert()?.signature_value.as_bytes()).unbind();
640 let _ = self.signature_value_cache.set(py_bytes.clone_ref(py));
641 Ok(py_bytes.into_bound(py))
642 }
643
644 /// notBefore time as a string.
645 #[getter]
646 fn not_before<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
647 if let Some(cached) = self.not_before_cache.get() {
648 return Ok(cached.clone_ref(py).into_bound(py));
649 }
650 let s = match &self.cert()?.tbs_certificate.validity.not_before {
651 Time::UtcTime(t) => t.to_string(),
652 Time::GeneralTime(t) => t.to_string(),
653 };
654 let py_str = PyString::new(py, &s).unbind();
655 let _ = self.not_before_cache.set(py_str.clone_ref(py));
656 Ok(py_str.into_bound(py))
657 }
658
659 /// notAfter time as a string.
660 #[getter]
661 fn not_after<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
662 if let Some(cached) = self.not_after_cache.get() {
663 return Ok(cached.clone_ref(py).into_bound(py));
664 }
665 let s = match &self.cert()?.tbs_certificate.validity.not_after {
666 Time::UtcTime(t) => t.to_string(),
667 Time::GeneralTime(t) => t.to_string(),
668 };
669 let py_str = PyString::new(py, &s).unbind();
670 let _ = self.not_after_cache.set(py_str.clone_ref(py));
671 Ok(py_str.into_bound(py))
672 }
673
674 /// Subject public-key algorithm name (e.g. "RSA", "ECDSA", "Ed25519", "ML-DSA-65"),
675 /// or the dotted OID notation for unrecognized algorithms.
676 #[getter]
677 fn public_key_algorithm<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
678 if let Some(cached) = self.public_key_algorithm_cache.get() {
679 return Ok(cached.clone_ref(py).into_bound(py));
680 }
681 let oid = &self
682 .cert()?
683 .tbs_certificate
684 .subject_public_key_info
685 .algorithm
686 .algorithm;
687 let s = synta_certificate::identify_public_key_algorithm(oid)
688 .map(|s| s.to_string())
689 .unwrap_or_else(|| oid.to_string());
690 let py_str = PyString::new(py, &s).unbind();
691 let _ = self.public_key_algorithm_cache.set(py_str.clone_ref(py));
692 Ok(py_str.into_bound(py))
693 }
694
695 /// Raw subject public-key bytes.
696 ///
697 /// The `bytes` object is created once on first access and the same Python
698 /// object is returned (by reference) on every subsequent call,
699 /// avoiding repeated allocation and PyO3 string construction.
700 #[getter]
701 fn public_key<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
702 if let Some(cached) = self.public_key_cache.get() {
703 return Ok(cached.clone_ref(py).into_bound(py));
704 }
705 let py_bytes = PyBytes::new(
706 py,
707 self.cert()?
708 .tbs_certificate
709 .subject_public_key_info
710 .subject_public_key
711 .as_bytes(),
712 )
713 .unbind();
714 let _ = self.public_key_cache.set(py_bytes.clone_ref(py));
715 Ok(py_bytes.into_bound(py))
716 }
717
718 /// Version field (0 = v1, 1 = v2, 2 = v3), or None if absent.
719 #[getter]
720 fn version(&self) -> PyResult<Option<i64>> {
721 Ok(self
722 .cert()?
723 .tbs_certificate
724 .version
725 .as_ref()
726 .and_then(|v| v.as_i64().ok()))
727 }
728
729 /// Complete DER encoding of this certificate (the original bytes passed to
730 /// ``from_der``).
731 fn to_der<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
732 self._data.clone_ref(py).into_bound(py)
733 }
734
735 /// Raw DER bytes of the TBSCertificate structure (the bytes that were
736 /// signed by the issuer's key).
737 #[getter]
738 fn tbs_bytes<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
739 self.tbs_bytes_cache
740 .get_or_init(|| {
741 let data = self._data.bind(py).as_bytes();
742 PyBytes::new(py, &data[self.tbs_range.clone()]).unbind()
743 })
744 .clone_ref(py)
745 .into_bound(py)
746 }
747
748 /// OID of the signature algorithm
749 /// (e.g. ``ObjectIdentifier("1.2.840.113549.1.1.11")`` for sha256WithRSAEncryption).
750 ///
751 /// Unlike ``signature_algorithm``, this always returns the machine-readable
752 /// OID, even for algorithms that synta does not recognise by name.
753 #[getter]
754 fn signature_algorithm_oid<'py>(
755 &self,
756 py: Python<'py>,
757 ) -> PyResult<Bound<'py, PyObjectIdentifier>> {
758 if let Some(cached) = self.signature_algorithm_oid_cache.get() {
759 return Ok(cached.clone_ref(py).into_bound(py));
760 }
761 let obj = Py::new(
762 py,
763 PyObjectIdentifier::from_oid(self.cert()?.signature_algorithm.algorithm.clone()),
764 )?;
765 let _ = self.signature_algorithm_oid_cache.set(obj.clone_ref(py));
766 Ok(obj.into_bound(py))
767 }
768
769 /// Raw DER bytes of the signature algorithm parameters, or ``None`` if
770 /// the AlgorithmIdentifier has no parameters field (e.g. Ed25519).
771 #[getter]
772 fn signature_algorithm_params<'py>(
773 &self,
774 py: Python<'py>,
775 ) -> PyResult<Option<Bound<'py, PyBytes>>> {
776 if let Some(cached) = self.signature_algorithm_params_cache.get() {
777 return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
778 }
779 let computed =
780 encode_element_opt(py, self.cert()?.signature_algorithm.parameters.as_ref())?;
781 let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
782 let _ = self.signature_algorithm_params_cache.set(to_store);
783 Ok(computed)
784 }
785
786 /// DER-encoded ``AlgorithmIdentifier`` SEQUENCE from the certificate's outer
787 /// ``signatureAlgorithm`` field.
788 ///
789 /// This is the raw DER form of the field — suitable for passing directly to
790 /// :meth:`PublicKey.verify_certificate_signature` without any re-encoding.
791 ///
792 /// ```python,ignore
793 /// pub = synta.PublicKey.from_der(issuer_cert.public_key)
794 /// pub.verify_certificate_signature(
795 /// cert.tbs_certificate_der,
796 /// cert.signature_algorithm_der,
797 /// cert.signature_value,
798 /// )
799 /// ```
800 #[getter]
801 fn signature_algorithm_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
802 if let Some(cached) = self.signature_algorithm_der_cache.get() {
803 return Ok(cached.clone_ref(py).into_bound(py));
804 }
805 let ranges = synta_certificate::cert_byte_ranges(self.raw).ok_or_else(|| {
806 pyo3::exceptions::PyValueError::new_err("certificate DER structure is invalid")
807 })?;
808 let py_bytes = PyBytes::new(py, &self.raw[ranges.signature_algorithm]).unbind();
809 let _ = self
810 .signature_algorithm_der_cache
811 .set(py_bytes.clone_ref(py));
812 Ok(py_bytes.into_bound(py))
813 }
814
815 /// OID of the subject public-key algorithm
816 /// (e.g. ``ObjectIdentifier("1.2.840.10045.2.1")`` for id-ecPublicKey).
817 #[getter]
818 fn public_key_algorithm_oid<'py>(
819 &self,
820 py: Python<'py>,
821 ) -> PyResult<Bound<'py, PyObjectIdentifier>> {
822 if let Some(cached) = self.public_key_algorithm_oid_cache.get() {
823 return Ok(cached.clone_ref(py).into_bound(py));
824 }
825 let obj = Py::new(
826 py,
827 PyObjectIdentifier::from_oid(
828 self.cert()?
829 .tbs_certificate
830 .subject_public_key_info
831 .algorithm
832 .algorithm
833 .clone(),
834 ),
835 )?;
836 let _ = self.public_key_algorithm_oid_cache.set(obj.clone_ref(py));
837 Ok(obj.into_bound(py))
838 }
839
840 /// Raw DER bytes of the public-key algorithm parameters, or ``None``.
841 ///
842 /// For EC keys this is an OID naming the curve (e.g. secp256r1).
843 /// For RSA and Ed25519 this is typically ``None`` or a NULL element.
844 #[getter]
845 fn public_key_algorithm_params<'py>(
846 &self,
847 py: Python<'py>,
848 ) -> PyResult<Option<Bound<'py, PyBytes>>> {
849 if let Some(cached) = self.public_key_algorithm_params_cache.get() {
850 return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
851 }
852 let computed = encode_element_opt(
853 py,
854 self.cert()?
855 .tbs_certificate
856 .subject_public_key_info
857 .algorithm
858 .parameters
859 .as_ref(),
860 )?;
861 let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
862 let _ = self.public_key_algorithm_params_cache.set(to_store);
863 Ok(computed)
864 }
865
866 /// Raw DER bytes of the extensions SEQUENCE OF, or ``None`` for v1/v2
867 /// certificates that carry no extensions.
868 ///
869 /// The returned bytes begin with the SEQUENCE tag (``0x30``) and contain
870 /// the full SEQUENCE OF Extension encoding. Pass them to a ``Decoder``
871 /// to iterate the individual extensions:
872 ///
873 /// ```python
874 /// ext_der = cert.extensions_der
875 /// if ext_der:
876 /// dec = synta.Decoder(ext_der, synta.Encoding.DER)
877 /// exts_dec = dec.decode_sequence()
878 /// while not exts_dec.is_empty():
879 /// ext_tlv = exts_dec.decode_raw_tlv()
880 /// ```
881 #[getter]
882 fn extensions_der<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
883 if let Some(cached) = self.extensions_der_cache.get() {
884 return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
885 }
886 let computed = self
887 .cert()?
888 .tbs_certificate
889 .extensions
890 .as_ref()
891 .map(|raw: &synta::RawDer<'_>| PyBytes::new(py, raw.as_bytes()));
892 let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
893 let _ = self.extensions_der_cache.set(to_store);
894 Ok(computed)
895 }
896
897 /// Return the DER content of a named extension's value, or ``None``.
898 ///
899 /// Searches the certificate's extension list for an extension whose
900 /// ``extnID`` equals *oid* (dotted-decimal notation, e.g.
901 /// ``"2.5.29.17"`` for SubjectAltName). When found, the bytes inside
902 /// the ``extnValue`` OCTET STRING — the DER-encoded extension-specific
903 /// structure — are returned. Returns ``None`` when the certificate
904 /// carries no extensions or the named OID is absent.
905 ///
906 /// Example — parse SubjectAltName:
907 ///
908 /// ```python
909 /// san_der = cert.get_extension_value_der("2.5.29.17")
910 /// if san_der:
911 /// dec = synta.Decoder(san_der, synta.Encoding.DER)
912 /// san_seq = dec.decode_sequence()
913 /// while not san_seq.is_empty():
914 /// tag_num, tag_class, _ = san_seq.peek_tag()
915 /// child = san_seq.decode_implicit_tag(tag_num, tag_class)
916 /// if tag_num == 2: # dNSName
917 /// print(child.remaining_bytes().decode("ascii"))
918 /// ```
919 fn get_extension_value_der<'py>(
920 &self,
921 py: Python<'py>,
922 oid: &Bound<'_, PyAny>,
923 ) -> PyResult<Option<Bound<'py, PyBytes>>> {
924 let target = oid_from_pyany(oid)?;
925 let raw = match self.cert()?.tbs_certificate.extensions.as_ref() {
926 Some(r) => r,
927 None => return Ok(None),
928 };
929 for ext in &synta_certificate::decode_extensions(raw.as_bytes()) {
930 if ext.extn_id == target {
931 return Ok(Some(PyBytes::new(py, ext.extn_value.as_bytes())));
932 }
933 }
934 Ok(None)
935 }
936
937 /// Return the Subject Alternative Names of this certificate as typed objects.
938 ///
939 /// Combines looking up the SAN extension (OID ``2.5.29.17``) and parsing
940 /// its ``GeneralName`` entries into a single call. Returns a
941 /// :class:`list` of typed :mod:`synta.general_name` objects in document
942 /// order; returns an empty list when the certificate carries no SAN
943 /// extension.
944 ///
945 /// Each element is one of:
946 ///
947 /// - :class:`~synta.general_name.OtherName` — tag 0
948 /// - :class:`~synta.general_name.RFC822Name` — tag 1 (e-mail address)
949 /// - :class:`~synta.general_name.DNSName` — tag 2
950 /// - :class:`~synta.general_name.X400Address` — tag 3 (raw DER)
951 /// - :class:`~synta.general_name.DirectoryName` — tag 4
952 /// - :class:`~synta.general_name.EDIPartyName` — tag 5 (raw DER)
953 /// - :class:`~synta.general_name.UniformResourceIdentifier` — tag 6
954 /// - :class:`~synta.general_name.IPAddress` — tag 7
955 /// - :class:`~synta.general_name.RegisteredID` — tag 8
956 ///
957 /// ```python,ignore
958 /// import ipaddress
959 /// import synta.general_name as gn
960 ///
961 /// for name in cert.subject_alt_names():
962 /// if isinstance(name, gn.DNSName):
963 /// print("DNS:", name.value)
964 /// elif isinstance(name, gn.IPAddress):
965 /// print("IP:", ipaddress.ip_address(name.address))
966 /// elif isinstance(name, gn.RFC822Name):
967 /// print("email:", name.value)
968 /// elif isinstance(name, gn.DirectoryName):
969 /// attrs = synta.parse_name_attrs(name.name_der)
970 /// print("DirName:", attrs)
971 /// elif isinstance(name, gn.UniformResourceIdentifier):
972 /// print("URI:", name.value)
973 /// ```
974 fn subject_alt_names<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
975 use super::general_name::decode_general_names_to_py;
976 use synta_certificate::find_extension_value;
977
978 let cert = self.cert()?;
979 let raw = match cert.tbs_certificate.extensions.as_ref() {
980 Some(r) => r,
981 None => return Ok(PyList::empty(py)),
982 };
983 let san_bytes =
984 match find_extension_value(raw.as_bytes(), synta_certificate::oids::SUBJECT_ALT_NAME) {
985 Some(b) => b,
986 None => return Ok(PyList::empty(py)),
987 };
988 decode_general_names_to_py(py, san_bytes)
989 }
990
991 /// Return the GeneralNames from any extension that encodes a
992 /// ``SEQUENCE OF GeneralName``, identified by OID.
993 ///
994 /// Looks up the extension with the given *oid* and decodes its value as a
995 /// ``SEQUENCE OF GeneralName``. Suitable for SAN (``2.5.29.17``), IAN
996 /// (``2.5.29.18``), or any private extension that uses the same format.
997 ///
998 /// Returns a :class:`list` of typed :mod:`synta.general_name` objects, or
999 /// an empty list when the extension is absent or cannot be decoded.
1000 ///
1001 /// ```python,ignore
1002 /// import synta
1003 /// import synta.general_name as gn
1004 ///
1005 /// cert = synta.Certificate.from_der(der)
1006 /// # Decode the Issuer Alternative Names extension:
1007 /// for name in cert.general_names("2.5.29.18"):
1008 /// if isinstance(name, gn.DirectoryName):
1009 /// print(synta.parse_name_attrs(name.name_der))
1010 /// ```
1011 fn general_names<'py>(
1012 &self,
1013 py: Python<'py>,
1014 oid: &Bound<'_, PyAny>,
1015 ) -> PyResult<Bound<'py, PyList>> {
1016 use super::general_name::decode_general_names_to_py;
1017 use super::oid_from_pyany;
1018 use synta_certificate::find_extension_value;
1019
1020 let target = oid_from_pyany(oid)?;
1021 let cert = self.cert()?;
1022 let raw = match cert.tbs_certificate.extensions.as_ref() {
1023 Some(r) => r,
1024 None => return Ok(PyList::empty(py)),
1025 };
1026 let ext_bytes = match find_extension_value(raw.as_bytes(), target.components()) {
1027 Some(b) => b,
1028 None => return Ok(PyList::empty(py)),
1029 };
1030 decode_general_names_to_py(py, ext_bytes)
1031 }
1032
1033 /// Return the CertificatePolicies entries from the extension (OID ``2.5.29.32``).
1034 ///
1035 /// Decodes the ``CertificatePolicies`` extension value
1036 /// (``SEQUENCE OF PolicyInformation``) and returns a list of
1037 /// :class:`PolicyInformation` objects, each containing the policy OID and
1038 /// any ``policyQualifiers``.
1039 ///
1040 /// Returns an empty list when the extension is absent or cannot be parsed.
1041 ///
1042 /// ```python,ignore
1043 /// import synta
1044 /// cert = synta.Certificate.from_der(der)
1045 /// for pi in cert.certificate_policies():
1046 /// print(str(pi.policy_oid)) # e.g. "1.3.6.1.4.1.311.21.8...."
1047 /// for q in pi.qualifiers:
1048 /// print(str(q.qualifier_oid))
1049 /// ```
1050 fn certificate_policies<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
1051 let list = PyList::empty(py);
1052 let cert = self.cert()?;
1053 let ext_raw = match cert.tbs_certificate.extensions.as_ref() {
1054 Some(r) => r,
1055 None => return Ok(list),
1056 };
1057 let ext_value_bytes = synta_certificate::find_extension_value(
1058 ext_raw.as_bytes(),
1059 synta_certificate::oids::CERTIFICATE_POLICIES,
1060 );
1061 let ext_bytes = match ext_value_bytes {
1062 Some(b) => b,
1063 None => return Ok(list),
1064 };
1065 // Decode using the code-generated PolicyInformation<'a> type.
1066 let mut decoder = synta::Decoder::new(ext_bytes, synta::Encoding::Der);
1067 let infos: Vec<synta_certificate::PolicyInformation<'_>> = match decoder.decode() {
1068 Ok(v) => v,
1069 Err(_) => return Ok(list),
1070 };
1071 for pi in infos {
1072 let qualifiers = pi
1073 .policy_qualifiers
1074 .unwrap_or_default()
1075 .into_iter()
1076 .filter_map(|qi: PolicyQualifierInfo<'_>| {
1077 let mut enc = synta::Encoder::new(synta::Encoding::Der);
1078 qi.qualifier.encode(&mut enc).ok()?;
1079 let qval = enc.finish().ok()?;
1080 Some((qi.policy_qualifier_id, qval))
1081 })
1082 .collect();
1083 let py_pi = Py::new(
1084 py,
1085 PyPolicyInformation {
1086 policy_oid: pi.policy_identifier,
1087 qualifiers,
1088 },
1089 )?;
1090 list.append(py_pi.into_bound(py))?;
1091 }
1092 Ok(list)
1093 }
1094
1095 /// Raw DER bytes of the issuer Name SEQUENCE.
1096 #[getter]
1097 fn issuer_raw_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
1098 if let Some(cached) = self.issuer_raw_der_cache.get() {
1099 return Ok(cached.clone_ref(py).into_bound(py));
1100 }
1101 let py_bytes = PyBytes::new(py, self.cert()?.tbs_certificate.issuer.as_bytes()).unbind();
1102 let _ = self.issuer_raw_der_cache.set(py_bytes.clone_ref(py));
1103 Ok(py_bytes.into_bound(py))
1104 }
1105
1106 /// Raw DER bytes of the subject Name SEQUENCE.
1107 #[getter]
1108 fn subject_raw_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
1109 if let Some(cached) = self.subject_raw_der_cache.get() {
1110 return Ok(cached.clone_ref(py).into_bound(py));
1111 }
1112 let py_bytes = PyBytes::new(py, self.cert()?.tbs_certificate.subject.as_bytes()).unbind();
1113 let _ = self.subject_raw_der_cache.set(py_bytes.clone_ref(py));
1114 Ok(py_bytes.into_bound(py))
1115 }
1116
1117 /// notBefore time as a UTC-aware ``datetime.datetime``.
1118 ///
1119 /// Equivalent to ``cert.not_valid_before.replace(tzinfo=datetime.timezone.utc)``
1120 /// from the ``cryptography`` library. Derives the value from the parsed
1121 /// ASN.1 ``Time`` field (``UTCTime`` or ``GeneralizedTime``) without calling
1122 /// any Python date-parsing code.
1123 ///
1124 /// ```python,ignore
1125 /// import datetime
1126 /// cert = synta.Certificate.from_der(der)
1127 /// dt = cert.not_before_utc
1128 /// print(dt.isoformat()) # e.g. "2023-01-01T00:00:00+00:00"
1129 /// print(dt.tzinfo) # datetime.timezone.utc
1130 /// ```
1131 #[getter]
1132 fn not_before_utc<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
1133 if let Some(cached) = self.not_before_utc_cache.get() {
1134 return Ok(cached.clone_ref(py).into_bound(py));
1135 }
1136 let unix_secs = synta_x509_verification::certificate::time_to_unix(
1137 &self.cert()?.tbs_certificate.validity.not_before,
1138 )
1139 .ok_or_else(|| {
1140 pyo3::exceptions::PyValueError::new_err("notBefore date is structurally invalid")
1141 })?;
1142 let datetime_mod = py.import("datetime")?;
1143 let utc = datetime_mod.getattr("timezone")?.getattr("utc")?;
1144 let dt = datetime_mod
1145 .getattr("datetime")?
1146 .call_method1("fromtimestamp", (unix_secs, &utc))?
1147 .unbind();
1148 let _ = self.not_before_utc_cache.set(dt.clone_ref(py));
1149 Ok(dt.into_bound(py))
1150 }
1151
1152 /// notAfter time as a UTC-aware ``datetime.datetime``.
1153 ///
1154 /// Equivalent to ``cert.not_valid_after.replace(tzinfo=datetime.timezone.utc)``
1155 /// from the ``cryptography`` library. Derives the value from the parsed
1156 /// ASN.1 ``Time`` field (``UTCTime`` or ``GeneralizedTime``) without calling
1157 /// any Python date-parsing code.
1158 ///
1159 /// ```python,ignore
1160 /// import datetime
1161 /// cert = synta.Certificate.from_der(der)
1162 /// dt = cert.not_after_utc
1163 /// print(dt.isoformat()) # e.g. "2033-01-01T00:00:00+00:00"
1164 /// print(dt.tzinfo) # datetime.timezone.utc
1165 /// ```
1166 #[getter]
1167 fn not_after_utc<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
1168 if let Some(cached) = self.not_after_utc_cache.get() {
1169 return Ok(cached.clone_ref(py).into_bound(py));
1170 }
1171 let unix_secs = synta_x509_verification::certificate::time_to_unix(
1172 &self.cert()?.tbs_certificate.validity.not_after,
1173 )
1174 .ok_or_else(|| {
1175 pyo3::exceptions::PyValueError::new_err("notAfter date is structurally invalid")
1176 })?;
1177 let datetime_mod = py.import("datetime")?;
1178 let utc = datetime_mod.getattr("timezone")?.getattr("utc")?;
1179 let dt = datetime_mod
1180 .getattr("datetime")?
1181 .call_method1("fromtimestamp", (unix_secs, &utc))?
1182 .unbind();
1183 let _ = self.not_after_utc_cache.set(dt.clone_ref(py));
1184 Ok(dt.into_bound(py))
1185 }
1186
1187 /// Hash algorithm name component of the certificate's outer signature algorithm.
1188 ///
1189 /// Returns the lowercase hash algorithm name (e.g. ``"sha256"``, ``"sha1"``,
1190 /// ``"sha384"``) implied by the signature algorithm OID. Returns ``None``
1191 /// for algorithms that do not use a traditional hash (Ed25519, Ed448,
1192 /// ML-DSA, etc.).
1193 ///
1194 /// For RSASSA-PSS the hash is encoded in the parameters field, which is not
1195 /// inspected here; ``"sha256"`` is returned as a conservative default.
1196 ///
1197 /// ```python,ignore
1198 /// cert = synta.Certificate.from_der(der)
1199 /// hash_name = cert.signature_hash_algorithm_name
1200 /// if hash_name:
1201 /// print(hash_name) # e.g. "sha256"
1202 /// else:
1203 /// print("no hash (e.g. Ed25519)")
1204 /// ```
1205 #[getter]
1206 fn signature_hash_algorithm_name(&self) -> PyResult<Option<&'static str>> {
1207 use synta_certificate::oids;
1208 let oid = &self.cert()?.signature_algorithm.algorithm;
1209 let c = oid.components();
1210 // RSA PKCS#1 v1.5 variants (1.2.840.113549.1.1.*)
1211 if c == oids::SHA1_WITH_RSA {
1212 return Ok(Some("sha1"));
1213 }
1214 if c == oids::SHA256_WITH_RSA {
1215 return Ok(Some("sha256"));
1216 }
1217 if c == oids::SHA384_WITH_RSA {
1218 return Ok(Some("sha384"));
1219 }
1220 if c == oids::SHA512_WITH_RSA {
1221 return Ok(Some("sha512"));
1222 }
1223 // RSA-PSS (hash is in params; return sha256 as safe default)
1224 if c == oids::RSASSA_PSS {
1225 return Ok(Some("sha256"));
1226 }
1227 // ECDSA variants
1228 if c == oids::ECDSA_WITH_SHA1 {
1229 return Ok(Some("sha1"));
1230 }
1231 if c == oids::ECDSA_WITH_SHA256 {
1232 return Ok(Some("sha256"));
1233 }
1234 if c == oids::ECDSA_WITH_SHA384 {
1235 return Ok(Some("sha384"));
1236 }
1237 if c == oids::ECDSA_WITH_SHA512 {
1238 return Ok(Some("sha512"));
1239 }
1240 // Ed25519, Ed448, ML-DSA-*, SLH-DSA-* and everything else: no hash
1241 Ok(None)
1242 }
1243
1244 /// Complete DER encoding of the SubjectPublicKeyInfo SEQUENCE.
1245 ///
1246 /// Returns the full ``SubjectPublicKeyInfo`` SEQUENCE TLV as ``bytes``,
1247 /// including the algorithm identifier and the BIT STRING carrying the
1248 /// actual public key. This is the DER encoding that
1249 /// ``cryptography``'s ``public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo)``
1250 /// produces.
1251 ///
1252 /// ```python,ignore
1253 /// cert = synta.Certificate.from_der(der)
1254 /// spki_der = cert.subject_public_key_info_der
1255 /// # Load the public key with cryptography:
1256 /// from cryptography.hazmat.primitives.serialization import load_der_public_key
1257 /// pub_key = load_der_public_key(spki_der)
1258 /// ```
1259 #[getter]
1260 fn subject_public_key_info_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
1261 if let Some(cached) = self.spki_der_cache.get() {
1262 return Ok(cached.clone_ref(py).into_bound(py));
1263 }
1264 let spki = &self.cert()?.tbs_certificate.subject_public_key_info;
1265 let mut enc = synta::Encoder::new(synta::Encoding::Der);
1266 spki.encode(&mut enc)
1267 .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
1268 let bytes = enc
1269 .finish()
1270 .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
1271 let py_bytes = PyBytes::new(py, &bytes).unbind();
1272 let _ = self.spki_der_cache.set(py_bytes.clone_ref(py));
1273 Ok(py_bytes.into_bound(py))
1274 }
1275
1276 /// Raw DER bytes of the ``TBSCertificate`` structure — the bytes that
1277 /// were signed by the issuer's key.
1278 ///
1279 /// This is the DER encoding of the full ``TBSCertificate`` SEQUENCE TLV,
1280 /// suitable for signature verification or as the input to a hash function.
1281 /// Equivalent to ``cryptography``'s ``cert.tbs_certificate_bytes``.
1282 ///
1283 /// ```python,ignore
1284 /// cert = synta.Certificate.from_der(der)
1285 /// tbs = cert.tbs_certificate_der
1286 /// # Verify signature manually:
1287 /// import synta.crypto
1288 /// digest = synta.digest("sha256", tbs)
1289 /// ```
1290 #[getter]
1291 fn tbs_certificate_der<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
1292 self.tbs_bytes(py)
1293 }
1294
1295 /// Verify that this certificate was directly issued by ``issuer``.
1296 ///
1297 /// Checks two things:
1298 ///
1299 /// 1. The ``issuer`` field of this certificate's TBS matches the
1300 /// ``subject`` field of the provided issuer certificate (byte-exact
1301 /// DER comparison).
1302 /// 2. The certificate's signature is valid under the issuer's public key.
1303 ///
1304 /// Raises :exc:`ValueError` if either check fails. This is the synta
1305 /// equivalent of ``cryptography``'s
1306 /// ``cert.verify_directly_issued_by(issuer)``.
1307 ///
1308 /// ```python,ignore
1309 /// root = synta.Certificate.from_pem(open("root.pem", "rb").read())
1310 /// leaf = synta.Certificate.from_pem(open("leaf.pem", "rb").read())
1311 /// leaf.verify_issued_by(root) # raises ValueError if not valid
1312 /// ```
1313 fn verify_issued_by(&self, issuer: &PyCertificate) -> PyResult<()> {
1314 use synta_certificate::{cert_byte_ranges, default_signature_verifier, SignatureVerifier};
1315
1316 // 1. Name check: our issuer Name must byte-equal their subject Name.
1317 let cert = self.cert()?;
1318 let issuer_cert = issuer.cert()?;
1319 if cert.tbs_certificate.issuer.as_bytes() != issuer_cert.tbs_certificate.subject.as_bytes()
1320 {
1321 return Err(pyo3::exceptions::PyValueError::new_err(
1322 "issuer name does not match subject of provided certificate",
1323 ));
1324 }
1325
1326 // 2. Locate byte ranges within each cert's DER without re-encoding.
1327 let my_ranges = cert_byte_ranges(self.raw).ok_or_else(|| {
1328 pyo3::exceptions::PyValueError::new_err(
1329 "failed to locate byte ranges in certificate DER",
1330 )
1331 })?;
1332 let issuer_ranges = cert_byte_ranges(issuer.raw).ok_or_else(|| {
1333 pyo3::exceptions::PyValueError::new_err(
1334 "failed to locate byte ranges in issuer certificate DER",
1335 )
1336 })?;
1337
1338 // 3. Signature bits (raw bytes, no unused-bits prefix).
1339 let signature_bits = cert.signature_value.as_bytes();
1340
1341 // 4. Verify using the backend-agnostic verifier.
1342 default_signature_verifier()
1343 .verify_certificate_signature(
1344 &self.raw[my_ranges.tbs],
1345 &self.raw[my_ranges.signature_algorithm],
1346 signature_bits,
1347 &issuer.raw[issuer_ranges.subject_public_key_info],
1348 )
1349 .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
1350
1351 Ok(())
1352 }
1353
1354 /// Compute a hash fingerprint of the complete certificate DER.
1355 ///
1356 /// ``algorithm`` must be one of ``"sha1"``, ``"sha224"``, ``"sha256"``,
1357 /// ``"sha384"``, ``"sha512"``, or ``"md5"``. Raises :exc:`ValueError` for
1358 /// unknown algorithm names.
1359 ///
1360 /// ```python,ignore
1361 /// cert = synta.Certificate.from_der(der)
1362 /// fp = cert.fingerprint("sha256")
1363 /// print(fp.hex()) # e.g. "3a2b1c..."
1364 /// ```
1365 fn fingerprint<'py>(&self, py: Python<'py>, algorithm: &str) -> PyResult<Bound<'py, PyBytes>> {
1366 use synta_certificate::{default_data_hasher, DataHasher};
1367 let digest = default_data_hasher()
1368 .hash_data(algorithm, self.raw)
1369 .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
1370 Ok(PyBytes::new(py, &digest))
1371 }
1372
1373 fn __repr__(&self) -> PyResult<String> {
1374 let cert = self.cert()?;
1375 let serial = &cert.tbs_certificate.serial_number;
1376 let serial_str = if let Ok(v) = serial.as_i64() {
1377 v.to_string()
1378 } else {
1379 format!("<{} bytes>", serial.as_bytes().len())
1380 };
1381 Ok(format!(
1382 "Certificate(subject={:?}, serial={})",
1383 decode_name_debug(cert.tbs_certificate.subject.as_bytes()),
1384 serial_str,
1385 ))
1386 }
1387}
1388
1389impl PyCertificate {
1390 /// Parse a DER-encoded X.509 certificate (internal helper visible to sibling modules).
1391 ///
1392 /// This is the same logic as the `#[staticmethod] from_der` exposed to Python but
1393 /// accessible from Rust sibling modules via `pub(super)` visibility.
1394 pub(crate) fn new_from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
1395 Self::from_der(py, data)
1396 }
1397}