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

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/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],

src/rust/src/pkcs12.rs

@@ -5,7 +5,7 @@
 use crate::backend::{ciphers, hashes, hmac, kdf, keys};
 use crate::buf::CffiBuf;
 use crate::error::{CryptographyError, CryptographyResult};
-use crate::padding::PKCS7PaddingContext;
+use crate::padding::{PKCS7PaddingContext, PKCS7UnpaddingContext};
 use crate::x509::certificate::Certificate;
 use crate::{types, x509};
 use cryptography_x509::common::Utf8StoredBMPString;
@@ -107,6 +107,40 @@ pub(crate) fn symmetric_encrypt(
     Ok(ciphertext)
 }
 
+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)
+}
+
 enum EncryptionAlgorithm {
     PBESv1SHA1And3KeyTripleDESCBC,
     PBESv2SHA256AndAES256CBC,