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

_synta/certificate/
ac.rs
