Emit warnings on BER PKCS#7 and PKCS#12

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

@@ -13,14 +13,14 @@ pub const X509_CERTIFICATE_OID: asn1::ObjectIdentifier = asn1::oid!(1, 2, 840, 1
 pub const FRIENDLY_NAME_OID: asn1::ObjectIdentifier = asn1::oid!(1, 2, 840, 113549, 1, 9, 20);
 pub const LOCAL_KEY_ID_OID: asn1::ObjectIdentifier = asn1::oid!(1, 2, 840, 113549, 1, 9, 21);
 
-#[derive(asn1::Asn1Write)]
+#[derive(asn1::Asn1Write, asn1::Asn1Read)]
 pub struct Pfx<'a> {
     pub version: u8,
     pub auth_safe: pkcs7::ContentInfo<'a>,
     pub mac_data: Option<MacData<'a>>,
 }
 
-#[derive(asn1::Asn1Write)]
+#[derive(asn1::Asn1Write, asn1::Asn1Read)]
 pub struct MacData<'a> {
     pub mac: pkcs7::DigestInfo<'a>,
     pub salt: &'a [u8],

src/rust/src/backend/ec.rs

@@ -10,7 +10,7 @@ use pyo3::types::{PyAnyMethods, PyDictMethods};
 use crate::backend::utils;
 use crate::buf::CffiBuf;
 use crate::error::{CryptographyError, CryptographyResult};
-use crate::x509::common::cstr_from_literal;
+use crate::utils::cstr_from_literal;
 use crate::{exceptions, types};
 
 #[pyo3::pyclass(frozen, module = "cryptography.hazmat.bindings._rust.openssl.ec")]

src/rust/src/lib.rs

@@ -23,6 +23,7 @@ mod pkcs12;
 mod pkcs7;
 mod test_support;
 pub(crate) mod types;
+pub(crate) mod utils;
 mod x509;
 
 #[cfg(CRYPTOGRAPHY_OPENSSL_300_OR_GREATER)]

src/rust/src/pkcs12.rs

@@ -6,11 +6,13 @@ use crate::backend::{ciphers, hashes, hmac, kdf, keys};
 use crate::buf::CffiBuf;
 use crate::error::{CryptographyError, CryptographyResult};
 use crate::padding::PKCS7PaddingContext;
+use crate::utils::cstr_from_literal;
 use crate::x509::certificate::Certificate;
 use crate::{types, x509};
 use cryptography_x509::common::Utf8StoredBMPString;
 use pyo3::types::{PyAnyMethods, PyBytesMethods, PyListMethods};
 use pyo3::IntoPyObject;
+use pyo3::PyTypeInfo;
 use std::collections::hash_map::DefaultHasher;
 use std::hash::{Hash, Hasher};
 
@@ -617,6 +619,7 @@ fn serialize_key_and_certificates<'p>(
 }
 
 fn decode_p12(
+    py: pyo3::Python<'_>,
     data: CffiBuf<'_>,
     password: Option<CffiBuf<'_>>,
 ) -> CryptographyResult<openssl::pkcs12::ParsedPkcs12_2> {
@@ -637,6 +640,12 @@ fn decode_p12(
         .parse2(password)
         .map_err(|_| pyo3::exceptions::PyValueError::new_err("Invalid password or PKCS12 data"))?;
 
+    if asn1::parse_single::<cryptography_x509::pkcs12::Pfx<'_>>(data.as_bytes()).is_err() {
+        let warning_cls = pyo3::exceptions::PyUserWarning::type_object(py);
+        let message = cstr_from_literal!("PKCS#12 bundle could not be parsed as DER, falling back to parsing as BER. Please file an issue at https://github.com/pyca/cryptography/issues explaining how your PKCS#12 bundle was created. In the future, this may become an exception.");
+        pyo3::PyErr::warn(py, &warning_cls, message, 1)?;
+    }
+
     Ok(parsed)
 }
 
