synta-python 0.2.5

//! Python binding for the CMS SignedData fluent builder:
//! [`PySignedDataBuilder`] (RFC 5652 §5).

use std::str::FromStr;

use pyo3::exceptions::PyValueError;
use pyo3::prelude::*;
use pyo3::types::PyBytes;
use synta::{
    Decoder, Encoding, ExplicitTag, Integer, ObjectIdentifier, OctetStringRef, RawDer, SetOf, Tag,
    ToDer,
};
use synta_certificate::{
    cms_2010_types::IssuerAndSerialNumber,
    cms_rfc5652_types::{Attribute, EncapsulatedContentInfo, SignedData, SignerInfo},
    digest_alg_id, pkcs7_types, AlgorithmIdentifier, Certificate, DataHasher as _, Name,
    PrivateKey as _,
};

use crate::crypto_keys::PyPrivateKey;

// ── SignerSpec ────────────────────────────────────────────────────────────────

struct SignerSpec {
    key: synta_certificate::BackendPrivateKey,
    cert_der: Vec<u8>,
    hash_algorithm: String,
}

// ── PySignedDataBuilder ───────────────────────────────────────────────────────

/// Builder for CMS ``SignedData`` (RFC 5652 §5).
///
/// Accumulates the signing ingredients and optional extra certificates, then
/// assembles the final DER on :meth:`build`.
///
/// ```python,ignore
/// import synta, synta.cms
///
/// # Attached (content included in SignedData):
/// sd_der = (
///     synta.cms.SignedDataBuilder(content_bytes, content_type="1.2.840.113549.1.7.1")
///     .add_signer(private_key, signing_cert_der, hash_algorithm="sha256")
///     .add_cert(intermediate_cert_der)
///     .build()
/// )
///
/// # Detached (content excluded from SignedData):
/// sd_der = (
///     synta.cms.SignedDataBuilder(content_bytes, detached=True)
///     .add_signer(private_key, signing_cert_der, hash_algorithm="sha256")
///     .build()
/// )
/// ```
///
/// Each ``add_signer`` / ``add_cert`` call returns the **same** builder object,
/// enabling chaining.
#[pyclass(name = "SignedDataBuilder")]
pub struct PySignedDataBuilder {
    content: Vec<u8>,
    content_oid: ObjectIdentifier,
    detached: bool,
    signers: Vec<SignerSpec>,
    extra_certs: Vec<Vec<u8>>,
}

#[pymethods]
impl PySignedDataBuilder {
    /// Create a ``SignedDataBuilder``.
    ///
    /// ``content`` is the data that will be signed.
    ///
    /// ``content_type`` (keyword-only) is the dotted-decimal OID string for
    /// ``eContentType``; defaults to ``id-data`` (``1.2.840.113549.1.7.1``).
    ///
    /// ``detached`` (keyword-only) controls whether the content is embedded in
    /// the output (``False``, default) or omitted (``True`` — detached signature).
    ///
    /// Raises :exc:`ValueError` if ``content_type`` is not a valid OID string.
    #[new]
    #[pyo3(signature = (content, *, content_type = None, detached = false))]
    fn new(content: &[u8], content_type: Option<&str>, detached: bool) -> PyResult<Self> {
        let oid_str = content_type.unwrap_or("1.2.840.113549.1.7.1");
        let content_oid = ObjectIdentifier::from_str(oid_str).map_err(|e| {
            PyValueError::new_err(format!("invalid content_type OID {oid_str:?}: {e}"))
        })?;
        Ok(Self {
            content: content.to_vec(),
            content_oid,
            detached,
            signers: Vec::new(),
            extra_certs: Vec::new(),
        })
    }

