test support pkcs7_encrypt with openssl

added algorithm to pkcs7_encrypt signature

refacto: decrypt function is clearer

flow is more natural

refacto: added all rust error tests

refacto: added another CA chain for checking

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_encrypt(
+    cert_recipients: list[x509.Certificate],
+    msg: bytes,
+    cipher: bytes,
+    options: list[pkcs7.PKCS7Options],
+    encoding: serialization.Encoding,
+) -> bytes: ...
 def pkcs7_decrypt(
     encoding: serialization.Encoding,
     msg: bytes,

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

@@ -440,7 +440,9 @@ def _smime_enveloped_encode(data: bytes) -> bytes:
 
 def _smime_enveloped_decode(data: bytes) -> bytes:
     m = email.message_from_bytes(data)
-    if m.get_content_type() != "application/pkcs7-mime":
+    # Content type can be application/x-pkcs7-mime or application/pkcs7-mime
+    # Could also be using the filename parameter, or no checks at all.
+    if "pkcs7-mime" not in m.get_content_type():
         raise ValueError("Not an S/MIME enveloped message")
     return bytes(m.get_payload(decode=True))
 

src/rust/src/pkcs7.rs

@@ -200,39 +200,14 @@ fn deserialize_and_decrypt<'p>(
         .getattr(pyo3::intern!(py, "_private_key"))?
         .extract()?;
 
-    //  Deserialize the content info
+    // Deserialize the content info
     let content_info =
         asn1::parse_single::<pkcs7::ContentInfo<'_>>(extracted_data.as_bytes()).unwrap();
     let plain_content = match content_info.content {
         pkcs7::Content::EnvelopedData(data) => {
-            // Extract enveloped data (if possible)
+            // Extract enveloped data
             let enveloped_data = data.into_inner();
 
-            // Get encrypted content
-            let encrypted_content = enveloped_data
-                .encrypted_content_info
-                .encrypted_content
-                .unwrap();
-
-            // Get algorithm
-            let algorithm = enveloped_data
-                .encrypted_content_info
-                .content_encryption_algorithm;
-
-            // Get initialization vector
-            let iv = match algorithm.params {
-                AlgorithmParameters::Aes128Cbc(iv) => pyo3::types::PyBytes::new_bound(py, &iv),
-                AlgorithmParameters::Aes256Cbc(iv) => pyo3::types::PyBytes::new_bound(py, &iv),
-                _ => {
-                    return Err(CryptographyError::from(
-                        exceptions::UnsupportedAlgorithm::new_err((
-                            "Only AES-128-CBC or AES-256-CBC are currently supported for decryption.",
-                            exceptions::Reasons::UNSUPPORTED_SERIALIZATION,
-                        )),
-                    ));
-                }
-            };
-
             // Get recipients, and the one matching with the given certificate (if any)
             let mut recipient_infos = enveloped_data.recipient_infos.unwrap_read().clone();
             let recipient_serial_number = recipient.get().raw.borrow_dependent().tbs_cert.serial;
@@ -242,13 +217,14 @@ fn deserialize_and_decrypt<'p>(
             });
 
             // Raise error when no recipient is found
+            // Unsure if this is the right exception to raise
             let recipient_info = match found_recipient_info {
                 Some(info) => info,
                 None => {
                     return Err(CryptographyError::from(
                         exceptions::AttributeNotFound::new_err((
                             "No recipient found that matches the given certificate.",
-                            exceptions::Reasons::UNSUPPORTED_CIPHER,
+                            exceptions::Reasons::UNSUPPORTED_X509,
                         )),
                     ));
                 }
@@ -263,24 +239,34 @@ fn deserialize_and_decrypt<'p>(
                 )?
                 .extract::<pyo3::pybacked::PyBackedBytes>()?;
 
-            // Decrypt the content using the key
-            // Should we use Python or Rust for the algorithm?
-            // TODO: are there any other algorithm to use for decryption in here?
-            let py_algorithm = match algorithm.params {
-                AlgorithmParameters::Aes128Cbc(_) => types::AES128.get(py)?.call1((key,))?,
-                AlgorithmParameters::Aes256Cbc(_) => types::AES256.get(py)?.call1((key,))?,
+            // Get algorithm
+            // TODO: implement all the possible algorithms
+            let algorithm_identifier = enveloped_data
+                .encrypted_content_info
+                .content_encryption_algorithm;
+            let (algorithm, mode) = match algorithm_identifier.params {
+                AlgorithmParameters::Aes128Cbc(iv) => (
+                    types::AES128.get(py)?.call1((key,))?,
+                    types::CBC
+                        .get(py)?
+                        .call1((pyo3::types::PyBytes::new_bound(py, &iv),))?,
+                ),
                 _ => {
                     return Err(CryptographyError::from(
                         exceptions::UnsupportedAlgorithm::new_err((
-                            "Only AES-128-CBC or AES-256-CBC are currently supported for decryption.",
+                            "Only AES-128-CBC is currently supported for decryption.",
                             exceptions::Reasons::UNSUPPORTED_SERIALIZATION,
                         )),
                     ));
                 }
             };
-            let cbc_mode = types::CBC.get(py)?.call1((iv,))?;
-            let decrypted_content =
-                symmetric_decrypt(py, py_algorithm, cbc_mode, encrypted_content)?;
+
+            // Decrypt the content using the key and proper algorithm
+            let encrypted_content = enveloped_data
+                .encrypted_content_info
+                .encrypted_content
+                .unwrap();
+            let decrypted_content = symmetric_decrypt(py, algorithm, mode, encrypted_content)?;
             pyo3::types::PyBytes::new_bound(py, decrypted_content.as_slice())
         }
         _ => {
@@ -302,7 +288,6 @@ fn deserialize_and_decrypt<'p>(
         pyo3::types::PyBytes::new_bound(py, decanonicalized.into_owned().as_slice())
     };
 
-    // Return the content
     Ok(plain_data)
 }
 

