_synta/certificate/

ac.rs

//! Python bindings for RFC 5755 Attribute Certificate types.
//!
//! Exposes ``AttributeCertificate`` as a Python class and installs OID constants
//! into the ``synta.ac`` submodule.

use std::sync::OnceLock;

use pyo3::prelude::*;
use pyo3::types::{PyBytes, PyString};

use synta::traits::Encode;
use synta::{Decoder, Encoding};

use crate::error::SyntaErr;
use crate::types::PyObjectIdentifier;

// ── helpers ───────────────────────────────────────────────────────────────────

/// Encode an arbitrary `T: Encode` value to DER bytes.
fn encode_to_der<T: Encode>(v: &T) -> Vec<u8> {
    let mut enc = synta::Encoder::new(Encoding::Der);
    if v.encode(&mut enc).is_err() {
        return Vec::new();
    }
    enc.finish().unwrap_or_default()
}

// ── PyAttributeCertificate ────────────────────────────────────────────────────

/// X.509 Attribute Certificate v2 (RFC 5755).
///
/// An Attribute Certificate (AC) binds a set of attributes (roles, clearances,
/// service-authentication information) to a holder identified by reference to
/// their Public Key Certificate (PKC), without requiring re-issuance of the PKC.
///
/// ```python,ignore
/// import synta.ac as ac
/// acer = ac.AttributeCertificate.from_der(open("attr.ac", "rb").read())
/// print(acer.serial_number.hex())
/// print(acer.not_before, "–", acer.not_after)
/// print(acer.signature_algorithm_oid)
/// ```
#[pyclass(frozen, name = "AttributeCertificate")]
pub struct PyAttributeCertificate {
    _data: Py<PyBytes>,
    raw: &'static [u8],
    inner: OnceLock<Box<synta_certificate::attribute_cert_types::AttributeCertificate<'static>>>,
    // Field caches
    serial_number_cache: OnceLock<Py<PyBytes>>,
    not_before_cache: OnceLock<Py<PyString>>,
    not_after_cache: OnceLock<Py<PyString>>,
    signature_algorithm_oid_cache: OnceLock<Py<PyObjectIdentifier>>,
    signature_cache: OnceLock<Py<PyBytes>>,
    holder_der_cache: OnceLock<Py<PyBytes>>,
    issuer_der_cache: OnceLock<Py<PyBytes>>,
    attributes_der_cache: OnceLock<Py<PyBytes>>,
}

impl PyAttributeCertificate {
    fn ac(
        &self,
    ) -> PyResult<&synta_certificate::attribute_cert_types::AttributeCertificate<'static>> {
        if let Some(v) = self.inner.get() {
            return Ok(v.as_ref());
        }
        let mut dec = Decoder::new(self.raw, Encoding::Der);
        let decoded = dec
            .decode::<synta_certificate::attribute_cert_types::AttributeCertificate<'_>>()
            .map_err(SyntaErr)?;
        // SAFETY: raw is pinned for the lifetime of self (kept alive by _data).
        let decoded: synta_certificate::attribute_cert_types::AttributeCertificate<'static> =
            unsafe { std::mem::transmute(decoded) };
        let _ = self.inner.set(Box::new(decoded));
        Ok(self.inner.get().unwrap().as_ref())
    }
}

