diff --git a/lib/Crypto/PublicKey/DSA.py b/lib/Crypto/PublicKey/DSA.py index 4c7f47b68..6373f6134 100644 --- a/lib/Crypto/PublicKey/DSA.py +++ b/lib/Crypto/PublicKey/DSA.py @@ -612,7 +612,8 @@ def import_key(extern_key, passphrase=None): - X.509 certificate (binary DER or PEM) - X.509 ``subjectPublicKeyInfo`` (binary DER or PEM) - - OpenSSH (ASCII one-liner, see `RFC4253`_) + - OpenSSH (ASCII one-liner, see `RFC4253`_), including public + certificate lines. The following formats are supported for a DSA **private** key: @@ -664,6 +665,24 @@ def import_key(extern_key, passphrase=None): tup = [Integer.from_bytes(keyparts[x]) for x in (4, 3, 1, 2)] return construct(tup) + if extern_key.startswith(b'ssh-dss-cert-v01@openssh.com '): + from ._openssh import (import_openssh_public_cert_generic, + read_bytes, + check_openssh_public_cert_footer) + + key_type, keystring = import_openssh_public_cert_generic(extern_key) + if key_type != b"ssh-dss-cert-v01@openssh.com": + raise ValueError("This SSH certificate is not DSA") + + p, keystring = read_bytes(keystring) + q, keystring = read_bytes(keystring) + g, keystring = read_bytes(keystring) + y, keystring = read_bytes(keystring) + check_openssh_public_cert_footer(keystring) + + tup = [Integer.from_bytes(x) for x in (y, g, p, q)] + return construct(tup) + if len(extern_key) > 0 and bord(extern_key[0]) == 0x30: # This is probably a DER encoded key return _import_key_der(extern_key, passphrase, None) diff --git a/lib/Crypto/PublicKey/ECC.py b/lib/Crypto/PublicKey/ECC.py index 46789f90a..8f06a3ac5 100644 --- a/lib/Crypto/PublicKey/ECC.py +++ b/lib/Crypto/PublicKey/ECC.py @@ -959,6 +959,49 @@ def _import_openssh_public(encoded): raise ValueError("Not an openssh public key") try: + if parts[0].endswith(b"-cert-v01@openssh.com"): + from ._openssh import (import_openssh_public_cert_generic, + read_bytes, + check_openssh_public_cert_footer) + + key_type, keystring = import_openssh_public_cert_generic(encoded) + + # NIST P curves + if key_type.startswith(b"ecdsa-sha2-"): + ecdsa_curve_name, keystring = read_bytes(keystring) + expected_type = (b"ecdsa-sha2-" + ecdsa_curve_name + + b"-cert-v01@openssh.com") + if key_type != expected_type: + raise ValueError("Mismatch in openssh public certificate") + + for curve_name, curve in _curves.items(): + if curve.openssh is None: + continue + if not curve.openssh.startswith("ecdsa-sha2"): + continue + middle = tobytes(curve.openssh.split("-")[2]) + if ecdsa_curve_name == middle: + break + else: + raise ValueError("Unsupported ECC curve: " + + tostr(ecdsa_curve_name)) + + public_key, keystring = read_bytes(keystring) + check_openssh_public_cert_footer(keystring) + ecc_key = _import_public_der(public_key, curve_oid=curve.oid) + + # EdDSA + elif key_type == b"ssh-ed25519-cert-v01@openssh.com": + public_key, keystring = read_bytes(keystring) + check_openssh_public_cert_footer(keystring) + x, y = _import_ed25519_public_key(public_key) + ecc_key = construct(curve="Ed25519", point_x=x, point_y=y) + else: + raise ValueError("Unsupported SSH certificate type: " + + tostr(key_type)) + + return ecc_key + keystring = binascii.a2b_base64(parts[1]) keyparts = [] @@ -982,7 +1025,7 @@ def _import_openssh_public(encoded): if keyparts[1] == middle: break else: - raise ValueError("Unsupported ECC curve: " + middle) + raise ValueError("Unsupported ECC curve: " + tostr(middle)) ecc_key = _import_public_der(keyparts[2], curve_oid=curve.oid) @@ -991,10 +1034,10 @@ def _import_openssh_public(encoded): x, y = _import_ed25519_public_key(keyparts[1]) ecc_key = construct(curve="Ed25519", point_x=x, point_y=y) else: - raise ValueError("Unsupported SSH key type: " + parts[0]) + raise ValueError("Unsupported SSH key type: " + tostr(parts[0])) except (IndexError, TypeError, binascii.Error): - raise ValueError("Error parsing SSH key type: " + parts[0]) + raise ValueError("Error parsing SSH key type: " + tostr(parts[0])) return ecc_key @@ -1212,6 +1255,8 @@ def import_key(encoded, passphrase=None, curve_name=None): You must also provide the ``curve_name`` (with a value from the `ECC table`_) * OpenSSH line, defined in RFC5656_ and RFC8709_ (ASCII). This is normally the content of files like ``~/.ssh/id_ecdsa.pub``. + OpenSSH public certificate lines are also accepted and imported + as regular public keys. Supported formats for an ECC **private** key: diff --git a/lib/Crypto/PublicKey/RSA.py b/lib/Crypto/PublicKey/RSA.py index bc8f9d2d5..31006b75b 100644 --- a/lib/Crypto/PublicKey/RSA.py +++ b/lib/Crypto/PublicKey/RSA.py @@ -792,6 +792,7 @@ def import_key(extern_key, passphrase=None): encoding) - `PKCS#1`_ ``RSAPublicKey`` DER SEQUENCE (binary or PEM encoding) - An OpenSSH line (e.g. the content of ``~/.ssh/id_ecdsa``, ASCII) + or an OpenSSH public certificate line. The following formats are supported for an RSA **private key**: @@ -850,6 +851,23 @@ def import_key(extern_key, passphrase=None): n = Integer.from_bytes(keyparts[2]) return construct([n, e]) + if extern_key.startswith(b'ssh-rsa-cert-v01@openssh.com '): + from ._openssh import (import_openssh_public_cert_generic, + read_bytes, + check_openssh_public_cert_footer) + + key_type, keystring = import_openssh_public_cert_generic(extern_key) + if key_type != b"ssh-rsa-cert-v01@openssh.com": + raise ValueError("This SSH certificate is not RSA") + + e, keystring = read_bytes(keystring) + n, keystring = read_bytes(keystring) + check_openssh_public_cert_footer(keystring) + + e = Integer.from_bytes(e) + n = Integer.from_bytes(n) + return construct([n, e]) + if len(extern_key) > 0 and bord(extern_key[0]) == 0x30: # This is probably a DER encoded key return _import_keyDER(extern_key, passphrase) diff --git a/lib/Crypto/PublicKey/_openssh.py b/lib/Crypto/PublicKey/_openssh.py index 88dacfc27..81d1eb6af 100644 --- a/lib/Crypto/PublicKey/_openssh.py +++ b/lib/Crypto/PublicKey/_openssh.py @@ -29,6 +29,7 @@ # =================================================================== import struct +import binascii from Crypto.Cipher import AES from Crypto.Hash import SHA512 @@ -44,6 +45,13 @@ def read_int4(data): return value, data[4:] +def read_int8(data): + if len(data) < 8: + raise ValueError("Insufficient data") + value = struct.unpack(">Q", data[:8])[0] + return value, data[8:] + + def read_bytes(data): size, data = read_int4(data) if len(data) < size: @@ -62,6 +70,46 @@ def check_padding(pad): raise ValueError("Incorrect padding") +def import_openssh_public_cert_generic(data): + # https://cvsweb.openbsd.org/src/PROTOCOL.certkeys + + parts = data.split(b' ') + if len(parts) not in (2, 3): + raise ValueError("Not an openssh public certificate") + + cert_type = parts[0] + if not cert_type.endswith(b"-cert-v01@openssh.com"): + raise ValueError("Not an openssh public certificate") + + try: + cert = binascii.a2b_base64(parts[1]) + except binascii.Error as exc: + raise ValueError("Invalid OpenSSH certificate encoding") from exc + + inner_type, cert = read_bytes(cert) + if inner_type != cert_type: + raise ValueError("Mismatch in openssh public certificate") + + _, cert = read_bytes(cert) # nonce + return cert_type, cert + + +def check_openssh_public_cert_footer(data): + _, data = read_int8(data) # serial + _, data = read_int4(data) # type + _, data = read_bytes(data) # key id + _, data = read_bytes(data) # valid principals + _, data = read_int8(data) # valid after + _, data = read_int8(data) # valid before + _, data = read_bytes(data) # critical options + _, data = read_bytes(data) # extensions + _, data = read_bytes(data) # reserved + _, data = read_bytes(data) # signature key + _, data = read_bytes(data) # signature + if data: + raise ValueError("Too much data") + + def import_openssh_private_generic(data, password): # https://cvsweb.openbsd.org/cgi-bin/cvsweb/src/usr.bin/ssh/PROTOCOL.key?annotate=HEAD # https://github.com/openssh/openssh-portable/blob/master/sshkey.c diff --git a/lib/Crypto/PublicKey/_openssh.pyi b/lib/Crypto/PublicKey/_openssh.pyi index 15f3677ad..15e27ee6b 100644 --- a/lib/Crypto/PublicKey/_openssh.pyi +++ b/lib/Crypto/PublicKey/_openssh.pyi @@ -1,7 +1,10 @@ from typing import Tuple def read_int4(data: bytes) -> Tuple[int, bytes]: ... +def read_int8(data: bytes) -> Tuple[int, bytes]: ... def read_bytes(data: bytes) -> Tuple[bytes, bytes]: ... def read_string(data: bytes) -> Tuple[str, bytes]: ... def check_padding(pad: bytes) -> None: ... +def import_openssh_public_cert_generic(data: bytes) -> Tuple[bytes, bytes]: ... +def check_openssh_public_cert_footer(data: bytes) -> None: ... def import_openssh_private_generic(data: bytes, password: bytes) -> Tuple[str, bytes]: ... diff --git a/lib/Crypto/SelfTest/PublicKey/test_import_DSA.py b/lib/Crypto/SelfTest/PublicKey/test_import_DSA.py index 266b46f01..2fe04394a 100644 --- a/lib/Crypto/SelfTest/PublicKey/test_import_DSA.py +++ b/lib/Crypto/SelfTest/PublicKey/test_import_DSA.py @@ -22,6 +22,8 @@ import unittest import re +import struct +import binascii from Crypto.PublicKey import DSA from Crypto.SelfTest.st_common import * @@ -29,6 +31,55 @@ from binascii import unhexlify + +def _ssh_string(data): + return struct.pack(">I", len(data)) + data + + +def _ssh_uint32(value): + return struct.pack(">I", value) + + +def _ssh_uint64(value): + return struct.pack(">Q", value) + + +def _read_ssh_string(data): + length = struct.unpack(">I", data[:4])[0] + return data[4:4 + length], data[4 + length:] + + +def _build_openssh_certificate(public_line, outer_type=None, inner_type=None, + trailing=b""): + public_type, public_blob = public_line.split()[:2] + if outer_type is None: + outer_type = public_type + b"-cert-v01@openssh.com" + if inner_type is None: + inner_type = outer_type + + public_key = binascii.a2b_base64(public_blob) + _, public_key_fields = _read_ssh_string(public_key) + + certificate = ( + _ssh_string(inner_type) + + _ssh_string(b"nonce") + + public_key_fields + + _ssh_uint64(1) + + _ssh_uint32(1) + + _ssh_string(b"key-id") + + _ssh_string(_ssh_string(b"user")) + + _ssh_uint64(0) + + _ssh_uint64(0xffffffffffffffff) + + _ssh_string(b"") + + _ssh_string(b"") + + _ssh_string(b"") + + _ssh_string(b"ca-key") + + _ssh_string(b"signature") + + trailing + ) + return outer_type + b" " + binascii.b2a_base64(certificate)[:-1] + + class ImportKeyTests(unittest.TestCase): y = 92137165128186062214622779787483327510946462589285775188003362705875131352591574106484271700740858696583623951844732128165434284507709057439633739849986759064015013893156866539696757799934634945787496920169462601722830899660681779448742875054459716726855443681559131362852474817534616736104831095601710736729 @@ -245,6 +296,38 @@ def testImportKey7(self): self.assertEqual(self.q, key_obj.q) self.assertEqual(self.g, key_obj.g) + def testImportKey7a(self): + key_ref = DSA.importKey(self.ssh_pub) + + certificate = _build_openssh_certificate(self.ssh_pub) + key_obj = DSA.importKey(certificate) + self.assertFalse(key_obj.has_private()) + self.assertEqual(key_ref.y, key_obj.y) + self.assertEqual(key_ref.p, key_obj.p) + self.assertEqual(key_ref.q, key_obj.q) + self.assertEqual(key_ref.g, key_obj.g) + + certificate += b" comment" + key_obj = DSA.importKey(tostr(certificate)) + self.assertEqual(key_ref.y, key_obj.y) + self.assertEqual(key_ref.p, key_obj.p) + self.assertEqual(key_ref.q, key_obj.q) + self.assertEqual(key_ref.g, key_obj.g) + + def testImportKey7b(self): + self.assertRaises(ValueError, DSA.importKey, + _build_openssh_certificate( + self.ssh_pub, + inner_type=b"ssh-rsa-cert-v01@openssh.com")) + self.assertRaises(ValueError, DSA.importKey, + _build_openssh_certificate(self.ssh_pub)[:-1]) + self.assertRaises(ValueError, DSA.importKey, + b"ssh-dss-cert-v01@openssh.com !!!") + self.assertRaises(ValueError, DSA.importKey, + _build_openssh_certificate( + self.ssh_pub, + trailing=b"x")) + def testExportKey7(self): tup = (self.y, self.g, self.p, self.q) key = DSA.construct(tup) @@ -551,4 +634,3 @@ def get_tests(config={}): if __name__ == '__main__': suite = lambda: unittest.TestSuite(get_tests()) unittest.main(defaultTest='suite') - diff --git a/lib/Crypto/SelfTest/PublicKey/test_import_ECC.py b/lib/Crypto/SelfTest/PublicKey/test_import_ECC.py index 16f21622a..2cfa7e43d 100644 --- a/lib/Crypto/SelfTest/PublicKey/test_import_ECC.py +++ b/lib/Crypto/SelfTest/PublicKey/test_import_ECC.py @@ -30,6 +30,8 @@ import os import errno +import struct +import binascii import warnings import unittest from binascii import unhexlify @@ -86,6 +88,54 @@ def compact(lines): return unhexlify(tostr(ext).replace(" ", "").replace(":", "")) +def _ssh_string(data): + return struct.pack(">I", len(data)) + data + + +def _ssh_uint32(value): + return struct.pack(">I", value) + + +def _ssh_uint64(value): + return struct.pack(">Q", value) + + +def _read_ssh_string(data): + length = struct.unpack(">I", data[:4])[0] + return data[4:4 + length], data[4 + length:] + + +def _build_openssh_certificate(public_line, outer_type=None, inner_type=None, + trailing=b""): + public_type, public_blob = public_line.split()[:2] + if outer_type is None: + outer_type = public_type + b"-cert-v01@openssh.com" + if inner_type is None: + inner_type = outer_type + + public_key = binascii.a2b_base64(public_blob) + _, public_key_fields = _read_ssh_string(public_key) + + certificate = ( + _ssh_string(inner_type) + + _ssh_string(b"nonce") + + public_key_fields + + _ssh_uint64(1) + + _ssh_uint32(1) + + _ssh_string(b"key-id") + + _ssh_string(_ssh_string(b"user")) + + _ssh_uint64(0) + + _ssh_uint64(0xffffffffffffffff) + + _ssh_string(b"") + + _ssh_string(b"") + + _ssh_string(b"") + + _ssh_string(b"ca-key") + + _ssh_string(b"signature") + + trailing + ) + return outer_type + b" " + binascii.b2a_base64(certificate)[:-1] + + def create_ref_keys_p192(): key_len = 24 key_lines = load_file("ecc_p192.txt").splitlines() @@ -213,6 +263,43 @@ def test_import_private_rfc5915_only_1(self): key = _import_rfc5915_der(unhexlify(data_hex), None, "1.2.840.10045.3.1.7") self.assertEqual(key.d, 0x5c4e4320ef260f91ed9fc597aee98c8236b60e0ced692cc7a057d5e45798a052) + def test_import_openssh_public_certificate(self): + public_point = unhexlify( + "04a40ad59a2050ebe92479bd5fb16bb2e45b6465eb3cb2b1effe423" + "fabe6cb7424db8219ef0bab80acf26fd70595b61fe4760d33eed" + "80dd03d2fd0dfb27b8ce75c") + public_line = ( + b"ecdsa-sha2-nistp256 " + + binascii.b2a_base64( + _ssh_string(b"ecdsa-sha2-nistp256") + + _ssh_string(b"nistp256") + + _ssh_string(public_point))[:-1]) + public_key = ECC.construct( + curve="P-256", + point_x=bytes_to_long(public_point[1:33]), + point_y=bytes_to_long(public_point[33:])) + + certificate = _build_openssh_certificate(public_line) + self.assertEqual(public_key, ECC.import_key(certificate)) + self.assertEqual(public_key, ECC._import_openssh_public(certificate)) + + self.assertRaises(ValueError, ECC.import_key, + _build_openssh_certificate( + public_line, + inner_type=b"ssh-ed25519-cert-v01@openssh.com")) + self.assertRaises(ValueError, ECC.import_key, + _build_openssh_certificate( + public_line, + outer_type=(b"ecdsa-sha2-nistp384-cert-v01" + b"@openssh.com"))) + self.assertRaises(ValueError, ECC.import_key, + _build_openssh_certificate(public_line)[:-1]) + self.assertRaises(ValueError, ECC.import_key, + b"ecdsa-sha2-nistp256-cert-v01@openssh.com !!!") + self.assertRaises(ValueError, ECC.import_key, + _build_openssh_certificate(public_line, + trailing=b"x")) + class TestImport_P192(unittest.TestCase): def __init__(self, *args, **kwargs): @@ -574,6 +661,29 @@ def test_import_openssh_public(self): key = ECC.import_key(key_file) self.assertEqual(self.ref_public, key) + def test_import_openssh_public_certificate(self): + key_file = load_file("ecc_p256_public_openssh.txt") + certificate = _build_openssh_certificate(key_file) + + key = ECC._import_openssh_public(certificate) + self.assertEqual(self.ref_public, key) + + key = ECC.import_key(certificate + b" comment") + self.assertEqual(self.ref_public, key) + + def test_import_openssh_public_certificate_error(self): + key_file = load_file("ecc_p256_public_openssh.txt") + self.assertRaises(ValueError, ECC.import_key, + _build_openssh_certificate( + key_file, + inner_type=b"ssh-ed25519-cert-v01@openssh.com")) + self.assertRaises(ValueError, ECC.import_key, + _build_openssh_certificate(key_file)[:-1]) + self.assertRaises(ValueError, ECC.import_key, + b"ecdsa-sha2-nistp256-cert-v01@openssh.com !!!") + self.assertRaises(ValueError, ECC.import_key, + _build_openssh_certificate(key_file, trailing=b"x")) + def test_import_openssh_private_clear(self): key_file = load_file("ecc_p256_private_openssh.pem") key_file_old = load_file("ecc_p256_private_openssh_old.pem") @@ -718,6 +828,13 @@ def test_import_openssh_public(self): key = ECC.import_key(key_file) self.assertEqual(self.ref_public, key) + def test_import_openssh_public_certificate(self): + key_file = load_file("ecc_p384_public_openssh.txt") + certificate = _build_openssh_certificate(key_file) + + key = ECC.import_key(certificate) + self.assertEqual(self.ref_public, key) + def test_import_openssh_private_clear(self): key_file = load_file("ecc_p384_private_openssh.pem") key_file_old = load_file("ecc_p384_private_openssh_old.pem") @@ -862,6 +979,13 @@ def test_import_openssh_public(self): key = ECC.import_key(key_file) self.assertEqual(self.ref_public, key) + def test_import_openssh_public_certificate(self): + key_file = load_file("ecc_p521_public_openssh.txt") + certificate = _build_openssh_certificate(key_file) + + key = ECC.import_key(certificate) + self.assertEqual(self.ref_public, key) + def test_import_openssh_private_clear(self): key_file = load_file("ecc_p521_private_openssh.pem") key_file_old = load_file("ecc_p521_private_openssh_old.pem") @@ -2374,6 +2498,17 @@ def test_import_openssh_public(self): key = ECC.import_key(key_file) self.assertFalse(key.has_private()) + def test_import_openssh_public_certificate(self): + key_file = load_file("ecc_ed25519_public_openssh.txt") + key_ref = ECC.import_key(key_file) + certificate = _build_openssh_certificate(key_file) + + key = ECC._import_openssh_public(certificate) + self.assertEqual(key_ref, key) + + key = ECC.import_key(certificate + b" comment") + self.assertEqual(key_ref, key) + def test_import_openssh_private_clear(self): key_file = load_file("ecc_ed25519_private_openssh.pem") key = ECC.import_key(key_file) diff --git a/lib/Crypto/SelfTest/PublicKey/test_import_RSA.py b/lib/Crypto/SelfTest/PublicKey/test_import_RSA.py index f6f7afd45..29878d873 100644 --- a/lib/Crypto/SelfTest/PublicKey/test_import_RSA.py +++ b/lib/Crypto/SelfTest/PublicKey/test_import_RSA.py @@ -23,6 +23,8 @@ import os import re import errno +import struct +import binascii import warnings import unittest from unittest import SkipTest @@ -76,6 +78,54 @@ def der2pem(der, text='PUBLIC'): return pem +def _ssh_string(data): + return struct.pack(">I", len(data)) + data + + +def _ssh_uint32(value): + return struct.pack(">I", value) + + +def _ssh_uint64(value): + return struct.pack(">Q", value) + + +def _read_ssh_string(data): + length = struct.unpack(">I", data[:4])[0] + return data[4:4 + length], data[4 + length:] + + +def _build_openssh_certificate(public_line, outer_type=None, inner_type=None, + trailing=b""): + public_type, public_blob = public_line.split()[:2] + if outer_type is None: + outer_type = public_type + b"-cert-v01@openssh.com" + if inner_type is None: + inner_type = outer_type + + public_key = binascii.a2b_base64(public_blob) + _, public_key_fields = _read_ssh_string(public_key) + + certificate = ( + _ssh_string(inner_type) + + _ssh_string(b"nonce") + + public_key_fields + + _ssh_uint64(1) + + _ssh_uint32(1) + + _ssh_string(b"key-id") + + _ssh_string(_ssh_string(b"user")) + + _ssh_uint64(0) + + _ssh_uint64(0xffffffffffffffff) + + _ssh_string(b"") + + _ssh_string(b"") + + _ssh_string(b"") + + _ssh_string(b"ca-key") + + _ssh_string(b"signature") + + trailing + ) + return outer_type + b" " + binascii.b2a_base64(certificate)[:-1] + + class ImportKeyTests(unittest.TestCase): # 512-bit RSA key generated with openssl rsaKeyPEM = u'''-----BEGIN RSA PRIVATE KEY----- @@ -259,6 +309,34 @@ def testImportKey7(self): self.assertEqual(key.n, self.n) self.assertEqual(key.e, self.e) + def testImportKey7a(self): + """Verify import of OpenSSH public certificate""" + key_ref = RSA.importKey(self.rsaPublicKeyOpenSSH) + + certificate = _build_openssh_certificate(self.rsaPublicKeyOpenSSH) + key = RSA.importKey(certificate) + self.assertEqual(key_ref, key) + + certificate += b" comment" + key = RSA.importKey(tostr(certificate)) + self.assertEqual(key_ref, key) + + def testImportKey7b(self): + """Reject malformed OpenSSH public certificates""" + self.assertRaises(ValueError, RSA.importKey, + _build_openssh_certificate( + self.rsaPublicKeyOpenSSH, + inner_type=b"ssh-dss-cert-v01@openssh.com")) + self.assertRaises(ValueError, RSA.importKey, + _build_openssh_certificate( + self.rsaPublicKeyOpenSSH)[:-1]) + self.assertRaises(ValueError, RSA.importKey, + b"ssh-rsa-cert-v01@openssh.com !!!") + self.assertRaises(ValueError, RSA.importKey, + _build_openssh_certificate( + self.rsaPublicKeyOpenSSH, + trailing=b"x")) + def testImportKey8(self): """Verify import of encrypted PrivateKeyInfo DER SEQUENCE""" for t in self.rsaKeyEncryptedPEM: