first python API proposition

first round-trip tests

feat: made asn1 structures readable

refacto: adapted existing functions accordingly

feat/pkcs12: added symmetric_decrypt

feat: deserialize 3 possible encodings

feat: handling AES-128

feat: raise error when no recipient is found

feat/pkcs7: added decanonicalize function

feat/asn1: added decode_der_data

feat/pkcs7: added smime_enveloped_decode

tests are the round-trip (encrypt & decrypt)

more tests for 100% python coverage

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

fix: const handling

Refactor PKCS7Decryptor to pkcs7_decrypt

refacto: removed SMIME_ENVELOPED_DECODE from rust code

refacto: removed decode_der_data

adapted tests accordingly

removed the PEM tag check

added tests for smime_decnonicalize

one more test case

Update src/rust/src/pkcs7.rs

Co-authored-by: Alex Gaynor <[email protected]>

took comments into account

pem to der is now outside of decrypt

fix: removed test_support pkcs7_encrypt

added vector for aes_256_cbc encrypted pkcs7

feat: not using test_support decrypt anymore

added new vectors for PKCS7 tests

feat: using pkcs7 vectors

removed previous ones

fix: changed wrong function

feat: added certificate issuer check

test: generating the RSA chain

removed the vectors accordingly

moved symmetric_decrypt to pkcs7.rs

Author: nitneuqr

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/src/pkcs7.rs

@@ -16,12 +16,14 @@ use openssl::pkcs7::Pkcs7;
 use pyo3::types::{PyAnyMethods, PyBytesMethods, PyListMethods};
 
 use crate::asn1::encode_der_data;
+use crate::backend::ciphers;
 use crate::buf::CffiBuf;
 use crate::error::{CryptographyError, CryptographyResult};
+use crate::padding::PKCS7UnpaddingContext;
 use crate::pkcs12::symmetric_encrypt;
 #[cfg(not(CRYPTOGRAPHY_IS_BORINGSSL))]
 use crate::x509::certificate::load_der_x509_certificate;
-use crate::{exceptions, types, x509};
+use crate::{backend, exceptions, types, x509};
 
 const PKCS7_CONTENT_TYPE_OID: asn1::ObjectIdentifier = asn1::oid!(1, 2, 840, 113549, 1, 9, 3);
 const PKCS7_MESSAGE_DIGEST_OID: asn1::ObjectIdentifier = asn1::oid!(1, 2, 840, 113549, 1, 9, 4);
@@ -164,6 +166,153 @@ fn encrypt_and_serialize<'p>(
     }
 }
 
+#[pyo3::pyfunction]
+fn pem_to_der<'p>(
+    py: pyo3::Python<'p>,
+    data: CffiBuf<'p>,
+) -> CryptographyResult<pyo3::Bound<'p, pyo3::types::PyBytes>> {
+    let pem_str = std::str::from_utf8(data.as_bytes())
+        .map_err(|_| pyo3::exceptions::PyValueError::new_err("Invalid PEM data"))?;
+    let pem = pem::parse(pem_str)
+        .map_err(|_| pyo3::exceptions::PyValueError::new_err("Failed to parse PEM data"))?;
+
+    Ok(pyo3::types::PyBytes::new_bound(py, &pem.into_contents()))
+}
+
+#[pyo3::pyfunction]
+fn deserialize_and_decrypt<'p>(
+    py: pyo3::Python<'p>,
+    data: CffiBuf<'p>,
+    certificate: pyo3::Bound<'p, x509::certificate::Certificate>,
+    private_key: pyo3::Bound<'p, backend::rsa::RsaPrivateKey>,
+    options: &pyo3::Bound<'p, pyo3::types::PyList>,
+) -> CryptographyResult<pyo3::Bound<'p, pyo3::types::PyBytes>> {
+    // Deserialize the content info
+    let content_info = asn1::parse_single::<pkcs7::ContentInfo<'_>>(data.as_bytes()).unwrap();
+    let plain_content = match content_info.content {
+        pkcs7::Content::EnvelopedData(data) => {
+            // Extract enveloped data
+            let enveloped_data = data.into_inner();
+
+            // 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_certificate = certificate.get().raw.borrow_dependent();
+            let recipient_serial_number = recipient_certificate.tbs_cert.serial;
+            let recipient_issuer = recipient_certificate.tbs_cert.issuer.clone();
+            let found_recipient_info = recipient_infos.find(|info| {
+                info.issuer_and_serial_number.serial_number == recipient_serial_number
+                    && info.issuer_and_serial_number.issuer == recipient_issuer
+            });
+
+            // 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_X509,
+                        )),
+                    ));
+                }
+            };
+
+            // Decrypt the key using the private key
+            let padding = types::PKCS1V15.get(py)?.call0()?;
+            let key = private_key
+                .call_method1(
+                    pyo3::intern!(py, "decrypt"),
+                    (recipient_info.encrypted_key, &padding),
+                )?
+                .extract::<pyo3::pybacked::PyBackedBytes>()?;
+
+            // 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 is currently supported for decryption.",
+                            exceptions::Reasons::UNSUPPORTED_SERIALIZATION,
+                        )),
+                    ));
+                }
+            };
+
+            // 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())
+        }
+        _ => {
+            return Err(CryptographyError::from(
+                exceptions::UnsupportedAlgorithm::new_err((
+                    "Only EnvelopedData structures are currently supported.",
+                    exceptions::Reasons::UNSUPPORTED_SERIALIZATION,
+                )),
+            ));
+        }
+    };
+
+    // Return content in different form, based on options
+    let text_mode = options.contains(types::PKCS7_TEXT.get(py)?)?;
+    let plain_data = if options.contains(types::PKCS7_BINARY.get(py)?)? {
+        plain_content
+    } else {
+        let decanonicalized = smime_decanonicalize(plain_content.as_bytes(), text_mode);
+        pyo3::types::PyBytes::new_bound(py, decanonicalized.into_owned().as_slice())
+    };
+
+    Ok(plain_data)
+}
+
+pub(crate) fn symmetric_decrypt(
+    py: pyo3::Python<'_>,
+    algorithm: pyo3::Bound<'_, pyo3::PyAny>,
+    mode: pyo3::Bound<'_, pyo3::PyAny>,
+    data: &[u8],
+) -> CryptographyResult<Vec<u8>> {
+    let block_size = algorithm
+        .getattr(pyo3::intern!(py, "block_size"))?
+        .extract()?;
+
+    let mut cipher =
+        ciphers::CipherContext::new(py, algorithm, mode, openssl::symm::Mode::Decrypt)?;
+
+    // Decrypt the data
+    let mut decrypted_data = vec![0; data.len() + (block_size / 8)];
+    let count = cipher.update_into(py, data, &mut decrypted_data)?;
+    let final_block = cipher.finalize(py)?;
+    assert!(final_block.as_bytes().is_empty());
+    decrypted_data.truncate(count);
+
+    // Unpad the data
+    let mut unpadder = PKCS7UnpaddingContext::new(block_size);
+    let unpadded_first_blocks = unpadder.update(py, CffiBuf::from_bytes(py, &decrypted_data))?;
+    let unpadded_last_block = unpadder.finalize(py)?;
+
+    let unpadded_data = [
+        unpadded_first_blocks.as_bytes(),
+        unpadded_last_block.as_bytes(),
+    ]
+    .concat();
+
+    Ok(unpadded_data)
+}
+
 #[pyo3::pyfunction]
 fn sign_and_serialize<'p>(
     py: pyo3::Python<'p>,
@@ -415,6 +564,39 @@ fn smime_canonicalize(data: &[u8], text_mode: bool) -> (Cow<'_, [u8]>, Cow<'_, [
     }
 }
 
+fn smime_decanonicalize(data: &[u8], text_mode: bool) -> Cow<'_, [u8]> {
+    let mut new_data = vec![];
+    let mut last_idx = 0;
+
+    // Remove the header if text_mode is true
+    let data = if text_mode {
+        let header = b"Content-Type: text/plain\r\n\r\n";
+        if data.starts_with(header) {
+            &data[header.len()..]
+        } else {
+            data
+        }
+    } else {
+        data
+    };
+
+    // Remove the \r in the data
+    for (i, c) in data.iter().copied().enumerate() {
+        if c == b'\n' && i > 0 && data[i - 1] == b'\r' {
+            new_data.extend_from_slice(&data[last_idx..i - 1]);
+            new_data.push(b'\n');
+            last_idx = i + 1;
+        }
+    }
+
+    // Copy the remaining data
+    if last_idx < data.len() {
+        new_data.extend_from_slice(&data[last_idx..]);
+    }
+
+    Cow::Owned(new_data)
+}
+
 #[cfg(not(CRYPTOGRAPHY_IS_BORINGSSL))]
 fn load_pkcs7_certificates(
     py: pyo3::Python<'_>,
@@ -507,8 +689,8 @@ fn load_der_pkcs7_certificates<'p>(
 pub(crate) mod pkcs7_mod {
     #[pymodule_export]
     use super::{
-        encrypt_and_serialize, load_der_pkcs7_certificates, load_pem_pkcs7_certificates,
-        serialize_certificates, sign_and_serialize,
+        deserialize_and_decrypt, encrypt_and_serialize, load_der_pkcs7_certificates,
+        load_pem_pkcs7_certificates, pem_to_der, serialize_certificates, sign_and_serialize,
     };
 }
 
@@ -517,7 +699,7 @@ mod tests {
     use std::borrow::Cow;
     use std::ops::Deref;
 
-    use super::smime_canonicalize;
+    use super::{smime_canonicalize, smime_decanonicalize};
 
     #[test]
     fn test_smime_canonicalize() {
@@ -577,4 +759,28 @@ mod tests {
             );
         }
     }
+
+    #[test]
+    fn test_smime_decanonicalize() {
+        for (input, text_mode, expected_output) in [
+            // Values with text_mode=false
+            (b"" as &[u8], false, b"" as &[u8]),
+            (b"abc\r\n", false, b"abc\n"),
+            (b"\r\nabc\n", false, b"\nabc\n"),
+            (b"abc\r\ndef\r\n", false, b"abc\ndef\n"),
+            (b"abc\r\ndef\nabc", false, b"abc\ndef\nabc"),
+            // Values with text_mode=true
+            (b"abc\r\n", true, b"abc\n"),
+            (b"Content-Type: text/plain\r\n\r\n", true, b""),
+            (b"Content-Type: text/plain\r\n\r\nabc", true, b"abc"),
+            (
+                b"Content-Type: text/plain\r\n\r\nabc\r\ndef\nabc",
+                true,
+                b"abc\ndef\nabc",
+            ),
+        ] {
+            let result = smime_decanonicalize(input, text_mode);
+            assert_eq!(result.deref(), expected_output);
+        }
+    }
 }