_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, 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/// X.509 Certificate accessible from Python.
120///
121/// Example (inside a Python extension that called `register_module`):
122///
123/// ```python
124/// cert = Certificate.from_der(open("cert.der", "rb").read())
125/// print(cert.subject)
126/// ```
127#[pyclass(frozen, name = "Certificate")]
128pub struct PyCertificate {
129 // Holds a strong reference to the Python bytes object that backs the
130 // DER data. The CPython refcount prevents the underlying bytes buffer
131 // from being freed for the full lifetime of this struct. After the
132 // struct is collected by Python's GC, `_data` drops (in Rust's
133 // declaration order) and decrements the refcount; by that point no
134 // Rust code can hold a borrow of this struct, so no read of `raw`
135 // can occur through `&self` after drop begins.
136 pub(super) _data: Py<PyBytes>,
137 // Raw slice pointing into `_data`'s buffer. SAFETY: valid as long as
138 // `_data` is alive; CPython bytes objects have a fixed-address buffer
139 // that is never relocated. Stored here so the OnceLock init closure in
140 // `cert()` can access the bytes without requiring a GIL token.
141 pub(super) raw: &'static [u8],
142 // Full decoded certificate, heap-allocated and initialised lazily on first
143 // field access. `Box` keeps the 568-byte `Certificate` off the struct so
144 // that `PyCertificate` stays within CPython's 512-byte `pymalloc`
145 // threshold (after adding the Python object header). Allocating through
146 // pymalloc instead of the system allocator is roughly 3× faster, which is
147 // measurable at parse-only speeds. The full recursive decode is deferred
148 // so that parse-only workloads pay only the shallow envelope-scan cost.
149 inner: OnceLock<Box<Certificate<'static>>>,
150 // Byte range within `_data` covering the complete TBSCertificate TLV.
151 tbs_range: std::ops::Range<usize>,
152 // Lazily-computed Python objects. Each field is initialised on first
153 // Python access and returned via clone_ref on every subsequent call,
154 // avoiding repeated allocation and PyO3 string construction.
155 issuer_cache: OnceLock<Py<PyString>>,
156 subject_cache: OnceLock<Py<PyString>>,
157 signature_algorithm_cache: OnceLock<Py<PyString>>,
158 not_before_cache: OnceLock<Py<PyString>>,
159 not_after_cache: OnceLock<Py<PyString>>,
160 public_key_algorithm_cache: OnceLock<Py<PyString>>,
161 serial_number_cache: OnceLock<Py<PyAny>>,
162 signature_value_cache: OnceLock<Py<PyBytes>>,
163 public_key_cache: OnceLock<Py<PyBytes>>,
164 tbs_bytes_cache: OnceLock<Py<PyBytes>>,
165 signature_algorithm_oid_cache: OnceLock<Py<PyObjectIdentifier>>,
166 public_key_algorithm_oid_cache: OnceLock<Py<PyObjectIdentifier>>,
167 signature_algorithm_params_cache: OnceLock<Option<Py<PyBytes>>>,
168 public_key_algorithm_params_cache: OnceLock<Option<Py<PyBytes>>>,
169 extensions_der_cache: OnceLock<Option<Py<PyBytes>>>,
170 issuer_raw_der_cache: OnceLock<Py<PyBytes>>,
171 subject_raw_der_cache: OnceLock<Py<PyBytes>>,
172 not_before_utc_cache: OnceLock<Py<PyAny>>,
173 not_after_utc_cache: OnceLock<Py<PyAny>>,
174 spki_der_cache: OnceLock<Py<PyBytes>>,
175}
176
177impl PyCertificate {
178 /// Return the fully-decoded certificate, triggering a full recursive
179 /// `Certificate::decode()` on the first call.
180 ///
181 /// The result is cached in `inner` so subsequent calls are a single
182 /// atomic load. Returns `Err(PyValueError)` if the full decode fails
183 /// (e.g. malformed inner content that passed the shallow envelope scan
184 /// in `from_der`). Unlike a panic, this surfaces as a catchable Python
185 /// `ValueError` rather than an uncatchable `PanicException`.
186 fn cert(&self) -> PyResult<&Certificate<'static>> {
187 if let Some(v) = self.inner.get() {
188 return Ok(v.as_ref());
189 }
190 let mut decoder = Decoder::new(self.raw, Encoding::Der);
191 let decoded = decoder.decode().map_err(|e| {
192 pyo3::exceptions::PyValueError::new_err(format!("Certificate DER decode failed: {e}"))
193 })?;
194 let _ = self.inner.set(Box::new(decoded));
195 Ok(self.inner.get().unwrap().as_ref())
196 }
197}
198
199#[pymethods]
200impl PyCertificate {
201 /// Parse a DER-encoded X.509 certificate.
202 #[staticmethod]
203 fn from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
204 // Store the Python bytes object directly — no copy of the DER data.
205 let py_bytes = data.unbind();
206
207 // SAFETY: `py_bytes` holds a strong reference (Py<PyBytes>) that
208 // keeps the Python bytes object alive for the lifetime of this struct.
209 // CPython's bytes objects have a fixed-address, non-relocating payload
210 // buffer (CPython has no moving GC for bytes objects). The slice
211 // lifetime is extended to 'static; the actual safety invariants are:
212 // (1) All reads of `raw` go through `&self` (a borrow of the struct).
213 // No borrow of the struct can outlive the struct, so `raw` is
214 // never read after the struct begins dropping.
215 // (2) `raw: &'static [u8]` has no destructor (it is a fat pointer
216 // with no heap allocation), so Rust dropping `_data` before `raw`
217 // (fields drop in declaration order) does not cause a
218 // use-after-free during the drop sequence itself.
219 // (3) `inner` contains `Certificate<'static>` references into the
220 // buffer; dropping `Box<Certificate<'static>>` frees the box
221 // allocation but does not read through the contained &'static
222 // slices (borrows have no destructors in Rust).
223 // CPython-only: this pattern does not hold for PyPy or GraalPy,
224 // which may relocate or compact heap objects.
225 let raw: &'static [u8] = unsafe {
226 let s = py_bytes.bind(py).as_bytes();
227 std::slice::from_raw_parts(s.as_ptr(), s.len())
228 };
229
230 // Shallow structural scan: validate the Certificate SEQUENCE envelope
231 // and locate the TBSCertificate TLV range. This is ~4 decoder
232 // operations — roughly 10× faster than a full Certificate::decode().
233 // The full decode is deferred to the first field access via `cert()`.
234 //
235 // Certificate ::= SEQUENCE {
236 // tbsCertificate TBSCertificate, -- child 0: SEQUENCE
237 // signatureAlgorithm AlgorithmIdentifier, -- child 1: SEQUENCE
238 // signature BIT STRING -- child 2
239 // }
240 let tbs_range = {
241 let mut d = Decoder::new(raw, Encoding::Der);
242 // Outer Certificate SEQUENCE tag + length.
243 d.read_tag()
244 .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
245 d.read_length()
246 .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
247 // TBSCertificate: record the full TLV range (tag + length + content).
248 let tbs_start = d.position();
249 d.read_tag()
250 .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
251 let tbs_len = d
252 .read_length()
253 .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
254 let tbs_content_len = tbs_len
255 .definite()
256 .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
257 tbs_start..(d.position() + tbs_content_len)
258 };
259
260 Ok(Self {
261 _data: py_bytes,
262 raw,
263 inner: OnceLock::new(),
264 tbs_range,
265 issuer_cache: OnceLock::new(),
266 subject_cache: OnceLock::new(),
267 signature_algorithm_cache: OnceLock::new(),
268 not_before_cache: OnceLock::new(),
269 not_after_cache: OnceLock::new(),
270 public_key_algorithm_cache: OnceLock::new(),
271 serial_number_cache: OnceLock::new(),
272 signature_value_cache: OnceLock::new(),
273 public_key_cache: OnceLock::new(),
274 tbs_bytes_cache: OnceLock::new(),
275 signature_algorithm_oid_cache: OnceLock::new(),
276 public_key_algorithm_oid_cache: OnceLock::new(),
277 signature_algorithm_params_cache: OnceLock::new(),
278 public_key_algorithm_params_cache: OnceLock::new(),
279 extensions_der_cache: OnceLock::new(),
280 issuer_raw_der_cache: OnceLock::new(),
281 subject_raw_der_cache: OnceLock::new(),
282 not_before_utc_cache: OnceLock::new(),
283 not_after_utc_cache: OnceLock::new(),
284 spki_der_cache: OnceLock::new(),
285 })
286 }
287
288 /// Parse a DER-encoded X.509 certificate and perform a full RFC 5280 decode immediately.
289 ///
290 /// Unlike :meth:`from_der`, which performs only a shallow 4-operation
291 /// envelope scan and defers the full :class:`Certificate` decode to the
292 /// first field access, this method triggers the complete recursive
293 /// ``Certificate::decode()`` at construction time.
294 ///
295 /// Use this when you need all fields to be available without any
296 /// lazy-decode latency on the first getter call, or when benchmarking the
297 /// true full-parse cost that is comparable to Criterion's ``rust_typed``
298 /// numbers.
299 ///
300 /// ```python
301 /// # Full parse happens here — no deferred work on first field access.
302 /// cert = Certificate.full_from_der(der)
303 /// print(cert.issuer) # warm path only
304 /// ```
305 #[staticmethod]
306 fn full_from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
307 let cert = Self::from_der(py, data)?;
308 // Prime the OnceLock: runs the full Certificate::decode() now.
309 cert.cert()?;
310 Ok(cert)
311 }
312
313 /// Parse a PEM-encoded X.509 certificate.
314 ///
315 /// Strips the ``-----BEGIN CERTIFICATE-----`` / ``-----END CERTIFICATE-----``
316 /// boundary lines, decodes the base64 body, and calls :meth:`from_der`.
317 /// No external dependencies are required — the decoder is implemented in
318 /// pure Rust inside ``synta-certificate``.
319 ///
320 /// Returns a single :class:`Certificate` when the input contains exactly
321 /// one PEM block. When multiple blocks are present (e.g. a certificate
322 /// chain file), returns a :class:`list` of :class:`Certificate` objects in
323 /// the order they appear. Raises :exc:`ValueError` if no PEM block is
324 /// found.
325 ///
326 /// ```python
327 /// # Single certificate — returns Certificate directly.
328 /// cert = Certificate.from_pem(open("cert.pem", "rb").read())
329 /// print(cert.subject)
330 ///
331 /// # Certificate chain — returns list[Certificate].
332 /// chain = Certificate.from_pem(open("chain.pem", "rb").read())
333 /// for cert in chain:
334 /// print(cert.subject)
335 /// ```
336 #[staticmethod]
337 fn from_pem<'py>(py: Python<'py>, data: Bound<'_, PyBytes>) -> PyResult<Bound<'py, PyAny>> {
338 pem_blocks_to_pyobject(py, data.as_bytes(), |py, bytes| {
339 let obj = Self::from_der(py, bytes)?;
340 Ok(Py::new(py, obj)?.into_bound(py).into_any())
341 })
342 }
343
344 /// Convert to a ``cryptography.x509.Certificate`` (PyCA) object.
345 ///
346 /// Passes the original DER bytes directly to
347 /// ``cryptography.x509.load_der_x509_certificate``; no re-encoding is
348 /// performed. Requires the ``cryptography`` package to be installed.
349 ///
350 /// ```python
351 /// synta_cert = synta.Certificate.from_der(der)
352 /// pyca_cert = synta_cert.to_pyca()
353 /// # Full cryptographic operations are now available via PyCA:
354 /// pyca_cert.public_key().verify(signature, message, ...)
355 /// ```
356 fn to_pyca<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
357 let m = py.import("cryptography.x509").map_err(|_| {
358 pyo3::exceptions::PyImportError::new_err(
359 "the 'cryptography' package is required; install it with: pip install cryptography",
360 )
361 })?;
362 m.call_method1(
363 "load_der_x509_certificate",
364 (self._data.clone_ref(py).into_bound(py),),
365 )
366 }
367
368 /// Construct a ``Certificate`` from a ``cryptography.x509.Certificate``.
369 ///
370 /// Serialises the PyCA certificate to DER via
371 /// ``cert.public_bytes(Encoding.DER)`` and parses the result with
372 /// :meth:`from_der`. Requires the ``cryptography`` package to be
373 /// installed.
374 ///
375 /// **Fast path** — if the object exposes a ``_synta_der_bytes`` attribute
376 /// whose value is a ``bytes`` object, that buffer is used directly and
377 /// the ``public_bytes()`` call is skipped entirely. Wrapper classes that
378 /// already hold the raw DER (e.g. an ``IPACertificate`` loaded from LDAP)
379 /// can opt in by setting the attribute at construction time:
380 ///
381 /// ```python
382 /// class IPACertificate:
383 /// def __init__(self, der: bytes):
384 /// self._synta_der_bytes = der # enables fast path
385 /// self._pyca = x509.load_der_x509_certificate(der)
386 ///
387 /// # Zero re-encoding cost — DER is read from _synta_der_bytes directly:
388 /// synta_cert = synta.Certificate.from_pyca(ipa_cert)
389 /// ```
390 ///
391 /// Without the attribute the standard path is used:
392 ///
393 /// ```python
394 /// pyca_cert = cryptography.x509.load_pem_x509_certificate(pem)
395 /// synta_cert = synta.Certificate.from_pyca(pyca_cert)
396 /// ```
397 #[staticmethod]
398 fn from_pyca(py: Python<'_>, pyca_cert: Bound<'_, PyAny>) -> PyResult<Self> {
399 // Optional fast path: if the caller has cached raw DER bytes on the
400 // object under the `_synta_der_bytes` attribute, use that directly to
401 // avoid the re-encoding cost of `public_bytes(Encoding.DER)`. This
402 // lets wrappers like FreeIPA's `IPACertificate` opt in by setting
403 // `self._synta_der_bytes = der` at construction time.
404 if let Ok(attr) = pyca_cert.getattr("_synta_der_bytes") {
405 if let Ok(der_bytes) = attr.cast_into::<PyBytes>() {
406 return Self::from_der(py, der_bytes);
407 }
408 }
409
410 let ser = py
411 .import("cryptography.hazmat.primitives.serialization")
412 .map_err(|_| {
413 pyo3::exceptions::PyImportError::new_err(
414 "the 'cryptography' package is required; install it with: pip install cryptography",
415 )
416 })?;
417 let der_bytes: Bound<'_, PyBytes> = pyca_cert
418 .call_method1("public_bytes", (ser.getattr("Encoding")?.getattr("DER")?,))?
419 .cast_into()?;
420 Self::from_der(py, der_bytes)
421 }
422
423 /// Serialize one certificate or a list of certificates to PEM format.
424 ///
425 /// The mirror of :meth:`from_pem`: accepts either a single
426 /// :class:`Certificate` or a :class:`list` of them, and returns a
427 /// :class:`bytes` value containing one or more
428 /// ``-----BEGIN CERTIFICATE-----`` blocks concatenated in order.
429 ///
430 /// ```python
431 /// # Single certificate:
432 /// pem = Certificate.to_pem(cert)
433 /// open("cert.pem", "wb").write(pem)
434 ///
435 /// # Certificate chain:
436 /// pem = Certificate.to_pem([leaf, intermediate, root])
437 /// open("chain.pem", "wb").write(pem)
438 /// ```
439 #[staticmethod]
440 fn to_pem<'py>(
441 py: Python<'py>,
442 obj_or_list: Bound<'_, PyAny>,
443 ) -> PyResult<Bound<'py, PyBytes>> {
444 pyobject_to_pem::<Self, _>(py, "CERTIFICATE", &obj_or_list, |c| c.raw)
445 }
446
447 /// Serial number as a Python int.
448 ///
449 /// Returns a native Python `int` regardless of size (X.509 serials can be
450 /// up to 20 bytes / 160 bits per RFC 5280). The Python object is created
451 /// once and cached; subsequent accesses return a reference to the same object.
452 #[getter]
453 fn serial_number<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
454 // `get_or_try_init` is not stable on `OnceLock`, so build the value
455 // outside the lock if the cache is empty, then store it.
456 if let Some(cached) = self.serial_number_cache.get() {
457 return Ok(cached.clone_ref(py).into_bound(py));
458 }
459 let serial = &self.cert()?.tbs_certificate.serial_number;
460 let py_int: Py<PyAny> = if let Ok(v) = serial.as_i64() {
461 // Fast path: fits in i64 (covers traditional short serials)
462 v.into_pyobject(py)?.into_any().unbind()
463 } else if let Ok(v) = serial.as_i128() {
464 // Fits in i128 (covers up to 16-byte random serials)
465 v.into_pyobject(py)?.into_any().unbind()
466 } else {
467 // Large serial (17–20 bytes): call int.from_bytes() directly.
468 // Using call_method avoids the parse+compile overhead of py.eval()
469 // on every cold-path invocation. `signed` is keyword-only in
470 // Python's int.from_bytes signature, so it must go in kwargs.
471 let bytes_obj = PyBytes::new(py, serial.as_bytes());
472 let kwargs = pyo3::types::PyDict::new(py);
473 kwargs.set_item(pyo3::intern!(py, "signed"), true)?;
474 py.get_type::<pyo3::types::PyInt>()
475 .call_method(
476 pyo3::intern!(py, "from_bytes"),
477 (bytes_obj, pyo3::intern!(py, "big")),
478 Some(&kwargs),
479 )
480 .map(|r| r.unbind())?
481 };
482 // Ignore a racing writer; both produce the same value.
483 let _ = self.serial_number_cache.set(py_int.clone_ref(py));
484 Ok(py_int.into_bound(py))
485 }
486
487 /// Issuer distinguished name as a string.
488 #[getter]
489 fn issuer<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
490 if let Some(cached) = self.issuer_cache.get() {
491 return Ok(cached.clone_ref(py).into_bound(py));
492 }
493 let s = decode_name_debug(self.cert()?.tbs_certificate.issuer.as_bytes());
494 let py_str = PyString::new(py, &s).unbind();
495 let _ = self.issuer_cache.set(py_str.clone_ref(py));
496 Ok(py_str.into_bound(py))
497 }
498
499 /// Subject distinguished name as a string.
500 #[getter]
501 fn subject<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
502 if let Some(cached) = self.subject_cache.get() {
503 return Ok(cached.clone_ref(py).into_bound(py));
504 }
505 let s = decode_name_debug(self.cert()?.tbs_certificate.subject.as_bytes());
506 let py_str = PyString::new(py, &s).unbind();
507 let _ = self.subject_cache.set(py_str.clone_ref(py));
508 Ok(py_str.into_bound(py))
509 }
510
511 /// Signature algorithm name (e.g. "RSA", "ECDSA", "Ed25519", "ML-DSA-65"),
512 /// or the dotted OID notation for unrecognized algorithms.
513 #[getter]
514 fn signature_algorithm<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
515 if let Some(cached) = self.signature_algorithm_cache.get() {
516 return Ok(cached.clone_ref(py).into_bound(py));
517 }
518 let oid = &self.cert()?.signature_algorithm.algorithm;
519 let name = synta_certificate::identify_signature_algorithm(oid);
520 let s = if name != "Other" {
521 name.to_string()
522 } else {
523 oid.to_string()
524 };
525 let py_str = PyString::new(py, &s).unbind();
526 let _ = self.signature_algorithm_cache.set(py_str.clone_ref(py));
527 Ok(py_str.into_bound(py))
528 }
529
530 /// Raw signature bytes.
531 ///
532 /// The `bytes` object is created once on first access and the same Python
533 /// object is returned (by reference) on every subsequent call,
534 /// avoiding repeated allocation and PyO3 string construction.
535 #[getter]
536 fn signature_value<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
537 if let Some(cached) = self.signature_value_cache.get() {
538 return Ok(cached.clone_ref(py).into_bound(py));
539 }
540 let py_bytes = PyBytes::new(py, self.cert()?.signature_value.as_bytes()).unbind();
541 let _ = self.signature_value_cache.set(py_bytes.clone_ref(py));
542 Ok(py_bytes.into_bound(py))
543 }
544
545 /// notBefore time as a string.
546 #[getter]
547 fn not_before<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
548 if let Some(cached) = self.not_before_cache.get() {
549 return Ok(cached.clone_ref(py).into_bound(py));
550 }
551 let s = match &self.cert()?.tbs_certificate.validity.not_before {
552 Time::UtcTime(t) => t.to_string(),
553 Time::GeneralTime(t) => t.to_string(),
554 };
555 let py_str = PyString::new(py, &s).unbind();
556 let _ = self.not_before_cache.set(py_str.clone_ref(py));
557 Ok(py_str.into_bound(py))
558 }
559
560 /// notAfter time as a string.
561 #[getter]
562 fn not_after<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
563 if let Some(cached) = self.not_after_cache.get() {
564 return Ok(cached.clone_ref(py).into_bound(py));
565 }
566 let s = match &self.cert()?.tbs_certificate.validity.not_after {
567 Time::UtcTime(t) => t.to_string(),
568 Time::GeneralTime(t) => t.to_string(),
569 };
570 let py_str = PyString::new(py, &s).unbind();
571 let _ = self.not_after_cache.set(py_str.clone_ref(py));
572 Ok(py_str.into_bound(py))
573 }
574
575 /// Subject public-key algorithm name (e.g. "RSA", "ECDSA", "Ed25519", "ML-DSA-65"),
576 /// or the dotted OID notation for unrecognized algorithms.
577 #[getter]
578 fn public_key_algorithm<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
579 if let Some(cached) = self.public_key_algorithm_cache.get() {
580 return Ok(cached.clone_ref(py).into_bound(py));
581 }
582 let oid = &self
583 .cert()?
584 .tbs_certificate
585 .subject_public_key_info
586 .algorithm
587 .algorithm;
588 let s = synta_certificate::identify_public_key_algorithm(oid)
589 .map(|s| s.to_string())
590 .unwrap_or_else(|| oid.to_string());
591 let py_str = PyString::new(py, &s).unbind();
592 let _ = self.public_key_algorithm_cache.set(py_str.clone_ref(py));
593 Ok(py_str.into_bound(py))
594 }
595
596 /// Raw subject public-key bytes.
597 ///
598 /// The `bytes` object is created once on first access and the same Python
599 /// object is returned (by reference) on every subsequent call,
600 /// avoiding repeated allocation and PyO3 string construction.
601 #[getter]
602 fn public_key<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
603 if let Some(cached) = self.public_key_cache.get() {
604 return Ok(cached.clone_ref(py).into_bound(py));
605 }
606 let py_bytes = PyBytes::new(
607 py,
608 self.cert()?
609 .tbs_certificate
610 .subject_public_key_info
611 .subject_public_key
612 .as_bytes(),
613 )
614 .unbind();
615 let _ = self.public_key_cache.set(py_bytes.clone_ref(py));
616 Ok(py_bytes.into_bound(py))
617 }
618
619 /// Version field (0 = v1, 1 = v2, 2 = v3), or None if absent.
620 #[getter]
621 fn version(&self) -> PyResult<Option<i64>> {
622 Ok(self
623 .cert()?
624 .tbs_certificate
625 .version
626 .as_ref()
627 .and_then(|v| v.as_i64().ok()))
628 }
629
630 /// Complete DER encoding of this certificate (the original bytes passed to
631 /// ``from_der``).
632 fn to_der<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
633 self._data.clone_ref(py).into_bound(py)
634 }
635
636 /// Raw DER bytes of the TBSCertificate structure (the bytes that were
637 /// signed by the issuer's key).
638 #[getter]
639 fn tbs_bytes<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
640 self.tbs_bytes_cache
641 .get_or_init(|| {
642 let data = self._data.bind(py).as_bytes();
643 PyBytes::new(py, &data[self.tbs_range.clone()]).unbind()
644 })
645 .clone_ref(py)
646 .into_bound(py)
647 }
648
649 /// OID of the signature algorithm
650 /// (e.g. ``ObjectIdentifier("1.2.840.113549.1.1.11")`` for sha256WithRSAEncryption).
651 ///
652 /// Unlike ``signature_algorithm``, this always returns the machine-readable
653 /// OID, even for algorithms that synta does not recognise by name.
654 #[getter]
655 fn signature_algorithm_oid<'py>(
656 &self,
657 py: Python<'py>,
658 ) -> PyResult<Bound<'py, PyObjectIdentifier>> {
659 if let Some(cached) = self.signature_algorithm_oid_cache.get() {
660 return Ok(cached.clone_ref(py).into_bound(py));
661 }
662 let obj = Py::new(
663 py,
664 PyObjectIdentifier::from_oid(self.cert()?.signature_algorithm.algorithm.clone()),
665 )?;
666 let _ = self.signature_algorithm_oid_cache.set(obj.clone_ref(py));
667 Ok(obj.into_bound(py))
668 }
669
670 /// Raw DER bytes of the signature algorithm parameters, or ``None`` if
671 /// the AlgorithmIdentifier has no parameters field (e.g. Ed25519).
672 #[getter]
673 fn signature_algorithm_params<'py>(
674 &self,
675 py: Python<'py>,
676 ) -> PyResult<Option<Bound<'py, PyBytes>>> {
677 if let Some(cached) = self.signature_algorithm_params_cache.get() {
678 return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
679 }
680 let computed =
681 encode_element_opt(py, self.cert()?.signature_algorithm.parameters.as_ref())?;
682 let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
683 let _ = self.signature_algorithm_params_cache.set(to_store);
684 Ok(computed)
685 }
686
687 /// OID of the subject public-key algorithm
688 /// (e.g. ``ObjectIdentifier("1.2.840.10045.2.1")`` for id-ecPublicKey).
689 #[getter]
690 fn public_key_algorithm_oid<'py>(
691 &self,
692 py: Python<'py>,
693 ) -> PyResult<Bound<'py, PyObjectIdentifier>> {
694 if let Some(cached) = self.public_key_algorithm_oid_cache.get() {
695 return Ok(cached.clone_ref(py).into_bound(py));
696 }
697 let obj = Py::new(
698 py,
699 PyObjectIdentifier::from_oid(
700 self.cert()?
701 .tbs_certificate
702 .subject_public_key_info
703 .algorithm
704 .algorithm
705 .clone(),
706 ),
707 )?;
708 let _ = self.public_key_algorithm_oid_cache.set(obj.clone_ref(py));
709 Ok(obj.into_bound(py))
710 }
711
712 /// Raw DER bytes of the public-key algorithm parameters, or ``None``.
713 ///
714 /// For EC keys this is an OID naming the curve (e.g. secp256r1).
715 /// For RSA and Ed25519 this is typically ``None`` or a NULL element.
716 #[getter]
717 fn public_key_algorithm_params<'py>(
718 &self,
719 py: Python<'py>,
720 ) -> PyResult<Option<Bound<'py, PyBytes>>> {
721 if let Some(cached) = self.public_key_algorithm_params_cache.get() {
722 return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
723 }
724 let computed = encode_element_opt(
725 py,
726 self.cert()?
727 .tbs_certificate
728 .subject_public_key_info
729 .algorithm
730 .parameters
731 .as_ref(),
732 )?;
733 let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
734 let _ = self.public_key_algorithm_params_cache.set(to_store);
735 Ok(computed)
736 }
737
738 /// Raw DER bytes of the extensions SEQUENCE OF, or ``None`` for v1/v2
739 /// certificates that carry no extensions.
740 ///
741 /// The returned bytes begin with the SEQUENCE tag (``0x30``) and contain
742 /// the full SEQUENCE OF Extension encoding. Pass them to a ``Decoder``
743 /// to iterate the individual extensions:
744 ///
745 /// ```python
746 /// ext_der = cert.extensions_der
747 /// if ext_der:
748 /// dec = synta.Decoder(ext_der, synta.Encoding.DER)
749 /// exts_dec = dec.decode_sequence()
750 /// while not exts_dec.is_empty():
751 /// ext_tlv = exts_dec.decode_raw_tlv()
752 /// ```
753 #[getter]
754 fn extensions_der<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
755 if let Some(cached) = self.extensions_der_cache.get() {
756 return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
757 }
758 let computed = self
759 .cert()?
760 .tbs_certificate
761 .extensions
762 .as_ref()
763 .map(|raw: &synta::RawDer<'_>| PyBytes::new(py, raw.as_bytes()));
764 let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
765 let _ = self.extensions_der_cache.set(to_store);
766 Ok(computed)
767 }
768
769 /// Return the DER content of a named extension's value, or ``None``.
770 ///
771 /// Searches the certificate's extension list for an extension whose
772 /// ``extnID`` equals *oid* (dotted-decimal notation, e.g.
773 /// ``"2.5.29.17"`` for SubjectAltName). When found, the bytes inside
774 /// the ``extnValue`` OCTET STRING — the DER-encoded extension-specific
775 /// structure — are returned. Returns ``None`` when the certificate
776 /// carries no extensions or the named OID is absent.
777 ///
778 /// Example — parse SubjectAltName:
779 ///
780 /// ```python
781 /// san_der = cert.get_extension_value_der("2.5.29.17")
782 /// if san_der:
783 /// dec = synta.Decoder(san_der, synta.Encoding.DER)
784 /// san_seq = dec.decode_sequence()
785 /// while not san_seq.is_empty():
786 /// tag_num, tag_class, _ = san_seq.peek_tag()
787 /// child = san_seq.decode_implicit_tag(tag_num, tag_class)
788 /// if tag_num == 2: # dNSName
789 /// print(child.remaining_bytes().decode("ascii"))
790 /// ```
791 fn get_extension_value_der<'py>(
792 &self,
793 py: Python<'py>,
794 oid: &Bound<'_, PyAny>,
795 ) -> PyResult<Option<Bound<'py, PyBytes>>> {
796 let target = oid_from_pyany(oid)?;
797 let raw = match self.cert()?.tbs_certificate.extensions.as_ref() {
798 Some(r) => r,
799 None => return Ok(None),
800 };
801 for ext in &synta_certificate::decode_extensions(raw.as_bytes()) {
802 if ext.extn_id == target {
803 return Ok(Some(PyBytes::new(py, ext.extn_value.as_bytes())));
804 }
805 }
806 Ok(None)
807 }
808
809 /// Return the Subject Alternative Names of this certificate.
810 ///
811 /// Combines looking up the SAN extension (OID ``2.5.29.17``) and parsing
812 /// its ``GeneralName`` entries into a single call. Returns a
813 /// :class:`list` of ``(tag_number: int, content: bytes)`` tuples in
814 /// document order; returns an empty list when the certificate carries no
815 /// SAN extension.
816 ///
817 /// Tag numbers follow RFC 5280 §4.2.1.6 — the same convention as
818 /// :func:`~synta.parse_general_names`. Named constants for every tag are
819 /// available in :mod:`synta.general_name`:
820 ///
821 /// | :attr:`~synta.general_name.RFC822_NAME` (1) | raw IA5String bytes (e-mail) |
822 /// | :attr:`~synta.general_name.DNS_NAME` (2) | raw IA5String bytes |
823 /// | :attr:`~synta.general_name.DIRECTORY_NAME` (4) | complete Name SEQUENCE TLV |
824 /// | :attr:`~synta.general_name.URI` (6) | raw IA5String bytes |
825 /// | :attr:`~synta.general_name.IP_ADDRESS` (7) | 4 bytes (IPv4) or 16 bytes (IPv6) |
826 /// | :attr:`~synta.general_name.REGISTERED_ID` (8) | raw OID value bytes |
827 ///
828 /// ```python,ignore
829 /// import ipaddress
830 /// import synta.general_name as gn
831 ///
832 /// for tag_num, content in cert.subject_alt_names():
833 /// if tag_num == gn.DNS_NAME:
834 /// print("DNS:", content.decode("ascii"))
835 /// elif tag_num == gn.IP_ADDRESS:
836 /// print("IP:", ipaddress.ip_address(content))
837 /// elif tag_num == gn.RFC822_NAME:
838 /// print("email:", content.decode("ascii"))
839 /// elif tag_num == gn.DIRECTORY_NAME:
840 /// attrs = synta.parse_name_attrs(content)
841 /// print("DirName:", attrs)
842 /// elif tag_num == gn.URI:
843 /// print("URI:", content.decode("ascii"))
844 /// ```
845 fn subject_alt_names<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
846 use pyo3::types::{PyBytes, PyList, PyTuple};
847 let list = PyList::empty(py);
848 for (tag_num, content) in self.cert()?.subject_alt_names() {
849 let tuple = PyTuple::new(
850 py,
851 [
852 tag_num.into_pyobject(py)?.into_any(),
853 PyBytes::new(py, &content).into_any(),
854 ],
855 )?;
856 list.append(tuple)?;
857 }
858 Ok(list)
859 }
860
861 /// Raw DER bytes of the issuer Name SEQUENCE.
862 #[getter]
863 fn issuer_raw_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
864 if let Some(cached) = self.issuer_raw_der_cache.get() {
865 return Ok(cached.clone_ref(py).into_bound(py));
866 }
867 let py_bytes = PyBytes::new(py, self.cert()?.tbs_certificate.issuer.as_bytes()).unbind();
868 let _ = self.issuer_raw_der_cache.set(py_bytes.clone_ref(py));
869 Ok(py_bytes.into_bound(py))
870 }
871
872 /// Raw DER bytes of the subject Name SEQUENCE.
873 #[getter]
874 fn subject_raw_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
875 if let Some(cached) = self.subject_raw_der_cache.get() {
876 return Ok(cached.clone_ref(py).into_bound(py));
877 }
878 let py_bytes = PyBytes::new(py, self.cert()?.tbs_certificate.subject.as_bytes()).unbind();
879 let _ = self.subject_raw_der_cache.set(py_bytes.clone_ref(py));
880 Ok(py_bytes.into_bound(py))
881 }
882
883 /// notBefore time as a UTC-aware ``datetime.datetime``.
884 ///
885 /// Equivalent to ``cert.not_valid_before.replace(tzinfo=datetime.timezone.utc)``
886 /// from the ``cryptography`` library. Derives the value from the parsed
887 /// ASN.1 ``Time`` field (``UTCTime`` or ``GeneralizedTime``) without calling
888 /// any Python date-parsing code.
889 ///
890 /// ```python,ignore
891 /// import datetime
892 /// cert = synta.Certificate.from_der(der)
893 /// dt = cert.not_before_utc
894 /// print(dt.isoformat()) # e.g. "2023-01-01T00:00:00+00:00"
895 /// print(dt.tzinfo) # datetime.timezone.utc
896 /// ```
897 #[getter]
898 fn not_before_utc<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
899 if let Some(cached) = self.not_before_utc_cache.get() {
900 return Ok(cached.clone_ref(py).into_bound(py));
901 }
902 let unix_secs = synta_x509_verification::certificate::time_to_unix(
903 &self.cert()?.tbs_certificate.validity.not_before,
904 )
905 .ok_or_else(|| {
906 pyo3::exceptions::PyValueError::new_err("notBefore date is structurally invalid")
907 })?;
908 let datetime_mod = py.import("datetime")?;
909 let utc = datetime_mod.getattr("timezone")?.getattr("utc")?;
910 let dt = datetime_mod
911 .getattr("datetime")?
912 .call_method1("fromtimestamp", (unix_secs, &utc))?
913 .unbind();
914 let _ = self.not_before_utc_cache.set(dt.clone_ref(py));
915 Ok(dt.into_bound(py))
916 }
917
918 /// notAfter time as a UTC-aware ``datetime.datetime``.
919 ///
920 /// Equivalent to ``cert.not_valid_after.replace(tzinfo=datetime.timezone.utc)``
921 /// from the ``cryptography`` library. Derives the value from the parsed
922 /// ASN.1 ``Time`` field (``UTCTime`` or ``GeneralizedTime``) without calling
923 /// any Python date-parsing code.
924 ///
925 /// ```python,ignore
926 /// import datetime
927 /// cert = synta.Certificate.from_der(der)
928 /// dt = cert.not_after_utc
929 /// print(dt.isoformat()) # e.g. "2033-01-01T00:00:00+00:00"
930 /// print(dt.tzinfo) # datetime.timezone.utc
931 /// ```
932 #[getter]
933 fn not_after_utc<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
934 if let Some(cached) = self.not_after_utc_cache.get() {
935 return Ok(cached.clone_ref(py).into_bound(py));
936 }
937 let unix_secs = synta_x509_verification::certificate::time_to_unix(
938 &self.cert()?.tbs_certificate.validity.not_after,
939 )
940 .ok_or_else(|| {
941 pyo3::exceptions::PyValueError::new_err("notAfter date is structurally invalid")
942 })?;
943 let datetime_mod = py.import("datetime")?;
944 let utc = datetime_mod.getattr("timezone")?.getattr("utc")?;
945 let dt = datetime_mod
946 .getattr("datetime")?
947 .call_method1("fromtimestamp", (unix_secs, &utc))?
948 .unbind();
949 let _ = self.not_after_utc_cache.set(dt.clone_ref(py));
950 Ok(dt.into_bound(py))
951 }
952
953 /// Hash algorithm name component of the certificate's outer signature algorithm.
954 ///
955 /// Returns the lowercase hash algorithm name (e.g. ``"sha256"``, ``"sha1"``,
956 /// ``"sha384"``) implied by the signature algorithm OID. Returns ``None``
957 /// for algorithms that do not use a traditional hash (Ed25519, Ed448,
958 /// ML-DSA, etc.).
959 ///
960 /// For RSASSA-PSS the hash is encoded in the parameters field, which is not
961 /// inspected here; ``"sha256"`` is returned as a conservative default.
962 ///
963 /// ```python,ignore
964 /// cert = synta.Certificate.from_der(der)
965 /// hash_name = cert.signature_hash_algorithm_name
966 /// if hash_name:
967 /// print(hash_name) # e.g. "sha256"
968 /// else:
969 /// print("no hash (e.g. Ed25519)")
970 /// ```
971 #[getter]
972 fn signature_hash_algorithm_name(&self) -> PyResult<Option<&'static str>> {
973 use synta_certificate::oids;
974 let oid = &self.cert()?.signature_algorithm.algorithm;
975 let c = oid.components();
976 // RSA PKCS#1 v1.5 variants (1.2.840.113549.1.1.*)
977 if c == oids::SHA1_WITH_RSA {
978 return Ok(Some("sha1"));
979 }
980 if c == oids::SHA256_WITH_RSA {
981 return Ok(Some("sha256"));
982 }
983 if c == oids::SHA384_WITH_RSA {
984 return Ok(Some("sha384"));
985 }
986 if c == oids::SHA512_WITH_RSA {
987 return Ok(Some("sha512"));
988 }
989 // RSA-PSS (hash is in params; return sha256 as safe default)
990 if c == oids::RSASSA_PSS {
991 return Ok(Some("sha256"));
992 }
993 // ECDSA variants
994 if c == oids::ECDSA_WITH_SHA1 {
995 return Ok(Some("sha1"));
996 }
997 if c == oids::ECDSA_WITH_SHA256 {
998 return Ok(Some("sha256"));
999 }
1000 if c == oids::ECDSA_WITH_SHA384 {
1001 return Ok(Some("sha384"));
1002 }
1003 if c == oids::ECDSA_WITH_SHA512 {
1004 return Ok(Some("sha512"));
1005 }
1006 // Ed25519, Ed448, ML-DSA-*, SLH-DSA-* and everything else: no hash
1007 Ok(None)
1008 }
1009
1010 /// Complete DER encoding of the SubjectPublicKeyInfo SEQUENCE.
1011 ///
1012 /// Returns the full ``SubjectPublicKeyInfo`` SEQUENCE TLV as ``bytes``,
1013 /// including the algorithm identifier and the BIT STRING carrying the
1014 /// actual public key. This is the DER encoding that
1015 /// ``cryptography``'s ``public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo)``
1016 /// produces.
1017 ///
1018 /// ```python,ignore
1019 /// cert = synta.Certificate.from_der(der)
1020 /// spki_der = cert.subject_public_key_info_der
1021 /// # Load the public key with cryptography:
1022 /// from cryptography.hazmat.primitives.serialization import load_der_public_key
1023 /// pub_key = load_der_public_key(spki_der)
1024 /// ```
1025 #[getter]
1026 fn subject_public_key_info_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
1027 if let Some(cached) = self.spki_der_cache.get() {
1028 return Ok(cached.clone_ref(py).into_bound(py));
1029 }
1030 let spki = &self.cert()?.tbs_certificate.subject_public_key_info;
1031 let mut enc = synta::Encoder::new(synta::Encoding::Der);
1032 spki.encode(&mut enc)
1033 .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
1034 let bytes = enc
1035 .finish()
1036 .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
1037 let py_bytes = PyBytes::new(py, &bytes).unbind();
1038 let _ = self.spki_der_cache.set(py_bytes.clone_ref(py));
1039 Ok(py_bytes.into_bound(py))
1040 }
1041
1042 /// Raw DER bytes of the ``TBSCertificate`` structure — the bytes that
1043 /// were signed by the issuer's key.
1044 ///
1045 /// This is the DER encoding of the full ``TBSCertificate`` SEQUENCE TLV,
1046 /// suitable for signature verification or as the input to a hash function.
1047 /// Equivalent to ``cryptography``'s ``cert.tbs_certificate_bytes``.
1048 ///
1049 /// ```python,ignore
1050 /// cert = synta.Certificate.from_der(der)
1051 /// tbs = cert.tbs_certificate_der
1052 /// # Verify signature manually:
1053 /// import synta.crypto
1054 /// digest = synta.digest("sha256", tbs)
1055 /// ```
1056 #[getter]
1057 fn tbs_certificate_der<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
1058 self.tbs_bytes(py)
1059 }
1060
1061 /// Verify that this certificate was directly issued by ``issuer``.
1062 ///
1063 /// Checks two things:
1064 ///
1065 /// 1. The ``issuer`` field of this certificate's TBS matches the
1066 /// ``subject`` field of the provided issuer certificate (byte-exact
1067 /// DER comparison).
1068 /// 2. The certificate's signature is valid under the issuer's public key.
1069 ///
1070 /// Raises :exc:`ValueError` if either check fails. This is the synta
1071 /// equivalent of ``cryptography``'s
1072 /// ``cert.verify_directly_issued_by(issuer)``.
1073 ///
1074 /// ```python,ignore
1075 /// root = synta.Certificate.from_pem(open("root.pem", "rb").read())
1076 /// leaf = synta.Certificate.from_pem(open("leaf.pem", "rb").read())
1077 /// leaf.verify_issued_by(root) # raises ValueError if not valid
1078 /// ```
1079 fn verify_issued_by(&self, issuer: &PyCertificate) -> PyResult<()> {
1080 use synta_certificate::{cert_byte_ranges, OpensslSignatureVerifier, SignatureVerifier};
1081
1082 // 1. Name check: our issuer Name must byte-equal their subject Name.
1083 let cert = self.cert()?;
1084 let issuer_cert = issuer.cert()?;
1085 if cert.tbs_certificate.issuer.as_bytes() != issuer_cert.tbs_certificate.subject.as_bytes()
1086 {
1087 return Err(pyo3::exceptions::PyValueError::new_err(
1088 "issuer name does not match subject of provided certificate",
1089 ));
1090 }
1091
1092 // 2. Locate byte ranges within each cert's DER without re-encoding.
1093 let my_ranges = cert_byte_ranges(self.raw).ok_or_else(|| {
1094 pyo3::exceptions::PyValueError::new_err(
1095 "failed to locate byte ranges in certificate DER",
1096 )
1097 })?;
1098 let issuer_ranges = cert_byte_ranges(issuer.raw).ok_or_else(|| {
1099 pyo3::exceptions::PyValueError::new_err(
1100 "failed to locate byte ranges in issuer certificate DER",
1101 )
1102 })?;
1103
1104 // 3. Signature bits (raw bytes, no unused-bits prefix).
1105 let signature_bits = cert.signature_value.as_bytes();
1106
1107 // 4. Verify.
1108 OpensslSignatureVerifier
1109 .verify_certificate_signature(
1110 &self.raw[my_ranges.tbs],
1111 &self.raw[my_ranges.signature_algorithm],
1112 signature_bits,
1113 &issuer.raw[issuer_ranges.subject_public_key_info],
1114 )
1115 .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
1116
1117 Ok(())
1118 }
1119
1120 /// Compute a hash fingerprint of the complete certificate DER.
1121 ///
1122 /// ``algorithm`` must be one of ``"sha1"``, ``"sha224"``, ``"sha256"``,
1123 /// ``"sha384"``, ``"sha512"``, or ``"md5"``. Raises :exc:`ValueError` for
1124 /// unknown algorithm names.
1125 ///
1126 /// ```python,ignore
1127 /// cert = synta.Certificate.from_der(der)
1128 /// fp = cert.fingerprint("sha256")
1129 /// print(fp.hex()) # e.g. "3a2b1c..."
1130 /// ```
1131 fn fingerprint<'py>(&self, py: Python<'py>, algorithm: &str) -> PyResult<Bound<'py, PyBytes>> {
1132 use synta_certificate::{default_data_hasher, DataHasher};
1133 let digest = default_data_hasher()
1134 .hash_data(algorithm, self.raw)
1135 .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
1136 Ok(PyBytes::new(py, &digest))
1137 }
1138
1139 fn __repr__(&self) -> PyResult<String> {
1140 let cert = self.cert()?;
1141 let serial = &cert.tbs_certificate.serial_number;
1142 let serial_str = if let Ok(v) = serial.as_i64() {
1143 v.to_string()
1144 } else {
1145 format!("<{} bytes>", serial.as_bytes().len())
1146 };
1147 Ok(format!(
1148 "Certificate(subject={:?}, serial={})",
1149 decode_name_debug(cert.tbs_certificate.subject.as_bytes()),
1150 serial_str,
1151 ))
1152 }
1153}
1154
1155impl PyCertificate {
1156 /// Parse a DER-encoded X.509 certificate (internal helper visible to sibling modules).
1157 ///
1158 /// This is the same logic as the `#[staticmethod] from_der` exposed to Python but
1159 /// accessible from Rust sibling modules via `pub(super)` visibility.
1160 pub(crate) fn new_from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
1161 Self::from_der(py, data)
1162 }
1163}