Initial APIs for making serialization encryption configurable

CHANGELOG.rst

@@ -49,7 +49,10 @@ Changelog
 * Fixed :rfc:`4514` name parsing to reverse the order of the RDNs according 
   to the section 2.1 of the RFC, affecting method 
   :meth:`~cryptography.x509.Name.from_rfc4514_string`.
-
+* It is now possible to customize some aspects of encryption when serializing
+  private keys, using
+  :meth:`~cryptography.hazmat.primitives.serialization.PrivateFormat.encryption_builder`.
+
 .. _v37-0-4:
 
 37.0.4 - 2022-07-05

docs/hazmat/primitives/asymmetric/serialization.rst

@@ -638,7 +638,7 @@ contain certificates, CRLs, and much more. PKCS7 files commonly have a ``p7b``,
     :param certs: A list of :class:`~cryptography.x509.Certificate`.
     :param encoding: :attr:`~cryptography.hazmat.primitives.serialization.Encoding.PEM`
         or :attr:`~cryptography.hazmat.primitives.serialization.Encoding.DER`.
-    :return bytes: The serialized PKCS7 data.
+    :returns bytes: The serialized PKCS7 data.
 
 .. testsetup::
 
@@ -730,7 +730,7 @@ contain certificates, CRLs, and much more. PKCS7 files commonly have a ``p7b``,
         :param options: A list of
             :class:`~cryptography.hazmat.primitives.serialization.pkcs7.PKCS7Options`.
 
-        :return bytes: The signed PKCS7 message.
+        :returns bytes: The signed PKCS7 message.
 
 
 .. class:: PKCS7Options
@@ -841,6 +841,30 @@ Serialization Formats
             ...
             -----END OPENSSH PRIVATE KEY-----
 
+    .. method:: encryption_builder()
+
+        .. versionadded:: 38.0.0
+
+        Returns a builder for configuring how values are encrypted with this
+        format.
+
+        For most use cases, :class:`BestAvailableEncryption` is preferred.
+
+        :returns KeySerializationEncryptionBuilder: A new builder.
+
+        .. doctest::
+
+            >>> from cryptography.hazmat.primitives import serialization
+            >>> encryption = (
+            ...     serialization.PrivateFormat.OpenSSH.encryption_builder().kdf_rounds(30).build(b"my password")
+            ... )
+            >>> key.private_bytes(
+            ...     encoding=serialization.Encoding.PEM,
+            ...     format=serialization.PrivateFormat.OpenSSH,
+            ...     encryption_algorithm=encryption
+            ... )
+            b'-----BEGIN OPENSSH PRIVATE KEY-----\n...\n-----END OPENSSH PRIVATE KEY-----\n'
+
 
 .. class:: PublicFormat
 
@@ -1007,6 +1031,28 @@ Serialization Encryption Types
     Do not encrypt.
 
 
+.. class:: KeySerializationEncryptionBuilder
+
+    A builder that can be used to configure how key data is encrypted. To
+    create one, call :meth:`PrivateFormat.encryption_builder`.
+
+    .. method:: kdf_rounds(rounds)
+
+        Set the number of rounds the Key Derivation Function should use. The
+        meaning of the number of rounds varies on the KDF being used.
+
+        :param int rounds: Number of rounds.
+        :returns KeySerializationEncryptionBuilder: A new builder.
+
+    .. method:: build(password)
+
+        Turns the builder into an instance of
+        :class:`KeySerializationEncryption` with a given password.
+
+        :param bytes password: The password.
+        :returns KeySerializationEncryption: A key key serialization
+            encryption that can be passed to ``private_bytes`` methods.
+
 .. _`a bug in Firefox`: https://bugzilla.mozilla.org/show_bug.cgi?id=773111
 .. _`PKCS3`: https://www.teletrust.de/fileadmin/files/oid/oid_pkcs-3v1-4.pdf
 .. _`SEC 1 v2.0`: https://www.secg.org/sec1-v2.pdf

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

