Add support for decrypting S/MIME messages

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

@@ -6,6 +6,7 @@ import typing
 
 from cryptography import x509
 from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
 from cryptography.hazmat.primitives.serialization import pkcs7
 
 def serialize_certificates(
@@ -22,6 +23,13 @@ def sign_and_serialize(
     encoding: serialization.Encoding,
     options: typing.Iterable[pkcs7.PKCS7Options],
 ) -> bytes: ...
+def pem_to_der(data: bytes) -> bytes: ...
+def deserialize_and_decrypt(
+    data: bytes,
+    certificate: x509.Certificate,
+    private_key: rsa.RSAPrivateKey,
+    options: typing.Iterable[pkcs7.PKCS7Options],
+) -> bytes: ...
 def load_pem_pkcs7_certificates(
     data: bytes,
 ) -> list[x509.Certificate]: ...

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

@@ -13,13 +13,6 @@ 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

@@ -263,6 +263,81 @@ def encrypt(
         return rust_pkcs7.encrypt_and_serialize(self, encoding, options)
 
 
+def pkcs7_decrypt_der(
+    data: bytes,
+    certificate: x509.Certificate,
+    private_key: rsa.RSAPrivateKey,
+    options: typing.Iterable[PKCS7Options],
+) -> bytes:
+    return _pkcs7_decrypt(data, certificate, private_key, options)
+
+
+def pkcs7_decrypt_pem(
+    data: bytes,
+    certificate: x509.Certificate,
+    private_key: rsa.RSAPrivateKey,
+    options: typing.Iterable[PKCS7Options],
+) -> bytes:
+    data = rust_pkcs7.pem_to_der(data)
+    return _pkcs7_decrypt(data, certificate, private_key, options)
+
+
+def pkcs7_decrypt_smime(
+    data: bytes,
+    certificate: x509.Certificate,
+    private_key: rsa.RSAPrivateKey,
+    options: typing.Iterable[PKCS7Options],
+) -> bytes:
+    data = _smime_enveloped_decode(data)
+    return _pkcs7_decrypt(data, certificate, private_key, options)
+
+
+def _pkcs7_decrypt(
+    data: bytes,
+    certificate: x509.Certificate,
+    private_key: rsa.RSAPrivateKey,
+    options: typing.Iterable[PKCS7Options],
+) -> bytes:
+    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,
+        )
+
+    options = list(options)
+    if not all(isinstance(x, PKCS7Options) for x in options):
+        raise ValueError("options must be from the PKCS7Options enum")
+    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")
+
+    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.")
+
+    if not isinstance(private_key, rsa.RSAPrivateKey):
+        raise TypeError("Only RSA private keys are supported at this time.")
+
+    return rust_pkcs7.deserialize_and_decrypt(
+        data, certificate, private_key, options
+    )
+
+
 def _smime_signed_encode(
     data: bytes, signature: bytes, micalg: str, text_mode: bool
 ) -> bytes:
@@ -328,6 +403,16 @@ def _smime_enveloped_encode(data: bytes) -> bytes:
     return m.as_bytes(policy=m.policy.clone(linesep="\n", max_line_length=0))
 
 
+def _smime_enveloped_decode(data: bytes) -> bytes:
+    m = email.message_from_bytes(data)
+    if m.get_content_type() not in {
+        "application/x-pkcs7-mime",
+        "application/pkcs7-mime",
+    }:
+        raise ValueError("Not an S/MIME enveloped message")
+    return bytes(m.get_payload(decode=True))
+
+
 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/pkcs7.rs

@@ -9,15 +9,15 @@ pub const PKCS7_SIGNED_DATA_OID: asn1::ObjectIdentifier = asn1::oid!(1, 2, 840,
 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)]
+#[derive(asn1::Asn1Write, asn1::Asn1Read)]
 pub struct ContentInfo<'a> {
     pub _content_type: asn1::DefinedByMarker<asn1::ObjectIdentifier>,
 
     #[defined_by(_content_type)]
     pub content: Content<'a>,
 }
 
