synta-python 0.2.5

//! Python bindings for building PKCS #10 Certificate Signing Requests.
//!
//! Delegates all DER construction and signing to
//! [`synta_certificate::csr_builder::CsrBuilder`].  OpenSSL is only used for
//! the signing step; all ASN.1 encoding is done inside the Rust builder.
//!
//! # Chaining
//!
//! Each setter returns the same `CsrBuilder` object, enabling a
//! single-expression build:
//!
//! ```python,ignore
//! import synta
//!
//! csr = (
//!     synta.CsrBuilder()
//!     .subject_name(name_der)
//!     .public_key(key.public_key())
//!     .add_extension("2.5.29.17", False, san_der)
//!     .sign(key, "sha256")
//! )
//! ```

use pyo3::exceptions::PyValueError;
use pyo3::prelude::*;
use pyo3::types::PyBytes;
use pyo3::Py;
use synta::ObjectIdentifier;
use synta_certificate::{CertificateSigner as _, CsrBuilder, PrivateKey as _};

use crate::certificate::pkix::PyCsr;
use crate::crypto_keys::PyPrivateKey;

// ── CsrBuilder ────────────────────────────────────────────────────────────────

/// Builder for PKCS #10 Certificate Signing Requests.
///
/// The subject Name and SubjectPublicKeyInfo bytes are stored as-is and spliced
/// verbatim into the ``CertificationRequestInfo`` DER at signing time — no
/// re-parse or re-encode.
///
/// Each setter method returns the **same** builder object, so calls can be
/// chained with ``.``:
///
/// ```python,ignore
/// import synta
///
/// csr = (
///     synta.CsrBuilder()
///     .subject_name(cert.subject_raw_der)
///     .public_key(key.public_key())
///     .add_extension("2.5.29.17", False, san_der)
///     .sign(key, "sha256")
/// )
/// ```
#[pyclass(name = "CsrBuilder")]
pub struct PyCsrBuilder {
    /// Subject Name TLV (pre-encoded DER).
    subject: Option<Vec<u8>>,
    /// SubjectPublicKeyInfo TLV (pre-encoded DER).
    spki: Option<Vec<u8>>,
    /// Extensions: (OID, critical, extension-value DER bytes).
    extensions: Vec<(ObjectIdentifier, bool, Vec<u8>)>,
}

#[pymethods]
impl PyCsrBuilder {
    /// Create a new, empty ``CsrBuilder``.
    #[new]
    fn new() -> Self {
        Self {
            subject: None,
            spki: None,
            extensions: Vec::new(),
        }
    }

