_synta/certificate/

cert_builder.rs

//! Python bindings for building X.509 certificates natively.
//!
//! Delegates all DER construction and signing to
//! [`synta_certificate::builder::CertificateBuilder`].  OpenSSL is only used
//! for the signing step; all ASN.1 encoding is done inside the Rust builder.
//!
//! # Chaining
//!
//! Each setter returns the same `CertificateBuilder` object, enabling a
//! single-expression build:
//!
//! ```python,ignore
//! import synta, datetime
//!
//! cert = (
//!     synta.CertificateBuilder()
//!     .issuer_name(ca_cert.subject_raw_der)      # zero re-encode
//!     .subject_name(csr.subject_raw_der)         # zero re-encode
//!     .public_key_der(csr.subject_public_key_info_der)  # zero re-encode
//!     .serial_number(42)
//!     .not_valid_before_utc(datetime.datetime(2024, 1, 1, tzinfo=datetime.timezone.utc))
//!     .not_valid_after_utc(datetime.datetime(2025, 1, 1, tzinfo=datetime.timezone.utc))
//!     .add_extension("2.5.29.19", True, basic_constraints_der)
//!     .sign(ca_key, "sha256")
//! )
//! ```

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

use crate::certificate::cert::PyCertificate;
use crate::crypto_keys::PyPrivateKey;

// ── Helpers ───────────────────────────────────────────────────────────────────

/// Convert a Python `datetime.datetime` (or any object with year/month/day/
/// hour/minute/second attributes) to a synta `Time`.
///
/// Uses `UtcTime` for years 1950–2049 (RFC 5280 §4.1.2.5), `GeneralizedTime`
/// otherwise.
fn py_to_synta_time(dt: &Bound<'_, PyAny>) -> PyResult<Time> {
    // Reject naive datetimes (tzinfo is None) to prevent silent UTC misinterpretation.
    if dt.getattr("tzinfo")?.is_none() {
        return Err(PyValueError::new_err(
            "datetime must be timezone-aware (tzinfo must not be None); \
             use datetime.timezone.utc to specify UTC",
        ));
    }
    let year = dt.getattr("year")?.extract::<u32>()?;
    let month = dt.getattr("month")?.extract::<u8>()?;
    let day = dt.getattr("day")?.extract::<u8>()?;
    let hour = dt.getattr("hour")?.extract::<u8>()?;
    let minute = dt.getattr("minute")?.extract::<u8>()?;
    let second = dt.getattr("second")?.extract::<u8>()?;
    if (1950..=2049).contains(&year) {
        Ok(Time::UtcTime(
            synta::UtcTime::new(year as u16, month, day, hour, minute, second)
                .map_err(|e| PyValueError::new_err(format!("invalid UTCTime: {e}")))?,
        ))
    } else {
        Ok(Time::GeneralTime(
            synta::GeneralizedTime::new(year as u16, month, day, hour, minute, second, None)
                .map_err(|e| PyValueError::new_err(format!("invalid GeneralizedTime: {e}")))?,
        ))
    }
}

// ── CertificateBuilder ────────────────────────────────────────────────────────

/// Builder for X.509 v3 certificates.
///
/// All Name, SubjectPublicKeyInfo, and extension-value bytes are stored as-is
/// and spliced verbatim into the TBS DER at signing time — no re-parse or
/// re-encode.  The only allocation per field is the initial copy of the byte
/// slice into the builder's owned storage.
///
/// Each setter method returns the **same** builder object, so calls can be
/// chained with ``.``:
///
/// ```python,ignore
/// import synta, datetime
///
/// now = datetime.datetime(2024, 1, 1, tzinfo=datetime.timezone.utc)
/// then = datetime.datetime(2025, 1, 1, tzinfo=datetime.timezone.utc)
///
/// cert = (
///     synta.CertificateBuilder()
///     .issuer_name(ca_cert.subject_raw_der)
///     .subject_name(csr.subject_raw_der)
///     .public_key_der(csr.subject_public_key_info_der)
///     .serial_number(42)
///     .not_valid_before_utc(now)
///     .not_valid_after_utc(then)
///     .add_extension("2.5.29.19", True, bc_der)
///     .sign(ca_key, "sha256")
/// )
/// ```
#[pyclass(name = "CertificateBuilder")]
pub struct PyCertificateBuilder {
    /// Issuer Name TLV (pre-encoded DER).
    issuer: Option<Vec<u8>>,
    /// Subject Name TLV (pre-encoded DER).
    subject: Option<Vec<u8>>,
    /// SubjectPublicKeyInfo TLV (pre-encoded DER).
    spki: Option<Vec<u8>>,
    /// Certificate serial number.
    serial: Option<Integer>,
    /// notBefore validity time.
    not_before: Option<Time>,
    /// notAfter validity time.
    not_after: Option<Time>,
    /// Extensions: (OID, critical, extension-value DER bytes).
    extensions: Vec<(ObjectIdentifier, bool, Vec<u8>)>,
}