    /// Add a signer to the ``SignedData``.
    ///
    /// ``key`` is the private key used for signing.
    ///
    /// ``cert_der`` is the DER-encoded signing certificate; the issuer name and
    /// serial number are extracted from it to populate ``IssuerAndSerialNumber``,
    /// and the certificate itself is included in the ``certificates`` field.
    ///
    /// ``hash_algorithm`` (keyword-only) is the digest algorithm name; must be
    /// one of ``"sha1"``, ``"sha256"`` (default), ``"sha384"``, or ``"sha512"``.
    ///
    /// Returns the same builder to allow chaining.
    ///
    /// Raises :exc:`ValueError` for unsupported hash algorithms.
    #[pyo3(signature = (key, cert_der, hash_algorithm = "sha256"))]
    fn add_signer<'py>(
        slf: Bound<'py, Self>,
        key: &PyPrivateKey,
        cert_der: &[u8],
        hash_algorithm: &str,
    ) -> PyResult<Bound<'py, Self>> {
        match hash_algorithm {
            "sha1" | "sha256" | "sha384" | "sha512" => {}
            other => {
                return Err(PyValueError::new_err(format!(
                    "unsupported hash_algorithm {other:?}; use sha1, sha256, sha384, or sha512"
                )));
            }
        }
        slf.borrow_mut().signers.push(SignerSpec {
            key: key.inner.clone(),
            cert_der: cert_der.to_vec(),
            hash_algorithm: hash_algorithm.to_owned(),
        });
        Ok(slf)
    }

    /// Add an extra certificate to the ``certificates [0]`` field.
    ///
    /// ``cert_der`` is the DER-encoded certificate to include.
    ///
    /// Returns the same builder to allow chaining.
    fn add_cert<'py>(slf: Bound<'py, Self>, cert_der: &[u8]) -> Bound<'py, Self> {
        slf.borrow_mut().extra_certs.push(cert_der.to_vec());
        slf
    }

    /// Assemble and return the DER-encoded ``ContentInfo`` wrapping a
    /// ``SignedData`` as :class:`bytes`.
    ///
    /// For each signer:
    ///
    /// 1. Computes the message digest of the content.
    /// 2. Builds ``signedAttrs`` using the generated ``Attribute`` type
    ///    (``contentType`` + ``messageDigest``).
    /// 3. Signs the DER-encoded ``signedAttrs`` SET (RFC 5652 §5.4).
    /// 4. Assembles the ``SignerInfo`` using the generated CMS types.
    ///
    /// Then constructs ``SignedData`` and wraps it in a ``ContentInfo``.
    ///
    /// Raises :exc:`ValueError` on certificate parse errors, unsupported
    /// algorithms, or crypto backend failures.
    fn build<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        if self.signers.is_empty() {
            return Err(PyValueError::new_err(
                "at least one signer must be added before calling build()",
            ));
        }

        let hasher = synta_certificate::default_data_hasher();

        // All intermediate byte buffers must outlive the generated type structs
        // that borrow from them.  Collect them all here first.
        struct SignerBuffers {
            ias_der: Vec<u8>,
            signed_attrs_content: Vec<u8>,
            digest_alg: AlgorithmIdentifier<'static>,
            sig_alg_bytes: Vec<u8>,
            signature: Vec<u8>,
            cert_der: Vec<u8>,
        }

        let mut signer_bufs: Vec<SignerBuffers> = Vec::with_capacity(self.signers.len());
        let mut seen_alg_oids: Vec<ObjectIdentifier> = Vec::new();
        let mut digest_algs: Vec<AlgorithmIdentifier<'static>> = Vec::new();

        for spec in &self.signers {
            let alg = digest_alg_id(&spec.hash_algorithm).ok_or_else(|| {
                PyValueError::new_err(format!(
                    "unsupported hash_algorithm {:?}; use sha1, sha256, sha384, or sha512",
                    spec.hash_algorithm
                ))
            })?;
            if !seen_alg_oids.contains(&alg.algorithm) {
                seen_alg_oids.push(alg.algorithm.clone());
                digest_algs.push(alg.clone());
            }

            let ias_der = issuer_and_serial_der(&spec.cert_der)?;

            let digest_bytes = hasher
                .hash_data(&spec.hash_algorithm, &self.content)
                .map_err(|e| PyValueError::new_err(format!("hash failed: {e}")))?;

            let (to_sign, signed_attrs_content) =
                build_signed_attrs(&self.content_oid, &digest_bytes)?;

            let signer = spec.key.as_signer(&spec.hash_algorithm);
            let sig_alg_bytes = signer
                .signature_algorithm_der_erased()
                .map_err(|e| PyValueError::new_err(format!("sig alg: {e}")))?;
            let signature = signer
                .sign_tbs_erased(&to_sign)
                .map_err(|e| PyValueError::new_err(format!("sign: {e}")))?;

            signer_bufs.push(SignerBuffers {
                ias_der,
                signed_attrs_content,
                digest_alg: alg,
                sig_alg_bytes,
                signature,
                cert_der: spec.cert_der.clone(),
            });
        }

        // Build SignerInfo structs using generated types.
        // Lifetimes: signer_bufs outlives signer_infos (declared after).
        let mut signer_infos: Vec<SignerInfo<'_>> = Vec::with_capacity(signer_bufs.len());
        for buf in &signer_bufs {
            let sig_alg = AlgorithmIdentifier::from_der(&buf.sig_alg_bytes)
                .map_err(|e| PyValueError::new_err(format!("decode sig alg: {e}")))?;
            signer_infos.push(SignerInfo {
                version: Integer::from_i64(1),
                sid: RawDer(&buf.ias_der),
                digest_algorithm: buf.digest_alg.clone(),
                signed_attrs: Some(RawDer(&buf.signed_attrs_content)),
                signature_algorithm: sig_alg,
                signature: OctetStringRef::new(&buf.signature),
                unsigned_attrs: None,
            });
        }

        // Collect all cert DER bytes (signing certs then extra certs).
        let mut all_cert_slices: Vec<&[u8]> =
            signer_bufs.iter().map(|b| b.cert_der.as_slice()).collect();
        for extra in &self.extra_certs {
            all_cert_slices.push(extra.as_slice());
        }
        let certs_content: Vec<u8> = all_cert_slices.concat();

        // EncapsulatedContentInfo.
        let eci = EncapsulatedContentInfo {
            e_content_type: self.content_oid.clone(),
            e_content: if self.detached {
                None
            } else {
                Some(OctetStringRef::new(&self.content))
            },
        };

        // SignedData — generated type handles all field encoding and tagging.
        let signed_data = SignedData {
            version: Integer::from_i64(1),
            digest_algorithms: SetOf::from_vec(digest_algs),
            encap_content_info: eci,
            certificates: if certs_content.is_empty() {
                None
            } else {
                Some(RawDer(&certs_content))
            },
            crls: None,
            signer_infos: SetOf::from_vec(signer_infos),
        };
        let sd_der = signed_data
            .to_der()
            .map_err(|e| PyValueError::new_err(format!("encode SignedData: {e}")))?;

        // ContentInfo — content is [0] EXPLICIT SignedData.
        // The PKCS7-CMS.asn1 schema defines content as ANY (no tag annotation),
        // so the [0] EXPLICIT wrapper must be supplied here explicitly.
        let explicit_0 = ExplicitTag::context_specific(0, RawDer(sd_der.as_slice()))
            .to_der()
            .map_err(|e| PyValueError::new_err(format!("encode [0] EXPLICIT wrapper: {e}")))?;
        let id_signed_data = ObjectIdentifier::new(pkcs7_types::ID_SIGNED_DATA)
            .map_err(|_| PyValueError::new_err("invalid id-signedData OID"))?;
        let content_info = pkcs7_types::ContentInfo {
            content_type: id_signed_data,
            content: RawDer(&explicit_0),
        };
        let ci_der = content_info
            .to_der()
            .map_err(|e| PyValueError::new_err(format!("encode ContentInfo: {e}")))?;

        Ok(PyBytes::new(py, &ci_der))
    }
}