@@ -654,7 +663,7 @@ fn load_key_and_certificates<'p>(
 )> {
     let _ = backend;
 
-    let p12 = decode_p12(data, password)?;
+    let p12 = decode_p12(py, data, password)?;
 
     let private_key = if let Some(pkey) = p12.pkey {
         let pkey_bytes = pkey.private_key_to_pkcs8()?;
@@ -702,7 +711,7 @@ fn load_pkcs12<'p>(
 ) -> CryptographyResult<pyo3::Bound<'p, pyo3::PyAny>> {
     let _ = backend;
 
-    let p12 = decode_p12(data, password)?;
+    let p12 = decode_p12(py, data, password)?;
 
     let private_key = if let Some(pkey) = p12.pkey {
         let pkey_bytes = pkey.private_key_to_pkcs8()?;

src/rust/src/pkcs7.rs

@@ -14,6 +14,8 @@ use once_cell::sync::Lazy;
 #[cfg(not(CRYPTOGRAPHY_IS_BORINGSSL))]
 use openssl::pkcs7::Pkcs7;
 use pyo3::types::{PyAnyMethods, PyBytesMethods, PyListMethods};
+#[cfg(not(CRYPTOGRAPHY_IS_BORINGSSL))]
+use pyo3::PyTypeInfo;
 
 use crate::asn1::encode_der_data;
 use crate::backend::ciphers;
@@ -22,6 +24,8 @@ use crate::error::{CryptographyError, CryptographyResult};
 use crate::padding::PKCS7UnpaddingContext;
 use crate::pkcs12::symmetric_encrypt;
 #[cfg(not(CRYPTOGRAPHY_IS_BORINGSSL))]
+use crate::utils::cstr_from_literal;
+#[cfg(not(CRYPTOGRAPHY_IS_BORINGSSL))]
 use crate::x509::certificate::load_der_x509_certificate;
 use crate::{exceptions, types, x509};
 
@@ -744,12 +748,16 @@ fn load_pem_pkcs7_certificates<'p>(
 ) -> CryptographyResult<pyo3::Bound<'p, pyo3::types::PyList>> {
     cfg_if::cfg_if! {
         if #[cfg(not(CRYPTOGRAPHY_IS_BORINGSSL))] {
-            let pkcs7_decoded = openssl::pkcs7::Pkcs7::from_pem(data).map_err(|_| {
-                CryptographyError::from(pyo3::exceptions::PyValueError::new_err(
-                    "Unable to parse PKCS7 data",
-                ))
-            })?;
-            load_pkcs7_certificates(py, pkcs7_decoded)
+            let pem_block = pem::parse(data)?;
+            if pem_block.tag() != "PKCS7" {
+                return Err(CryptographyError::from(
+                    pyo3::exceptions::PyValueError::new_err(
+                        "The provided PEM data does not have the PKCS7 tag.",
+                    ),
+                ));
+            }
+
+            load_der_pkcs7_certificates(py, pem_block.contents())
         } else {
             let _ = py;
             let _ = data;
@@ -775,7 +783,14 @@ fn load_der_pkcs7_certificates<'p>(
                     "Unable to parse PKCS7 data",
                 ))
             })?;
-            load_pkcs7_certificates(py, pkcs7_decoded)
+            let result = load_pkcs7_certificates(py, pkcs7_decoded)?;
+            if asn1::parse_single::<pkcs7::ContentInfo<'_>>(data).is_err() {
+                let warning_cls = pyo3::exceptions::PyUserWarning::type_object(py);
+                let message = cstr_from_literal!("PKCS#7 certificates could not be parsed as DER, falling back to parsing as BER. Please file an issue at https://github.com/pyca/cryptography/issues explaining how your PKCS#7 certificates was created. In the future, this may become an exception.");
+                pyo3::PyErr::warn(py, &warning_cls, message, 1)?;
+            }
+
+            Ok(result)
         } else {
             let _ = py;
             let _ = data;

src/rust/src/utils.rs

@@ -0,0 +1,11 @@
+// This file is dual licensed under the terms of the Apache License, Version
+// 2.0, and the BSD License. See the LICENSE file in the root of this repository
+// for complete details.
+
+macro_rules! cstr_from_literal {
+    ($str:expr) => {
+        std::ffi::CStr::from_bytes_with_nul(concat!($str, "\0").as_bytes()).unwrap()
+    };
+}
+
+pub(crate) use cstr_from_literal;

src/rust/src/x509/certificate.rs

@@ -25,7 +25,7 @@ use crate::asn1::{
 };
 use crate::backend::{hashes, keys};
 use crate::error::{CryptographyError, CryptographyResult};
-use crate::x509::common::cstr_from_literal;
+use crate::utils::cstr_from_literal;
 use crate::x509::verify::PyCryptoOps;
 use crate::x509::{extensions, sct, sign};
 use crate::{exceptions, types, x509};

src/rust/src/x509/common.rs

@@ -530,11 +530,3 @@ pub(crate) fn datetime_now(py: pyo3::Python<'_>) -> pyo3::PyResult<asn1::DateTim
             .call_method1(pyo3::intern!(py, "now"), (utc,))?,
     )
 }