    /// Set the subject Name from pre-encoded DER bytes.
    ///
    /// Accepts any bytes object that is a valid DER-encoded ``Name`` SEQUENCE
    /// TLV.  The bytes from ``Certificate.subject_raw_der`` or
    /// ``CertificationRequest.subject_raw_der`` are suitable directly:
    ///
    /// ```python,ignore
    /// builder.subject_name(cert.subject_raw_der)
    /// ```
    fn subject_name<'py>(slf: Bound<'py, Self>, name_der: &[u8]) -> Bound<'py, Self> {
        slf.borrow_mut().subject = Some(name_der.to_vec());
        slf
    }

    /// Set the SubjectPublicKeyInfo from a :class:`PublicKey` object.
    ///
    /// Serializes the key to SPKI DER once and stores the bytes.
    ///
    /// ```python,ignore
    /// builder.public_key(key.public_key())
    /// ```
    fn public_key<'py>(
        slf: Bound<'py, Self>,
        key: &crate::crypto_keys::PyPublicKey,
    ) -> PyResult<Bound<'py, Self>> {
        let der = key
            .inner
            .to_der()
            .map_err(|e| PyValueError::new_err(format!("{e}")))?;
        slf.borrow_mut().spki = Some(der);
        Ok(slf)
    }

    /// Set the SubjectPublicKeyInfo from pre-encoded SPKI DER bytes.
    ///
    /// Stores the bytes verbatim — no re-parsing.  Suitable for passing
    /// ``Certificate.subject_public_key_info_der`` or
    /// ``CertificationRequest.subject_public_key_info_der`` directly:
    ///
    /// ```python,ignore
    /// builder.public_key_der(csr.subject_public_key_info_der)
    /// ```
    fn public_key_der<'py>(slf: Bound<'py, Self>, spki_der: &[u8]) -> Bound<'py, Self> {
        slf.borrow_mut().spki = Some(spki_der.to_vec());
        slf
    }

    /// Add an X.509 v3 extension to the ``extensionRequest`` attribute.
    ///
    /// ``oid`` is the extension OID in dotted-decimal notation.
    /// ``critical`` is ``True`` if the extension is critical.
    /// ``value_der`` is the DER-encoded extension value — the raw bytes that
    /// ``Certificate.get_extension_value_der(oid)`` returns (i.e., the
    /// content of the OCTET STRING, *not* the OCTET STRING TLV itself).
    ///
    /// All extensions added with this method are collected into a single
    /// ``extensionRequest`` attribute in the ``CertificationRequestInfo``.
    ///
    /// ```python,ignore
    /// builder.add_extension("2.5.29.17", False, san_der)
    /// ```
    fn add_extension<'py>(
        slf: Bound<'py, Self>,
        oid: &str,
        critical: bool,
        value_der: &[u8],
    ) -> PyResult<Bound<'py, Self>> {
        use std::str::FromStr;
        let oid = ObjectIdentifier::from_str(oid)
            .map_err(|_| PyValueError::new_err(format!("invalid OID: {oid}")))?;
        slf.borrow_mut()
            .extensions
            .push((oid, critical, value_der.to_vec()));
        Ok(slf)
    }

    /// Sign the CSR and return a :class:`CertificationRequest`.
    ///
    /// ``key`` is the subject's :class:`PrivateKey`.  ``algorithm`` is the
    /// hash algorithm name — one of ``"sha1"``, ``"sha256"``, ``"sha384"``,
    /// ``"sha512"`` for RSA and ECDSA keys.  For Ed25519 / Ed448 keys the
    /// argument is ignored (no pre-hash is used).
    ///
    /// ``context`` is the ML-DSA context string (FIPS 204 domain separator).
    /// It defaults to ``None`` (equivalent to an empty context).  Ignored for
    /// non-ML-DSA keys.  When ``context`` is a non-empty ``bytes`` object and
    /// the key is an ML-DSA key, the manual signing path is used:
    /// ``build_cri`` → ``sign_ml_dsa_with_context`` → ``assemble``.
    ///
    /// Both subject name and public key must have been set; a :exc:`ValueError`
    /// is raised if either is missing.
    ///
    /// Raises :exc:`ValueError` on encoding or signing errors.
    ///
    /// ```python,ignore
    /// csr = builder.sign(key, "sha256")
    /// ml_csr = builder.sign(ml_dsa_key, "sha256", context=b"my-app")
    /// ```
    #[pyo3(signature = (key, algorithm, context = None))]
    fn sign<'py>(
        &self,
        py: Python<'py>,
        key: &PyPrivateKey,
        algorithm: &str,
        context: Option<&[u8]>,
    ) -> PyResult<Bound<'py, PyCsr>> {
        // Build the Rust-level builder from our stored fields.
        let mut builder = CsrBuilder::new();
        if let Some(ref b) = self.subject {
            builder = builder.subject_name(b);
        }
        if let Some(ref b) = self.spki {
            builder = builder.public_key_der(b);
        }
        for (oid, critical, value_bytes) in &self.extensions {
            builder = builder.add_extension(oid.clone(), *critical, value_bytes);
        }

        // If a non-empty context is provided and this is an ML-DSA key, take the
        // manual path: build_cri → sign_ml_dsa_with_context → assemble.
        let ctx = context.unwrap_or(b"");
        let is_ml_dsa = matches!(
            key.inner.key_type(),
            "ml-dsa-44" | "ml-dsa-65" | "ml-dsa-87"
        );
        let csr_der = if !ctx.is_empty() && is_ml_dsa {
            let signer = key.as_signer(algorithm);
            let sig_alg_der = signer
                .signature_algorithm_der()
                .map_err(|e| PyValueError::new_err(format!("{e}")))?;
            let cri_der = builder
                .build_cri(&sig_alg_der)
                .map_err(|e| PyValueError::new_err(format!("{e}")))?;
            let signature = key
                .inner
                .sign_ml_dsa_with_context(&cri_der, ctx)
                .map_err(|e| PyValueError::new_err(format!("{e}")))?;
            CsrBuilder::assemble(&cri_der, &sig_alg_der, &signature)
                .map_err(|e| PyValueError::new_err(format!("{e}")))?
        } else {
            // Standard path: delegate to the signer trait object.
            let signer = key.as_signer(algorithm);
            builder
                .sign(&signer)
                .map_err(|e| PyValueError::new_err(format!("{e}")))?
        };

        // Parse the assembled DER back into a PyCsr.
        let py_bytes = PyBytes::new(py, &csr_der);
        let csr = PyCsr::new_from_der(py, py_bytes)?;
        Py::new(py, csr).map(|py_csr| py_csr.into_bound(py))
    }

    fn __repr__(&self) -> String {
        let subject = self
            .subject
            .as_ref()
            .map(|b| format!("{} bytes", b.len()))
            .unwrap_or_else(|| "not set".to_string());
        format!("CsrBuilder(subject={subject})")
    }
}