Add support for post-quantum ML-KEM key exchange algorithms

This commit adds support for the post-quantum kex algorithms
mlkem768nistp256-sha256, mlkem1024nistp384-sha384, and
mlkem768x25519-sha256. The latter is now also supported in
OpenSSH 9.9, and interoperability has been tested against that.

This commit also makes sntrup761x25519-sha512 available without the
"@openssh.com" suffix, now that OpenSSH 9.9 supports both names.

Author: ronf

.github/workflows/run_tests.yml

@@ -38,7 +38,7 @@ jobs:
 
     runs-on: ${{ matrix.os }}
     env:
-      liboqs_version: '0.7.2'
+      liboqs_version: '0.10.1'
       nettle_version: nettle_3.8.1_release_20220727
 
     steps:

README.rst

@@ -57,8 +57,7 @@ Features
 * Byte and string based I/O with settable encoding
 * A variety of `key exchange`__, `encryption`__, and `MAC`__ algorithms
 
-  * Including OpenSSH post-quantum kex algorithm
-    sntrup761x25519-sha512\@openssh.com
+  * Including post-quantum kex algorithms ML-KEM and SNTRUP
 
 * Support for `gzip compression`__
 
@@ -159,7 +158,7 @@ functionality:
 
 * Install liboqs from https://github.com/open-quantum-safe/liboqs
   if you want support for the OpenSSH post-quantum key exchange
-  algorithm sntrup761x25519-sha512\@openssh.com.
+  algorithms based on ML-KEM and SNTRUP.
 
 * Install libsodium from https://github.com/jedisct1/libsodium
   and libnacl from https://pypi.python.org/pypi/libnacl if you have

asyncssh/crypto/__init__.py

@@ -40,9 +40,7 @@
 
 from .rsa import RSAPrivateKey, RSAPublicKey
 
-from .sntrup import sntrup761_available
-from .sntrup import sntrup761_pubkey_bytes, sntrup761_ciphertext_bytes
-from .sntrup import sntrup761_keypair, sntrup761_encaps, sntrup761_decaps
+from .pq import mlkem_available, sntrup_available, PQDH
 
 # Import chacha20-poly1305 cipher if available
 from .chacha import ChachaCipher, chacha_available

asyncssh/crypto/ec.py

@@ -194,12 +194,15 @@ def get_public(self) -> bytes:
         return pub_key.public_bytes(Encoding.X962,
                                     PublicFormat.UncompressedPoint)
 
-    def get_shared(self, peer_public: bytes) -> int:
-        """Return the shared key from the peer's public key"""
+    def get_shared_bytes(self, peer_public: bytes) -> bytes:
+        """Return the shared key from the peer's public key as bytes"""
 
         peer_key = ec.EllipticCurvePublicKey.from_encoded_point(
             self._priv_key.curve, peer_public)
 
-        shared_key = self._priv_key.exchange(ec.ECDH(), peer_key)
+        return self._priv_key.exchange(ec.ECDH(), peer_key)
+
+    def get_shared(self, peer_public: bytes) -> int:
+        """Return the shared key from the peer's public key"""
 
-        return int.from_bytes(shared_key, 'big')
+        return int.from_bytes(self.get_shared_bytes(peer_public), 'big')

asyncssh/crypto/ed.py

@@ -245,13 +245,13 @@ def get_public(self) -> bytes:
                                                             PublicFormat.Raw)
 
         def get_shared_bytes(self, peer_public: bytes) -> bytes:
-            """Return the shared key from the peer's public key"""
+            """Return the shared key from the peer's public key as bytes"""
 
             peer_key = x25519.X25519PublicKey.from_public_bytes(peer_public)
             return self._priv_key.exchange(peer_key)
 
         def get_shared(self, peer_public: bytes) -> int:
-            """Return the shared key from the peer's public key as bytes"""
+            """Return the shared key from the peer's public key"""
 
             return int.from_bytes(self.get_shared_bytes(peer_public), 'big')
 else: # pragma: no cover

asyncssh/crypto/pq.py