src/rust/src/test_support.rs

@@ -103,6 +103,57 @@ fn pkcs7_verify(
     Ok(())
 }
 
+#[cfg(not(CRYPTOGRAPHY_IS_BORINGSSL))]
+#[pyo3::pyfunction]
+#[pyo3(signature = (cert_recipients, msg, cipher, options, encoding))]
+fn pkcs7_encrypt<'p>(
+    py: pyo3::Python<'p>,
+    cert_recipients: Vec<pyo3::Bound<'p, PyCertificate>>,
+    msg: CffiBuf<'p>,
+    cipher: CffiBuf<'p>,
+    options: pyo3::Bound<'p, pyo3::types::PyList>,
+    encoding: pyo3::Bound<'p, pyo3::PyAny>,
+) -> CryptographyResult<pyo3::Bound<'p, pyo3::types::PyBytes>> {
+    // Prepare the certificates
+    let mut certs_stack = openssl::stack::Stack::new()?;
+    for cert in &cert_recipients {
+        let der = asn1::write_single(cert.get().raw.borrow_dependent())?;
+        certs_stack.push(openssl::x509::X509::from_der(&der)?)?;
+    }
+
+    // Prepare the cipher
+    // SAFETY: No pre-conditions
+    let cipher = unsafe {
+        let ptr = openssl_sys::EVP_get_cipherbyname(cipher.as_bytes().as_ptr() as *const i8);
+        openssl::symm::Cipher::from_ptr(ptr as *mut _)
+    };
+
+    // Prepare the options
+    let mut flags = openssl::pkcs7::Pkcs7Flags::empty();
+    if options.contains(types::PKCS7_TEXT.get(py)?)? {
+        flags |= openssl::pkcs7::Pkcs7Flags::TEXT;
+    }
+    if options.contains(types::PKCS7_BINARY.get(py)?)? {
+        flags |= openssl::pkcs7::Pkcs7Flags::BINARY;
+    }
+
+    // Encrypt the message
+    let p7 = openssl::pkcs7::Pkcs7::encrypt(&certs_stack, msg.as_bytes(), cipher, flags).unwrap();
+
+    // Return the result in the correct format
+    if encoding.is(&types::ENCODING_DER.get(py)?) {
+        Ok(pyo3::types::PyBytes::new_bound(py, &p7.to_der().unwrap()))
+    } else if encoding.is(&types::ENCODING_PEM.get(py)?) {
+        Ok(pyo3::types::PyBytes::new_bound(py, &p7.to_pem().unwrap()))
+    } else {
+        Ok(pyo3::types::PyBytes::new_bound(
+            py,
+            &p7.to_smime(&[], openssl::pkcs7::Pkcs7Flags::empty())
+                .unwrap(),
+        ))
+    }
+}
+
 #[cfg(not(CRYPTOGRAPHY_IS_BORINGSSL))]
 #[pyo3::pyfunction]
 #[pyo3(signature = (encoding, msg, pkey, cert_recipient, options))]