// ── Private helpers ───────────────────────────────────────────────────────────

/// Parse a DER-encoded certificate and return the DER bytes of
/// `IssuerAndSerialNumber { issuer, serialNumber }`.
fn issuer_and_serial_der(cert_der: &[u8]) -> PyResult<Vec<u8>> {
    let mut dec = Decoder::new(cert_der, Encoding::Ber);
    let cert: Certificate = dec
        .decode()
        .map_err(|e| PyValueError::new_err(format!("failed to parse signing certificate: {e}")))?;
    let issuer_name = Name::from_der(cert.tbs_certificate.issuer.as_bytes())
        .map_err(|e| PyValueError::new_err(format!("parse issuer Name: {e}")))?;
    IssuerAndSerialNumber {
        issuer: issuer_name,
        serial_number: cert.tbs_certificate.serial_number.clone(),
    }
    .to_der()
    .map_err(|e| PyValueError::new_err(format!("encode IssuerAndSerialNumber: {e}")))
}

/// Build the signed attributes for one signer and return:
/// - `to_sign`: the DER-encoded `SET OF Attribute` (with `0x31` tag), used for
///   signing per RFC 5652 §5.4.
/// - `signed_attrs_content`: the value bytes of that SET (without outer tag+len),
///   passed as `RawDer` to `SignerInfo.signed_attrs` where the encoder adds the
///   `[0] IMPLICIT` context tag.
fn build_signed_attrs(
    content_oid: &ObjectIdentifier,
    digest: &[u8],
) -> PyResult<(Vec<u8>, Vec<u8>)> {
    let id_ct = ObjectIdentifier::new(synta_certificate::oids::PKCS9_CONTENT_TYPE)
        .map_err(|_| PyValueError::new_err("invalid id-contentType OID"))?;
    let id_md = ObjectIdentifier::new(synta_certificate::oids::PKCS9_MESSAGE_DIGEST)
        .map_err(|_| PyValueError::new_err("invalid id-messageDigest OID"))?;

    // contentType Attribute: attrValues = SET { eContentType OID }
    let ct_vals = SetOf::from_vec(vec![content_oid.clone()])
        .to_der()
        .map_err(|e| PyValueError::new_err(format!("encode contentType attrValues: {e}")))?;
    let ct_attr_der = Attribute {
        attr_type: id_ct,
        attr_values: RawDer(&ct_vals),
    }
    .to_der()
    .map_err(|e| PyValueError::new_err(format!("encode contentType Attribute: {e}")))?;

    // messageDigest Attribute: attrValues = SET { OCTET STRING digest }
    let md_vals = SetOf::from_vec(vec![OctetStringRef::new(digest)])
        .to_der()
        .map_err(|e| PyValueError::new_err(format!("encode messageDigest attrValues: {e}")))?;
    let md_attr_der = Attribute {
        attr_type: id_md,
        attr_values: RawDer(&md_vals),
    }
    .to_der()
    .map_err(|e| PyValueError::new_err(format!("encode messageDigest Attribute: {e}")))?;

    // SET OF attributes — DER-sorted by SetOf (RFC 5652 §5.4).
    let to_sign = SetOf::from_vec(vec![
        RawDer(ct_attr_der.as_slice()),
        RawDer(md_attr_der.as_slice()),
    ])
    .to_der()
    .map_err(|e| PyValueError::new_err(format!("encode signedAttrs SET: {e}")))?;

    // Extract SET value bytes (without 0x31 tag+len) for signed_attrs: [0] IMPLICIT.
    let mut tmp = Decoder::new(&to_sign, Encoding::Der);
    let inner = tmp
        .enter_constructed(Tag::universal_constructed(17))
        .map_err(|e| PyValueError::new_err(format!("strip SET outer tag: {e}")))?;
    let signed_attrs_content = inner.remaining().to_vec();

    Ok((to_sign, signed_attrs_content))
}