_synta/certificate/
ac.rs

1//! Python bindings for RFC 5755 Attribute Certificate types.
2//!
3//! Exposes ``AttributeCertificate`` as a Python class and installs OID constants
4//! into the ``synta.ac`` submodule.
5
6use std::sync::OnceLock;
7
8use pyo3::prelude::*;
9use pyo3::types::{PyBytes, PyString};
10
11use synta::traits::Encode;
12use synta::{Decoder, Encoding};
13
14use crate::error::SyntaErr;
15use crate::types::PyObjectIdentifier;
16
17// ── helpers ───────────────────────────────────────────────────────────────────
18
19/// Encode an arbitrary `T: Encode` value to DER bytes.
20fn encode_to_der<T: Encode>(v: &T) -> Vec<u8> {
21    let mut enc = synta::Encoder::new(Encoding::Der);
22    if v.encode(&mut enc).is_err() {
23        return Vec::new();
24    }
25    enc.finish().unwrap_or_default()
26}
27
28// ── PyAttributeCertificate ────────────────────────────────────────────────────
29
30/// X.509 Attribute Certificate v2 (RFC 5755).
31///
32/// An Attribute Certificate (AC) binds a set of attributes (roles, clearances,
33/// service-authentication information) to a holder identified by reference to
34/// their Public Key Certificate (PKC), without requiring re-issuance of the PKC.
35///
36/// ```python,ignore
37/// import synta.ac as ac
38/// acer = ac.AttributeCertificate.from_der(open("attr.ac", "rb").read())
39/// print(acer.serial_number.hex())
40/// print(acer.not_before, "–", acer.not_after)
41/// print(acer.signature_algorithm_oid)
42/// ```
43#[pyclass(frozen, name = "AttributeCertificate")]
44pub struct PyAttributeCertificate {
45    _data: Py<PyBytes>,
46    raw: &'static [u8],
47    inner: OnceLock<Box<synta_certificate::attribute_cert_types::AttributeCertificate<'static>>>,
48    // Field caches
49    serial_number_cache: OnceLock<Py<PyBytes>>,
50    not_before_cache: OnceLock<Py<PyString>>,
51    not_after_cache: OnceLock<Py<PyString>>,
52    signature_algorithm_oid_cache: OnceLock<Py<PyObjectIdentifier>>,
53    signature_cache: OnceLock<Py<PyBytes>>,
54    holder_der_cache: OnceLock<Py<PyBytes>>,
55    issuer_der_cache: OnceLock<Py<PyBytes>>,
56    attributes_der_cache: OnceLock<Py<PyBytes>>,
57}
58
59impl PyAttributeCertificate {
60    fn ac(
61        &self,
62    ) -> PyResult<&synta_certificate::attribute_cert_types::AttributeCertificate<'static>> {
63        if let Some(v) = self.inner.get() {
64            return Ok(v.as_ref());
65        }
66        let mut dec = Decoder::new(self.raw, Encoding::Der);
67        let decoded = dec
68            .decode::<synta_certificate::attribute_cert_types::AttributeCertificate<'_>>()
69            .map_err(SyntaErr)?;
70        // SAFETY: raw is pinned for the lifetime of self (kept alive by _data).
71        let decoded: synta_certificate::attribute_cert_types::AttributeCertificate<'static> =
72            unsafe { std::mem::transmute(decoded) };
73        let _ = self.inner.set(Box::new(decoded));
74        Ok(self.inner.get().unwrap().as_ref())
75    }
76}
77
78#[pymethods]
79impl PyAttributeCertificate {
80    /// Parse a DER-encoded ``AttributeCertificate`` SEQUENCE.
81    ///
82    /// :param data: DER bytes of the ``AttributeCertificate``.
83    /// :raises ValueError: if the bytes cannot be decoded.
84    #[staticmethod]
85    fn from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
86        let py_bytes = data.unbind();
87        {
88            let raw = py_bytes.as_bytes(py);
89            Decoder::new(raw, Encoding::Der)
90                .decode::<synta_certificate::attribute_cert_types::AttributeCertificate<'_>>()
91                .map_err(SyntaErr)?;
92        }
93        let raw: &'static [u8] = unsafe { std::mem::transmute(py_bytes.as_bytes(py)) };
94        Ok(Self {
95            _data: py_bytes,
96            raw,
97            inner: OnceLock::new(),
98            serial_number_cache: OnceLock::new(),
99            not_before_cache: OnceLock::new(),
100            not_after_cache: OnceLock::new(),
101            signature_algorithm_oid_cache: OnceLock::new(),
102            signature_cache: OnceLock::new(),
103            holder_der_cache: OnceLock::new(),
104            issuer_der_cache: OnceLock::new(),
105            attributes_der_cache: OnceLock::new(),
106        })
107    }
108
109    /// Return the DER encoding of this ``AttributeCertificate``.
110    fn to_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
111        Ok(PyBytes::new(py, &self.ac()?.to_der().map_err(SyntaErr)?))
112    }
113
114    /// Parse the first ``ATTRIBUTE CERTIFICATE`` PEM block from ``data``.
115    ///
116    /// ```python,ignore
117    /// import synta.ac as ac
118    /// acer = ac.AttributeCertificate.from_pem(open("attr.pem", "rb").read())
119    /// print(acer.serial_number.hex())
120    /// ```
121    ///
122    /// :raises ValueError: if no valid PEM block is found or the DER is invalid.
123    #[staticmethod]
124    fn from_pem(py: Python<'_>, data: &[u8]) -> PyResult<Self> {
125        let blocks = synta_certificate::pem_blocks(data);
126        let der = match blocks.as_slice() {
127            [] => {
128                return Err(pyo3::exceptions::PyValueError::new_err(
129                    "no PEM block found in input",
130                ))
131            }
132            [(_, first), ..] => first,
133        };
134        let py_bytes = pyo3::types::PyBytes::new(py, der).unbind();
135        {
136            let raw = py_bytes.as_bytes(py);
137            synta::Decoder::new(raw, Encoding::Der)
138                .decode::<synta_certificate::attribute_cert_types::AttributeCertificate<'_>>()
139                .map_err(SyntaErr)?;
140        }
141        let raw: &'static [u8] = unsafe { std::mem::transmute(py_bytes.as_bytes(py)) };
142        Ok(Self {
143            _data: py_bytes,
144            raw,
145            inner: OnceLock::new(),
146            serial_number_cache: OnceLock::new(),
147            not_before_cache: OnceLock::new(),
148            not_after_cache: OnceLock::new(),
149            signature_algorithm_oid_cache: OnceLock::new(),
150            signature_cache: OnceLock::new(),
151            holder_der_cache: OnceLock::new(),
152            issuer_der_cache: OnceLock::new(),
153            attributes_der_cache: OnceLock::new(),
154        })
155    }
156
157    /// Return the PEM encoding of this ``AttributeCertificate``.
158    ///
159    /// ```python,ignore
160    /// import synta.ac as ac
161    /// acer = ac.AttributeCertificate.from_der(open("attr.ac", "rb").read())
162    /// open("attr.pem", "wb").write(acer.to_pem())
163    /// ```
164    fn to_pem<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
165        let pem = synta_certificate::der_to_pem("ATTRIBUTE CERTIFICATE", self.raw);
166        PyBytes::new(py, &pem)
167    }
168
169    /// ``AttCertVersion`` integer (always ``1`` for v2 per RFC 5755).
170    #[getter]
171    fn version(&self) -> PyResult<i64> {
172        Ok(self.ac()?.acinfo.version.as_i64().unwrap_or(1))
173    }
174
175    /// Certificate serial number as big-endian bytes.
176    #[getter]
177    fn serial_number<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
178        if let Some(c) = self.serial_number_cache.get() {
179            return Ok(c.clone_ref(py).into_bound(py));
180        }
181        let b = PyBytes::new(py, self.ac()?.acinfo.serial_number.as_bytes());
182        let _ = self.serial_number_cache.set(b.as_unbound().clone_ref(py));
183        Ok(b)
184    }
185
186    /// Validity period start time (GeneralizedTime string, e.g. ``"20240101120000Z"``).
187    #[getter]
188    fn not_before<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
189        if let Some(c) = self.not_before_cache.get() {
190            return Ok(c.clone_ref(py).into_bound(py));
191        }
192        let s = self
193            .ac()?
194            .acinfo
195            .attr_cert_validity_period
196            .not_before_time
197            .to_string();
198        let ps = PyString::new(py, &s);
199        let _ = self.not_before_cache.set(ps.as_unbound().clone_ref(py));
200        Ok(ps)
201    }
202
203    /// Validity period end time (GeneralizedTime string).
204    #[getter]
205    fn not_after<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
206        if let Some(c) = self.not_after_cache.get() {
207            return Ok(c.clone_ref(py).into_bound(py));
208        }
209        let s = self
210            .ac()?
211            .acinfo
212            .attr_cert_validity_period
213            .not_after_time
214            .to_string();
215        let ps = PyString::new(py, &s);
216        let _ = self.not_after_cache.set(ps.as_unbound().clone_ref(py));
217        Ok(ps)
218    }
219
220    /// Signature algorithm OID.
221    #[getter]
222    fn signature_algorithm_oid(&self, py: Python<'_>) -> PyResult<Py<PyObjectIdentifier>> {
223        if let Some(c) = self.signature_algorithm_oid_cache.get() {
224            return Ok(c.clone_ref(py));
225        }
226        let oid = self.ac()?.acinfo.signature.algorithm.clone();
227        let obj = Py::new(py, PyObjectIdentifier::from_oid(oid))?;
228        let _ = self.signature_algorithm_oid_cache.set(obj.clone_ref(py));
229        Ok(obj)
230    }
231
232    /// Raw signature bytes (the bit-string value, zero-byte padding stripped).
233    #[getter]
234    fn signature<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
235        if let Some(c) = self.signature_cache.get() {
236            return Ok(c.clone_ref(py).into_bound(py));
237        }
238        let b = PyBytes::new(py, self.ac()?.signature.as_bytes());
239        let _ = self.signature_cache.set(b.as_unbound().clone_ref(py));
240        Ok(b)
241    }
242
243    /// Raw DER bytes of the ``Holder`` SEQUENCE (for re-decoding or inspection).
244    #[getter]
245    fn holder_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
246        if let Some(c) = self.holder_der_cache.get() {
247            return Ok(c.clone_ref(py).into_bound(py));
248        }
249        let der = encode_to_der(&self.ac()?.acinfo.holder);
250        let b = PyBytes::new(py, &der);
251        let _ = self.holder_der_cache.set(b.as_unbound().clone_ref(py));
252        Ok(b)
253    }
254
255    /// Raw DER bytes of the ``AttCertIssuer`` CHOICE (for re-decoding or inspection).
256    #[getter]
257    fn issuer_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
258        if let Some(c) = self.issuer_der_cache.get() {
259            return Ok(c.clone_ref(py).into_bound(py));
260        }
261        let der = encode_to_der(&self.ac()?.acinfo.issuer);
262        let b = PyBytes::new(py, &der);
263        let _ = self.issuer_der_cache.set(b.as_unbound().clone_ref(py));
264        Ok(b)
265    }
266
267    /// Raw DER bytes of the ``SEQUENCE OF Attribute`` attributes field.
268    ///
269    /// Each ``Attribute`` in the sequence can carry roles, clearances,
270    /// or service-authentication information.  Re-decode with a ``Decoder``
271    /// to inspect individual attributes.
272    #[getter]
273    fn attributes_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
274        if let Some(c) = self.attributes_der_cache.get() {
275            return Ok(c.clone_ref(py).into_bound(py));
276        }
277        let der = encode_to_der(&self.ac()?.acinfo.attributes);
278        let b = PyBytes::new(py, &der);
279        let _ = self.attributes_der_cache.set(b.as_unbound().clone_ref(py));
280        Ok(b)
281    }
282
283    /// Verify that this Attribute Certificate was signed by ``issuer``.
284    ///
285    /// Verifies the outer signature against the issuer's public key.  Note that
286    /// AC issuers are represented as ``AttCertIssuer`` (a GeneralNames CHOICE),
287    /// not a plain Name, so no issuer-to-subject name matching is performed
288    /// here — only the cryptographic signature is checked.
289    ///
290    /// :raises ValueError: if the AC carries no ``responseBytes``, or the
291    ///     signature is invalid.
292    ///
293    /// ```python,ignore
294    /// import synta.ac as ac
295    /// ca_cert = synta.Certificate.from_pem(open("ca.pem", "rb").read())
296    /// acer = ac.AttributeCertificate.from_der(open("attr.ac", "rb").read())
297    /// acer.verify_issued_by(ca_cert)   # raises ValueError if not valid
298    /// ```
299    fn verify_issued_by(&self, issuer: &super::cert::PyCertificate) -> PyResult<()> {
300        use synta_certificate::{default_signature_verifier, SignatureVerifier};
301
302        let ac = self.ac()?;
303        let issuer_cert = issuer.cert()?;
304
305        // Re-encode AttributeCertificateInfo (TBS).
306        let tbs_der = encode_to_der(&ac.acinfo);
307        if tbs_der.is_empty() {
308            return Err(pyo3::exceptions::PyValueError::new_err(
309                "failed to encode AttributeCertificateInfo",
310            ));
311        }
312
313        // Re-encode outer signatureAlgorithm.
314        let sig_alg_der = encode_to_der(&ac.signature_algorithm);
315        if sig_alg_der.is_empty() {
316            return Err(pyo3::exceptions::PyValueError::new_err(
317                "failed to encode AC signatureAlgorithm",
318            ));
319        }
320
321        // Re-encode issuer SPKI.
322        let spki_der = encode_to_der(&issuer_cert.tbs_certificate.subject_public_key_info);
323        if spki_der.is_empty() {
324            return Err(pyo3::exceptions::PyValueError::new_err(
325                "failed to encode issuer SubjectPublicKeyInfo",
326            ));
327        }
328
329        // Verify using the backend-agnostic verifier.
330        default_signature_verifier()
331            .verify_certificate_signature(
332                &tbs_der,
333                &sig_alg_der,
334                ac.signature.as_bytes(),
335                &spki_der,
336            )
337            .map_err(|e| {
338                pyo3::exceptions::PyValueError::new_err(format!("AC signature invalid: {e}"))
339            })
340    }
341
342    fn __repr__(&self) -> PyResult<String> {
343        let ac = self.ac()?;
344        Ok(format!(
345            "AttributeCertificate(serial={})",
346            ac.acinfo
347                .serial_number
348                .as_bytes()
349                .iter()
350                .map(|b| format!("{b:02x}"))
351                .collect::<String>(),
352        ))
353    }
354}
355
356// ── register_ac_submodule ─────────────────────────────────────────────────────
357
358/// Build and register the ``synta.ac`` submodule.
359pub(super) fn register_ac_submodule(parent: &Bound<'_, PyModule>) -> PyResult<()> {
360    let py = parent.py();
361    let m = PyModule::new(py, "ac")?;
362
363    m.add_class::<PyAttributeCertificate>()?;
364    m.add_class::<PyAttributeCertificateBuilder>()?;
365
366    // ── RFC 5755 OIDs ─────────────────────────────────────────────────────────
367    m.add(
368        "ID_PE_AC_AUDIT_IDENTITY",
369        super::oid_const(
370            py,
371            synta_certificate::attribute_cert_types::ID_PE_AC_AUDIT_IDENTITY,
372        ),
373    )?;
374    m.add(
375        "ID_PE_AA_CONTROLS",
376        super::oid_const(
377            py,
378            synta_certificate::attribute_cert_types::ID_PE_AA_CONTROLS,
379        ),
380    )?;
381    m.add(
382        "ID_PE_AC_PROXYING",
383        super::oid_const(
384            py,
385            synta_certificate::attribute_cert_types::ID_PE_AC_PROXYING,
386        ),
387    )?;
388    m.add(
389        "ID_CE_TARGET_INFORMATION",
390        super::oid_const(
391            py,
392            synta_certificate::attribute_cert_types::ID_CE_TARGET_INFORMATION,
393        ),
394    )?;
395    m.add(
396        "ID_ACA_AUTHENTICATION_INFO",
397        super::oid_const(
398            py,
399            synta_certificate::attribute_cert_types::ID_ACA_AUTHENTICATION_INFO,
400        ),
401    )?;
402    m.add(
403        "ID_ACA_ACCESS_IDENTITY",
404        super::oid_const(
405            py,
406            synta_certificate::attribute_cert_types::ID_ACA_ACCESS_IDENTITY,
407        ),
408    )?;
409    m.add(
410        "ID_ACA_CHARGING_IDENTITY",
411        super::oid_const(
412            py,
413            synta_certificate::attribute_cert_types::ID_ACA_CHARGING_IDENTITY,
414        ),
415    )?;
416    m.add(
417        "ID_ACA_GROUP",
418        super::oid_const(py, synta_certificate::attribute_cert_types::ID_ACA_GROUP),
419    )?;
420    m.add(
421        "ID_ACA_ENC_ATTRS",
422        super::oid_const(
423            py,
424            synta_certificate::attribute_cert_types::ID_ACA_ENC_ATTRS,
425        ),
426    )?;
427    m.add(
428        "ID_AT_ROLE",
429        super::oid_const(py, synta_certificate::attribute_cert_types::ID_AT_ROLE),
430    )?;
431    m.add(
432        "ID_AT_CLEARANCE",
433        super::oid_const(py, synta_certificate::attribute_cert_types::ID_AT_CLEARANCE),
434    )?;
435
436    crate::install_submodule(
437        parent,
438        &m,
439        "synta.ac",
440        Some(concat!(
441            "synta.ac — RFC 5755 Attribute Certificate v2 types.\n\n",
442            "Provides AttributeCertificate for decoding X.509 Attribute\n",
443            "Certificates that bind roles, clearances, or service-auth\n",
444            "attributes to a holder's PKC, along with OID constants for\n",
445            "RFC 5755 extensions and attribute types.",
446        )),
447    )
448}
449
450// ── PyAttributeCertificateBuilder ────────────────────────────────────────────
451
452/// Builder for RFC 5755 Attribute Certificate TBS encoding.
453///
454/// Use the fluent setter methods to configure the certificate, then call
455/// :meth:`build` to obtain the DER-encoded ``AttributeCertificateInfo``
456/// (TBS) SEQUENCE.
457///
458/// ```python,ignore
459/// import synta.ac as ac
460///
461/// tbs_der = (
462///     ac.AttributeCertificateBuilder()
463///     .serial_number(42)
464///     .not_before("20240101120000Z")
465///     .not_after("20250101120000Z")
466///     .issuer_rfc822("ca@example.com")
467///     .holder_entity_name_rfc822("user@example.com")
468///     .build()
469/// )
470/// ```
471#[pyclass(name = "AttributeCertificateBuilder")]
472pub struct PyAttributeCertificateBuilder {
473    inner: synta_certificate::AttributeCertificateBuilder,
474}
475
476#[pymethods]
477impl PyAttributeCertificateBuilder {
478    /// Create a new, empty ``AttributeCertificateBuilder``.
479    #[new]
480    fn new() -> Self {
481        Self {
482            inner: synta_certificate::AttributeCertificateBuilder::new(),
483        }
484    }
485
486    /// Set the AC serial number.
487    ///
488    /// ```python,ignore
489    /// builder.serial_number(42)
490    /// ```
491    fn serial_number<'py>(slf: Bound<'py, Self>, n: i64) -> Bound<'py, Self> {
492        let old = std::mem::replace(
493            &mut slf.borrow_mut().inner,
494            synta_certificate::AttributeCertificateBuilder::new(),
495        );
496        slf.borrow_mut().inner = old.serial_number(n);
497        slf
498    }
499
500    /// Set the validity start time as a GeneralizedTime string (``"YYYYMMDDHHmmssZ"``).
501    ///
502    /// :raises ValueError: if the string is not a valid GeneralizedTime.
503    ///
504    /// ```python,ignore
505    /// builder.not_before("20240101120000Z")
506    /// ```
507    fn not_before<'py>(slf: Bound<'py, Self>, s: &str) -> Bound<'py, Self> {
508        let old = std::mem::replace(
509            &mut slf.borrow_mut().inner,
510            synta_certificate::AttributeCertificateBuilder::new(),
511        );
512        slf.borrow_mut().inner = old.not_before(s);
513        slf
514    }
515
516    /// Set the validity end time as a GeneralizedTime string (``"YYYYMMDDHHmmssZ"``).
517    ///
518    /// :raises ValueError: if the string is not a valid GeneralizedTime.
519    ///
520    /// ```python,ignore
521    /// builder.not_after("20250101120000Z")
522    /// ```
523    fn not_after<'py>(slf: Bound<'py, Self>, s: &str) -> Bound<'py, Self> {
524        let old = std::mem::replace(
525            &mut slf.borrow_mut().inner,
526            synta_certificate::AttributeCertificateBuilder::new(),
527        );
528        slf.borrow_mut().inner = old.not_after(s);
529        slf
530    }
531
532    /// Add an ``rfc822Name`` GeneralName to the ``AttCertIssuer`` ``v1Form``.
533    ///
534    /// ```python,ignore
535    /// builder.issuer_rfc822("ca@example.com")
536    /// ```
537    fn issuer_rfc822<'py>(slf: Bound<'py, Self>, email: &str) -> Bound<'py, Self> {
538        let old = std::mem::replace(
539            &mut slf.borrow_mut().inner,
540            synta_certificate::AttributeCertificateBuilder::new(),
541        );
542        slf.borrow_mut().inner = old.issuer_rfc822(email);
543        slf
544    }
545
546    /// Add a ``dNSName`` GeneralName to the ``AttCertIssuer`` ``v1Form``.
547    ///
548    /// ```python,ignore
549    /// builder.issuer_dns("ca.example.com")
550    /// ```
551    fn issuer_dns<'py>(slf: Bound<'py, Self>, name: &str) -> Bound<'py, Self> {
552        let old = std::mem::replace(
553            &mut slf.borrow_mut().inner,
554            synta_certificate::AttributeCertificateBuilder::new(),
555        );
556        slf.borrow_mut().inner = old.issuer_dns(name);
557        slf
558    }
559
560    /// Add an ``rfc822Name`` GeneralName to the ``Holder.entityName``.
561    ///
562    /// ```python,ignore
563    /// builder.holder_entity_name_rfc822("user@example.com")
564    /// ```
565    fn holder_entity_name_rfc822<'py>(slf: Bound<'py, Self>, email: &str) -> Bound<'py, Self> {
566        let old = std::mem::replace(
567            &mut slf.borrow_mut().inner,
568            synta_certificate::AttributeCertificateBuilder::new(),
569        );
570        slf.borrow_mut().inner = old.holder_entity_name_rfc822(email);
571        slf
572    }
573
574    /// Encode the ``AttributeCertificateInfo`` SEQUENCE to DER bytes.
575    ///
576    /// :raises ValueError: if any required field is missing or encoding fails.
577    ///
578    /// ```python,ignore
579    /// tbs_der = builder.build()
580    /// ```
581    fn build<'py>(&mut self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
582        let inner = std::mem::replace(
583            &mut self.inner,
584            synta_certificate::AttributeCertificateBuilder::new(),
585        );
586        let der = inner
587            .build()
588            .map_err(pyo3::exceptions::PyValueError::new_err)?;
589        Ok(PyBytes::new(py, &der))
590    }
591
592    fn __repr__(&self) -> String {
593        "AttributeCertificateBuilder()".to_string()
594    }
595}
_synta/certificate/ac.rs

_synta/certificate/
ac.rs
