synta-python 0.3.0

//! 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<'static>>()
            .map_err(SyntaErr)?;
        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 {
            let s = py_bytes.bind(py).as_bytes();
            std::slice::from_raw_parts(s.as_ptr(), s.len())
        };
        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>> {
        Ok(PyBytes::new(py, &self.ac()?.to_der().map_err(SyntaErr)?))
    }

    /// Parse the first ``ATTRIBUTE CERTIFICATE`` PEM block from ``data``.
    ///
    /// ```python,ignore
    /// import synta.ac as ac
    /// acer = ac.AttributeCertificate.from_pem(open("attr.pem", "rb").read())
    /// print(acer.serial_number.hex())
    /// ```
    ///
    /// :raises ValueError: if no valid PEM block is found or the DER is invalid.
    #[staticmethod]
    fn from_pem(py: Python<'_>, data: &[u8]) -> PyResult<Self> {
        let blocks = synta_certificate::pem_blocks(data);
        let der = match blocks.as_slice() {
            [] => {
                return Err(pyo3::exceptions::PyValueError::new_err(
                    "no PEM block found in input",
                ))
            }
            [(_, first), ..] => first,
        };
        let py_bytes = pyo3::types::PyBytes::new(py, der).unbind();
        {
            let raw = py_bytes.as_bytes(py);
            synta::Decoder::new(raw, Encoding::Der)
                .decode::<synta_certificate::attribute_cert_types::AttributeCertificate<'_>>()
                .map_err(SyntaErr)?;
        }
        let raw: &'static [u8] = unsafe {
            let s = py_bytes.bind(py).as_bytes();
            std::slice::from_raw_parts(s.as_ptr(), s.len())
        };
        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 PEM encoding of this ``AttributeCertificate``.
    ///
    /// ```python,ignore
    /// import synta.ac as ac
    /// acer = ac.AttributeCertificate.from_der(open("attr.ac", "rb").read())
    /// open("attr.pem", "wb").write(acer.to_pem())
    /// ```
    fn to_pem<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
        let pem = synta_certificate::der_to_pem("ATTRIBUTE CERTIFICATE", self.raw);
        PyBytes::new(py, &pem)
    }

    /// ``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)
    }

    /// Verify that this Attribute Certificate was signed by ``issuer``.
    ///
    /// Verifies the outer signature against the issuer's public key.  Note that
    /// AC issuers are represented as ``AttCertIssuer`` (a GeneralNames CHOICE),
    /// not a plain Name, so no issuer-to-subject name matching is performed
    /// here — only the cryptographic signature is checked.
    ///
    /// :raises ValueError: if the AC carries no ``responseBytes``, or the
    ///     signature is invalid.
    ///
    /// ```python,ignore
    /// import synta.ac as ac
    /// ca_cert = synta.Certificate.from_pem(open("ca.pem", "rb").read())
    /// acer = ac.AttributeCertificate.from_der(open("attr.ac", "rb").read())
    /// acer.verify_issued_by(ca_cert)   # raises ValueError if not valid
    /// ```
    fn verify_issued_by(&self, issuer: &super::cert::PyCertificate) -> PyResult<()> {
        use synta_certificate::{default_signature_verifier, SignatureVerifier};

        let ac = self.ac()?;
        let issuer_cert = issuer.cert()?;

        // Re-encode AttributeCertificateInfo (TBS).
        let tbs_der = encode_to_der(&ac.acinfo);
        if tbs_der.is_empty() {
            return Err(pyo3::exceptions::PyValueError::new_err(
                "failed to encode AttributeCertificateInfo",
            ));
        }

        // Re-encode outer signatureAlgorithm.
        let sig_alg_der = encode_to_der(&ac.signature_algorithm);
        if sig_alg_der.is_empty() {
            return Err(pyo3::exceptions::PyValueError::new_err(
                "failed to encode AC signatureAlgorithm",
            ));
        }

        // Re-encode issuer SPKI.
        let spki_der = encode_to_der(&issuer_cert.tbs_certificate.subject_public_key_info);
        if spki_der.is_empty() {
            return Err(pyo3::exceptions::PyValueError::new_err(
                "failed to encode issuer SubjectPublicKeyInfo",
            ));
        }

        // Verify using the backend-agnostic verifier.
        default_signature_verifier()
            .verify_certificate_signature(
                &tbs_der,
                &sig_alg_der,
                ac.signature.as_bytes(),
                &spki_der,
            )
            .map_err(|e| {
                pyo3::exceptions::PyValueError::new_err(format!("AC signature invalid: {e}"))
            })
    }

    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>()?;
    m.add_class::<PyAttributeCertificateBuilder>()?;

    // ── 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.",
        )),
    )
}

// ── PyAttributeCertificateBuilder ────────────────────────────────────────────