@@ -154,6 +205,9 @@ pub(crate) mod test_support {
     use super::pkcs7_decrypt;
     #[cfg(not(CRYPTOGRAPHY_IS_BORINGSSL))]
     #[pymodule_export]
+    use super::pkcs7_encrypt;
+    #[cfg(not(CRYPTOGRAPHY_IS_BORINGSSL))]
+    #[pymodule_export]
     use super::pkcs7_verify;
     #[pymodule_export]
     use super::test_parse_certificate;

tests/hazmat/primitives/test_pkcs7.py

@@ -10,7 +10,7 @@
 
 import pytest
 
-from cryptography import x509
+from cryptography import exceptions, x509
 from cryptography.exceptions import _Reasons
 from cryptography.hazmat.bindings._rust import test_support
 from cryptography.hazmat.primitives import hashes, serialization
@@ -851,6 +851,22 @@ def _load_rsa_cert_key():
     return cert, key
 
 
+def _load_another_rsa_cert_key():
+    key = load_vectors_from_file(
+        os.path.join("pkcs7", "rsa_key.pem"),
+        lambda pemfile: serialization.load_pem_private_key(
+            pemfile.read(), None, unsafe_skip_rsa_key_validation=True
+        ),
+        mode="rb",
+    )
+    cert = load_vectors_from_file(
+        os.path.join("pkcs7", "rsa_ca.pem"),
+        loader=lambda pemfile: x509.load_pem_x509_certificate(pemfile.read()),
+        mode="rb",
+    )
+    return cert, key
+
+
 @pytest.mark.supported(
     only_if=lambda backend: backend.pkcs7_supported()
     and backend.rsa_encryption_supported(padding.PKCS1v15()),
@@ -1194,10 +1210,9 @@ def test_smime_decrypt(self, backend, encoding, options):
         # Encrypt some data
         plain = b"hello world\n"
         cert, private_key = _load_rsa_cert_key()
-        builder = (
-            pkcs7.PKCS7EnvelopeBuilder().set_data(plain).add_recipient(cert)
+        enveloped = test_support.pkcs7_encrypt(
+            [cert], plain, b"aes-128-cbc", options, encoding
         )
-        enveloped = builder.encrypt(encoding, options)
 
         # Test decryption
         decryptor = (
@@ -1210,7 +1225,69 @@ def test_smime_decrypt(self, backend, encoding, options):
         decrypted = decryptor.decrypt(encoding, options)
         assert decrypted == plain
 
-    def test_smime_decrypt_not_encrypted(self, backend):
+    def test_smime_decrypt_no_recipient_match(self, backend):
+        # Encrypt some data with one RSA chain
+        plain = b"hello world\n"
+        cert, _ = _load_rsa_cert_key()
+        enveloped = test_support.pkcs7_encrypt(
+            [cert], plain, b"aes-128-cbc", [], serialization.Encoding.DER
+        )
+
+        # Test decryption with another RSA chain
+        another_cert, another_private_key = _load_another_rsa_cert_key()
+        decryptor = (
+            pkcs7.PKCS7EnvelopeDecryptor()
+            .set_data(enveloped)
+            .set_recipient(another_cert)
+            .set_private_key(another_private_key)
+        )
+
+        with pytest.raises(x509.AttributeNotFound):
+            decryptor.decrypt(serialization.Encoding.DER, [])
+
+    def test_smime_decrypt_algorithm_not_supported(self, backend):
+        # Encrypt some data
+        plain = b"hello world\n"
+        cert, private_key = _load_rsa_cert_key()
+        enveloped = test_support.pkcs7_encrypt(
+            [cert], plain, b"aes-256-cbc", [], serialization.Encoding.DER
+        )
+
+        # Test decryption
+        decryptor = (
+            pkcs7.PKCS7EnvelopeDecryptor()
+            .set_data(enveloped)
+            .set_recipient(cert)
+            .set_private_key(private_key)
+        )
+
+        with pytest.raises(exceptions.UnsupportedAlgorithm):
+            decryptor.decrypt(serialization.Encoding.DER, [])
+
+    def test_smime_decrypt_not_enveloped(self, backend):
+        # Create a signed email
+        cert, key = _load_cert_key()
+        options = [pkcs7.PKCS7Options.DetachedSignature]
+        builder = (
+            pkcs7.PKCS7SignatureBuilder()
+            .set_data(b"hello world")
+            .add_signer(cert, key, hashes.SHA256())
+        )
+        signed = builder.sign(serialization.Encoding.DER, options)
+
+        # Test decryption failure with signed email
+        cert, private_key = _load_rsa_cert_key()
+        decryptor = (
+            pkcs7.PKCS7EnvelopeDecryptor()
+            .set_data(signed)
+            .set_recipient(cert)
+            .set_private_key(private_key)
+        )
+
+        with pytest.raises(exceptions.UnsupportedAlgorithm):
+            decryptor.decrypt(serialization.Encoding.DER, [])
+
+    def test_smime_decrypt_smime_not_encrypted(self, backend):
         # Create a plain email
         email_message = EmailMessage()
         email_message.set_content("hello world\n")

vectors/cryptography_vectors/pkcs7/rsa_ca.pem

@@ -0,0 +1,19 @@
+-----BEGIN CERTIFICATE-----
+MIIDFzCCAf+gAwIBAgIUbV+9yyz0+DsvQNP81cYOpdIuPUkwDQYJKoZIhvcNAQEL
+BQAwGjEYMBYGA1UEAwwPY3J5cHRvZ3JhcGh5IENBMCAXDTI0MDkxNjEyMDc0N1oY
+DzIwNTIwMjAyMTIwNzQ3WjAaMRgwFgYDVQQDDA9jcnlwdG9ncmFwaHkgQ0EwggEi
+MA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQCWXjSWWQr3Im5Wj2Jr80PRWA5a
+9Ou8fbAOnsYa9Fof+hpU8zUxUJb3QaHedQJn98/ypjglMCvamDfiObF+uxUMOPP9
+MfxuLzKoYX0XFwffx4RuqxBPZQ9Mrd1+p/3CZ+/j2Up3DWmB9oGBuhlBJd2428Zz
+nA+TOyRt3Q4qa3oIrhR3dbJ47TU+u4qQdJ/wrcMbfKlWFY2AaO91DXZxPQdOb9Cf
+B97FBNqK1Jq1zXiRk/dgMfTmomKVDGRruGy5f/042KEHJB4xBamlaPOeqjnw1sEz
+lgGq0iVi0bbQxgLzSVZJ9zU4CatOKdzb7pfDEGsfP6Rmj4jYTiZJaabrdF7fAgMB
+AAGjUzBRMB0GA1UdDgQWBBQQ51DtVzCTX0l2ycbae/JWmNd8XDAfBgNVHSMEGDAW
+gBQQ51DtVzCTX0l2ycbae/JWmNd8XDAPBgNVHRMBAf8EBTADAQH/MA0GCSqGSIb3
+DQEBCwUAA4IBAQCHHg6wNJXIZH7YUgM7mGkw1IKIXNWSdw5mvESglk77DXuo8gxg
+rCgkJtQ64FuWo55cqfG5lxoOdYYIwVcEIDvEUa1/OQ+BdcX2WSVl1wnps4ZP1MWh
+33vtzmhcfcEYE79j8BUtSIjYpvVf5T/pxEL9DXbYkvG8PoS2YOOtQ0hoyzdMDzOd
+d0U2YQn2vgcXMBvpt3A32Ol5cCZRLid4NdZVTMfihGeu+r8/fpnpchS+bSJMzO5J
+CtO6nDrqxxQ+h7GssAmo3R7VR6Ii5e8sFV/D5E0FihgxsacSseJeGjWqJcahu9K4
+GIyrSZM0hWacVlOFUJ69FbN/VNb3qNe8LS3/
+-----END CERTIFICATE-----

vectors/cryptography_vectors/pkcs7/rsa_key.pem

@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQCWXjSWWQr3Im5W
+j2Jr80PRWA5a9Ou8fbAOnsYa9Fof+hpU8zUxUJb3QaHedQJn98/ypjglMCvamDfi
+ObF+uxUMOPP9MfxuLzKoYX0XFwffx4RuqxBPZQ9Mrd1+p/3CZ+/j2Up3DWmB9oGB
+uhlBJd2428ZznA+TOyRt3Q4qa3oIrhR3dbJ47TU+u4qQdJ/wrcMbfKlWFY2AaO91
+DXZxPQdOb9CfB97FBNqK1Jq1zXiRk/dgMfTmomKVDGRruGy5f/042KEHJB4xBaml
+aPOeqjnw1sEzlgGq0iVi0bbQxgLzSVZJ9zU4CatOKdzb7pfDEGsfP6Rmj4jYTiZJ
+aabrdF7fAgMBAAECggEASepKEGP562zt7zfxih4dRrFx8Q03OkNQgYdT25klGlXv
+jrW0/qZeZ3HIV5mOErxy6JUDwWgSDpTH17lDgogFOypd6aciLqmdzb6KZojIMp0Z
+gOsvC9AiOq+20xVSPLep6Qd9vtMFe8DIfZQU6PzrpnzGazsaPuYSLL0+wuRtKwli
+CQPYmLnHRkBnedl/WKLBHtbFfOf41dk9ZhUOdiA/xACwlFGlR+1MSknjaBPrZsAN
+OnhlHCy2TUpjNqJij9lw2khdDZjCJwXzApgdhMu549kV8WMMai4mbhDMrbmXayws
+lBI5TDy2hH6obCs/LhoCpX3UTgoTneQKvhg35IGNeQKBgQDT4J+rvWhibFoyK4NR
+oqkDQR3ktozkqL0wTwJUJNrito+CfT/xK88qcEq/wPRy/gFoGtUqvlJ6BQIppvqm
+nS1C1pf6insHXpsDSzSjUMjpkjKP+4N+rFrbr5/Un7mna9lI9DYFBRLaU7Pd9wsP
+ia/AmKTwTM0CfcNjeEDgRGsWNwKBgQC1rnU5rUWT02doGjznOM367zPn9bSxbRSB
+18dm2eunXNOCvdkgo19ltDngWCwYMZe7pNm1Ma+6M+1Arzu9cGy9QTKJ510KD8ZK
+2iplRIuHuGi26C2sosBHinGGZ9A0meF5aMGaP8wweT3h2d+lwim0zPNlVw9br27e
+Jt/V1iComQKBgDpKO+NVspgRUycmTXRyGalpir3Xcx+dRJ87vVpE4Wsp6oVty9+f
+u9jjF811WksgkmZ/q1GdFre3FQc8hwy7hS40N4+X9YzrfB4F3K9plDyeGgb+nQiG
+hmCWCpTGdJ8YzqiBdzMeWDd0e7F/O63EpXOJCyeJGYWr78s31DFe4U81AoGAfXbO
+yVfZNq7FOGT2Lg9SX7oUtFk/wU883DXxJBvV2ywFfMIyUZHA7XE6jq0VNlf5GsJK
+/hAj9IirjQH2VWpwDXoatpUp8SrXmxVyNaDSYpV86uMHt7Jk6oK2x0Slyc2Cod8v
+45+vGMKBc5iME+Iz/wGSDCfFGWHbRFrANOG8h/ECgYEAzRfFncoM6pcmq6O6pdjX
+BLwjBrow40lUcHChc4ZEnmvffVYezZ6CFqjoPj1NtcgHdwbTp7uzevIDkHBG4jYG
+nJa3GYfx+g35K68eLahiV6Dt8qiuVHaTRadsvZCnt9+9u2HiH9TU7q4Hg/840TtN
+K9LHFDW6OzV7IP/Ew0sz1BM=
+-----END PRIVATE KEY-----