Add Encryption2021 serialization encryption type

docs/hazmat/primitives/asymmetric/serialization.rst

@@ -978,6 +978,17 @@ Serialization Encryption Types
 
     :param bytes password: The password to use for encryption.
 
+.. class:: Encryption2021(password)
+
+    Encrypt using the best available encryption for a given key in year 2021.
+    The algorithm will never change. In the future, encryption may fail or the
+    class may be removed if the encryption algorithm is considered too weak.
+
+    As of now it uses the same algorithms as
+    :class:`~cryptography.hazmat.primitives.serialization.BestAvailableEncryption`.
+
+    :param bytes password: The password to use for encryption.
+
 .. class:: NoEncryption
 
     Do not encrypt.

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

@@ -1397,6 +1397,25 @@ def _private_key_bytes(
         else:
             raise ValueError("Unsupported encryption type")
 
+        # get cipher
+        pkcs8_cipher: typing.Optional[bytes] = None
+        traditionalopenssl_cipher: typing.Optional[bytes] = None
+        openssh_cipher: typing.Optional[bytes] = None
+
+        if isinstance(encryption_algorithm, serialization.Encryption2021):
+            # fixed values, deprecate and eventually remove the class
+            # if these algorithms become too weak.
+            pkcs8_cipher = b"aes-256-cbc"
+            traditionalopenssl_cipher = b"aes-256-cbc"
+            openssh_cipher = b"aes256-ctr"
+        elif isinstance(
+            encryption_algorithm, serialization.BestAvailableEncryption
+        ):
+            # Curated values that we will update over time.
+            pkcs8_cipher = b"aes-256-cbc"
+            traditionalopenssl_cipher = b"aes-256-cbc"
+            openssh_cipher = b"aes256-ctr"
+
         # PKCS8 + PEM/DER
         if format is serialization.PrivateFormat.PKCS8:
             if encoding is serialization.Encoding.PEM:
@@ -1406,7 +1425,7 @@ def _private_key_bytes(
             else:
                 raise ValueError("Unsupported encoding for PKCS8")
             return self._private_key_bytes_via_bio(
-                write_bio, evp_pkey, password
+                write_bio, evp_pkey, password, pkcs8_cipher
             )
 
         # TraditionalOpenSSL + PEM/DER
@@ -1432,7 +1451,7 @@ def _private_key_bytes(
                         "Unsupported key type for TraditionalOpenSSL"
                     )
                 return self._private_key_bytes_via_bio(
-                    write_bio, cdata, password
+                    write_bio, cdata, password, traditionalopenssl_cipher
                 )
 
             if encoding is serialization.Encoding.DER:
@@ -1458,7 +1477,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, ciphername=openssh_cipher
+                )
 
             raise ValueError(
                 "OpenSSH private key format can only be used"
@@ -1469,12 +1490,13 @@ def _private_key_bytes(
         # like Raw.
         raise ValueError("format is invalid with this key")
 
-    def _private_key_bytes_via_bio(self, write_bio, evp_pkey, password):
+    def _private_key_bytes_via_bio(
+        self, write_bio, evp_pkey, password, ciphername
+    ):
         if not password:
             evp_cipher = self._ffi.NULL
         else:
-            # This is a curated value that we will update over time.
-            evp_cipher = self._lib.EVP_get_cipherbyname(b"aes-256-cbc")
+            evp_cipher = self._lib.EVP_get_cipherbyname(ciphername)
 
         return self._bio_func_output(
             write_bio,

src/cryptography/hazmat/primitives/_serialization.py

@@ -51,5 +51,9 @@ def __init__(self, password: bytes):
         self.password = password
 
 
+class Encryption2021(BestAvailableEncryption):
+    pass
+
+
 class NoEncryption(KeySerializationEncryption):
     pass

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

@@ -6,6 +6,7 @@
 from cryptography.hazmat.primitives._serialization import (
     BestAvailableEncryption,
     Encoding,
+    Encryption2021,
     KeySerializationEncryption,
     NoEncryption,
     ParameterFormat,
@@ -41,5 +42,6 @@
     "ParameterFormat",
     "KeySerializationEncryption",
     "BestAvailableEncryption",
+    "Encryption2021",
     "NoEncryption",
 ]

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

@@ -550,6 +550,8 @@ def load_ssh_private_key(
 def serialize_ssh_private_key(
     private_key: _SSH_PRIVATE_KEY_TYPES,
     password: typing.Optional[bytes] = None,
+    *,
+    ciphername: typing.Optional[bytes] = None,
 ) -> bytes:
     """Serialize private key with OpenSSH custom encoding."""
     if password is not None:
@@ -575,7 +577,8 @@ def serialize_ssh_private_key(
     # setup parameters
     f_kdfoptions = _FragList()
     if password:
-        ciphername = _DEFAULT_CIPHER
+        if ciphername is None:
+            ciphername = _DEFAULT_CIPHER
         blklen = _SSH_CIPHERS[ciphername][3]
         kdfname = _BCRYPT
         rounds = _DEFAULT_ROUNDS

tests/hazmat/primitives/test_rsa.py

@@ -2180,7 +2180,7 @@ def test_invalid_recover_prime_factors(self):
 
 class TestRSAPrivateKeySerialization(object):
     @pytest.mark.parametrize(
-        ("fmt", "password"),
+        ("fmt", "password", "encryption"),
         itertools.product(
             [
                 serialization.PrivateFormat.TraditionalOpenSSL,
@@ -2192,15 +2192,21 @@ class TestRSAPrivateKeySerialization(object):
                 b"!*$&(@#$*&($T@%_somesymbols",
                 b"\x01" * 1000,
             ],
+            [
+                serialization.BestAvailableEncryption,
+                serialization.Encryption2021,
+            ],
         ),
     )
-    def test_private_bytes_encrypted_pem(self, backend, fmt, password):
+    def test_private_bytes_encrypted_pem(
+        self, backend, fmt, password, encryption
+    ):
         skip_fips_traditional_openssl(backend, fmt)
         key = RSA_KEY_2048.private_key(backend)
         serialized = key.private_bytes(
             serialization.Encoding.PEM,
             fmt,
-            serialization.BestAvailableEncryption(password),
+            encryption(password),
         )
         loaded_key = serialization.load_pem_private_key(
             serialized, password, backend
@@ -2225,20 +2231,29 @@ def test_private_bytes_rejects_invalid(self, encoding, fmt, backend):
             key.private_bytes(encoding, fmt, serialization.NoEncryption())
 
     @pytest.mark.parametrize(
-        ("fmt", "password"),
-        [
-            [serialization.PrivateFormat.PKCS8, b"s"],
-            [serialization.PrivateFormat.PKCS8, b"longerpassword"],
-            [serialization.PrivateFormat.PKCS8, b"!*$&(@#$*&($T@%_somesymbol"],
-            [serialization.PrivateFormat.PKCS8, b"\x01" * 1000],
-        ],
+        ("fmt", "password", "encryption"),
+        itertools.product(
+            [serialization.PrivateFormat.PKCS8],
+            [
+                b"s",
+                b"longerpassword",
+                b"!*$&(@#$*&($T@%_somesymbols",
+                b"\x01" * 1000,
+            ],
+            [
+                serialization.BestAvailableEncryption,
+                serialization.Encryption2021,
+            ],
+        ),
     )
-    def test_private_bytes_encrypted_der(self, backend, fmt, password):
+    def test_private_bytes_encrypted_der(
+        self, backend, fmt, password, encryption
+    ):
         key = RSA_KEY_2048.private_key(backend)
         serialized = key.private_bytes(
             serialization.Encoding.DER,
             fmt,
-            serialization.BestAvailableEncryption(password),
+            encryption(password),
         )
         loaded_key = serialization.load_der_private_key(
             serialized, password, backend

tests/hazmat/primitives/test_serialization.py

@@ -23,6 +23,7 @@
 from cryptography.hazmat.primitives.serialization import (
     BestAvailableEncryption,
     Encoding,
+    Encryption2021,
     KeySerializationEncryption,
     NoEncryption,
     PrivateFormat,
@@ -1372,13 +1373,19 @@ def test_load_ssh_public_key_trailing_data(self, backend):
 
 
 class TestKeySerializationEncryptionTypes(object):
-    def test_non_bytes_password(self):
+    @pytest.mark.parametrize(
+        "encryption_class", [BestAvailableEncryption, Encryption2021]
+    )
+    def test_non_bytes_password(self, encryption_class):
         with pytest.raises(ValueError):
-            BestAvailableEncryption(object())  # type:ignore[arg-type]
+            encryption_class(object())  # type:ignore[arg-type]
 
-    def test_encryption_with_zero_length_password(self):
+    @pytest.mark.parametrize(
+        "encryption_class", [BestAvailableEncryption, Encryption2021]
+    )
+    def test_encryption_with_zero_length_password(self, encryption_class):
         with pytest.raises(ValueError):
-            BestAvailableEncryption(b"")
+            encryption_class(b"")
 
 
 @pytest.mark.supported(
@@ -1990,15 +1997,16 @@ def test_bcrypt_encryption(self, backend):
             b"x" * 72,
         ):
             # BestAvailableEncryption does not handle bytes-like?
-            best = BestAvailableEncryption(psw)
-            encdata = private_key.private_bytes(
-                Encoding.PEM, PrivateFormat.OpenSSH, best
-            )
-            decoded_key = load_ssh_private_key(encdata, psw, backend)
-            pub2 = decoded_key.public_key().public_bytes(
-                Encoding.OpenSSH, PublicFormat.OpenSSH
-            )
-            assert pub1 == pub2
+            for encryption_class in [BestAvailableEncryption, Encryption2021]:
+                encryption = encryption_class(psw)
+                encdata = private_key.private_bytes(
+                    Encoding.PEM, PrivateFormat.OpenSSH, encryption
+                )
+                decoded_key = load_ssh_private_key(encdata, psw, backend)
+                pub2 = decoded_key.public_key().public_bytes(
+                    Encoding.OpenSSH, PublicFormat.OpenSSH
+                )
+                assert pub1 == pub2
 
             # bytearray
             decoded_key2 = load_ssh_private_key(