synta-python 0.3.1

//! Python bindings for CMS encryption types:
//! [`PyEnvelopedData`] (RFC 5652 §6) and [`PyEncryptedData`] (RFC 5652 §8).

use std::sync::OnceLock;

use pyo3::prelude::*;
use pyo3::types::PyBytes;

use synta::traits::Encode;
use synta::{Decoder, Encoding};

use crate::types::PyObjectIdentifier;

// ── PyEnvelopedData ───────────────────────────────────────────────────────────

/// CMS ``EnvelopedData`` (RFC 5652 §6) accessible from Python.
///
/// Obtained by parsing the inner content of a :class:`ContentInfo` whose
/// ``content_type_oid`` equals :data:`~synta.cms.ID_ENVELOPED_DATA`.
#[pyclass(frozen, name = "EnvelopedData")]
pub struct PyEnvelopedData {
    _data: Py<PyBytes>,
    pub(super) raw: &'static [u8],
    inner: OnceLock<Box<synta_certificate::cms_rfc5652_types::EnvelopedData<'static>>>,
    originator_info_cache: OnceLock<Option<Py<PyBytes>>>,
    recipient_infos_cache: OnceLock<Py<PyBytes>>,
    content_type_cache: OnceLock<Py<PyObjectIdentifier>>,
    content_encryption_algorithm_oid_cache: OnceLock<Py<PyObjectIdentifier>>,
    content_encryption_algorithm_params_cache: OnceLock<Option<Py<PyBytes>>>,
    encrypted_content_cache: OnceLock<Option<Py<PyBytes>>>,
    unprotected_attrs_cache: OnceLock<Option<Py<PyBytes>>>,
}

impl PyEnvelopedData {
    fn enveloped_data(
        &self,
    ) -> PyResult<&synta_certificate::cms_rfc5652_types::EnvelopedData<'static>> {
        if let Some(v) = self.inner.get() {
            return Ok(v.as_ref());
        }
        let mut decoder = Decoder::new(self.raw, Encoding::Ber);
        let decoded = decoder.decode().map_err(|e| {
            pyo3::exceptions::PyValueError::new_err(format!("EnvelopedData BER decode failed: {e}"))
        })?;
        let _ = self.inner.set(Box::new(decoded));
        Ok(self.inner.get().unwrap().as_ref())
    }

    /// Construct from a raw PyBytes containing DER — used by `create()` and builder.
    pub(super) fn from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
        let py_bytes = data.unbind();
        let raw: &'static [u8] = unsafe {
            let s = py_bytes.bind(py).as_bytes();
            std::slice::from_raw_parts(s.as_ptr(), s.len())
        };
        {
            let mut d = Decoder::new(raw, Encoding::Ber);
            d.read_tag()
                .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
            d.read_length()
                .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
        }
        Ok(Self {
            _data: py_bytes,
            raw,
            inner: OnceLock::new(),
            originator_info_cache: OnceLock::new(),
            recipient_infos_cache: OnceLock::new(),
            content_type_cache: OnceLock::new(),
            content_encryption_algorithm_oid_cache: OnceLock::new(),
            content_encryption_algorithm_params_cache: OnceLock::new(),
            encrypted_content_cache: OnceLock::new(),
            unprotected_attrs_cache: OnceLock::new(),
        })
    }
}

