Parses an OpenSSH Private Key file

parse_openssh_key.py

#!/usr/bin/env python3

# /// script
# dependencies = [
#   "bcrypt",
#   "cryptography >= 43.0.0",
#   "pyyaml",
# ]
# ///

"""Unpacks OpenSSH Private Keys.

This script can be used to parse OpenSSH private key files. OpenSSH private
keys start with the header '-----BEGIN OPENSSH PRIVATE KEY-----'. It will
output the key data as a YAML string.

If the key is encrypted, specify the password with the '--password' argument.

OpenSSH key algorithms supported:
    ssh-rsa
    ecdsa-sha2-*
    ssh-ed25519

OpenSSH encryption ciphers supported:
    none
    3des-cbc
    aes128-cbc
    aes128-ctr
    [email protected]
    aes192-cbc
    aes192-ctr
    aes256-cbc
    aes256-ctr
    [email protected]
    [email protected]

This bash script can be used to generate some test keys:

ciphers="$( ssh -Q ciphers )"

for key_type in "rsa" "ecdsa" "ed25519"; do
    filename="${key_type}"
    ssh-keygen -C "${filename}" -f "${filename}" -N "" -m RFC4716 -t "${key_type}"

    while IFS= read -r cipher; do
        ssh-keygen -C "${filename}-${cipher}" -f "${filename}-${cipher}" -N password -m RFC4716 -t "${key_type}" -Z "${cipher}"
    done <<< "${ciphers}"
done

ssh-keygen -f "rsa-PEM" -N "" -m PEM -t rsa
for cipher in "aes128" "aes256" "des3"; do
    openssl genrsa -out "rsa-PEM-${cipher}" -passout pass:password -"${cipher}" 2048
    ssh-keygen -f "rsa-PEM-${cipher}" -y -P password > "rsa-PEM-${cipher}.pub"
done

openssl ecparam -name prime256v1 -genkey -noout -out ecdsa-PEM
ssh-keygen -f ecdsa-PEM -y > ecdsa-PEM.pub

for cipher in "aes128" "aes256" "des3"; do
    openssl ec -in ecdsa-PEM -out "ecdsa-PEM-${cipher}" -"${cipher}" -passout pass:password
    ssh-keygen -f "ecdsa-PEM-${cipher}" -y -P password > "ecdsa-PEM-${cipher}.pub"
done
"""

from __future__ import annotations

import argparse
import base64
import collections.abc
import dataclasses
import pathlib
import sys
import typing as t

import bcrypt
import yaml
from cryptography.hazmat.decrepit.ciphers import algorithms as decrepit_algorithms
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes

HAS_ARGCOMPLETE = True
try:
    import argcomplete
except ImportError:
    HAS_ARGCOMPLETE = False


@dataclasses.dataclass(frozen=True)
class RFC4253PubKey:
    identifier: str

    @classmethod
    def from_bytes(
        cls,
        data: bytes,
    ) -> RFC4253PubKey:
        view = memoryview(data)
        b_identifier, view = _read_string(view)
        identifier = b_identifier.decode()

        if identifier == "ssh-rsa":
            e, view = _read_mpint(view)
            n, view = _read_mpint(view)

            if view:
                raise ValueError(f"RSA Public Key had {len(view)} bytes left over")

            return SSHRsaPubKey(
                identifier=identifier,
                e=e,
                n=n,
            )

        elif identifier.startswith("ecdsa-sha2-"):
            ecdsa_id, view = _read_string(view)
            q, view = _read_string(view)

            if view:
                raise ValueError(f"ECDSA Public Key had {len(view)} bytes left over")

            return SSHEcdsaSha2PubKey(
                identifier=identifier,
                ec_identifier=ecdsa_id.decode(),
                q=q,
            )

        elif identifier.startswith("ssh-ed25519"):
            b, view = _read_string(view)

            if view:
                raise ValueError(f"Ed25519 Public Key had {len(view)} bytes left over")

            return SSHEd25519PubKey(
                identifier=identifier,
                b=b,
            )
        else:
            raise ValueError(f"Unsupported public key {identifier}")