@@ -1528,6 +1528,15 @@ def _private_key_bytes(
                     "Passwords longer than 1023 bytes are not supported by "
                     "this backend"
                 )
+        elif (
+            isinstance(
+                encryption_algorithm, serialization._KeySerializationEncryption
+            )
+            and encryption_algorithm._format
+            is format
+            is serialization.PrivateFormat.OpenSSH
+        ):
+            password = encryption_algorithm.password
         else:
             raise ValueError("Unsupported encryption type")
 
@@ -1592,7 +1601,9 @@ def _private_key_bytes(
         # OpenSSH + PEM
         if format is serialization.PrivateFormat.OpenSSH:
             if encoding is serialization.Encoding.PEM:
-                return ssh.serialize_ssh_private_key(key, password)
+                return ssh._serialize_ssh_private_key(
+                    key, password, encryption_algorithm
+                )
 
             raise ValueError(
                 "OpenSSH private key format can only be used"

src/cryptography/hazmat/primitives/_serialization.py

@@ -3,6 +3,7 @@
 # for complete details.
 
 import abc
+import typing
 
 from cryptography import utils
 
@@ -25,6 +26,13 @@ class PrivateFormat(utils.Enum):
     Raw = "Raw"
     OpenSSH = "OpenSSH"
 
+    def encryption_builder(self) -> "KeySerializationEncryptionBuilder":
+        if self is not PrivateFormat.OpenSSH:
+            raise ValueError(
+                "encryption_builder only supported with PrivateFormat.OpenSSH"
+            )
+        return KeySerializationEncryptionBuilder(self)
+
 
 class PublicFormat(utils.Enum):
     SubjectPublicKeyInfo = "X.509 subjectPublicKeyInfo with PKCS#1"
@@ -53,3 +61,44 @@ def __init__(self, password: bytes):
 
 class NoEncryption(KeySerializationEncryption):
     pass
+
+
+class KeySerializationEncryptionBuilder(object):
+    def __init__(
+        self,
+        format: PrivateFormat,
+        *,
+        _kdf_rounds: typing.Optional[int] = None,
+    ) -> None:
+        self._format = format
+
+        self._kdf_rounds = _kdf_rounds
+
+    def kdf_rounds(self, rounds: int) -> "KeySerializationEncryptionBuilder":
+        if self._kdf_rounds is not None:
+            raise ValueError("kdf_rounds already set")
+        return KeySerializationEncryptionBuilder(
+            self._format, _kdf_rounds=rounds
+        )
+
+    def build(self, password: bytes) -> KeySerializationEncryption:
+        if not isinstance(password, bytes) or len(password) == 0:
+            raise ValueError("Password must be 1 or more bytes.")
+
+        return _KeySerializationEncryption(
+            self._format, password, kdf_rounds=self._kdf_rounds
+        )
+
+
+class _KeySerializationEncryption(KeySerializationEncryption):
+    def __init__(
+        self,
+        format: PrivateFormat,
+        password: bytes,
+        *,
+        kdf_rounds: typing.Optional[int],
+    ):
+        self._format = format
+        self.password = password
+
+        self._kdf_rounds = kdf_rounds

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

@@ -11,6 +11,7 @@
     ParameterFormat,
     PrivateFormat,
     PublicFormat,
+    _KeySerializationEncryption,
 )
 from cryptography.hazmat.primitives.serialization.base import (
     load_der_parameters,
@@ -42,4 +43,5 @@
     "KeySerializationEncryption",
     "BestAvailableEncryption",
     "NoEncryption",
+    "_KeySerializationEncryption",
 ]

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

@@ -15,9 +15,11 @@
 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
 from cryptography.hazmat.primitives.serialization import (
     Encoding,
+    KeySerializationEncryption,
     NoEncryption,
     PrivateFormat,
     PublicFormat,
+    _KeySerializationEncryption,
 )
 
 try:
@@ -601,13 +603,13 @@ def load_ssh_private_key(
     return private_key
 
 
-def serialize_ssh_private_key(
+def _serialize_ssh_private_key(
     private_key: _SSH_PRIVATE_KEY_TYPES,
-    password: typing.Optional[bytes] = None,
+    password: bytes,
+    encryption_algorithm: KeySerializationEncryption,
 ) -> bytes:
     """Serialize private key with OpenSSH custom encoding."""
-    if password is not None:
-        utils._check_bytes("password", password)
+    utils._check_bytes("password", password)
 
     if isinstance(private_key, ec.EllipticCurvePrivateKey):
         key_type = _ecdsa_key_type(private_key.public_key())
@@ -628,6 +630,11 @@ def serialize_ssh_private_key(
         blklen = _SSH_CIPHERS[ciphername][3]
         kdfname = _BCRYPT
         rounds = _DEFAULT_ROUNDS
+        if (
+            isinstance(encryption_algorithm, _KeySerializationEncryption)
+            and encryption_algorithm._kdf_rounds is not None
+        ):
+            rounds = encryption_algorithm._kdf_rounds
         salt = os.urandom(16)
         f_kdfoptions.put_sshstr(salt)
         f_kdfoptions.put_u32(rounds)

tests/hazmat/primitives/test_serialization.py

@@ -2335,9 +2335,10 @@ def test_serialize_ssh_private_key_errors(self, backend):
 
         # bad object type
         with pytest.raises(ValueError):
-            ssh.serialize_ssh_private_key(
+            ssh._serialize_ssh_private_key(
                 object(),  # type:ignore[arg-type]
-                None,
+                b"",
+                NoEncryption(),
             )
 
         private_key = ec.generate_private_key(ec.SECP256R1(), backend)
@@ -2362,11 +2363,26 @@ def test_serialize_ssh_private_key_errors(self, backend):
             b"x" * 100,
         ),
     )
-    def test_serialize_ssh_private_key_with_password(self, password, backend):
+    @pytest.mark.parametrize(
+        "kdf_rounds",
+        [
+            1,
+            10,
+            30,
+        ],
+    )
+    def test_serialize_ssh_private_key_with_password(
+        self, password, kdf_rounds, backend
+    ):
         original_key = ec.generate_private_key(ec.SECP256R1(), backend)
-        encoded_key_data = ssh.serialize_ssh_private_key(
-            private_key=original_key,
-            password=password,
+        encoded_key_data = original_key.private_bytes(
+            Encoding.PEM,
+            PrivateFormat.OpenSSH,
+            (
+                PrivateFormat.OpenSSH.encryption_builder()
+                .kdf_rounds(kdf_rounds)
+                .build(password)
+            ),
         )
 
         decoded_key = load_ssh_private_key(
@@ -2416,3 +2432,22 @@ def test_dsa_private_key_sizes(self, key_path, supported, backend):
                 key.private_bytes(
                     Encoding.PEM, PrivateFormat.OpenSSH, NoEncryption()
                 )
+
+
+class TestEncryptionBuilder:
+    def test_unsupported_format(self):
+        f = PrivateFormat.PKCS8
+        with pytest.raises(ValueError):
+            f.encryption_builder()
+
+    def test_duplicate_kdf_rounds(self):
+        b = PrivateFormat.OpenSSH.encryption_builder().kdf_rounds(12)
+        with pytest.raises(ValueError):
+            b.kdf_rounds(12)
+
+    def test_invalid_password(self):
+        b = PrivateFormat.OpenSSH.encryption_builder()
+        with pytest.raises(ValueError):
+            b.build(12)  # type: ignore[arg-type]
+        with pytest.raises(ValueError):
+            b.build(b"")

tox.ini

@@ -21,6 +21,7 @@ extras =
     docs
     docstest
     sdist
+    ssh
 basepython = python3
 commands =
     sphinx-build -T -W -b html -d {envtmpdir}/doctrees docs docs/_build/html