-#[derive(asn1::Asn1DefinedByWrite)]
+#[derive(asn1::Asn1DefinedByWrite, asn1::Asn1DefinedByRead)]
 pub enum Content<'a> {
     #[defined_by(PKCS7_ENVELOPED_DATA_OID)]
     EnvelopedData(asn1::Explicit<Box<EnvelopedData<'a>>, 0>),
@@ -29,22 +29,38 @@ pub enum Content<'a> {
     EncryptedData(asn1::Explicit<EncryptedData<'a>, 0>),
 }
 
-#[derive(asn1::Asn1Write)]
+#[derive(asn1::Asn1Write, asn1::Asn1Read)]
 pub struct SignedData<'a> {
     pub version: u8,
-    pub digest_algorithms: asn1::SetOfWriter<'a, common::AlgorithmIdentifier<'a>>,
+    pub digest_algorithms: common::Asn1ReadableOrWritable<
+        asn1::SetOf<'a, common::AlgorithmIdentifier<'a>>,
+        asn1::SetOfWriter<'a, common::AlgorithmIdentifier<'a>>,
+    >,
     pub content_info: ContentInfo<'a>,
     #[implicit(0)]
-    pub certificates: Option<asn1::SetOfWriter<'a, &'a certificate::Certificate<'a>>>,
+    pub certificates: Option<
+        common::Asn1ReadableOrWritable<
+            asn1::SetOf<'a, certificate::Certificate<'a>>,
+            asn1::SetOfWriter<'a, &'a certificate::Certificate<'a>>,
+        >,
+    >,
 
     // We don't ever supply any of these, so for now, don't fill out the fields.
     #[implicit(1)]
-    pub crls: Option<asn1::SetOfWriter<'a, asn1::Sequence<'a>>>,
-
-    pub signer_infos: asn1::SetOfWriter<'a, SignerInfo<'a>>,
+    pub crls: Option<
+        common::Asn1ReadableOrWritable<
+            asn1::SetOf<'a, asn1::Sequence<'a>>,
+            asn1::SetOfWriter<'a, asn1::Sequence<'a>>,
+        >,
+    >,
+
+    pub signer_infos: common::Asn1ReadableOrWritable<
+        asn1::SetOf<'a, SignerInfo<'a>>,
+        asn1::SetOfWriter<'a, SignerInfo<'a>>,
+    >,
 }
 
-#[derive(asn1::Asn1Write)]
+#[derive(asn1::Asn1Write, asn1::Asn1Read)]
 pub struct SignerInfo<'a> {
     pub version: u8,
     pub issuer_and_serial_number: IssuerAndSerialNumber<'a>,
@@ -59,42 +75,45 @@ pub struct SignerInfo<'a> {
     pub unauthenticated_attributes: Option<csr::Attributes<'a>>,
 }
 
-#[derive(asn1::Asn1Write)]
+#[derive(asn1::Asn1Write, asn1::Asn1Read)]
 pub struct EnvelopedData<'a> {
     pub version: u8,
-    pub recipient_infos: asn1::SetOfWriter<'a, RecipientInfo<'a>>,
+    pub recipient_infos: common::Asn1ReadableOrWritable<
+        asn1::SetOf<'a, RecipientInfo<'a>>,
+        asn1::SetOfWriter<'a, RecipientInfo<'a>>,
+    >,
     pub encrypted_content_info: EncryptedContentInfo<'a>,
 }
 
-#[derive(asn1::Asn1Write)]
+#[derive(asn1::Asn1Write, asn1::Asn1Read)]
 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)]
+#[derive(asn1::Asn1Write, asn1::Asn1Read)]
 pub struct IssuerAndSerialNumber<'a> {
     pub issuer: name::Name<'a>,
     pub serial_number: asn1::BigInt<'a>,
 }
 
-#[derive(asn1::Asn1Write)]
+#[derive(asn1::Asn1Write, asn1::Asn1Read)]
 pub struct EncryptedData<'a> {
     pub version: u8,
     pub encrypted_content_info: EncryptedContentInfo<'a>,
 }
 
-#[derive(asn1::Asn1Write)]
+#[derive(asn1::Asn1Write, asn1::Asn1Read)]
 pub struct EncryptedContentInfo<'a> {
     pub content_type: asn1::ObjectIdentifier,
     pub content_encryption_algorithm: common::AlgorithmIdentifier<'a>,
     #[implicit(0)]
     pub encrypted_content: Option<&'a [u8]>,
 }
 
-#[derive(asn1::Asn1Write)]
+#[derive(asn1::Asn1Write, asn1::Asn1Read)]
 pub struct DigestInfo<'a> {
     pub algorithm: common::AlgorithmIdentifier<'a>,
     pub digest: &'a [u8],