/// Builder for RFC 5755 Attribute Certificate TBS encoding.
///
/// Use the fluent setter methods to configure the certificate, then call
/// :meth:`build` to obtain the DER-encoded ``AttributeCertificateInfo``
/// (TBS) SEQUENCE.
///
/// ```python,ignore
/// import synta.ac as ac
///
/// tbs_der = (
///     ac.AttributeCertificateBuilder()
///     .serial_number(42)
///     .not_before("20240101120000Z")
///     .not_after("20250101120000Z")
///     .issuer_rfc822("ca@example.com")
///     .holder_entity_name_rfc822("user@example.com")
///     .build()
/// )
/// ```
#[pyclass(name = "AttributeCertificateBuilder")]
pub struct PyAttributeCertificateBuilder {
    inner: synta_certificate::AttributeCertificateBuilder,
}

#[pymethods]
impl PyAttributeCertificateBuilder {
    /// Create a new, empty ``AttributeCertificateBuilder``.
    #[new]
    fn new() -> Self {
        Self {
            inner: synta_certificate::AttributeCertificateBuilder::new(),
        }
    }

    /// Set the AC serial number.
    ///
    /// ```python,ignore
    /// builder.serial_number(42)
    /// ```
    fn serial_number<'py>(slf: Bound<'py, Self>, n: i64) -> Bound<'py, Self> {
        let old = std::mem::replace(
            &mut slf.borrow_mut().inner,
            synta_certificate::AttributeCertificateBuilder::new(),
        );
        slf.borrow_mut().inner = old.serial_number(n);
        slf
    }

    /// Set the validity start time as a GeneralizedTime string (``"YYYYMMDDHHmmssZ"``).
    ///
    /// :raises ValueError: if the string is not a valid GeneralizedTime.
    ///
    /// ```python,ignore
    /// builder.not_before("20240101120000Z")
    /// ```
    fn not_before<'py>(slf: Bound<'py, Self>, s: &str) -> Bound<'py, Self> {
        let old = std::mem::replace(
            &mut slf.borrow_mut().inner,
            synta_certificate::AttributeCertificateBuilder::new(),
        );
        slf.borrow_mut().inner = old.not_before(s);
        slf
    }

    /// Set the validity end time as a GeneralizedTime string (``"YYYYMMDDHHmmssZ"``).
    ///
    /// :raises ValueError: if the string is not a valid GeneralizedTime.
    ///
    /// ```python,ignore
    /// builder.not_after("20250101120000Z")
    /// ```
    fn not_after<'py>(slf: Bound<'py, Self>, s: &str) -> Bound<'py, Self> {
        let old = std::mem::replace(
            &mut slf.borrow_mut().inner,
            synta_certificate::AttributeCertificateBuilder::new(),
        );
        slf.borrow_mut().inner = old.not_after(s);
        slf
    }

    /// Add an ``rfc822Name`` GeneralName to the ``AttCertIssuer`` ``v1Form``.
    ///
    /// ```python,ignore
    /// builder.issuer_rfc822("ca@example.com")
    /// ```
    fn issuer_rfc822<'py>(slf: Bound<'py, Self>, email: &str) -> Bound<'py, Self> {
        let old = std::mem::replace(
            &mut slf.borrow_mut().inner,
            synta_certificate::AttributeCertificateBuilder::new(),
        );
        slf.borrow_mut().inner = old.issuer_rfc822(email);
        slf
    }

    /// Add a ``dNSName`` GeneralName to the ``AttCertIssuer`` ``v1Form``.
    ///
    /// ```python,ignore
    /// builder.issuer_dns("ca.example.com")
    /// ```
    fn issuer_dns<'py>(slf: Bound<'py, Self>, name: &str) -> Bound<'py, Self> {
        let old = std::mem::replace(
            &mut slf.borrow_mut().inner,
            synta_certificate::AttributeCertificateBuilder::new(),
        );
        slf.borrow_mut().inner = old.issuer_dns(name);
        slf
    }

    /// Add an ``rfc822Name`` GeneralName to the ``Holder.entityName``.
    ///
    /// ```python,ignore
    /// builder.holder_entity_name_rfc822("user@example.com")
    /// ```
    fn holder_entity_name_rfc822<'py>(slf: Bound<'py, Self>, email: &str) -> Bound<'py, Self> {
        let old = std::mem::replace(
            &mut slf.borrow_mut().inner,
            synta_certificate::AttributeCertificateBuilder::new(),
        );
        slf.borrow_mut().inner = old.holder_entity_name_rfc822(email);
        slf
    }

    /// Encode the ``AttributeCertificateInfo`` SEQUENCE to DER bytes.
    ///
    /// :raises ValueError: if any required field is missing or encoding fails.
    ///
    /// ```python,ignore
    /// tbs_der = builder.build()
    /// ```
    fn build<'py>(&mut self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        let inner = std::mem::replace(
            &mut self.inner,
            synta_certificate::AttributeCertificateBuilder::new(),
        );
        let der = inner
            .build()
            .map_err(pyo3::exceptions::PyValueError::new_err)?;
        Ok(PyBytes::new(py, &der))
    }

    fn __repr__(&self) -> String {
        "AttributeCertificateBuilder()".to_string()
    }
}