@dataclasses.dataclass(frozen=True)
class SSHRsaPubKey(RFC4253PubKey):
    e: int
    n: int


@dataclasses.dataclass(frozen=True)
class SSHEcdsaSha2PubKey(RFC4253PubKey):
    ec_identifier: str
    q: bytes


@dataclasses.dataclass(frozen=True)
class SSHEd25519PubKey(RFC4253PubKey):
    b: bytes


@dataclasses.dataclass(frozen=True)
class BCryptKDFOptions:
    salt: bytes
    rounds: int

    @classmethod
    def from_bytes(
        cls,
        data: bytes,
    ) -> BCryptKDFOptions:
        view = memoryview(data)
        salt, view = _read_string(view)
        rounds, view = _read_uint32(view)

        if view:
            raise ValueError(f"BCrypt KDF Options had {len(view[4:])} bytes left over")

        return BCryptKDFOptions(
            salt=salt,
            rounds=rounds,
        )


@dataclasses.dataclass(frozen=True)
class SSHAgentKey:
    key_type: str
    comment: str


@dataclasses.dataclass(frozen=True)
class SSHRsaKey(SSHAgentKey):
    n: int
    e: int
    d: int
    iqmp: int
    p: int
    q: int


@dataclasses.dataclass(frozen=True)
class SSHEcdsaKey(SSHAgentKey):
    curve_name: str
    q: bytes
    d: int


@dataclasses.dataclass(frozen=True)
class SSHEddsaKey(SSHAgentKey):
    pub_key: bytes
    priv_key: bytes


