PKCS12 support (#5325)

* generate_pkcs12 (#4952)

* pkcs12 support

* simplify

* remove fixtures

* reorg and other improvements. memleak check

* ugh

* more fixes

* last changes hopefully

Co-authored-by: Tomer Shalev <[email protected]>

CHANGELOG.rst

@@ -49,6 +49,8 @@ Changelog
 * On OpenSSL 1.1.1d and higher ``cryptography`` now uses OpenSSL's
   built-in CSPRNG instead of its own OS random engine because these versions of
   OpenSSL properly reseed on fork.
+* Added initial support for creating PKCS12 files with
+  :func:`~cryptography.hazmat.primitives.serialization.pkcs12.serialize_key_and_certificates`.
 
 .. _v2-9-2:
 

docs/hazmat/primitives/asymmetric/serialization.rst

@@ -489,6 +489,45 @@ file suffix.
         ``additional_certificates`` is a list of all other
         :class:`~cryptography.x509.Certificate` instances in the PKCS12 object.
 
+.. function:: serialize_key_and_certificates(name, key, cert, cas, encryption_algorithm)
+
+    .. versionadded:: 3.0
+
+    .. warning::
+
+        PKCS12 encryption is not secure and should not be used as a security
+        mechanism. Wrap a PKCS12 blob in a more secure envelope if you need
+        to store or send it safely. Encryption is provided for compatibility
+        reasons only.
+
+    Serialize a PKCS12 blob.
+
+    :param name: The friendly name to use for the supplied certificate and key.
+    :type name: bytes
+
+    :param key: The private key to include in the structure.
+    :type key: An
+        :class:`~cryptography.hazmat.primitives.asymmetric.rsa.RSAPrivateKeyWithSerialization`
+        ,
+        :class:`~cryptography.hazmat.primitives.asymmetric.ec.EllipticCurvePrivateKeyWithSerialization`
+        , or
+        :class:`~cryptography.hazmat.primitives.asymmetric.dsa.DSAPrivateKeyWithSerialization`
+        object.
+
+    :param cert: The certificate associated with the private key.
+    :type cert: :class:`~cryptography.x509.Certificate` or ``None``
+
+    :param cas: An optional set of certificates to also include in the structure.
+    :type cas: list of :class:`~cryptography.x509.Certificate` or ``None``
+
+    :param encryption_algorithm: The encryption algorithm that should be used
+        for the key and certificate. An instance of an object conforming to the
+        :class:`~cryptography.hazmat.primitives.serialization.KeySerializationEncryption`
+        interface. PKCS12 encryption is **very weak** and should not be used
+        as a security boundary.
+
+    :return bytes: Serialized PKCS12.
+
 Serialization Formats
 ~~~~~~~~~~~~~~~~~~~~~
 

src/cryptography/hazmat/backends/openssl/backend.py

@@ -2397,6 +2397,61 @@ def load_key_and_certificates_from_pkcs12(self, data, password):
 
         return (key, cert, additional_certificates)
 
+    def serialize_key_and_certificates_to_pkcs12(self, name, key, cert, cas,
+                                                 encryption_algorithm):
+        password = None
+        if name is not None:
+            utils._check_bytes("name", name)
+
+        if isinstance(encryption_algorithm, serialization.NoEncryption):
+            nid_cert = -1
+            nid_key = -1
+            pkcs12_iter = 0
+            mac_iter = 0
+        elif isinstance(encryption_algorithm,
+                        serialization.BestAvailableEncryption):
+            # PKCS12 encryption is hopeless trash and can never be fixed.
+            # This is the least terrible option.
+            nid_cert = self._lib.NID_pbe_WithSHA1And3_Key_TripleDES_CBC
+            nid_key = self._lib.NID_pbe_WithSHA1And3_Key_TripleDES_CBC
+            # At least we can set this higher than OpenSSL's default
+            pkcs12_iter = 20000
+            # mac_iter chosen for compatibility reasons, see:
+            # https://www.openssl.org/docs/man1.1.1/man3/PKCS12_create.html
+            # Did we mention how lousy PKCS12 encryption is?
+            mac_iter = 1
+            password = encryption_algorithm.password
+        else:
+            raise ValueError("Unsupported key encryption type")
+
+        if cas is None or len(cas) == 0:
+            sk_x509 = self._ffi.NULL
+        else:
+            sk_x509 = self._lib.sk_X509_new_null()
+            sk_x509 = self._ffi.gc(sk_x509, self._lib.sk_X509_free)
+
+            # reverse the list when building the stack so that they're encoded
+            # in the order they were originally provided. it is a mystery
+            for ca in reversed(cas):
+                res = self._lib.sk_X509_push(sk_x509, ca._x509)
+                backend.openssl_assert(res >= 1)
+
+        with self._zeroed_null_terminated_buf(password) as password_buf:
+            with self._zeroed_null_terminated_buf(name) as name_buf:
+                p12 = self._lib.PKCS12_create(
+                    password_buf, name_buf,
+                    key._evp_pkey if key else self._ffi.NULL,
+                    cert._x509 if cert else self._ffi.NULL,
+                    sk_x509, nid_key, nid_cert, pkcs12_iter, mac_iter, 0)
+
+        self.openssl_assert(p12 != self._ffi.NULL)
+        p12 = self._ffi.gc(p12, self._lib.PKCS12_free)
+
+        bio = self._create_mem_bio_gc()
+        res = self._lib.i2d_PKCS12_bio(bio, p12)
+        self.openssl_assert(res > 0)
+        return self._read_mem_bio(bio)
+
     def poly1305_supported(self):
         return self._lib.Cryptography_HAS_POLY1305 == 1
 

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

@@ -4,6 +4,43 @@
 
 from __future__ import absolute_import, division, print_function
 
+from cryptography import x509
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import dsa, ec, rsa
+
 
 def load_key_and_certificates(data, password, backend):
     return backend.load_key_and_certificates_from_pkcs12(data, password)
+
+
+def serialize_key_and_certificates(name, key, cert, cas, encryption_algorithm):
+    if key is not None and not isinstance(
+        key, (rsa.RSAPrivateKeyWithSerialization,
+              dsa.DSAPrivateKeyWithSerialization,
+              ec.EllipticCurvePrivateKeyWithSerialization)):
+        raise TypeError(
+            "Key must be RSA, DSA, or EllipticCurve private key."
+        )
+    if cert is not None and not isinstance(cert, x509.Certificate):
+        raise TypeError("cert must be a certificate")
+
+    if cas is not None:
+        cas = list(cas)
+        if not all(isinstance(val, x509.Certificate) for val in cas):
+            raise TypeError("all values in cas must be certificates")
+
+    if not isinstance(
+        encryption_algorithm, serialization.KeySerializationEncryption
+    ):
+        raise TypeError(
+            "Key encryption algorithm must be a "
+            "KeySerializationEncryption instance"
+        )
+
+    if key is None and cert is None and not cas:
+        raise ValueError("You must supply at least one of key, cert, or cas")
+
+    return default_backend().serialize_key_and_certificates_to_pkcs12(
+        name, key, cert, cas, encryption_algorithm
+    )

tests/hazmat/backends/test_openssl_memleak.py

@@ -449,3 +449,38 @@ def func():
             cert = builder.sign(private_key, hashes.SHA256(), backend)
             cert.extensions
         """))
+
+    def test_write_pkcs12_key_and_certificates(self):
+        assert_no_memory_leaks(textwrap.dedent("""
+        def func():
+            import os
+            from cryptography import x509
+            from cryptography.hazmat.backends.openssl import backend
+            from cryptography.hazmat.primitives import serialization
+            from cryptography.hazmat.primitives.serialization import pkcs12
+            import cryptography_vectors
+
+            path = os.path.join('x509', 'custom', 'ca', 'ca.pem')
+            with cryptography_vectors.open_vector_file(path, "rb") as f:
+                cert = x509.load_pem_x509_certificate(
+                    f.read(), backend
+                )
+            path2 = os.path.join('x509', 'custom', 'dsa_selfsigned_ca.pem')
+            with cryptography_vectors.open_vector_file(path2, "rb") as f:
+                cert2 = x509.load_pem_x509_certificate(
+                    f.read(), backend
+                )
+            path3 = os.path.join('x509', 'letsencryptx3.pem')
+            with cryptography_vectors.open_vector_file(path3, "rb") as f:
+                cert3 = x509.load_pem_x509_certificate(
+                    f.read(), backend
+                )
+            key_path = os.path.join("x509", "custom", "ca", "ca_key.pem")
+            with cryptography_vectors.open_vector_file(key_path, "rb") as f:
+                key = serialization.load_pem_private_key(
+                    f.read(), None, backend
+                )
+            encryption = serialization.NoEncryption()
+            pkcs12.serialize_key_and_certificates(
+                b"name", key, cert, [cert2, cert3], encryption)
+        """))

tests/hazmat/primitives/test_pkcs12.py

@@ -11,16 +11,18 @@
 from cryptography import x509
 from cryptography.hazmat.backends.interfaces import DERSerializationBackend
 from cryptography.hazmat.backends.openssl.backend import _RC2
+from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.serialization import load_pem_private_key
 from cryptography.hazmat.primitives.serialization.pkcs12 import (
-    load_key_and_certificates
+    load_key_and_certificates, serialize_key_and_certificates
 )
 
 from .utils import load_vectors_from_file
+from ...doubles import DummyKeySerializationEncryption
 
 
 @pytest.mark.requires_backend_interface(interface=DERSerializationBackend)
-class TestPKCS12(object):
+class TestPKCS12Loading(object):
     def _test_load_pkcs12_ec_keys(self, filename, password, backend):
         cert = load_vectors_from_file(
             os.path.join("x509", "custom", "ca", "ca.pem"),
@@ -137,3 +139,113 @@ def test_buffer_protocol(self, backend):
         assert parsed_key is not None
         assert parsed_cert is not None
         assert parsed_more_certs == []
+
+
+def _load_cert(backend, path):
+    return load_vectors_from_file(
+        path,
+        lambda pemfile: x509.load_pem_x509_certificate(
+            pemfile.read(), backend
+        ), mode='rb'
+    )
+
+
+def _load_ca(backend):
+    cert = _load_cert(backend, os.path.join('x509', 'custom', 'ca', 'ca.pem'))
+    key = load_vectors_from_file(
+        os.path.join('x509', 'custom', 'ca', 'ca_key.pem'),
+        lambda pemfile: load_pem_private_key(
+            pemfile.read(), None, backend
+        ), mode='rb'
+    )
+    return cert, key
+
+
+class TestPKCS12Creation(object):
+    @pytest.mark.parametrize('name', [None, b'name'])
+    @pytest.mark.parametrize(('encryption_algorithm', 'password'), [
+        (serialization.BestAvailableEncryption(b'password'), b'password'),
+        (serialization.NoEncryption(), None)
+    ])
+    def test_generate(self, backend, name, encryption_algorithm, password):
+        cert, key = _load_ca(backend)
+        p12 = serialize_key_and_certificates(
+            name, key, cert, None, encryption_algorithm)
+
+        parsed_key, parsed_cert, parsed_more_certs = \
+            load_key_and_certificates(p12, password, backend)
+        assert parsed_cert == cert
+        assert parsed_key.private_numbers() == key.private_numbers()
+        assert parsed_more_certs == []
+
+    def test_generate_with_cert_key_ca(self, backend):
+        cert, key = _load_ca(backend)
+        cert2 = _load_cert(
+            backend, os.path.join('x509', 'custom', 'dsa_selfsigned_ca.pem')
+        )
+        cert3 = _load_cert(backend, os.path.join('x509', 'letsencryptx3.pem'))
+        encryption = serialization.NoEncryption()
+        p12 = serialize_key_and_certificates(
+            None, key, cert, [cert2, cert3], encryption)
+
+        parsed_key, parsed_cert, parsed_more_certs = \
+            load_key_and_certificates(p12, None, backend)
+        assert parsed_cert == cert
+        assert parsed_key.private_numbers() == key.private_numbers()
+        assert parsed_more_certs == [cert2, cert3]
+
+    def test_generate_wrong_types(self, backend):
+        cert, key = _load_ca(backend)
+        cert2 = _load_cert(backend, os.path.join('x509', 'letsencryptx3.pem'))
+        encryption = serialization.NoEncryption()
+        with pytest.raises(TypeError) as exc:
+            serialize_key_and_certificates(
+                b'name', cert, cert, None, encryption)
+        assert str(exc.value) == \
+            'Key must be RSA, DSA, or EllipticCurve private key.'
+
+        with pytest.raises(TypeError) as exc:
+            serialize_key_and_certificates(b'name', key, key, None, encryption)
+        assert str(exc.value) == 'cert must be a certificate'
+
+        with pytest.raises(TypeError) as exc:
+            serialize_key_and_certificates(
+                b'name', key, cert, None, key)
+        assert str(
+            exc.value) == ('Key encryption algorithm must be a '
+                           'KeySerializationEncryption instance')
+
+        with pytest.raises(TypeError) as exc:
+            serialize_key_and_certificates(None, key, cert, cert2, encryption)
+
+        with pytest.raises(TypeError) as exc:
+            serialize_key_and_certificates(None, key, cert, [key], encryption)
+        assert str(exc.value) == 'all values in cas must be certificates'
+
+    def test_generate_no_cert(self, backend):
+        _, key = _load_ca(backend)
+        p12 = serialize_key_and_certificates(
+            None, key, None, None, serialization.NoEncryption())
+        parsed_key, parsed_cert, parsed_more_certs = \
+            load_key_and_certificates(p12, None, backend)
+        assert parsed_cert is None
+        assert parsed_key.private_numbers() == key.private_numbers()
+        assert parsed_more_certs == []
+
+    def test_must_supply_something(self):
+        with pytest.raises(ValueError) as exc:
+            serialize_key_and_certificates(
+                None, None, None, None, serialization.NoEncryption()
+            )
+        assert str(exc.value) == (
+            'You must supply at least one of key, cert, or cas'
+        )
+
+    def test_generate_unsupported_encryption_type(self, backend):
+        cert, key = _load_ca(backend)
+        with pytest.raises(ValueError) as exc:
+            serialize_key_and_certificates(
+                None, key, cert, None,
+                DummyKeySerializationEncryption(),
+            )
+        assert str(exc.value) == 'Unsupported key encryption type'