#[pymethods]
impl PyCertificateBuilder {
    /// Create a new, empty ``CertificateBuilder``.
    #[new]
    fn new() -> Self {
        Self {
            issuer: None,
            subject: None,
            spki: None,
            serial: None,
            not_before: None,
            not_after: None,
            extensions: Vec::new(),
        }
    }

    /// Set the issuer 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
    /// ``Certificate.issuer_raw_der`` are suitable directly:
    ///
    /// ```python,ignore
    /// builder.issuer_name(ca_cert.subject_raw_der)
    /// ```
    fn issuer_name<'py>(slf: Bound<'py, Self>, name_der: &[u8]) -> Bound<'py, Self> {
        slf.borrow_mut().issuer = Some(name_der.to_vec());
        slf
    }

    /// Set the subject Name from pre-encoded DER bytes.
    ///
    /// ```python,ignore
    /// builder.subject_name(csr.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(csr_pub_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
    /// ``csr.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
    }

    /// Set the certificate serial number.
    ///
    /// Accepts a Python ``int`` (any size) or ``bytes``
    /// (big-endian two's-complement encoding).
    ///
    /// ```python,ignore
    /// import os
    /// builder.serial_number(42)
    /// builder.serial_number(int.from_bytes(os.urandom(20), "big"))
    /// ```
    fn serial_number<'py>(
        slf: Bound<'py, Self>,
        n: &Bound<'py, PyAny>,
    ) -> PyResult<Bound<'py, Self>> {
        let serial = if let Ok(i) = n.extract::<i64>() {
            Integer::from_i64(i)
        } else if let Ok(b) = n.cast::<PyBytes>() {
            Integer::from_bytes(b.as_bytes())
        } else if n.is_instance_of::<pyo3::types::PyInt>() {
            // Large Python int that doesn't fit in i64 — convert via to_bytes().
            let bit_length: usize = n.call_method0("bit_length")?.extract()?;
            let byte_length = bit_length.div_ceil(8).max(1);
            let bytes_obj = n.call_method1("to_bytes", (byte_length, "big"))?;
            let b = bytes_obj.cast::<PyBytes>()?;
            Integer::from_bytes(b.as_bytes())
        } else {
            return Err(PyValueError::new_err("serial_number expects int or bytes"));
        };
        slf.borrow_mut().serial = Some(serial);
        Ok(slf)
    }

    /// Set the ``notBefore`` validity time.
    ///
    /// Accepts a timezone-aware ``datetime.datetime`` object (or any object
    /// with ``year``, ``month``, ``day``, ``hour``, ``minute``, ``second``,
    /// and ``tzinfo`` attributes).  A naive datetime (``tzinfo=None``) raises
    /// :exc:`ValueError`.  The time fields are treated as UTC.
    ///
    /// ```python,ignore
    /// import datetime
    /// builder.not_valid_before_utc(
    ///     datetime.datetime(2024, 1, 1, tzinfo=datetime.timezone.utc)
    /// )
    /// ```
    fn not_valid_before_utc<'py>(
        slf: Bound<'py, Self>,
        dt: &Bound<'py, PyAny>,
    ) -> PyResult<Bound<'py, Self>> {
        slf.borrow_mut().not_before = Some(py_to_synta_time(dt)?);
        Ok(slf)
    }

    /// Set the ``notAfter`` validity time.
    ///
    /// ```python,ignore
    /// import datetime
    /// builder.not_valid_after_utc(
    ///     datetime.datetime(2025, 1, 1, tzinfo=datetime.timezone.utc)
    /// )
    /// ```
    fn not_valid_after_utc<'py>(
        slf: Bound<'py, Self>,
        dt: &Bound<'py, PyAny>,
    ) -> PyResult<Bound<'py, Self>> {
        slf.borrow_mut().not_after = Some(py_to_synta_time(dt)?);
        Ok(slf)
    }

    /// Add an X.509 v3 extension.
    ///
    /// ``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).
    ///
    /// ```python,ignore
    /// # Copy an extension from an existing cert:
    /// bc_der = ca_cert.get_extension_value_der("2.5.29.19")
    /// builder.add_extension("2.5.29.19", True, bc_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 certificate and return a :class:`Certificate`.
    ///
    /// ``key`` is the issuer'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_tbs`` → ``sign_ml_dsa_with_context`` → ``assemble``.
    ///
    /// All required fields (issuer name, subject name, public key, serial
    /// number, notBefore, notAfter) must have been set; a :exc:`ValueError`
    /// is raised if any is missing.
    ///
    /// Raises :exc:`ValueError` on encoding or signing errors.
    ///
    /// ```python,ignore
    /// cert = builder.sign(ca_key, "sha256")
    /// ml_cert = 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, PyCertificate>> {
        // Build the Rust-level builder from our stored fields.
        let mut builder = CertificateBuilder::new();
        if let Some(ref b) = self.issuer {
            builder = builder.issuer_name(b);
        }
        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);
        }
        if let Some(ref s) = self.serial {
            builder = builder.serial_number(s.clone());
        }
        if let Some(ref t) = self.not_before {
            builder = builder.not_valid_before(t.clone());
        }
        if let Some(ref t) = self.not_after {
            builder = builder.not_valid_after(t.clone());
        }
        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_tbs → 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 cert_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 tbs_der = builder
                .build_tbs(&sig_alg_der)
                .map_err(|e| PyValueError::new_err(format!("{e}")))?;
            let signature = key
                .inner
                .sign_ml_dsa_with_context(&tbs_der, ctx)
                .map_err(|e| PyValueError::new_err(format!("{e}")))?;
            CertificateBuilder::assemble(&tbs_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 PyCertificate.
        let py_bytes = PyBytes::new(py, &cert_der);
        let cert = PyCertificate::new_from_der(py, py_bytes)?;
        Py::new(py, cert).map(|py_cert| py_cert.into_bound(py))
    }

    /// Sign the certificate using the RFC 9925 unsigned algorithm and return a
    /// :class:`Certificate`.
    ///
    /// No private key is required.  The resulting certificate carries
    /// ``id-alg-unsigned`` (1.3.6.1.5.5.7.6.36) in both
    /// ``TBSCertificate.signature`` and the outer ``signatureAlgorithm``, with
    /// a zero-length BIT STRING (``03 01 00``) as the ``signatureValue``.
    ///
    /// All required fields (issuer name, subject name, public key, serial
    /// number, notBefore, notAfter) must have been set; a :exc:`ValueError`
    /// is raised if any is missing.
    ///
    /// ```python,ignore
    /// # Build an unsigned root CA certificate:
    /// unsigned_root = (
    ///     synta.CertificateBuilder()
    ///     .issuer_name(issuer_name_der)
    ///     .subject_name(subject_name_der)
    ///     .public_key(root_key.public_key)
    ///     .serial_number(1)
    ///     .not_valid_before_utc(now)
    ///     .not_valid_after_utc(expires)
    ///     .add_extension("2.5.29.19", True, bc_der)
    ///     .sign_unsigned()
    /// )
    /// ```
    fn sign_unsigned<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyCertificate>> {
        let mut builder = CertificateBuilder::new();
        if let Some(ref b) = self.issuer {
            builder = builder.issuer_name(b);
        }
        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);
        }
        if let Some(ref s) = self.serial {
            builder = builder.serial_number(s.clone());
        }
        if let Some(ref t) = self.not_before {
            builder = builder.not_valid_before(t.clone());
        }
        if let Some(ref t) = self.not_after {
            builder = builder.not_valid_after(t.clone());
        }
        for (oid, critical, value_bytes) in &self.extensions {
            builder = builder.add_extension(oid.clone(), *critical, value_bytes);
        }

        let cert_der = builder
            .sign(&UnsignedCertificateSigner)
            .map_err(|e| PyValueError::new_err(format!("{e}")))?;

        let py_bytes = PyBytes::new(py, &cert_der);
        let cert = PyCertificate::new_from_der(py, py_bytes)?;
        Py::new(py, cert).map(|py_cert| py_cert.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!("CertificateBuilder(subject={subject})")
    }
}