#[pymethods]
impl PyEnvelopedData {
    /// Parse a DER- or BER-encoded CMS ``EnvelopedData`` SEQUENCE.
    #[staticmethod]
    #[pyo3(name = "from_der")]
    fn py_from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
        Self::from_der(py, data)
    }

    /// Return the original bytes passed to :meth:`from_der`.
    fn to_der<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
        self._data.clone_ref(py).into_bound(py)
    }

    /// Decrypt this ``EnvelopedData`` using an RSA private key.
    ///
    /// ``key`` must be a :class:`~synta.PrivateKey` holding the recipient's RSA
    /// private key.  The method iterates the ``RecipientInfos`` SET, unwraps
    /// the content-encryption key using the private key, and returns the
    /// decrypted plaintext.
    ///
    /// Supports RSA-OAEP SHA-256 (:data:`~synta.cms.ID_RSAES_OAEP`) and
    /// RSA PKCS\#1 v1.5 (:data:`~synta.cms.RSA_ENCRYPTION`) key-transport.
    ///
    /// Raises :exc:`ValueError` if no matching recipient info is found or if
    /// decryption fails.
    /// Raises :exc:`NotImplementedError` when built without a crypto backend.
    ///
    /// ```python,ignore
    /// import synta
    /// import synta.cms as cms
    ///
    /// # Parse CMS EnvelopedData from DER (e.g. from the ContentInfo content field)
    /// ed = cms.EnvelopedData.from_der(enveloped_data_der)
    ///
    /// # Load the recipient's RSA private key
    /// with open("recipient_key.pem", "rb") as f:
    ///     priv = synta.PrivateKey.from_pem(f.read())
    ///
    /// plaintext = ed.decrypt(priv)
    /// ```
    fn decrypt<'py>(
        &self,
        py: Python<'py>,
        key: &crate::crypto_keys::PyPrivateKey,
    ) -> PyResult<Bound<'py, PyBytes>> {
        use synta_certificate::EnvelopedDataDecryptor as _;
        let pkcs8_der = key
            .inner
            .to_der()
            .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
        let plaintext = synta_certificate::DefaultEnvelopedDataDecryptor::new(&pkcs8_der)
            .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?
            .decrypt_enveloped(self.enveloped_data()?)
            .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
        Ok(PyBytes::new(py, &plaintext))
    }

    /// Build a CMS ``EnvelopedData`` (RFC 5652 §6) with key-transport recipients.
    ///
    /// ``plaintext`` is the data to encrypt.
    ///
    /// ``recipients`` is a list of ``(cert, key_wrap_oid)`` pairs where:
    ///
    /// - *cert* is a :class:`~synta.Certificate` object or raw DER ``bytes``.
    /// - *key_wrap_oid* is an :class:`~synta.ObjectIdentifier` (or dotted-decimal
    ///   string) selecting how the CEK is encrypted under the recipient's public
    ///   key:
    ///
    ///   - :data:`~synta.RSAES_OAEP` — RSA-OAEP with SHA-256 (recommended).
    ///   - :data:`~synta.RSA_ENCRYPTION` — RSA PKCS#1 v1.5 (legacy).
    ///
    /// ``content_enc_alg`` selects the symmetric content-encryption algorithm.
    /// Accepted values: :data:`~synta.cms.ID_AES128_CBC`,
    /// :data:`~synta.cms.ID_AES192_CBC`, :data:`~synta.cms.ID_AES256_CBC`
    /// (default).
    ///
    /// Returns the ``EnvelopedData`` **SEQUENCE** only (not wrapped in a
    /// ``ContentInfo``).
    ///
    /// Raises :exc:`NotImplementedError` when built without a crypto backend.
    #[staticmethod]
    #[pyo3(signature = (plaintext, recipients, *, content_enc_alg = None))]
    fn create(
        py: Python<'_>,
        plaintext: &[u8],
        recipients: &Bound<'_, pyo3::types::PyList>,
        content_enc_alg: Option<&Bound<'_, PyAny>>,
    ) -> PyResult<Self> {
        // Resolve content-encryption OID (default: id-aes256-CBC).
        let content_enc_oid = match content_enc_alg {
            Some(obj) => super::super::oid_from_pyany(obj)?,
            None => synta::ObjectIdentifier::new(synta_certificate::pkcs12_types::ID_AES256_CBC)
                .expect("id-aes256-cbc is a valid OID"),
        };

        use pyo3::types::PyTuple;
        use synta_certificate::KeyWrapAlgorithm;

        // Collect (cert_der, KeyWrapAlgorithm) pairs from the Python list.
        let mut recipient_pairs: Vec<(Vec<u8>, KeyWrapAlgorithm)> =
            Vec::with_capacity(recipients.len());

        for item in recipients.iter() {
            let tup = item.cast::<PyTuple>().map_err(|_| {
                pyo3::exceptions::PyTypeError::new_err(
                    "each recipient entry must be a (cert, key_wrap_oid) tuple",
                )
            })?;
            if tup.len() != 2 {
                return Err(pyo3::exceptions::PyTypeError::new_err(
                    "each recipient tuple must have exactly 2 elements: (cert, key_wrap_oid)",
                ));
            }

            // First element: certificate (Certificate object or raw DER bytes).
            let cert_item = tup.get_item(0)?;
            let cert_der: Vec<u8> =
                if let Ok(py_cert) = cert_item.cast::<super::super::cert::PyCertificate>() {
                    py_cert.get().raw.to_vec()
                } else if let Ok(py_bytes) = cert_item.cast::<PyBytes>() {
                    py_bytes.as_bytes().to_vec()
                } else {
                    return Err(pyo3::exceptions::PyTypeError::new_err(
                        "cert must be a synta.Certificate or bytes",
                    ));
                };

            // Second element: key-wrap OID → KeyWrapAlgorithm enum.
            let kw_oid = super::super::oid_from_pyany(&tup.get_item(1)?)?;
            let key_wrap = if kw_oid.components() == synta_certificate::oids::RSAES_OAEP {
                KeyWrapAlgorithm::RsaOaepSha256
            } else if kw_oid.components() == synta_certificate::oids::RSA_ENCRYPTION {
                KeyWrapAlgorithm::RsaPkcs1v15
            } else {
                return Err(pyo3::exceptions::PyValueError::new_err(format!(
                    "unsupported key-wrap OID {:?}; use RSAES_OAEP or RSA_ENCRYPTION",
                    kw_oid.components(),
                )));
            };

            recipient_pairs.push((cert_der, key_wrap));
        }

        if recipient_pairs.is_empty() {
            return Err(pyo3::exceptions::PyValueError::new_err(
                "recipients list must not be empty",
            ));
        }

        let pairs_ref: Vec<(&[u8], KeyWrapAlgorithm)> = recipient_pairs
            .iter()
            .map(|(d, kw)| (d.as_slice(), *kw))
            .collect();
        let der = synta_certificate::default_create_enveloped_data(
            plaintext,
            &pairs_ref,
            content_enc_oid.components(),
        )
        .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
        let py_bytes = PyBytes::new(py, &der).unbind();
        Self::from_der(py, py_bytes.into_bound(py))
    }

    /// CMS version integer.
    #[getter]
    fn version(&self) -> PyResult<i64> {
        self.enveloped_data()?.version.as_i64().map_err(|_| {
            pyo3::exceptions::PyValueError::new_err(
                "EnvelopedData version field is out of i64 range",
            )
        })
    }

    /// DER-encoded ``OriginatorInfo`` SEQUENCE, or ``None`` if absent.
    #[getter]
    fn originator_info<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
        if let Some(cached) = self.originator_info_cache.get() {
            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
        }
        let computed = match &self.enveloped_data()?.originator_info {
            None => None,
            Some(oi) => {
                let mut enc = synta::Encoder::new(Encoding::Der);
                oi.encode(&mut enc)
                    .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
                let bytes = enc
                    .finish()
                    .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
                Some(PyBytes::new(py, &bytes))
            }
        };
        let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
        let _ = self.originator_info_cache.set(to_store);
        Ok(computed)
    }

    /// Raw TLV bytes of the ``RecipientInfos`` SET field.
    #[getter]
    fn recipient_infos<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
        if let Some(cached) = self.recipient_infos_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let py_bytes = PyBytes::new(py, self.enveloped_data()?.recipient_infos.as_bytes()).unbind();
        let _ = self.recipient_infos_cache.set(py_bytes.clone_ref(py));
        Ok(py_bytes.into_bound(py))
    }

    /// OID identifying the encrypted content type.
    #[getter]
    fn content_type<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyObjectIdentifier>> {
        if let Some(cached) = self.content_type_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let obj = Py::new(
            py,
            PyObjectIdentifier::from_oid(
                self.enveloped_data()?
                    .encrypted_content_info
                    .content_type
                    .clone(),
            ),
        )?;
        let _ = self.content_type_cache.set(obj.clone_ref(py));
        Ok(obj.into_bound(py))
    }

    /// OID of the content-encryption algorithm.
    #[getter]
    fn content_encryption_algorithm_oid<'py>(
        &self,
        py: Python<'py>,
    ) -> PyResult<Bound<'py, PyObjectIdentifier>> {
        if let Some(cached) = self.content_encryption_algorithm_oid_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let obj = Py::new(
            py,
            PyObjectIdentifier::from_oid(
                self.enveloped_data()?
                    .encrypted_content_info
                    .content_encryption_algorithm
                    .algorithm
                    .clone(),
            ),
        )?;
        let _ = self
            .content_encryption_algorithm_oid_cache
            .set(obj.clone_ref(py));
        Ok(obj.into_bound(py))
    }

    /// Raw DER bytes of the content-encryption algorithm parameters, or ``None``.
    #[getter]
    fn content_encryption_algorithm_params<'py>(
        &self,
        py: Python<'py>,
    ) -> PyResult<Option<Bound<'py, PyBytes>>> {
        if let Some(cached) = self.content_encryption_algorithm_params_cache.get() {
            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
        }
        let computed = super::encode_element_opt(
            py,
            self.enveloped_data()?
                .encrypted_content_info
                .content_encryption_algorithm
                .parameters
                .as_ref(),
        )?;
        let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
        let _ = self.content_encryption_algorithm_params_cache.set(to_store);
        Ok(computed)
    }

    /// Raw bytes of the encrypted content OCTET STRING value (IMPLICIT ``[0]``),
    /// or ``None`` if absent.
    #[getter]
    fn encrypted_content<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
        if let Some(cached) = self.encrypted_content_cache.get() {
            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
        }
        let computed = self
            .enveloped_data()?
            .encrypted_content_info
            .encrypted_content
            .as_ref()
            .map(|c| PyBytes::new(py, c.as_bytes()));
        let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
        let _ = self.encrypted_content_cache.set(to_store);
        Ok(computed)
    }

    /// Raw content bytes of the ``unprotectedAttrs`` field (IMPLICIT ``[1]``),
    /// or ``None`` if absent.
    #[getter]
    fn unprotected_attrs<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
        if let Some(cached) = self.unprotected_attrs_cache.get() {
            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
        }
        let computed = self
            .enveloped_data()?
            .unprotected_attrs
            .as_ref()
            .map(|a| PyBytes::new(py, a.as_bytes()));
        let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
        let _ = self.unprotected_attrs_cache.set(to_store);
        Ok(computed)
    }

    fn __repr__(&self) -> PyResult<String> {
        Ok(format!(
            "EnvelopedData(version={})",
            self.enveloped_data()?.version.as_i64().unwrap_or(0),
        ))
    }
}

