Support importing OpenSSH public certificates

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)

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:
 

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)

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

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]: ...

lib/Crypto/SelfTest/PublicKey/test_import_DSA.py

@@ -22,13 +22,64 @@
 
 import unittest
 import re
+import struct
+import binascii
 
 from Crypto.PublicKey import DSA
 from Crypto.SelfTest.st_common import *
 from Crypto.Util.py3compat import *
 
 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')
-