add support for aes256-gcm@openssh.com decryption for SSH keys (#8738)

* add support for aes256-gcm@openssh.com decryption for SSH keys

* review feedback

* skip when bcrypt isn't present

CHANGELOG.rst

@@ -17,6 +17,8 @@ Changelog
 * Added support for the :class:`~cryptography.x509.MSCertificateTemplate`
   proprietary Microsoft certificate extension.
 * Implemented support for equality checks on all asymmetric public key types.
+* Added support for ``aes256-gcm@openssh.com`` encrypted keys in
+  :func:`~cryptography.hazmat.primitives.serialization.load_ssh_private_key`.
 
 .. _v40-0-2:
 

docs/development/test-vectors.rst

@@ -803,6 +803,10 @@ Custom PKCS7 Test Vectors
 Custom OpenSSH Test Vectors
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
+* ``ed25519-aesgcm-psw.key`` and ``ed25519-aesgcm-psw.key.pub`` generated by
+  exporting an Ed25519 key from ``1password 8`` with the password "password".
+  This key is encrypted using the ``aes256-gcm@openssh.com`` algorithm.
+
 Generated by
 ``asymmetric/OpenSSH/gen.sh``
 using command-line tools from OpenSSH_7.6p1 package.

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

@@ -11,6 +11,7 @@ import re
 import typing
 import warnings
 from base64 import encodebytes as _base64_encode
+from dataclasses import dataclass
 
 from cryptography import utils
 from cryptography.exceptions import UnsupportedAlgorithm
@@ -23,7 +24,12 @@ from cryptography.hazmat.primitives.asymmetric import (
     rsa,
 )
 from cryptography.hazmat.primitives.asymmetric import utils as asym_utils
-from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+from cryptography.hazmat.primitives.ciphers import (
+    AEADDecryptionContext,
+    Cipher,
+    algorithms,
+    modes,
+)
 from cryptography.hazmat.primitives.serialization import (
     Encoding,
     KeySerializationEncryption,
@@ -78,18 +84,51 @@ _PEM_RC = re.compile(_SK_START + b"(.*?)" + _SK_END, re.DOTALL)
 # padding for max blocksize
 _PADDING = memoryview(bytearray(range(1, 1 + 16)))
 
+
+@dataclass
+class _SSHCipher:
+    alg: typing.Type[algorithms.AES]
+    key_len: int
+    mode: typing.Union[
+        typing.Type[modes.CTR],
+        typing.Type[modes.CBC],
+        typing.Type[modes.GCM],
+    ]
+    block_len: int
+    iv_len: int
+    tag_len: typing.Optional[int]
+    is_aead: bool
+
+
 # ciphers that are actually used in key wrapping
-_SSH_CIPHERS: typing.Dict[
-    bytes,
-    typing.Tuple[
-        typing.Type[algorithms.AES],
-        int,
-        typing.Union[typing.Type[modes.CTR], typing.Type[modes.CBC]],
-        int,
-    ],
-] = {
-    b"aes256-ctr": (algorithms.AES, 32, modes.CTR, 16),
-    b"aes256-cbc": (algorithms.AES, 32, modes.CBC, 16),
+_SSH_CIPHERS: typing.Dict[bytes, _SSHCipher] = {
+    b"aes256-ctr": _SSHCipher(
+        alg=algorithms.AES,
+        key_len=32,
+        mode=modes.CTR,
+        block_len=16,
+        iv_len=16,
+        tag_len=None,
+        is_aead=False,
+    ),
+    b"aes256-cbc": _SSHCipher(
+        alg=algorithms.AES,
+        key_len=32,
+        mode=modes.CBC,
+        block_len=16,
+        iv_len=16,
+        tag_len=None,
+        is_aead=False,
+    ),
+    b"aes256-gcm@openssh.com": _SSHCipher(
+        alg=algorithms.AES,
+        key_len=32,
+        mode=modes.GCM,
+        block_len=16,
+        iv_len=12,
+        tag_len=16,
+        is_aead=True,
+    ),
 }
 
 # map local curve name to key type
@@ -156,14 +195,19 @@ def _init_cipher(
     password: typing.Optional[bytes],
     salt: bytes,
     rounds: int,
-) -> Cipher[typing.Union[modes.CBC, modes.CTR]]:
+) -> Cipher[typing.Union[modes.CBC, modes.CTR, modes.GCM]]:
     """Generate key + iv and return cipher."""
     if not password:
         raise ValueError("Key is password-protected.")
 
-    algo, key_len, mode, iv_len = _SSH_CIPHERS[ciphername]
-    seed = _bcrypt_kdf(password, salt, key_len + iv_len, rounds, True)
-    return Cipher(algo(seed[:key_len]), mode(seed[key_len:]))
+    ciph = _SSH_CIPHERS[ciphername]
+    seed = _bcrypt_kdf(
+        password, salt, ciph.key_len + ciph.iv_len, rounds, True
+    )
+    return Cipher(
+        ciph.alg(seed[: ciph.key_len]),
+        ciph.mode(seed[ciph.key_len :]),
+    )
 
 
 def _get_u32(data: memoryview) -> typing.Tuple[int, memoryview]:
@@ -604,10 +648,6 @@ def load_ssh_private_key(
     pubfields, pubdata = kformat.get_public(pubdata)
     _check_empty(pubdata)
 
-    # load secret data
-    edata, data = _get_sshstr(data)
-    _check_empty(data)
-
     if (ciphername, kdfname) != (_NONE, _NONE):
         ciphername_bytes = ciphername.tobytes()
         if ciphername_bytes not in _SSH_CIPHERS:
@@ -616,14 +656,36 @@ def load_ssh_private_key(
             )
         if kdfname != _BCRYPT:
             raise UnsupportedAlgorithm(f"Unsupported KDF: {kdfname!r}")
-        blklen = _SSH_CIPHERS[ciphername_bytes][3]
+        blklen = _SSH_CIPHERS[ciphername_bytes].block_len
+        tag_len = _SSH_CIPHERS[ciphername_bytes].tag_len
+        # load secret data
+        edata, data = _get_sshstr(data)
+        # see https://bugzilla.mindrot.org/show_bug.cgi?id=3553 for
+        # information about how OpenSSH handles AEAD tags
+        if _SSH_CIPHERS[ciphername_bytes].is_aead:
+            tag = bytes(data)
+            if len(tag) != tag_len:
+                raise ValueError("Corrupt data: invalid tag length for cipher")
+        else:
+            _check_empty(data)
         _check_block_size(edata, blklen)
         salt, kbuf = _get_sshstr(kdfoptions)
         rounds, kbuf = _get_u32(kbuf)
         _check_empty(kbuf)
         ciph = _init_cipher(ciphername_bytes, password, salt.tobytes(), rounds)
-        edata = memoryview(ciph.decryptor().update(edata))
+        dec = ciph.decryptor()
+        edata = memoryview(dec.update(edata))
+        if _SSH_CIPHERS[ciphername_bytes].is_aead:
+            assert isinstance(dec, AEADDecryptionContext)
+            _check_empty(dec.finalize_with_tag(tag))
+        else:
+            # _check_block_size requires data to be a full block so there
+            # should be no output from finalize
+            _check_empty(dec.finalize())
     else:
+        # load secret data
+        edata, data = _get_sshstr(data)
+        _check_empty(data)
         blklen = 8
         _check_block_size(edata, blklen)
     ck1, edata = _get_u32(edata)
@@ -676,7 +738,7 @@ def _serialize_ssh_private_key(
     f_kdfoptions = _FragList()
     if password:
         ciphername = _DEFAULT_CIPHER
-        blklen = _SSH_CIPHERS[ciphername][3]
+        blklen = _SSH_CIPHERS[ciphername].block_len
         kdfname = _BCRYPT
         rounds = _DEFAULT_ROUNDS
         if (

tests/hazmat/primitives/test_ssh.py

@@ -10,7 +10,7 @@ import os
 import pytest
 
 from cryptography import utils
-from cryptography.exceptions import InvalidSignature
+from cryptography.exceptions import InvalidSignature, InvalidTag
 from cryptography.hazmat.primitives.asymmetric import (
     dsa,
     ec,
@@ -153,6 +153,7 @@ class TestOpenSSHSerialization:
             ("ecdsa-psw.key",),
             ("ed25519-nopsw.key",),
             ("ed25519-psw.key",),
+            ("ed25519-aesgcm-psw.key",),
         ],
     )
     def test_load_ssh_private_key(self, key_file, backend):
@@ -243,6 +244,48 @@ class TestOpenSSHSerialization:
         maxline = max(map(len, priv_data2.split(b"\n")))
         assert maxline < 80
 
+    @pytest.mark.supported(
+        only_if=lambda backend: backend.ed25519_supported(),
+        skip_message="Requires Ed25519 support",
+    )
+    @pytest.mark.supported(
+        only_if=lambda backend: ssh._bcrypt_supported,
+        skip_message="Requires that bcrypt exists",
+    )
+    def test_load_ssh_private_key_invalid_tag(self, backend):
+        priv_data = bytearray(
+            load_vectors_from_file(
+                os.path.join(
+                    "asymmetric", "OpenSSH", "ed25519-aesgcm-psw.key"
+                ),
+                lambda f: f.read(),
+                mode="rb",
+            )
+        )
+        # mutate one byte to break the tag
+        priv_data[-38] = 82
+        with pytest.raises(InvalidTag):
+            load_ssh_private_key(priv_data, b"password")
+
+    @pytest.mark.supported(
+        only_if=lambda backend: backend.ed25519_supported(),
+        skip_message="Requires Ed25519 support",
+    )
+    @pytest.mark.supported(
+        only_if=lambda backend: ssh._bcrypt_supported,
+        skip_message="Requires that bcrypt exists",
+    )
+    def test_load_ssh_private_key_tag_incorrect_length(self, backend):
+        priv_data = load_vectors_from_file(
+            os.path.join("asymmetric", "OpenSSH", "ed25519-aesgcm-psw.key"),
+            lambda f: f.read(),
+            mode="rb",
+        )
+        # clip out a byte
+        broken_data = priv_data[:-37] + priv_data[-38:]
+        with pytest.raises(ValueError):
+            load_ssh_private_key(broken_data, b"password")
+
     @pytest.mark.supported(
         only_if=lambda backend: ssh._bcrypt_supported,
         skip_message="Requires that bcrypt exists",

vectors/cryptography_vectors/asymmetric/OpenSSH/ed25519-aesgcm-psw.key

@@ -0,0 +1,8 @@
+-----BEGIN OPENSSH PRIVATE KEY-----
+b3BlbnNzaC1rZXktdjEAAAAAFmFlczI1Ni1nY21Ab3BlbnNzaC5jb20AAAAGYmNyeXB0AA
+AAGAAAABBxwbaftabtGFPlzbCIuqOIAAAAIAAAAAEAAAAzAAAAC3NzaC1lZDI1NTE5AAAA
+ICuPdFT6OORNyXh9rMfOx3LUCm9yANYovOfNlGd2hg01AAAAkBl0VICPNwd88NHm9w10X0
+bn0WTOJMzyQBw8cNZvswPvczViEFmW0pZwDmeVrBBTLmktn4b3D7IfCMJIbfAq+N+rRZ0p
+xhPi6toZopq1wP4dE44DYQ1dr2K4evLv5pRCLJUkmNny/7jFEOggVx8N5o8pOSuf0tNhYd
+SCn7oNc1syjS2w0Zjb2ZTiX4L9d60tSLDwLOolS1Xc0nPUMnzC5hM=
+-----END OPENSSH PRIVATE KEY-----

vectors/cryptography_vectors/asymmetric/OpenSSH/ed25519-aesgcm-psw.key.pub

@@ -0,0 +1 @@
+ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAICuPdFT6OORNyXh9rMfOx3LUCm9yANYovOfNlGd2hg01