-
-macro_rules! cstr_from_literal {
-    ($str:expr) => {
-        std::ffi::CStr::from_bytes_with_nul(concat!($str, "\0").as_bytes()).unwrap()
-    };
-}
-
-pub(crate) use cstr_from_literal;

src/rust/src/x509/crl.rs

@@ -20,7 +20,7 @@ use crate::asn1::{
 };
 use crate::backend::hashes::Hash;
 use crate::error::{CryptographyError, CryptographyResult};
-use crate::x509::common::cstr_from_literal;
+use crate::utils::cstr_from_literal;
 use crate::x509::{certificate, extensions, sign};
 use crate::{exceptions, types, x509};
 

src/rust/src/x509/csr.rs

@@ -13,7 +13,8 @@ use pyo3::types::{PyAnyMethods, PyListMethods};
 use crate::asn1::{encode_der_data, oid_to_py_oid, py_oid_to_oid};
 use crate::backend::keys;
 use crate::error::{CryptographyError, CryptographyResult};
-use crate::x509::{certificate, common::cstr_from_literal, sign};
+use crate::utils::cstr_from_literal;
+use crate::x509::{certificate, sign};
 use crate::{exceptions, types, x509};
 
 self_cell::self_cell!(

src/rust/src/x509/ocsp_resp.rs

@@ -14,7 +14,7 @@ use pyo3::types::{PyAnyMethods, PyBytesMethods, PyListMethods};
 
 use crate::asn1::{big_byte_slice_to_py_int, oid_to_py_oid};
 use crate::error::{CryptographyError, CryptographyResult};
-use crate::x509::common::cstr_from_literal;
+use crate::utils::cstr_from_literal;
 use crate::x509::{certificate, crl, extensions, ocsp, py_to_datetime, sct};
 use crate::{exceptions, types, x509};
 

tests/hazmat/primitives/test_pkcs12.py

@@ -3,7 +3,9 @@
 # for complete details.
 
 
+import contextlib
 import os
+import typing
 from datetime import datetime, timezone
 
 import pytest
@@ -87,7 +89,13 @@ def test_load_pkcs12_ec_keys(self, filename, password, backend):
         skip_message="Does not support RC2",
     )
     def test_load_pkcs12_ec_keys_rc2(self, filename, password, backend):
-        self._test_load_pkcs12_ec_keys(filename, password, backend)
+        if filename == "no-password.p12":
+            ctx: typing.Any = pytest.warns(UserWarning)
+        else:
+            ctx = contextlib.nullcontext()
+
+        with ctx:
+            self._test_load_pkcs12_ec_keys(filename, password, backend)
 
     def test_load_key_and_cert_cert_only(self, backend):
         cert, _ = _load_ca(backend)

tests/hazmat/primitives/test_pkcs7.py

@@ -3,6 +3,7 @@
 # for complete details.
 
 
+import contextlib
 import email.parser
 import os
 import typing
@@ -43,6 +44,12 @@ def test_load_invalid_pem_pkcs7(self, backend):
         with pytest.raises(ValueError):
             pkcs7.load_pem_pkcs7_certificates(b"nonsense")
 
+        with pytest.raises(ValueError):
+            pkcs7.load_pem_pkcs7_certificates(b"""
+-----BEGIN CERTIFICATE-----
+-----END CERTIFICATE-----
+            """)
+
     def test_not_bytes_der(self, backend):
         with pytest.raises(TypeError):
             pkcs7.load_der_pkcs7_certificates(38)  # type: ignore[arg-type]
@@ -70,11 +77,19 @@ def test_load_pkcs7_pem(self, backend):
         ],
     )
     def test_load_pkcs7_der(self, filepath, backend):
-        certs = load_vectors_from_file(
-            filepath,
-            lambda derfile: pkcs7.load_der_pkcs7_certificates(derfile.read()),
-            mode="rb",
-        )
+        if filepath.endswith("p7b"):
+            ctx: typing.Any = pytest.warns(UserWarning)
+        else:
+            ctx = contextlib.nullcontext()
+
+        with ctx:
+            certs = load_vectors_from_file(
+                filepath,
+                lambda derfile: pkcs7.load_der_pkcs7_certificates(
+                    derfile.read()
+                ),
+                mode="rb",
+            )
         assert len(certs) == 2
         assert certs[0].subject.get_attributes_for_oid(
             x509.oid.NameOID.COMMON_NAME