#[pymethods]
impl PyAttributeCertificate {
    /// Parse a DER-encoded ``AttributeCertificate`` SEQUENCE.
    ///
    /// :param data: DER bytes of the ``AttributeCertificate``.
    /// :raises ValueError: if the bytes cannot be decoded.
    #[staticmethod]
    fn from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
        let py_bytes = data.unbind();
        {
            let raw = py_bytes.as_bytes(py);
            Decoder::new(raw, Encoding::Der)
                .decode::<synta_certificate::attribute_cert_types::AttributeCertificate<'_>>()
                .map_err(SyntaErr)?;
        }
        let raw: &'static [u8] = unsafe { std::mem::transmute(py_bytes.as_bytes(py)) };
        Ok(Self {
            _data: py_bytes,
            raw,
            inner: OnceLock::new(),
            serial_number_cache: OnceLock::new(),
            not_before_cache: OnceLock::new(),
            not_after_cache: OnceLock::new(),
            signature_algorithm_oid_cache: OnceLock::new(),
            signature_cache: OnceLock::new(),
            holder_der_cache: OnceLock::new(),
            issuer_der_cache: OnceLock::new(),
            attributes_der_cache: OnceLock::new(),
        })
    }

    /// Return the DER encoding of this ``AttributeCertificate``.
    fn to_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        let mut enc = synta::Encoder::new(Encoding::Der);
        self.ac()?.encode(&mut enc).map_err(SyntaErr)?;
        Ok(PyBytes::new(py, &enc.finish().map_err(SyntaErr)?))
    }

    /// ``AttCertVersion`` integer (always ``1`` for v2 per RFC 5755).
    #[getter]
    fn version(&self) -> PyResult<i64> {
        Ok(self.ac()?.acinfo.version.as_i64().unwrap_or(1))
    }

    /// Certificate serial number as big-endian bytes.
    #[getter]
    fn serial_number<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        if let Some(c) = self.serial_number_cache.get() {
            return Ok(c.clone_ref(py).into_bound(py));
        }
        let b = PyBytes::new(py, self.ac()?.acinfo.serial_number.as_bytes());
        let _ = self.serial_number_cache.set(b.as_unbound().clone_ref(py));
        Ok(b)
    }

    /// Validity period start time (GeneralizedTime string, e.g. ``"20240101120000Z"``).
    #[getter]
    fn not_before<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
        if let Some(c) = self.not_before_cache.get() {
            return Ok(c.clone_ref(py).into_bound(py));
        }
        let s = self
            .ac()?
            .acinfo
            .attr_cert_validity_period
            .not_before_time
            .to_string();
        let ps = PyString::new(py, &s);
        let _ = self.not_before_cache.set(ps.as_unbound().clone_ref(py));
        Ok(ps)
    }

    /// Validity period end time (GeneralizedTime string).
    #[getter]
    fn not_after<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyString>> {
        if let Some(c) = self.not_after_cache.get() {
            return Ok(c.clone_ref(py).into_bound(py));
        }
        let s = self
            .ac()?
            .acinfo
            .attr_cert_validity_period
            .not_after_time
            .to_string();
        let ps = PyString::new(py, &s);
        let _ = self.not_after_cache.set(ps.as_unbound().clone_ref(py));
        Ok(ps)
    }

    /// Signature algorithm OID.
    #[getter]
    fn signature_algorithm_oid(&self, py: Python<'_>) -> PyResult<Py<PyObjectIdentifier>> {
        if let Some(c) = self.signature_algorithm_oid_cache.get() {
            return Ok(c.clone_ref(py));
        }
        let oid = self.ac()?.acinfo.signature.algorithm.clone();
        let obj = Py::new(py, PyObjectIdentifier::from_oid(oid))?;
        let _ = self.signature_algorithm_oid_cache.set(obj.clone_ref(py));
        Ok(obj)
    }

    /// Raw signature bytes (the bit-string value, zero-byte padding stripped).
    #[getter]
    fn signature<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        if let Some(c) = self.signature_cache.get() {
            return Ok(c.clone_ref(py).into_bound(py));
        }
        let b = PyBytes::new(py, self.ac()?.signature.as_bytes());
        let _ = self.signature_cache.set(b.as_unbound().clone_ref(py));
        Ok(b)
    }

    /// Raw DER bytes of the ``Holder`` SEQUENCE (for re-decoding or inspection).
    #[getter]
    fn holder_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        if let Some(c) = self.holder_der_cache.get() {
            return Ok(c.clone_ref(py).into_bound(py));
        }
        let der = encode_to_der(&self.ac()?.acinfo.holder);
        let b = PyBytes::new(py, &der);
        let _ = self.holder_der_cache.set(b.as_unbound().clone_ref(py));
        Ok(b)
    }

    /// Raw DER bytes of the ``AttCertIssuer`` CHOICE (for re-decoding or inspection).
    #[getter]
    fn issuer_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        if let Some(c) = self.issuer_der_cache.get() {
            return Ok(c.clone_ref(py).into_bound(py));
        }
        let der = encode_to_der(&self.ac()?.acinfo.issuer);
        let b = PyBytes::new(py, &der);
        let _ = self.issuer_der_cache.set(b.as_unbound().clone_ref(py));
        Ok(b)
    }

    /// Raw DER bytes of the ``SEQUENCE OF Attribute`` attributes field.
    ///
    /// Each ``Attribute`` in the sequence can carry roles, clearances,
    /// or service-authentication information.  Re-decode with a ``Decoder``
    /// to inspect individual attributes.
    #[getter]
    fn attributes_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        if let Some(c) = self.attributes_der_cache.get() {
            return Ok(c.clone_ref(py).into_bound(py));
        }
        let der = encode_to_der(&self.ac()?.acinfo.attributes);
        let b = PyBytes::new(py, &der);
        let _ = self.attributes_der_cache.set(b.as_unbound().clone_ref(py));
        Ok(b)
    }

    fn __repr__(&self) -> PyResult<String> {
        let ac = self.ac()?;
        Ok(format!(
            "AttributeCertificate(serial={})",
            ac.acinfo
                .serial_number
                .as_bytes()
                .iter()
                .map(|b| format!("{b:02x}"))
                .collect::<String>(),
        ))
    }
}

