diff options
| -rw-r--r-- | docs/hazmat/primitives/asymmetric/serialization.rst | 6 | ||||
| -rw-r--r-- | src/cryptography/hazmat/backends/openssl/backend.py | 50 | ||||
| -rw-r--r-- | tests/hazmat/primitives/test_pkcs12.py | 13 |
3 files changed, 54 insertions, 15 deletions
diff --git a/docs/hazmat/primitives/asymmetric/serialization.rst b/docs/hazmat/primitives/asymmetric/serialization.rst index 04bc705a..85aeced3 100644 --- a/docs/hazmat/primitives/asymmetric/serialization.rst +++ b/docs/hazmat/primitives/asymmetric/serialization.rst @@ -417,10 +417,12 @@ file suffix. Deserialize a PKCS12 blob. - :param bytes data: The binary data. + :param data: The binary data. + :type data: :term:`bytes-like` - :param bytes password: The password to use to decrypt the data. ``None`` + :param password: The password to use to decrypt the data. ``None`` if the PKCS12 is not encrypted. + :type password: :term:`bytes-like` :param backend: A backend instance. diff --git a/src/cryptography/hazmat/backends/openssl/backend.py b/src/cryptography/hazmat/backends/openssl/backend.py index 9eca5f8a..89356d3c 100644 --- a/src/cryptography/hazmat/backends/openssl/backend.py +++ b/src/cryptography/hazmat/backends/openssl/backend.py @@ -450,13 +450,13 @@ class Backend(object): The char* is the storage for the BIO and it must stay alive until the BIO is finished with. """ - data_char_p = self._ffi.new("char[]", data) + data_ptr = self._ffi.from_buffer(data) bio = self._lib.BIO_new_mem_buf( - data_char_p, len(data) + data_ptr, len(data) ) self.openssl_assert(bio != self._ffi.NULL) - return _MemoryBIO(self._ffi.gc(bio, self._lib.BIO_free), data_char_p) + return _MemoryBIO(self._ffi.gc(bio, self._lib.BIO_free), data_ptr) def _create_mem_bio_gc(self): """ @@ -1225,12 +1225,12 @@ class Backend(object): mem_bio = self._bytes_to_bio(data) if password is not None: - utils._check_bytes("password", password) + utils._check_byteslike("password", password) userdata = self._ffi.new("CRYPTOGRAPHY_PASSWORD_DATA *") if password is not None: - password_buf = self._ffi.new("char []", password) - userdata.password = password_buf + password_ptr = self._ffi.from_buffer(password) + userdata.password = password_ptr userdata.length = len(password) evp_pkey = openssl_read_func( @@ -2196,11 +2196,33 @@ class Backend(object): self._lib.EVP_get_cipherbyname(cipher_name) != self._ffi.NULL ) - def load_key_and_certificates_from_pkcs12(self, data, password): - if password is None: - password = self._ffi.NULL + @contextlib.contextmanager + def _zeroed_null_terminated_buf(self, data): + """ + This method takes bytes, which can be a bytestring or a mutable + buffer like a bytearray, and yields a null-terminated version of that + data. This is required because PKCS12_parse doesn't take a length with + its password char * and ffi.from_buffer doesn't provide null + termination. So, to support zeroing the data via bytearray we + need to build this ridiculous construct that copies the memory, but + zeroes it after use. + """ + if data is None: + yield self._ffi.NULL else: - utils._check_bytes("password", password) + data_len = len(data) + buf = self._ffi.new("char[]", data_len + 1) + self._ffi.memmove(buf, data, data_len) + try: + yield buf + finally: + # TODO: this could be done with memset to avoid the Python + # bytestring alloc. + self._ffi.memmove(buf, b"\x00" * data_len, data_len) + + def load_key_and_certificates_from_pkcs12(self, data, password): + if password is not None: + utils._check_byteslike("password", password) bio = self._bytes_to_bio(data) p12 = self._lib.d2i_PKCS12_bio(bio.bio, self._ffi.NULL) @@ -2212,9 +2234,11 @@ class Backend(object): evp_pkey_ptr = self._ffi.new("EVP_PKEY **") x509_ptr = self._ffi.new("X509 **") sk_x509_ptr = self._ffi.new("Cryptography_STACK_OF_X509 **") - res = self._lib.PKCS12_parse( - p12, password, evp_pkey_ptr, x509_ptr, sk_x509_ptr - ) + with self._zeroed_null_terminated_buf(password) as password_buf: + res = self._lib.PKCS12_parse( + p12, password_buf, evp_pkey_ptr, x509_ptr, sk_x509_ptr + ) + if res == 0: self._consume_errors() raise ValueError("Invalid password or PKCS12 data") diff --git a/tests/hazmat/primitives/test_pkcs12.py b/tests/hazmat/primitives/test_pkcs12.py index 85be3b51..f084d578 100644 --- a/tests/hazmat/primitives/test_pkcs12.py +++ b/tests/hazmat/primitives/test_pkcs12.py @@ -108,3 +108,16 @@ class TestPKCS12(object): derfile.read(), b"invalid", backend ), mode="rb" ) + + def test_buffer_protocol(self, backend): + p12 = load_vectors_from_file( + os.path.join("pkcs12", "cert-key-aes256cbc.p12"), + lambda derfile: derfile.read(), mode="rb" + ) + p12buffer = bytearray(p12) + parsed_key, parsed_cert, parsed_more_certs = load_key_and_certificates( + p12buffer, bytearray(b"cryptography"), backend + ) + assert parsed_key is not None + assert parsed_cert is not None + assert parsed_more_certs == [] |