@@ -0,0 +1,103 @@
+# Copyright (c) 2022-2024 by Ron Frederick <ronf@timeheart.net> and others.
+#
+# This program and the accompanying materials are made available under
+# the terms of the Eclipse Public License v2.0 which accompanies this
+# distribution and is available at:
+#
+#     http://www.eclipse.org/legal/epl-2.0/
+#
+# This program may also be made available under the following secondary
+# licenses when the conditions for such availability set forth in the
+# Eclipse Public License v2.0 are satisfied:
+#
+#    GNU General Public License, Version 2.0, or any later versions of
+#    that license
+#
+# SPDX-License-Identifier: EPL-2.0 OR GPL-2.0-or-later
+#
+# Contributors:
+#     Ron Frederick - initial implementation, API, and documentation
+
+"""A shim around liboqs for Streamlined NTRU Prime post-quantum encryption"""
+
+import ctypes
+import ctypes.util
+from typing import Mapping, Tuple
+
+
+_pq_algs: Mapping[bytes, Tuple[int, int, int, int, str]] = {
+    b'mlkem768':  (1184, 2400, 1088, 32, 'KEM_ml_kem_768'),
+    b'mlkem1024': (1568, 3168, 1568, 32, 'KEM_ml_kem_1024'),
+    b'sntrup761': (1158, 1763, 1039, 32, 'KEM_ntruprime_sntrup761')
+}
+
+mlkem_available = False
+sntrup_available = False
+
+for lib in ('oqs', 'liboqs'):
+    _oqs_lib = ctypes.util.find_library(lib)
+
+    if _oqs_lib: # pragma: no branch
+        break
+else: # pragma: no cover
+    _oqs_lib = None
+
+if _oqs_lib: # pragma: no branch
+    _oqs = ctypes.cdll.LoadLibrary(_oqs_lib)
+
+    mlkem_available = (hasattr(_oqs, 'OQS_KEM_ml_kem_768_keypair') or
+                       hasattr(_oqs, 'OQS_KEM_ml_kem_768_ipd_keypair'))
+    sntrup_available = hasattr(_oqs, 'OQS_KEM_ntruprime_sntrup761_keypair')
+
+
+class PQDH:
+    """A shim around liboqs for post-quantum key exchange algorithms"""
+
+    def __init__(self, alg_name: bytes):
+        try:
+            self.pubkey_bytes, self.privkey_bytes, \
+            self.ciphertext_bytes, self.secret_bytes, \
+            oqs_name = _pq_algs[alg_name]
+        except KeyError: # pragma: no cover, other algs not registered
+            raise ValueError('Unknown PQ algorithm %s' % oqs_name) from None
+
+        if not hasattr(_oqs, 'OQS_' + oqs_name + '_keypair'): # pragma: no cover
+            oqs_name += '_ipd'
+
+        self._keypair = getattr(_oqs, 'OQS_' + oqs_name + '_keypair')
+        self._encaps = getattr(_oqs, 'OQS_' + oqs_name + '_encaps')
+        self._decaps = getattr(_oqs, 'OQS_' + oqs_name + '_decaps')
+
+    def keypair(self) -> Tuple[bytes, bytes]:
+        """Make a new key pair"""
+
+        pubkey = ctypes.create_string_buffer(self.pubkey_bytes)
+        privkey = ctypes.create_string_buffer(self.privkey_bytes)
+        self._keypair(pubkey, privkey)
+
+        return pubkey.raw, privkey.raw
+
+    def encaps(self, pubkey: bytes) -> Tuple[bytes, bytes]:
+        """Generate a random secret and encrypt it with a public key"""
+
+        if len(pubkey) != self.pubkey_bytes:
+            raise ValueError('Invalid public key')
+
+        ciphertext = ctypes.create_string_buffer(self.ciphertext_bytes)
+        secret = ctypes.create_string_buffer(self.secret_bytes)
+
+        self._encaps(ciphertext, secret, pubkey)
+
+        return secret.raw, ciphertext.raw
+
+    def decaps(self, ciphertext: bytes, privkey: bytes) -> bytes:
+        """Decrypt an encrypted secret using a private key"""
+
+        if len(ciphertext) != self.ciphertext_bytes:
+            raise ValueError('Invalid ciphertext')
+
+        secret = ctypes.create_string_buffer(self.secret_bytes)
+
+        self._decaps(secret, ciphertext, privkey)
+
+        return secret.raw

asyncssh/crypto/sntrup.py

@@ -25,11 +25,9 @@
 from typing_extensions import Protocol
 
 from .constants import DEFAULT_LANG
+from .crypto import Curve25519DH, Curve448DH, DH, ECDH, PQDH
 from .crypto import curve25519_available, curve448_available
-from .crypto import Curve25519DH, Curve448DH, DH, ECDH
-from .crypto import sntrup761_available
-from .crypto import sntrup761_pubkey_bytes, sntrup761_ciphertext_bytes
-from .crypto import sntrup761_keypair, sntrup761_encaps, sntrup761_decaps
+from .crypto import mlkem_available, sntrup_available
 from .gss import GSSError
 from .kex import Kex, register_kex_alg, register_gss_kex_alg
 from .misc import HashType, KeyExchangeFailed, ProtocolError
@@ -50,6 +48,9 @@ class DHKey(Protocol):
     def get_public(self) -> bytes:
         """Return the public key to send to the peer"""
 
+    def get_shared_bytes(self, peer_public: bytes) -> bytes:
+        """Return the shared key from the peer's public key in bytes"""
+
     def get_shared(self, peer_public: bytes) -> int:
         """Return the shared key from the peer's public key"""
 
@@ -467,58 +468,56 @@ async def start(self) -> None:
     }
 
 