@dataclasses.dataclass(frozen=True)
class OpenSSHPrivateKey:
    # https://github.com/openssh/openssh-portable/blob/master/PROTOCOL.key
    AUTH_MAGIC = "openssh-key-v1"

    cipher_name: str
    kdf_name: str
    kdf_options: BCryptKDFOptions | None
    pub_keys: list[RFC4253PubKey]
    enc_priv_keys: bytes
    mac: bytes

    def get_private_keys(
        self,
        password: str | None = None,
    ) -> list[SSHAgentKey]:
        if self.cipher_name == "none":
            data = self.enc_priv_keys

        elif not password:
            raise ValueError(
                "Private keys are encrypted, specify the password to decrypt them."
            )

        else:
            algo_factory: collections.abc.Callable[[bytes], algorithms.CipherAlgorithm]
            mode_factory: collections.abc.Callable[[bytes], modes.Mode | None]
            if self.cipher_name == "3des-cbc":
                key_size = 24
                iv_size = 8
                algo_factory = decrepit_algorithms.TripleDES
                mode_factory = modes.CBC

            elif self.cipher_name in [
                "aes128-cbc",
                "aes128-ctr",
                "aes192-cbc",
                "aes192-ctr",
                "aes256-cbc",
                "aes256-ctr",
                "[email protected]",
                "[email protected]",
            ]:
                key_size = int(self.cipher_name[3:6]) // 8
                iv_size = 12 if self.cipher_name.endswith("[email protected]") else 16
                algo_factory = algorithms.AES

                if self.cipher_name.endswith("cbc"):
                    mode_factory = modes.CBC
                elif self.cipher_name.endswith("ctr"):
                    mode_factory = modes.CTR
                else:
                    mode_factory = lambda iv: modes.GCM(iv, tag=self.mac)

            elif self.cipher_name == "[email protected]":
                key_size = 64
                iv_size = 0
                algo_factory = lambda key: algorithms.ChaCha20(
                    key[:32],
                    b"\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00",
                )
                mode_factory = lambda iv: None

            else:
                raise ValueError(
                    f"Decryption has not been implemented for {self.cipher_name}"
                )

            if not self.kdf_options:
                raise ValueError("Key Derivation Function options are missing")

            bcrypt_key = bcrypt.kdf(
                password=password.encode(),
                salt=self.kdf_options.salt,
                desired_key_bytes=key_size + iv_size,
                rounds=self.kdf_options.rounds,
                ignore_few_rounds=True,
            )
            key = bcrypt_key[:key_size]
            iv = bcrypt_key[key_size:]

            algorithm = algo_factory(key)
            mode = mode_factory(iv)

            cipher = Cipher(algorithm, mode=mode)
            decryptor = cipher.decryptor()
            data = decryptor.update(self.enc_priv_keys) + decryptor.finalize()

        # The list of privatekey/comment pairs is padded with the
        # bytes 1, 2, 3, ... until the total length is a multiple
        # of the cipher block size.

        #     uint32	checkint
        #     uint32	checkint
        #     byte[]	privatekey1
        #     string	comment1
        #     byte[]	privatekey2
        #     string	comment2
        #     ...
        #     byte[]	privatekeyN
        #     string	commentN
        #     byte	1
        #     byte	2
        #     byte	3
        #     ...
        #     byte	padlen % 255

        view = memoryview(data)
        check_int1, view = _read_uint32(view)
        check_int2, view = _read_uint32(view)
        if check_int1 != check_int2:
            raise ValueError("Failed to decrypt private key data, checkint mismatch")

        keys = []
        for _ in range(len(self.pub_keys)):
            # https://datatracker.ietf.org/doc/html/draft-miller-ssh-agent
            b_key_type, view = _read_string(view)
            key_type = b_key_type.decode()

            agent_key: SSHAgentKey
            if key_type == "ssh-rsa":
                n, view = _read_mpint(view)
                e, view = _read_mpint(view)
                d, view = _read_mpint(view)
                iqmp, view = _read_mpint(view)
                p, view = _read_mpint(view)
                q, view = _read_mpint(view)
                comment, view = _read_string(view)

                agent_key = SSHRsaKey(
                    key_type=key_type,
                    comment=comment.decode(),
                    n=n,
                    e=e,
                    d=d,
                    iqmp=iqmp,
                    p=p,
                    q=q,
                )

            elif key_type.startswith("ecdsa-sha2-"):
                curve_name, view = _read_string(view)
                pub_key, view = _read_string(view)
                ecdsa_key, view = _read_mpint(view)
                comment, view = _read_string(view)

                agent_key = SSHEcdsaKey(
                    key_type=key_type,
                    comment=comment.decode(),
                    curve_name=curve_name.decode(),
                    q=pub_key,
                    d=ecdsa_key,
                )

            elif key_type == "ssh-ed25519":
                pub_key, view = _read_string(view)
                ed25519_key, view = _read_string(view)
                comment, view = _read_string(view)

                agent_key = SSHEddsaKey(
                    key_type=key_type,
                    comment=comment.decode(),
                    pub_key=pub_key,
                    priv_key=ed25519_key[: -(len(pub_key))],
                )

            else:
                raise ValueError(f"Unknown private key type '{key_type}'")

            keys.append(agent_key)

        return keys

    @classmethod
    def from_bytes(
        cls,
        data: bytes,
    ) -> OpenSSHPrivateKey:
        # #define AUTH_MAGIC      "openssh-key-v1"

        #     byte[]	AUTH_MAGIC
        #     string	ciphername
        #     string	kdfname
        #     string	kdfoptions
        #     uint32	number of keys N
        #     string	publickey1
        #     string	publickey2
        #     ...
        #     string	publickeyN
        #     string	encrypted, padded list of private keys

        view = memoryview(data)
        if not data.startswith(f"{cls.AUTH_MAGIC}\x00".encode()):
            raise ValueError("Invalid OpenSSH private key format")

        view = view[15:]
        ciphername, view = _read_string(view)
        b_kdf_name, view = _read_string(view)
        kdf_name = b_kdf_name.decode()

        b_kdf_opts, view = _read_string(view)
        kdf_opts = None
        if kdf_name == "bcrypt":
            kdf_opts = BCryptKDFOptions.from_bytes(b_kdf_opts)
        elif kdf_name != "none":
            raise ValueError(f"Unsupported KDF Name {kdf_name}")

        num_pub_keys, view = _read_uint32(view)
        pub_keys = []
        for _ in range(num_pub_keys):
            pkey, view = _read_string(view)
            pub_keys.append(RFC4253PubKey.from_bytes(pkey))

        enc_pkeys, view = _read_string(view)

        return OpenSSHPrivateKey(
            cipher_name=ciphername.decode("utf-8"),
            kdf_name=kdf_name,
            kdf_options=kdf_opts,
            pub_keys=pub_keys,
            enc_priv_keys=enc_pkeys,
            mac=view.tobytes(),
        )


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        prog=sys.argv[0],
        description="Parse a key file.",
    )
    parser.add_argument(
        "key_file",
        type=pathlib.Path,
        help="The key file to parse.",
    )
    parser.add_argument(
        "--password",
        type=str,
        required=False,
        help="The password used to decrypt an encrypted key.",
    )

    if HAS_ARGCOMPLETE:
        argcomplete.autocomplete(parser)

    return parser.parse_args(sys.argv[1:])