// ── PyEncryptedData ───────────────────────────────────────────────────────────

/// CMS ``EncryptedData`` (RFC 5652 §8) accessible from Python.
///
/// Obtained by parsing the inner content of a :class:`ContentInfo` whose
/// ``content_type_oid`` equals :data:`~synta.cms.ID_ENCRYPTED_DATA`.
#[pyclass(frozen, name = "EncryptedData")]
pub struct PyEncryptedData {
    _data: Py<PyBytes>,
    raw: &'static [u8],
    inner: OnceLock<Box<synta_certificate::cms_rfc5652_types::EncryptedData<'static>>>,
    content_type_cache: OnceLock<Py<PyObjectIdentifier>>,
    content_encryption_algorithm_oid_cache: OnceLock<Py<PyObjectIdentifier>>,
    content_encryption_algorithm_params_cache: OnceLock<Option<Py<PyBytes>>>,
    encrypted_content_cache: OnceLock<Option<Py<PyBytes>>>,
    unprotected_attrs_cache: OnceLock<Option<Py<PyBytes>>>,
}

impl PyEncryptedData {
    fn encrypted_data(
        &self,
    ) -> PyResult<&synta_certificate::cms_rfc5652_types::EncryptedData<'static>> {
        if let Some(v) = self.inner.get() {
            return Ok(v.as_ref());
        }
        let mut decoder = Decoder::new(self.raw, Encoding::Ber);
        let decoded = decoder.decode().map_err(|e| {
            pyo3::exceptions::PyValueError::new_err(format!("EncryptedData BER decode failed: {e}"))
        })?;
        let _ = self.inner.set(Box::new(decoded));
        Ok(self.inner.get().unwrap().as_ref())
    }

    fn from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
        let py_bytes = data.unbind();
        let raw: &'static [u8] = unsafe {
            let s = py_bytes.bind(py).as_bytes();
            std::slice::from_raw_parts(s.as_ptr(), s.len())
        };
        {
            let mut d = Decoder::new(raw, Encoding::Ber);
            d.read_tag()
                .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
            d.read_length()
                .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
        }
        Ok(Self {
            _data: py_bytes,
            raw,
            inner: OnceLock::new(),
            content_type_cache: OnceLock::new(),
            content_encryption_algorithm_oid_cache: OnceLock::new(),
            content_encryption_algorithm_params_cache: OnceLock::new(),
            encrypted_content_cache: OnceLock::new(),
            unprotected_attrs_cache: OnceLock::new(),
        })
    }
}