// ── register_ac_submodule ─────────────────────────────────────────────────────

/// Build and register the ``synta.ac`` submodule.
pub(super) fn register_ac_submodule(parent: &Bound<'_, PyModule>) -> PyResult<()> {
    let py = parent.py();
    let m = PyModule::new(py, "ac")?;

    m.add_class::<PyAttributeCertificate>()?;

    // ── RFC 5755 OIDs ─────────────────────────────────────────────────────────
    m.add(
        "ID_PE_AC_AUDIT_IDENTITY",
        super::oid_const(
            py,
            synta_certificate::attribute_cert_types::ID_PE_AC_AUDIT_IDENTITY,
        ),
    )?;
    m.add(
        "ID_PE_AA_CONTROLS",
        super::oid_const(
            py,
            synta_certificate::attribute_cert_types::ID_PE_AA_CONTROLS,
        ),
    )?;
    m.add(
        "ID_PE_AC_PROXYING",
        super::oid_const(
            py,
            synta_certificate::attribute_cert_types::ID_PE_AC_PROXYING,
        ),
    )?;
    m.add(
        "ID_CE_TARGET_INFORMATION",
        super::oid_const(
            py,
            synta_certificate::attribute_cert_types::ID_CE_TARGET_INFORMATION,
        ),
    )?;
    m.add(
        "ID_ACA_AUTHENTICATION_INFO",
        super::oid_const(
            py,
            synta_certificate::attribute_cert_types::ID_ACA_AUTHENTICATION_INFO,
        ),
    )?;
    m.add(
        "ID_ACA_ACCESS_IDENTITY",
        super::oid_const(
            py,
            synta_certificate::attribute_cert_types::ID_ACA_ACCESS_IDENTITY,
        ),
    )?;
    m.add(
        "ID_ACA_CHARGING_IDENTITY",
        super::oid_const(
            py,
            synta_certificate::attribute_cert_types::ID_ACA_CHARGING_IDENTITY,
        ),
    )?;
    m.add(
        "ID_ACA_GROUP",
        super::oid_const(py, synta_certificate::attribute_cert_types::ID_ACA_GROUP),
    )?;
    m.add(
        "ID_ACA_ENC_ATTRS",
        super::oid_const(
            py,
            synta_certificate::attribute_cert_types::ID_ACA_ENC_ATTRS,
        ),
    )?;
    m.add(
        "ID_AT_ROLE",
        super::oid_const(py, synta_certificate::attribute_cert_types::ID_AT_ROLE),
    )?;
    m.add(
        "ID_AT_CLEARANCE",
        super::oid_const(py, synta_certificate::attribute_cert_types::ID_AT_CLEARANCE),
    )?;

    crate::install_submodule(
        parent,
        &m,
        "synta.ac",
        Some(concat!(
            "synta.ac — RFC 5755 Attribute Certificate v2 types.\n\n",
            "Provides AttributeCertificate for decoding X.509 Attribute\n",
            "Certificates that bind roles, clearances, or service-auth\n",
            "attributes to a holder's PKC, along with OID constants for\n",
            "RFC 5755 extensions and attribute types.",
        )),
    )
}