-class _KexSNTRUP761(_KexECDH):
-    """Handler for Streamlined NTRU Prime post-quantum key exchange"""
+class _KexHybridECDH(_KexECDH):
+    """Handler for post-quantum key exchange"""
 
     def __init__(self, alg: bytes, conn: 'SSHConnection', hash_alg: HashType,
-                 *args: object):
-        super().__init__(alg, conn, hash_alg, Curve25519DH)
+                 pq_alg_name: bytes, ecdh_class: _ECDHClass, *args: object):
+        super().__init__(alg, conn, hash_alg, ecdh_class, *args)
+
+        self._pq = PQDH(pq_alg_name)
 
         if conn.is_client():
-            sntrup_pub, self._sntrup_priv = sntrup761_keypair()
-            self._client_pub = sntrup_pub + self._client_pub
+            pq_pub, self._pq_priv = self._pq.keypair()
+            self._client_pub = pq_pub + self._client_pub
 
     def _compute_client_shared(self) -> bytes:
         """Compute client shared key"""
 
-        ciphertext = self._server_pub[:sntrup761_ciphertext_bytes]
-        curve25519_pub = self._server_pub[sntrup761_ciphertext_bytes:]
+        pq_ciphertext = self._server_pub[:self._pq.ciphertext_bytes]
+        ec_pub = self._server_pub[self._pq.ciphertext_bytes:]
 
         try:
-            sntrup_secret = sntrup761_decaps(ciphertext, self._sntrup_priv)
+            pq_secret = self._pq.decaps(pq_ciphertext, self._pq_priv)
         except ValueError:
-            raise ProtocolError('Invalid SNTRUP server ciphertext') from None
+            raise ProtocolError('Invalid PQ server ciphertext') from None
 
         try:
-            priv = cast(Curve25519DH, self._priv)
-            curve25519_shared = priv.get_shared_bytes(curve25519_pub)
+            ec_shared = self._priv.get_shared_bytes(ec_pub)
         except ValueError:
             raise ProtocolError('Invalid ECDH server public key') from None
 
-        return String(self._hash_alg(sntrup_secret +
-                                     curve25519_shared).digest())
+        return String(self._hash_alg(pq_secret + ec_shared).digest())
 
     def _compute_server_shared(self) -> bytes:
         """Compute server shared key"""
 
-        sntrup_pub = self._client_pub[:sntrup761_pubkey_bytes]
-        curve25519_pub = self._client_pub[sntrup761_pubkey_bytes:]
+        pq_pub = self._client_pub[:self._pq.pubkey_bytes]
+        ec_pub = self._client_pub[self._pq.pubkey_bytes:]
 
         try:
-            sntrup_secret, ciphertext = sntrup761_encaps(sntrup_pub)
+            pq_secret, pq_ciphertext = self._pq.encaps(pq_pub)
         except ValueError:
-            raise ProtocolError('Invalid SNTRUP client public key') from None
+            raise ProtocolError('Invalid PQ client public key') from None
 
         try:
-            priv = cast(Curve25519DH, self._priv)
-            curve25519_shared = priv.get_shared_bytes(curve25519_pub)
+            ec_shared = self._priv.get_shared_bytes(ec_pub)
         except ValueError:
             raise ProtocolError('Invalid ECDH client public key') from None
 
-        self._server_pub = ciphertext + self._server_pub
+        self._server_pub = pq_ciphertext + self._server_pub
 
-        return String(self._hash_alg(sntrup_secret +
-                                     curve25519_shared).digest())
+        return String(self._hash_alg(pq_secret + ec_shared).digest())
 
 
 class _KexGSSBase(_KexDHBase):
@@ -737,10 +736,22 @@ class _KexGSSECDH(_KexGSSBase, _KexECDH):
     }
 
 
+if mlkem_available: # pragma: no branch
+    if curve25519_available: # pragma: no branch
+        register_kex_alg(b'mlkem768x25519-sha256', _KexHybridECDH,
+                         sha256, (b'mlkem768', Curve25519DH), True)
+
+    register_kex_alg(b'mlkem768nistp256-sha256', _KexHybridECDH,
+                     sha256, (b'mlkem768', ECDH, b'nistp256'), True)
+    register_kex_alg(b'mlkem1024nistp384-sha384', _KexHybridECDH,
+                     sha384, (b'mlkem1024', ECDH, b'nistp384'), True)
+
 if curve25519_available: # pragma: no branch
-    if sntrup761_available: # pragma: no branch
-        register_kex_alg(b'sntrup761x25519-sha512@openssh.com', _KexSNTRUP761,
-                         sha512, (), True)
+    if sntrup_available: # pragma: no branch
+        register_kex_alg(b'sntrup761x25519-sha512', _KexHybridECDH,
+                         sha512, (b'sntrup761', Curve25519DH), True)
+        register_kex_alg(b'sntrup761x25519-sha512@openssh.com', _KexHybridECDH,
+                         sha512, (b'sntrup761', Curve25519DH), True)
 
     register_kex_alg(b'curve25519-sha256', _KexECDH, sha256,
                      (Curve25519DH,), True)