Add support for encrypting S/MIME messages

src/cryptography/hazmat/bindings/_rust/pkcs7.pyi

@@ -12,6 +12,11 @@ def serialize_certificates(
     certs: list[x509.Certificate],
     encoding: serialization.Encoding,
 ) -> bytes: ...
+def encrypt_and_serialize(
+    builder: pkcs7.PKCS7EnvelopeBuilder,
+    encoding: serialization.Encoding,
+    options: typing.Iterable[pkcs7.PKCS7Options],
+) -> bytes: ...
 def sign_and_serialize(
     builder: pkcs7.PKCS7SignatureBuilder,
     encoding: serialization.Encoding,

src/cryptography/hazmat/bindings/_rust/test_support.pyi

@@ -13,6 +13,13 @@ class TestCertificate:
     subject_value_tags: list[int]
 
 def test_parse_certificate(data: bytes) -> TestCertificate: ...
+def pkcs7_decrypt(
+    encoding: serialization.Encoding,
+    msg: bytes,
+    pkey: serialization.pkcs7.PKCS7PrivateKeyTypes,
+    cert_recipient: x509.Certificate,
+    options: list[pkcs7.PKCS7Options],
+) -> bytes: ...
 def pkcs7_verify(
     encoding: serialization.Encoding,
     sig: bytes,

src/cryptography/hazmat/primitives/serialization/pkcs7.py

@@ -12,6 +12,7 @@
 import typing
 
 from cryptography import utils, x509
+from cryptography.exceptions import UnsupportedAlgorithm, _Reasons
 from cryptography.hazmat.bindings._rust import pkcs7 as rust_pkcs7
 from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.hazmat.primitives.asymmetric import ec, padding, rsa
@@ -177,7 +178,92 @@ def sign(
         return rust_pkcs7.sign_and_serialize(self, encoding, options)
 
 
-def _smime_encode(
+class PKCS7EnvelopeBuilder:
+    def __init__(
+        self,
+        *,
+        _data: bytes | None = None,
+        _recipients: list[x509.Certificate] | None = None,
+    ):
+        from cryptography.hazmat.backends.openssl.backend import (
+            backend as ossl,
+        )
+
+        if not ossl.rsa_encryption_supported(padding=padding.PKCS1v15()):
+            raise UnsupportedAlgorithm(
+                "RSA with PKCS1 v1.5 padding is not supported by this version"
+                " of OpenSSL.",
+                _Reasons.UNSUPPORTED_PADDING,
+            )
+        self._data = _data
+        self._recipients = _recipients if _recipients is not None else []
+
+    def set_data(self, data: bytes) -> PKCS7EnvelopeBuilder:
+        _check_byteslike("data", data)
+        if self._data is not None:
+            raise ValueError("data may only be set once")
+
+        return PKCS7EnvelopeBuilder(_data=data, _recipients=self._recipients)
+
+    def add_recipient(
+        self,
+        certificate: x509.Certificate,
+    ) -> PKCS7EnvelopeBuilder:
+        if not isinstance(certificate, x509.Certificate):
+            raise TypeError("certificate must be a x509.Certificate")
+
+        if not isinstance(certificate.public_key(), rsa.RSAPublicKey):
+            raise TypeError("Only RSA keys are supported at this time.")
+
+        return PKCS7EnvelopeBuilder(
+            _data=self._data,
+            _recipients=[
+                *self._recipients,
+                certificate,
+            ],
+        )
+
+    def encrypt(
+        self,
+        encoding: serialization.Encoding,
+        options: typing.Iterable[PKCS7Options],
+    ) -> bytes:
+        if len(self._recipients) == 0:
+            raise ValueError("Must have at least one recipient")
+        if self._data is None:
+            raise ValueError("You must add data to encrypt")
+        options = list(options)
+        if not all(isinstance(x, PKCS7Options) for x in options):
+            raise ValueError("options must be from the PKCS7Options enum")
+        if encoding not in (
+            serialization.Encoding.PEM,
+            serialization.Encoding.DER,
+            serialization.Encoding.SMIME,
+        ):
+            raise ValueError(
+                "Must be PEM, DER, or SMIME from the Encoding enum"
+            )
+
+        # Only allow options that make sense for encryption
+        if any(
+            opt not in [PKCS7Options.Text, PKCS7Options.Binary]
+            for opt in options
+        ):
+            raise ValueError(
+                "Only the following options are supported for encryption: "
+                "Text, Binary"
+            )
+        elif PKCS7Options.Text in options and PKCS7Options.Binary in options:
+            # OpenSSL accepts both options at the same time, but ignores Text.
+            # We fail defensively to avoid unexpected outputs.
+            raise ValueError(
+                "Cannot use Binary and Text options at the same time"
+            )
+
+        return rust_pkcs7.encrypt_and_serialize(self, encoding, options)
+
+
+def _smime_signed_encode(
     data: bytes, signature: bytes, micalg: str, text_mode: bool
 ) -> bytes:
     # This function works pretty hard to replicate what OpenSSL does
@@ -225,6 +311,23 @@ def _smime_encode(
     return fp.getvalue()
 
 
+def _smime_enveloped_encode(data: bytes) -> bytes:
+    m = email.message.Message()
+    m.add_header("MIME-Version", "1.0")
+    m.add_header("Content-Disposition", "attachment", filename="smime.p7m")
+    m.add_header(
+        "Content-Type",
+        "application/pkcs7-mime",
+        smime_type="enveloped-data",
+        name="smime.p7m",
+    )
+    m.add_header("Content-Transfer-Encoding", "base64")
+
+    m.set_payload(email.base64mime.body_encode(data, maxlinelen=65))
+
+    return m.as_bytes(policy=m.policy.clone(linesep="\n", max_line_length=0))
+
+
 class OpenSSLMimePart(email.message.MIMEPart):
     # A MIMEPart subclass that replicates OpenSSL's behavior of not including
     # a newline if there are no headers.

src/rust/cryptography-x509/src/common.rs

@@ -136,6 +136,16 @@ pub enum AlgorithmParameters<'a> {
     #[defined_by(oid::HMAC_WITH_SHA256_OID)]
     HmacWithSha256(asn1::Null),
 
+    // Used only in PKCS#7 AlgorithmIdentifiers
+    // https://datatracker.ietf.org/doc/html/rfc3565#section-4.1
+    //
+    // From RFC 3565 section 4.1:
+    // The AlgorithmIdentifier parameters field MUST be present, and the
+    // parameters field MUST contain a AES-IV:
+    //
+    // AES-IV ::= OCTET STRING (SIZE(16))
+    #[defined_by(oid::AES_128_CBC_OID)]
+    Aes128Cbc([u8; 16]),
     #[defined_by(oid::AES_256_CBC_OID)]
     Aes256Cbc([u8; 16]),
 

src/rust/cryptography-x509/src/pkcs7.rs

@@ -6,6 +6,7 @@ use crate::{certificate, common, csr, name};
 
 pub const PKCS7_DATA_OID: asn1::ObjectIdentifier = asn1::oid!(1, 2, 840, 113549, 1, 7, 1);
 pub const PKCS7_SIGNED_DATA_OID: asn1::ObjectIdentifier = asn1::oid!(1, 2, 840, 113549, 1, 7, 2);
+pub const PKCS7_ENVELOPED_DATA_OID: asn1::ObjectIdentifier = asn1::oid!(1, 2, 840, 113549, 1, 7, 3);
 pub const PKCS7_ENCRYPTED_DATA_OID: asn1::ObjectIdentifier = asn1::oid!(1, 2, 840, 113549, 1, 7, 6);
 
 #[derive(asn1::Asn1Write)]
@@ -18,6 +19,8 @@ pub struct ContentInfo<'a> {
 
 #[derive(asn1::Asn1DefinedByWrite)]
 pub enum Content<'a> {
+    #[defined_by(PKCS7_ENVELOPED_DATA_OID)]
+    EnvelopedData(asn1::Explicit<Box<EnvelopedData<'a>>, 0>),
     #[defined_by(PKCS7_SIGNED_DATA_OID)]
     SignedData(asn1::Explicit<Box<SignedData<'a>>, 0>),
     #[defined_by(PKCS7_DATA_OID)]
@@ -56,6 +59,21 @@ pub struct SignerInfo<'a> {
     pub unauthenticated_attributes: Option<csr::Attributes<'a>>,
 }
 
+#[derive(asn1::Asn1Write)]
+pub struct EnvelopedData<'a> {
+    pub version: u8,
+    pub recipient_infos: asn1::SetOfWriter<'a, RecipientInfo<'a>>,
+    pub encrypted_content_info: EncryptedContentInfo<'a>,
+}
+
+#[derive(asn1::Asn1Write)]
+pub struct RecipientInfo<'a> {
+    pub version: u8,
+    pub issuer_and_serial_number: IssuerAndSerialNumber<'a>,
+    pub key_encryption_algorithm: common::AlgorithmIdentifier<'a>,
+    pub encrypted_key: &'a [u8],
+}
+
 #[derive(asn1::Asn1Write)]
 pub struct IssuerAndSerialNumber<'a> {
     pub issuer: name::Name<'a>,

src/rust/src/pkcs12.rs

@@ -79,7 +79,7 @@ impl PKCS12Certificate {
     }
 }
 
-fn symmetric_encrypt(
+pub(crate) fn symmetric_encrypt(
     py: pyo3::Python<'_>,
     algorithm: pyo3::Bound<'_, pyo3::PyAny>,
     mode: pyo3::Bound<'_, pyo3::PyAny>,

src/rust/src/pkcs7.rs

@@ -6,7 +6,9 @@ use std::borrow::Cow;
 use std::collections::HashMap;
 use std::ops::Deref;
 
+use cryptography_x509::common::{AlgorithmIdentifier, AlgorithmParameters};
 use cryptography_x509::csr::Attribute;
+use cryptography_x509::pkcs7::PKCS7_DATA_OID;
 use cryptography_x509::{common, oid, pkcs7};
 use once_cell::sync::Lazy;
 #[cfg(not(CRYPTOGRAPHY_IS_BORINGSSL))]
@@ -18,6 +20,7 @@ use pyo3::IntoPy;
 use crate::asn1::encode_der_data;
 use crate::buf::CffiBuf;
 use crate::error::{CryptographyError, CryptographyResult};
+use crate::pkcs12::symmetric_encrypt;
 #[cfg(not(CRYPTOGRAPHY_IS_BORINGSSL))]
 use crate::x509::certificate::load_der_x509_certificate;
 use crate::{exceptions, types, x509};
@@ -75,6 +78,90 @@ fn serialize_certificates<'p>(
     encode_der_data(py, "PKCS7".to_string(), content_info_bytes, encoding)
 }
 
+#[pyo3::pyfunction]
+fn encrypt_and_serialize<'p>(
+    py: pyo3::Python<'p>,
+    builder: &pyo3::Bound<'p, pyo3::PyAny>,
+    encoding: &pyo3::Bound<'p, pyo3::PyAny>,
+    options: &pyo3::Bound<'p, pyo3::types::PyList>,
+) -> CryptographyResult<pyo3::Bound<'p, pyo3::types::PyBytes>> {
+    let raw_data: CffiBuf<'p> = builder.getattr(pyo3::intern!(py, "_data"))?.extract()?;
+    let text_mode = options.contains(types::PKCS7_TEXT.get(py)?)?;
+    let data_with_header = if options.contains(types::PKCS7_BINARY.get(py)?)? {
+        Cow::Borrowed(raw_data.as_bytes())
+    } else {
+        smime_canonicalize(raw_data.as_bytes(), text_mode).0
+    };
+
+    // The message is encrypted with AES-128-CBC, which the S/MIME v3.2 RFC
+    // specifies as MUST support (https://datatracker.ietf.org/doc/html/rfc5751#section-2.7)
+    let key = types::OS_URANDOM.get(py)?.call1((16,))?;
+    let aes128_algorithm = types::AES128.get(py)?.call1((&key,))?;
+    let iv = types::OS_URANDOM.get(py)?.call1((16,))?;
+    let cbc_mode = types::CBC.get(py)?.call1((&iv,))?;
+
+    let encrypted_content = symmetric_encrypt(py, aes128_algorithm, cbc_mode, &data_with_header)?;
+
+    let py_recipients: Vec<pyo3::Bound<'p, x509::certificate::Certificate>> = builder
+        .getattr(pyo3::intern!(py, "_recipients"))?
+        .extract()?;
+
+    let mut recipient_infos = vec![];
+    let padding = types::PKCS1V15.get(py)?.call0()?;
+    let ka_bytes = cryptography_keepalive::KeepAlive::new();
+    for cert in py_recipients.iter() {
+        // Currently, keys are encrypted with RSA (PKCS #1 v1.5), which the S/MIME v3.2 RFC
+        // specifies as MUST support (https://datatracker.ietf.org/doc/html/rfc5751#section-2.3)
+        let encrypted_key = cert
+            .call_method0(pyo3::intern!(py, "public_key"))?
+            .call_method1(pyo3::intern!(py, "encrypt"), (&key, &padding))?
+            .extract::<pyo3::pybacked::PyBackedBytes>()?;
+
+        recipient_infos.push(pkcs7::RecipientInfo {
+            version: 0,
+            issuer_and_serial_number: pkcs7::IssuerAndSerialNumber {
+                issuer: cert.get().raw.borrow_dependent().tbs_cert.issuer.clone(),
+                serial_number: cert.get().raw.borrow_dependent().tbs_cert.serial,
+            },
+            key_encryption_algorithm: AlgorithmIdentifier {
+                oid: asn1::DefinedByMarker::marker(),
+                params: AlgorithmParameters::Rsa(Some(())),
+            },
+            encrypted_key: ka_bytes.add(encrypted_key),
+        });
+    }
+
+    let enveloped_data = pkcs7::EnvelopedData {
+        version: 0,
+        recipient_infos: asn1::SetOfWriter::new(&recipient_infos),
+
+        encrypted_content_info: pkcs7::EncryptedContentInfo {
+            content_type: PKCS7_DATA_OID,
+            content_encryption_algorithm: AlgorithmIdentifier {
+                oid: asn1::DefinedByMarker::marker(),
+                params: AlgorithmParameters::Aes128Cbc(iv.extract()?),
+            },
+            encrypted_content: Some(&encrypted_content),
+        },
+    };
+
+    let content_info = pkcs7::ContentInfo {
+        _content_type: asn1::DefinedByMarker::marker(),
+        content: pkcs7::Content::EnvelopedData(asn1::Explicit::new(Box::new(enveloped_data))),
+    };
+    let ci_bytes = asn1::write_single(&content_info)?;
+
+    if encoding.is(&types::ENCODING_SMIME.get(py)?) {
+        Ok(types::SMIME_ENVELOPED_ENCODE
+            .get(py)?
+            .call1((&*ci_bytes,))?
+            .extract()?)
+    } else {
+        // Handles the DER, PEM, and error cases
+        encode_der_data(py, "PKCS7".to_string(), ci_bytes, encoding)
+    }
+}
+
 #[pyo3::pyfunction]
 fn sign_and_serialize<'p>(
     py: pyo3::Python<'p>,
@@ -256,7 +343,7 @@ fn sign_and_serialize<'p>(
             .map(|d| OIDS_TO_MIC_NAME[&d.oid()])
             .collect::<Vec<_>>()
             .join(",");
-        Ok(types::SMIME_ENCODE
+        Ok(types::SMIME_SIGNED_ENCODE
             .get(py)?
             .call1((&*data_without_header, &*ci_bytes, mic_algs, text_mode))?
             .extract()?)
@@ -412,8 +499,8 @@ fn load_der_pkcs7_certificates<'p>(
 pub(crate) mod pkcs7_mod {
     #[pymodule_export]
     use super::{
-        load_der_pkcs7_certificates, load_pem_pkcs7_certificates, serialize_certificates,
-        sign_and_serialize,
+        encrypt_and_serialize, load_der_pkcs7_certificates, load_pem_pkcs7_certificates,
+        serialize_certificates, sign_and_serialize,
     };
 }