_synta/certificate/

csr_builder.rs

//! 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::{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).
    ///
    /// 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")
    /// ```
    fn sign<'py>(
        &self,
        py: Python<'py>,
        key: &PyPrivateKey,
        algorithm: &str,
    ) -> 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);
        }

        // Sign and assemble the CSR DER.
        let signer = key.as_signer(algorithm);
        let csr_der = 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})")
    }
}