def _read_mpint(view: memoryview) -> tuple[int, memoryview]:
    int_len = int.from_bytes(view[:4], byteorder="big", signed=False)
    view = view[4:]

    int_val = 0
    if int_len:
        int_val = int.from_bytes(view[:int_len], byteorder="big", signed=True)
        view = view[int_len:]

    return int_val, view


def _read_uint32(view: memoryview) -> tuple[int, memoryview]:
    val = int.from_bytes(view[:4], byteorder="big", signed=False)
    return val, view[4:]


def _read_string(view: memoryview) -> tuple[bytes, memoryview]:
    string_len = int.from_bytes(view[:4], byteorder="big", signed=False)
    view = view[4:]

    if string_len:
        return view[:string_len].tobytes(), view[string_len:]
    else:
        return b"", view


def main() -> None:
    args = parse_args()

    key_file = t.cast(pathlib.Path, args.key_file)
    if not key_file.exists():
        raise ValueError(f"Key file '{key_file}' does not exist.")

    if not key_file.is_file():
        raise ValueError(f"Key file '{key_file}' is not a file.")

    begin_label = ""
    end_label = ""
    reached_end = False
    key_lines = []
    metadata = {}
    for line in open(key_file, mode="r"):
        line = line.strip()

        if reached_end:
            raise ValueError("Key had extra data after end label")

        if line.startswith("-----BEGIN "):
            if begin_label:
                raise ValueError("Key contains multiple begin segments")

            begin_label = line
            end_label = begin_label.replace("BEGIN", "END")

        elif not begin_label:
            raise ValueError("Key did not start with begin segment")

        elif line.startswith("-----END "):
            if line != end_label:
                raise ValueError(
                    f"Key did not end with expected end label. Expecting '{end_label}' but got '{line}'"
                )

            reached_end = True

        elif ":" in line:
            key, value = line.split(":", 1)
            metadata[key.strip()] = value.strip()

        else:
            key_lines.append(line)

    if not reached_end:
        raise ValueError(f"Key did not end with expected end label '{end_label}'")

    key_data = base64.b64decode("".join(key_lines))
    if begin_label == "-----BEGIN OPENSSH PRIVATE KEY-----":
        key = OpenSSHPrivateKey.from_bytes(key_data)
        key_dict = dataclasses.asdict(key)
        key_dict = {"auth_magic": key.AUTH_MAGIC} | key_dict

        dec_key: str | list[dict[str, t.Any]]
        try:
            dec_keys = key.get_private_keys(password=args.password)
            dec_key = [dataclasses.asdict(k) for k in dec_keys]

        except ValueError as e:
            dec_key = str(e)

        key_dict |= {"decrypted_keys": dec_key}

        yaml.dump(key_dict, sys.stdout, sort_keys=False)

    else:
        raise ValueError(f"Unsupported key format '{begin_label}'")


if __name__ == "__main__":
    main()