#[pymethods]
impl PyEncryptedData {
    /// Parse a DER- or BER-encoded CMS ``EncryptedData`` SEQUENCE.
    #[staticmethod]
    #[pyo3(name = "from_der")]
    fn py_from_der(py: Python<'_>, data: Bound<'_, PyBytes>) -> PyResult<Self> {
        Self::from_der(py, data)
    }

    /// Return the original bytes passed to :meth:`from_der`.
    fn to_der<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
        self._data.clone_ref(py).into_bound(py)
    }

    /// CMS version integer.
    #[getter]
    fn version(&self) -> PyResult<i64> {
        Ok(self.encrypted_data()?.version.as_i64().unwrap_or(0))
    }

    /// OID identifying the encrypted content type.
    #[getter]
    fn content_type<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyObjectIdentifier>> {
        if let Some(cached) = self.content_type_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let obj = Py::new(
            py,
            PyObjectIdentifier::from_oid(
                self.encrypted_data()?
                    .encrypted_content_info
                    .content_type
                    .clone(),
            ),
        )?;
        let _ = self.content_type_cache.set(obj.clone_ref(py));
        Ok(obj.into_bound(py))
    }

    /// OID of the content-encryption algorithm.
    #[getter]
    fn content_encryption_algorithm_oid<'py>(
        &self,
        py: Python<'py>,
    ) -> PyResult<Bound<'py, PyObjectIdentifier>> {
        if let Some(cached) = self.content_encryption_algorithm_oid_cache.get() {
            return Ok(cached.clone_ref(py).into_bound(py));
        }
        let obj = Py::new(
            py,
            PyObjectIdentifier::from_oid(
                self.encrypted_data()?
                    .encrypted_content_info
                    .content_encryption_algorithm
                    .algorithm
                    .clone(),
            ),
        )?;
        let _ = self
            .content_encryption_algorithm_oid_cache
            .set(obj.clone_ref(py));
        Ok(obj.into_bound(py))
    }

    /// Raw DER bytes of the content-encryption algorithm parameters, or ``None``.
    #[getter]
    fn content_encryption_algorithm_params<'py>(
        &self,
        py: Python<'py>,
    ) -> PyResult<Option<Bound<'py, PyBytes>>> {
        if let Some(cached) = self.content_encryption_algorithm_params_cache.get() {
            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
        }
        let computed = super::encode_element_opt(
            py,
            self.encrypted_data()?
                .encrypted_content_info
                .content_encryption_algorithm
                .parameters
                .as_ref(),
        )?;
        let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
        let _ = self.content_encryption_algorithm_params_cache.set(to_store);
        Ok(computed)
    }

    /// Raw bytes of the encrypted content OCTET STRING value (IMPLICIT ``[0]``),
    /// or ``None`` if absent.
    #[getter]
    fn encrypted_content<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
        if let Some(cached) = self.encrypted_content_cache.get() {
            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
        }
        let computed = self
            .encrypted_data()?
            .encrypted_content_info
            .encrypted_content
            .as_ref()
            .map(|c| PyBytes::new(py, c.as_bytes()));
        let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
        let _ = self.encrypted_content_cache.set(to_store);
        Ok(computed)
    }

    /// Raw content bytes of the ``unprotectedAttrs`` field (IMPLICIT ``[1]``),
    /// or ``None`` if absent.
    #[getter]
    fn unprotected_attrs<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyBytes>>> {
        if let Some(cached) = self.unprotected_attrs_cache.get() {
            return Ok(cached.as_ref().map(|b| b.clone_ref(py).into_bound(py)));
        }
        let computed = self
            .encrypted_data()?
            .unprotected_attrs
            .as_ref()
            .map(|a| PyBytes::new(py, a.as_bytes()));
        let to_store = computed.as_ref().map(|b| b.as_unbound().clone_ref(py));
        let _ = self.unprotected_attrs_cache.set(to_store);
        Ok(computed)
    }

    /// Build a new :class:`EncryptedData` by encrypting ``plaintext`` with
    /// ``key``.
    ///
    /// - ``algorithm_oid``: dotted-decimal OID string or
    ///   :class:`~synta.ObjectIdentifier` for the content-encryption algorithm
    ///   (e.g. ``"2.16.840.1.101.3.4.1.2"`` for AES-128-CBC).
    /// - ``content_type_oid``: optional content-type OID (default:
    ///   ``id-data`` = ``"1.2.840.113549.1.7.1"``).
    ///
    /// A fresh random IV is generated for each call.
    ///
    /// Raises :exc:`NotImplementedError` when built without a crypto backend.
    #[staticmethod]
    #[pyo3(signature = (plaintext, key, algorithm_oid, content_type_oid = None))]
    fn create(
        py: Python<'_>,
        plaintext: &[u8],
        key: &[u8],
        algorithm_oid: &Bound<'_, PyAny>,
        content_type_oid: Option<&Bound<'_, PyAny>>,
    ) -> PyResult<Self> {
        let enc_alg_oid = super::super::oid_from_pyany(algorithm_oid)?;

        let ct_oid = match content_type_oid {
            Some(obj) => super::super::oid_from_pyany(obj)?,
            None => synta::ObjectIdentifier::new(synta_certificate::pkcs7_types::ID_DATA)
                .expect("id-data is a valid OID"),
        };

        use synta_certificate::CmsEncryptor as _;
        let der = synta_certificate::DefaultCrypto
            .create_encrypted_data(
                ct_oid.components(),
                enc_alg_oid.components(),
                plaintext,
                key,
            )
            .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
        let py_bytes = PyBytes::new(py, &der).unbind();
        Self::from_der(py, py_bytes.into_bound(py))
    }

    /// Decrypt the encrypted content using a raw symmetric key.
    ///
    /// ``key`` must be the raw symmetric key bytes matching the
    /// content-encryption algorithm (e.g. 16 bytes for AES-128-CBC,
    /// 32 bytes for AES-256-CBC).
    ///
    /// Raises :exc:`NotImplementedError` when built without a crypto backend.
    fn decrypt<'py>(&self, py: Python<'py>, key: &[u8]) -> PyResult<Bound<'py, PyBytes>> {
        use synta_certificate::CmsDecryptor as _;
        let ed = self.encrypted_data()?;
        let mut enc = synta::Encoder::new(synta::Encoding::Der);
        ed.encrypted_content_info
            .content_encryption_algorithm
            .encode(&mut enc)
            .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
        let algorithm_der = enc
            .finish()
            .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
        let ciphertext = ed
            .encrypted_content_info
            .encrypted_content
            .as_ref()
            .ok_or_else(|| {
                pyo3::exceptions::PyValueError::new_err(
                    "EncryptedData has no encryptedContent field",
                )
            })?
            .as_bytes();
        let plaintext = synta_certificate::DefaultCrypto
            .decrypt(&algorithm_der, ciphertext, key)
            .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
        Ok(PyBytes::new(py, &plaintext))
    }

    fn __repr__(&self) -> PyResult<String> {
        Ok(format!(
            "EncryptedData(version={})",
            self.encrypted_data()?.version.as_i64().unwrap_or(0),
        ))
    }
}