Add support for RSA SHA2 host and public keys

Includes a handful of refactors and new semiprivate
attributes on Transport and AuthHandler for better
test visibility.

19 files changed, 631 insertions, 75 deletions

diff --git a/paramiko/__init__.py b/paramiko/__init__.py
index 8642f84a..5318cc9c 100644
--- a/paramiko/__init__.py
+++ b/paramiko/__init__.py
@@ -42,6 +42,7 @@ from paramiko.ssh_exception import (
     ChannelException,
     ConfigParseError,
     CouldNotCanonicalize,
+    IncompatiblePeer,
     PasswordRequiredException,
     ProxyCommandFailure,
     SSHException,
diff --git a/paramiko/agent.py b/paramiko/agent.py
index c7c8b7cb..3a02c06c 100644
--- a/paramiko/agent.py
+++ b/paramiko/agent.py
@@ -411,7 +411,7 @@ class AgentKey(PKey):
     def _fields(self):
         raise NotImplementedError
 
-    def sign_ssh_data(self, data):
+    def sign_ssh_data(self, data, algorithm=None):
         msg = Message()
         msg.add_byte(cSSH2_AGENTC_SIGN_REQUEST)
         msg.add_string(self.blob)
diff --git a/paramiko/auth_handler.py b/paramiko/auth_handler.py
index 011e57f3..845b9143 100644
--- a/paramiko/auth_handler.py
+++ b/paramiko/auth_handler.py
@@ -61,7 +61,7 @@ from paramiko.common import (
     cMSG_USERAUTH_BANNER,
 )
 from paramiko.message import Message
-from paramiko.py3compat import b
+from paramiko.py3compat import b, u
 from paramiko.ssh_exception import (
     SSHException,
     AuthenticationException,
@@ -206,6 +206,23 @@ class AuthHandler(object):
         self.transport._send_message(m)
         self.transport.close()
 
+    def _get_algorithm_and_bits(self, key):
+        """
+        Given any key, return appropriate signing algorithm & bits-to-sign.
+
+        Intended for input to or verification of, key signatures.
+        """
+        key_type, bits = None, None
+        # Use certificate contents, if available, plain pubkey otherwise
+        if key.public_blob:
+            key_type = key.public_blob.key_type
+            bits = key.public_blob.key_blob
+        else:
+            key_type = key.get_name()
+            bits = key
+        algorithm = self._finalize_pubkey_algorithm(key_type)
+        return algorithm, bits
+
     def _get_session_blob(self, key, service, username):
         m = Message()
         m.add_string(self.transport.session_id)
@@ -214,13 +231,9 @@ class AuthHandler(object):
         m.add_string(service)
         m.add_string("publickey")
         m.add_boolean(True)
-        # Use certificate contents, if available, plain pubkey otherwise
-        if key.public_blob:
-            m.add_string(key.public_blob.key_type)
-            m.add_string(key.public_blob.key_blob)
-        else:
-            m.add_string(key.get_name())
-            m.add_string(key)
+        algorithm, bits = self._get_algorithm_and_bits(key)
+        m.add_string(algorithm)
+        m.add_string(bits)
         return m.asbytes()
 
     def wait_for_response(self, event):
@@ -269,6 +282,39 @@ class AuthHandler(object):
         # dunno this one
         self._disconnect_service_not_available()
 
+    def _finalize_pubkey_algorithm(self, key_type):
+        # Short-circuit for non-RSA keys
+        if "rsa" not in key_type:
+            return key_type
+        self._log(
+            DEBUG,
+            "Finalizing pubkey algorithm for key of type {!r}".format(
+                key_type
+            ),
+        )
+        # Only consider RSA algos from our list, lest we agree on another!
+        my_algos = [x for x in self.transport.preferred_pubkeys if "rsa" in x]
+        self._log(DEBUG, "Our pubkey algorithm list: {}".format(my_algos))
+        # Short-circuit negatively if user disabled all RSA algos (heh)
+        if not my_algos:
+            raise SSHException(
+                "An RSA key was specified, but no RSA pubkey algorithms are configured!"  # noqa
+            )
+        # Check for server-sig-algs if supported & sent
+        server_algos = u(
+            self.transport.server_extensions.get("server-sig-algs", b(""))
+        ).split(",")
+        self._log(DEBUG, "Server-side algorithm list: {}".format(server_algos))
+        # Only use algos from our list that the server likes, in our own
+        # preference order. (NOTE: purposefully using same style as in
+        # Transport...expect to refactor later)
+        agreement = list(filter(server_algos.__contains__, my_algos))
+        # Fallback: first one in our (possibly tweaked by caller) list
+        final = agreement[0] if agreement else my_algos[0]
+        self.transport._agreed_pubkey_algorithm = final
+        self._log(DEBUG, "Agreed upon {!r} pubkey algorithm".format(final))
+        return final
+
     def _parse_service_accept(self, m):
         service = m.get_text()
         if service == "ssh-userauth":
@@ -287,18 +333,15 @@ class AuthHandler(object):
                 m.add_string(password)
             elif self.auth_method == "publickey":
                 m.add_boolean(True)
-                # Use certificate contents, if available, plain pubkey
-                # otherwise
-                if self.private_key.public_blob:
-                    m.add_string(self.private_key.public_blob.key_type)
-                    m.add_string(self.private_key.public_blob.key_blob)
-                else:
-                    m.add_string(self.private_key.get_name())
-                    m.add_string(self.private_key)
+                algorithm, bits = self._get_algorithm_and_bits(
+                    self.private_key
+                )
+                m.add_string(algorithm)
+                m.add_string(bits)
                 blob = self._get_session_blob(
                     self.private_key, "ssh-connection", self.username
                 )
-                sig = self.private_key.sign_ssh_data(blob)
+                sig = self.private_key.sign_ssh_data(blob, algorithm)
                 m.add_string(sig)
             elif self.auth_method == "keyboard-interactive":
                 m.add_string("")
@@ -529,13 +572,15 @@ Error Message: {}
                 username, key
             )
             if result != AUTH_FAILED:
+                sig_algo = self._finalize_pubkey_algorithm(keytype)
                 # key is okay, verify it
                 if not sig_attached:
                     # client wants to know if this key is acceptable, before it
                     # signs anything...  send special "ok" message
                     m = Message()
                     m.add_byte(cMSG_USERAUTH_PK_OK)
-                    m.add_string(keytype)
+                    # TODO: suspect we're not testing this
+                    m.add_string(sig_algo)
                     m.add_string(keyblob)
                     self.transport._send_message(m)
                     return
diff --git a/paramiko/common.py b/paramiko/common.py
index 7bd0cb10..55dd4bdf 100644
--- a/paramiko/common.py
+++ b/paramiko/common.py
@@ -29,7 +29,8 @@ from paramiko.py3compat import byte_chr, PY2, long, b
     MSG_DEBUG,
     MSG_SERVICE_REQUEST,
     MSG_SERVICE_ACCEPT,
-) = range(1, 7)
+    MSG_EXT_INFO,
+) = range(1, 8)
 (MSG_KEXINIT, MSG_NEWKEYS) = range(20, 22)
 (
     MSG_USERAUTH_REQUEST,
@@ -68,6 +69,7 @@ cMSG_UNIMPLEMENTED = byte_chr(MSG_UNIMPLEMENTED)
 cMSG_DEBUG = byte_chr(MSG_DEBUG)
 cMSG_SERVICE_REQUEST = byte_chr(MSG_SERVICE_REQUEST)
 cMSG_SERVICE_ACCEPT = byte_chr(MSG_SERVICE_ACCEPT)
+cMSG_EXT_INFO = byte_chr(MSG_EXT_INFO)
 cMSG_KEXINIT = byte_chr(MSG_KEXINIT)
 cMSG_NEWKEYS = byte_chr(MSG_NEWKEYS)
 cMSG_USERAUTH_REQUEST = byte_chr(MSG_USERAUTH_REQUEST)
@@ -109,6 +111,7 @@ MSG_NAMES = {
     MSG_SERVICE_REQUEST: "service-request",
     MSG_SERVICE_ACCEPT: "service-accept",
     MSG_KEXINIT: "kexinit",
+    MSG_EXT_INFO: "ext-info",
     MSG_NEWKEYS: "newkeys",
     30: "kex30",
     31: "kex31",
diff --git a/paramiko/dsskey.py b/paramiko/dsskey.py
index 09d6f648..1a0c4797 100644
--- a/paramiko/dsskey.py
+++ b/paramiko/dsskey.py
@@ -105,7 +105,7 @@ class DSSKey(PKey):
     def can_sign(self):
         return self.x is not None
 
-    def sign_ssh_data(self, data):
+    def sign_ssh_data(self, data, algorithm=None):
         key = dsa.DSAPrivateNumbers(
             x=self.x,
             public_numbers=dsa.DSAPublicNumbers(
diff --git a/paramiko/ecdsakey.py b/paramiko/ecdsakey.py
index b609d130..c4e2b1af 100644
--- a/paramiko/ecdsakey.py
+++ b/paramiko/ecdsakey.py
@@ -211,7 +211,7 @@ class ECDSAKey(PKey):
     def can_sign(self):
         return self.signing_key is not None
 
-    def sign_ssh_data(self, data):
+    def sign_ssh_data(self, data, algorithm=None):
         ecdsa = ec.ECDSA(self.ecdsa_curve.hash_object())
         sig = self.signing_key.sign(data, ecdsa)
         r, s = decode_dss_signature(sig)
diff --git a/paramiko/ed25519key.py b/paramiko/ed25519key.py
index 7b19e352..d322a0c1 100644
--- a/paramiko/ed25519key.py
+++ b/paramiko/ed25519key.py
@@ -191,7 +191,7 @@ class Ed25519Key(PKey):
     def can_sign(self):
         return self._signing_key is not None
 
-    def sign_ssh_data(self, data):
+    def sign_ssh_data(self, data, algorithm=None):
         m = Message()
         m.add_string("ssh-ed25519")
         m.add_string(self._signing_key.sign(data).signature)
diff --git a/paramiko/kex_curve25519.py b/paramiko/kex_curve25519.py
index 59710c1a..3420fb4f 100644
--- a/paramiko/kex_curve25519.py
+++ b/paramiko/kex_curve25519.py
@@ -89,7 +89,9 @@ class KexCurve25519(object):
         hm.add_mpint(K)
         H = self.hash_algo(hm.asbytes()).digest()
         self.transport._set_K_H(K, H)
-        sig = self.transport.get_server_key().sign_ssh_data(H)
+        sig = self.transport.get_server_key().sign_ssh_data(
+            H, self.transport.host_key_type
+        )
         # construct reply
         m = Message()
         m.add_byte(c_MSG_KEXECDH_REPLY)
diff --git a/paramiko/kex_ecdh_nist.py b/paramiko/kex_ecdh_nist.py
index ad5c9c79..19de2431 100644
--- a/paramiko/kex_ecdh_nist.py
+++ b/paramiko/kex_ecdh_nist.py
@@ -90,7 +90,9 @@ class KexNistp256:
         hm.add_mpint(long(K))
         H = self.hash_algo(hm.asbytes()).digest()
         self.transport._set_K_H(K, H)
-        sig = self.transport.get_server_key().sign_ssh_data(H)
+        sig = self.transport.get_server_key().sign_ssh_data(
+            H, self.transport.host_key_type
+        )
         # construct reply
         m = Message()
         m.add_byte(c_MSG_KEXECDH_REPLY)
diff --git a/paramiko/kex_gex.py b/paramiko/kex_gex.py
index fb8f01fd..ab462e6d 100644
--- a/paramiko/kex_gex.py
+++ b/paramiko/kex_gex.py
@@ -240,7 +240,9 @@ class KexGex(object):
         H = self.hash_algo(hm.asbytes()).digest()
         self.transport._set_K_H(K, H)
         # sign it
-        sig = self.transport.get_server_key().sign_ssh_data(H)
+        sig = self.transport.get_server_key().sign_ssh_data(
+            H, self.transport.host_key_type
+        )
         # send reply
         m = Message()
         m.add_byte(c_MSG_KEXDH_GEX_REPLY)
diff --git a/paramiko/kex_group1.py b/paramiko/kex_group1.py
index dce3fd91..6d548b01 100644
--- a/paramiko/kex_group1.py
+++ b/paramiko/kex_group1.py
@@ -143,7 +143,9 @@ class KexGroup1(object):
         H = self.hash_algo(hm.asbytes()).digest()
         self.transport._set_K_H(K, H)
         # sign it
-        sig = self.transport.get_server_key().sign_ssh_data(H)
+        sig = self.transport.get_server_key().sign_ssh_data(
+            H, self.transport.host_key_type
+        )
         # send reply
         m = Message()
         m.add_byte(c_MSG_KEXDH_REPLY)
diff --git a/paramiko/pkey.py b/paramiko/pkey.py
index 5bdfb1d4..7865a6ea 100644
--- a/paramiko/pkey.py
+++ b/paramiko/pkey.py
@@ -196,13 +196,20 @@ class PKey(object):
         """
         return u(encodebytes(self.asbytes())).replace("\n", "")
 
-    def sign_ssh_data(self, data):
+    def sign_ssh_data(self, data, algorithm=None):
         """
         Sign a blob of data with this private key, and return a `.Message`
         representing an SSH signature message.
 
-        :param str data: the data to sign.
+        :param str data:
+            the data to sign.
+        :param str algorithm:
+            the signature algorithm to use, if different from the key's
+            internal name. Default: ``None``.
         :return: an SSH signature `message <.Message>`.
+
+        .. versionchanged:: 2.9
+            Added the ``algorithm`` kwarg.
         """
         return bytes()
 
diff --git a/paramiko/rsakey.py b/paramiko/rsakey.py
index 292d0ccc..26c5313c 100644
--- a/paramiko/rsakey.py
+++ b/paramiko/rsakey.py
@@ -37,6 +37,15 @@ class RSAKey(PKey):
     data.
     """
 
+    HASHES = {
+        "ssh-rsa": hashes.SHA1,
+        "ssh-rsa-cert-v01@openssh.com": hashes.SHA1,
+        "rsa-sha2-256": hashes.SHA256,
+        "rsa-sha2-256-cert-v01@openssh.com": hashes.SHA256,
+        "rsa-sha2-512": hashes.SHA512,
+        "rsa-sha2-512-cert-v01@openssh.com": hashes.SHA512,
+    }
+
     def __init__(
         self,
         msg=None,
@@ -61,6 +70,8 @@ class RSAKey(PKey):
         else:
             self._check_type_and_load_cert(
                 msg=msg,
+                # NOTE: this does NOT change when using rsa2 signatures; it's
+                # purely about key loading, not exchange or verification
                 key_type="ssh-rsa",
                 cert_type="ssh-rsa-cert-v01@openssh.com",
             )
@@ -111,18 +122,20 @@ class RSAKey(PKey):
     def can_sign(self):
         return isinstance(self.key, rsa.RSAPrivateKey)
 
-    def sign_ssh_data(self, data):
+    def sign_ssh_data(self, data, algorithm="ssh-rsa"):
         sig = self.key.sign(
-            data, padding=padding.PKCS1v15(), algorithm=hashes.SHA1()
+            data,
+            padding=padding.PKCS1v15(),
+            algorithm=self.HASHES[algorithm](),
         )
-
         m = Message()
-        m.add_string("ssh-rsa")
+        m.add_string(algorithm)
         m.add_string(sig)
         return m
 
     def verify_ssh_sig(self, data, msg):
-        if msg.get_text() != "ssh-rsa":
+        sig_algorithm = msg.get_text()
+        if sig_algorithm not in self.HASHES:
             return False
         key = self.key
         if isinstance(key, rsa.RSAPrivateKey):
@@ -130,7 +143,10 @@ class RSAKey(PKey):
 
         try:
             key.verify(
-                msg.get_binary(), data, padding.PKCS1v15(), hashes.SHA1()
+                msg.get_binary(),
+                data,
+                padding.PKCS1v15(),
+                self.HASHES[sig_algorithm](),
             )
         except InvalidSignature:
             return False
diff --git a/paramiko/ssh_exception.py b/paramiko/ssh_exception.py
index 2789be99..39fcb10d 100644
--- a/paramiko/ssh_exception.py
+++ b/paramiko/ssh_exception.py
@@ -135,6 +135,21 @@ class BadHostKeyException(SSHException):
         )
 
 
+class IncompatiblePeer(SSHException):
+    """
+    A disagreement arose regarding an algorithm required for key exchange.
+
+    .. versionadded:: 2.9
+    """
+
+    # TODO 3.0: consider making this annotate w/ 1..N 'missing' algorithms,
+    # either just the first one that would halt kex, or even updating the
+    # Transport logic so we record /all/ that /could/ halt kex.
+    # TODO: update docstrings where this may end up raised so they are more
+    # specific.
+    pass
+
+
 class ProxyCommandFailure(SSHException):
     """
     The "ProxyCommand" found in the .ssh/config file returned an error.
diff --git a/paramiko/transport.py b/paramiko/transport.py
index d4f0b149..b99b3278 100644
--- a/paramiko/transport.py
+++ b/paramiko/transport.py
@@ -84,6 +84,8 @@ from paramiko.common import (
     HIGHEST_USERAUTH_MESSAGE_ID,
     MSG_UNIMPLEMENTED,
     MSG_NAMES,
+    MSG_EXT_INFO,
+    cMSG_EXT_INFO,
 )
 from paramiko.compress import ZlibCompressor, ZlibDecompressor
 from paramiko.dsskey import DSSKey
@@ -107,6 +109,7 @@ from paramiko.ssh_exception import (
     SSHException,
     BadAuthenticationType,
     ChannelException,
+    IncompatiblePeer,
     ProxyCommandFailure,
 )
 from paramiko.util import retry_on_signal, ClosingContextManager, clamp_value
@@ -168,11 +171,25 @@ class Transport(threading.Thread, ClosingContextManager):
         "hmac-sha1-96",
         "hmac-md5-96",
     )
+    # ~= HostKeyAlgorithms in OpenSSH land
     _preferred_keys = (
         "ssh-ed25519",
         "ecdsa-sha2-nistp256",
         "ecdsa-sha2-nistp384",
         "ecdsa-sha2-nistp521",
+        "rsa-sha2-512",
+        "rsa-sha2-256",
+        "ssh-rsa",
+        "ssh-dss",
+    )
+    # ~= PubKeyAcceptedAlgorithms
+    _preferred_pubkeys = (
+        "ssh-ed25519",
+        "ecdsa-sha2-nistp256",
+        "ecdsa-sha2-nistp384",
+        "ecdsa-sha2-nistp521",
+        "rsa-sha2-512",
+        "rsa-sha2-256",
         "ssh-rsa",
         "ssh-dss",
     )
@@ -259,8 +276,16 @@ class Transport(threading.Thread, ClosingContextManager):
     }
 
     _key_info = {
+        # TODO: at some point we will want to drop this as it's no longer
+        # considered secure due to using SHA-1 for signatures. OpenSSH 8.8 no
+        # longer supports it. Question becomes at what point do we want to
+        # prevent users with older setups from using this?
         "ssh-rsa": RSAKey,
         "ssh-rsa-cert-v01@openssh.com": RSAKey,
+        "rsa-sha2-256": RSAKey,
+        "rsa-sha2-256-cert-v01@openssh.com": RSAKey,
+        "rsa-sha2-512": RSAKey,
+        "rsa-sha2-512-cert-v01@openssh.com": RSAKey,
         "ssh-dss": DSSKey,
         "ssh-dss-cert-v01@openssh.com": DSSKey,
         "ecdsa-sha2-nistp256": ECDSAKey,
@@ -310,6 +335,7 @@ class Transport(threading.Thread, ClosingContextManager):
         gss_kex=False,
         gss_deleg_creds=True,
         disabled_algorithms=None,
+        server_sig_algs=True,
     ):
         """
         Create a new SSH session over an existing socket, or socket-like
@@ -372,6 +398,10 @@ class Transport(threading.Thread, ClosingContextManager):
             your code talks to a server which implements it differently from
             Paramiko), specify ``disabled_algorithms={"kex":
             ["diffie-hellman-group16-sha512"]}``.
+        :param bool server_sig_algs:
+            Whether to send an extra message to compatible clients, in server
+            mode, with a list of supported pubkey algorithms. Default:
+            ``True``.
 
         .. versionchanged:: 1.15
             Added the ``default_window_size`` and ``default_max_packet_size``
@@ -380,9 +410,12 @@ class Transport(threading.Thread, ClosingContextManager):
             Added the ``gss_kex`` and ``gss_deleg_creds`` kwargs.
         .. versionchanged:: 2.6
             Added the ``disabled_algorithms`` kwarg.
+        .. versionchanged:: 2.9
+            Added the ``server_sig_algs`` kwarg.
         """
         self.active = False
         self.hostname = None
+        self.server_extensions = {}
 
         if isinstance(sock, string_types):
             # convert "host:port" into (host, port)
@@ -488,6 +521,7 @@ class Transport(threading.Thread, ClosingContextManager):
         # how long (seconds) to wait for the auth response.
         self.auth_timeout = 30
         self.disabled_algorithms = disabled_algorithms or {}
+        self.server_sig_algs = server_sig_algs
 
         # server mode:
         self.server_mode = False
@@ -518,6 +552,10 @@ class Transport(threading.Thread, ClosingContextManager):
         return self._filter_algorithm("keys")
 
     @property
+    def preferred_pubkeys(self):
+        return self._filter_algorithm("pubkeys")
+
+    @property
     def preferred_kex(self):
         return self._filter_algorithm("kex")
 
@@ -743,6 +781,12 @@ class Transport(threading.Thread, ClosingContextManager):
             the host key to add, usually an `.RSAKey` or `.DSSKey`.
         """
         self.server_key_dict[key.get_name()] = key
+        # Handle SHA-2 extensions for RSA by ensuring that lookups into
+        # self.server_key_dict will yield this key for any of the algorithm
+        # names.
+        if isinstance(key, RSAKey):
+            self.server_key_dict["rsa-sha2-256"] = key
+            self.server_key_dict["rsa-sha2-512"] = key
 
     def get_server_key(self):
         """
@@ -1280,7 +1324,17 @@ class Transport(threading.Thread, ClosingContextManager):
             Added the ``gss_trust_dns`` argument.
         """
         if hostkey is not None:
-            self._preferred_keys = [hostkey.get_name()]
+            # TODO: a more robust implementation would be to ask each key class
+            # for its nameS plural, and just use that.
+            # TODO: that could be used in a bunch of other spots too
+            if isinstance(hostkey, RSAKey):
+                self._preferred_keys = [
+                    "rsa-sha2-512",
+                    "rsa-sha2-256",
+                    "ssh-rsa",
+                ]
+            else:
+                self._preferred_keys = [hostkey.get_name()]
 
         self.set_gss_host(
             gss_host=gss_host,
@@ -2126,7 +2180,12 @@ class Transport(threading.Thread, ClosingContextManager):
                             self._send_message(msg)
                     self.packetizer.complete_handshake()
             except SSHException as e:
-                self._log(ERROR, "Exception: " + str(e))
+                self._log(
+                    ERROR,
+                    "Exception ({}): {}".format(
+                        "server" if self.server_mode else "client", e
+                    ),
+                )
                 self._log(ERROR, util.tb_strings())
                 self.saved_exception = e
             except EOFError as e:
@@ -2237,7 +2296,7 @@ class Transport(threading.Thread, ClosingContextManager):
         client = segs[2]
         if version != "1.99" and version != "2.0":
             msg = "Incompatible version ({} instead of 2.0)"
-            raise SSHException(msg.format(version))
+            raise IncompatiblePeer(msg.format(version))
         msg = "Connected (version {}, client {})".format(version, client)
         self._log(INFO, msg)
 
@@ -2253,13 +2312,10 @@ class Transport(threading.Thread, ClosingContextManager):
             self.clear_to_send_lock.release()
         self.gss_kex_used = False
         self.in_kex = True
+        kex_algos = list(self.preferred_kex)
         if self.server_mode:
             mp_required_prefix = "diffie-hellman-group-exchange-sha"
-            kex_mp = [
-                k
-                for k in self.preferred_kex
-                if k.startswith(mp_required_prefix)
-            ]
+            kex_mp = [k for k in kex_algos if k.startswith(mp_required_prefix)]
             if (self._modulus_pack is None) and (len(kex_mp) > 0):
                 # can't do group-exchange if we don't have a pack of potential
                 # primes
@@ -2285,11 +2341,16 @@ class Transport(threading.Thread, ClosingContextManager):
             )
         else:
             available_server_keys = self.preferred_keys
+            # Signal support for MSG_EXT_INFO.
+            # NOTE: doing this here handily means we don't even consider this
+            # value when agreeing on real kex algo to use (which is a common
+            # pitfall when adding this apparently).
+            kex_algos.append("ext-info-c")
 
         m = Message()
         m.add_byte(cMSG_KEXINIT)
         m.add_bytes(os.urandom(16))
-        m.add_list(self.preferred_kex)
+        m.add_list(kex_algos)
         m.add_list(available_server_keys)
         m.add_list(self.preferred_ciphers)
         m.add_list(self.preferred_ciphers)
@@ -2299,29 +2360,49 @@ class Transport(threading.Thread, ClosingContextManager):
         m.add_list(self.preferred_compression)
         m.add_string(bytes())
         m.add_string(bytes())
-        # TODO: guess Robey never implemented the "guessing" part of the
-        # protocol. (Transport also never stores or acts on this flag's value
-        # in _parse_kex_init(), besides logging it to DEBUG.)
         m.add_boolean(False)
         m.add_int(0)
         # save a copy for later (needed to compute a hash)
-        self.local_kex_init = m.asbytes()
+        self.local_kex_init = self._latest_kex_init = m.asbytes()
         self._send_message(m)
 
-    def _parse_kex_init(self, m):
+    def _really_parse_kex_init(self, m, ignore_first_byte=False):
+        parsed = {}
+        if ignore_first_byte:
+            m.get_byte()
         m.get_bytes(16)  # cookie, discarded
-        kex_algo_list = m.get_list()
-        server_key_algo_list = m.get_list()
-        client_encrypt_algo_list = m.get_list()
-        server_encrypt_algo_list = m.get_list()
-        client_mac_algo_list = m.get_list()
-        server_mac_algo_list = m.get_list()
-        client_compress_algo_list = m.get_list()
-        server_compress_algo_list = m.get_list()
-        client_lang_list = m.get_list()
-        server_lang_list = m.get_list()
-        kex_follows = m.get_boolean()
+        parsed["kex_algo_list"] = m.get_list()
+        parsed["server_key_algo_list"] = m.get_list()
+        parsed["client_encrypt_algo_list"] = m.get_list()
+        parsed["server_encrypt_algo_list"] = m.get_list()
+        parsed["client_mac_algo_list"] = m.get_list()
+        parsed["server_mac_algo_list"] = m.get_list()
+        parsed["client_compress_algo_list"] = m.get_list()
+        parsed["server_compress_algo_list"] = m.get_list()
+        parsed["client_lang_list"] = m.get_list()
+        parsed["server_lang_list"] = m.get_list()
+        parsed["kex_follows"] = m.get_boolean()
         m.get_int()  # unused
+        return parsed
+
+    def _get_latest_kex_init(self):
+        return self._really_parse_kex_init(
+            Message(self._latest_kex_init), ignore_first_byte=True
+        )
+
+    def _parse_kex_init(self, m):
+        parsed = self._really_parse_kex_init(m)
+        kex_algo_list = parsed["kex_algo_list"]
+        server_key_algo_list = parsed["server_key_algo_list"]
+        client_encrypt_algo_list = parsed["client_encrypt_algo_list"]
+        server_encrypt_algo_list = parsed["server_encrypt_algo_list"]
+        client_mac_algo_list = parsed["client_mac_algo_list"]
+        server_mac_algo_list = parsed["server_mac_algo_list"]
+        client_compress_algo_list = parsed["client_compress_algo_list"]
+        server_compress_algo_list = parsed["server_compress_algo_list"]
+        client_lang_list = parsed["client_lang_list"]
+        server_lang_list = parsed["server_lang_list"]
+        kex_follows = parsed["kex_follows"]
 
         self._log(DEBUG, "=== Key exchange possibilities ===")
         for prefix, value in (
@@ -2345,6 +2426,11 @@ class Transport(threading.Thread, ClosingContextManager):
         self._log(DEBUG, "kex follows: {}".format(kex_follows))
         self._log(DEBUG, "=== Key exchange agreements ===")
 
+        # Strip out ext-info "kex algo"
+        self._remote_ext_info = None
+        if kex_algo_list[-1].startswith("ext-info-"):
+            self._remote_ext_info = kex_algo_list.pop()
+
         # as a server, we pick the first item in the client's list that we
         # support.
         # as a client, we pick the first item in our list that the server
@@ -2361,7 +2447,7 @@ class Transport(threading.Thread, ClosingContextManager):
             # TODO: do an auth-overhaul style aggregate exception here?
             # TODO: would let us streamline log output & show all failures up
             # front
-            raise SSHException(
+            raise IncompatiblePeer(
                 "Incompatible ssh peer (no acceptable kex algorithm)"
             )  # noqa
         self.kex_engine = self._kex_info[agreed_kex[0]](self)
@@ -2384,12 +2470,12 @@ class Transport(threading.Thread, ClosingContextManager):
                 filter(server_key_algo_list.__contains__, self.preferred_keys)
             )
         if len(agreed_keys) == 0:
-            raise SSHException(
+            raise IncompatiblePeer(
                 "Incompatible ssh peer (no acceptable host key)"
             )  # noqa
         self.host_key_type = agreed_keys[0]
         if self.server_mode and (self.get_server_key() is None):
-            raise SSHException(
+            raise IncompatiblePeer(
                 "Incompatible ssh peer (can't match requested host key type)"
             )  # noqa
         self._log_agreement("HostKey", agreed_keys[0], agreed_keys[0])
@@ -2421,7 +2507,7 @@ class Transport(threading.Thread, ClosingContextManager):
                 )
             )
         if len(agreed_local_ciphers) == 0 or len(agreed_remote_ciphers) == 0:
-            raise SSHException(
+            raise IncompatiblePeer(
                 "Incompatible ssh server (no acceptable ciphers)"
             )  # noqa
         self.local_cipher = agreed_local_ciphers[0]
@@ -2445,7 +2531,9 @@ class Transport(threading.Thread, ClosingContextManager):
                 filter(server_mac_algo_list.__contains__, self.preferred_macs)
             )
         if (len(agreed_local_macs) == 0) or (len(agreed_remote_macs) == 0):
-            raise SSHException("Incompatible ssh server (no acceptable macs)")
+            raise IncompatiblePeer(
+                "Incompatible ssh server (no acceptable macs)"
+            )
         self.local_mac = agreed_local_macs[0]
         self.remote_mac = agreed_remote_macs[0]
         self._log_agreement(
@@ -2484,7 +2572,7 @@ class Transport(threading.Thread, ClosingContextManager):
         ):
             msg = "Incompatible ssh server (no acceptable compression)"
             msg += " {!r} {!r} {!r}"
-            raise SSHException(
+            raise IncompatiblePeer(
                 msg.format(
                     agreed_local_compression,
                     agreed_remote_compression,
@@ -2584,6 +2672,20 @@ class Transport(threading.Thread, ClosingContextManager):
             self.packetizer.set_outbound_compressor(compress_out())
         if not self.packetizer.need_rekey():
             self.in_kex = False
+        # If client indicated extension support, send that packet immediately
+        if (
+            self.server_mode
+            and self.server_sig_algs
+            and self._remote_ext_info == "ext-info-c"
+        ):
+            extensions = {"server-sig-algs": ",".join(self.preferred_pubkeys)}
+            m = Message()
+            m.add_byte(cMSG_EXT_INFO)
+            m.add_int(len(extensions))
+            for name, value in sorted(extensions.items()):
+                m.add_string(name)
+                m.add_string(value)
+            self._send_message(m)
         # we always expect to receive NEWKEYS now
         self._expect_packet(MSG_NEWKEYS)
 
@@ -2599,6 +2701,20 @@ class Transport(threading.Thread, ClosingContextManager):
             self._log(DEBUG, "Switching on inbound compression ...")
             self.packetizer.set_inbound_compressor(compress_in())
 
+    def _parse_ext_info(self, msg):
+        # Packet is a count followed by that many key-string to possibly-bytes
+        # pairs.
+        extensions = {}
+        for _ in range(msg.get_int()):
+            name = msg.get_text()
+            value = msg.get_string()
+            extensions[name] = value
+        self._log(DEBUG, "Got EXT_INFO: {}".format(extensions))
+        # NOTE: this should work ok in cases where a server sends /two/ such
+        # messages; the RFC explicitly states a 2nd one should overwrite the
+        # 1st.
+        self.server_extensions = extensions
+
     def _parse_newkeys(self, m):
         self._log(DEBUG, "Switch to new keys ...")
         self._activate_inbound()
@@ -2866,6 +2982,7 @@ class Transport(threading.Thread, ClosingContextManager):
             self.lock.release()
 
     _handler_table = {
+        MSG_EXT_INFO: _parse_ext_info,
         MSG_NEWKEYS: _parse_newkeys,
         MSG_GLOBAL_REQUEST: _parse_global_request,
         MSG_REQUEST_SUCCESS: _parse_request_success,
diff --git a/sites/www/changelog.rst b/sites/www/changelog.rst
index c423f5a5..016a5ac9 100644
--- a/sites/www/changelog.rst
+++ b/sites/www/changelog.rst
@@ -2,6 +2,70 @@
 Changelog
 =========
 
+- :feature:`1643` Add support for SHA-2 variants of RSA key verification
+  algorithms (as described in :rfc:`8332`) as well as limited SSH extension
+  negotiation (:rfc:`8308`). How SSH servers/clients decide when and how to use
+  this functionality can be complicated; Paramiko's support is as follows:
+
+  - Client verification of server host key during key exchange will now prefer
+    ``rsa-sha2-512``, ``rsa-sha2-256``, and legacy ``ssh-rsa`` algorithms, in
+    that order, instead of just ``ssh-rsa``.
+
+      - Note that the preference order of other algorithm families such as
+        ``ed25519`` and ``ecdsa`` has not changed; for example, those two
+        groups are still preferred over RSA.
+
+  - Server mode will now offer all 3 RSA algorithms for host key verification
+    during key exchange, similar to client mode, if it has been configured with
+    an RSA host key.
+  - Client mode key exchange now sends the ``ext-info-c`` flag signaling
+    support for ``MSG_EXT_INFO``, and support for parsing the latter
+    (specifically, its ``server-sig-algs`` flag) has been added.
+  - Client mode, when performing public key authentication with an RSA key or
+    cert, will act as follows:
+
+    - In all cases, the list of algorithms to consider is based on the new
+      ``preferred_pubkeys`` list (see below) and ``disabled_algorithms``; this
+      list, like with host keys, prefers SHA2-512, SHA2-256 and SHA1, in that
+      order.
+    - When the server does not send ``server-sig-algs``, Paramiko will attempt
+      the first algorithm in the above list. Clients connecting to legacy
+      servers should thus use ``disabled_algorithms`` to turn off SHA2.
+    - When the server does send ``server-sig-algs``, the first algorithm
+      supported by both ends is used, or if there is none, it falls back to the
+      previous behavior.
+
+  - Server mode is now capable of pubkey auth involving SHA-2 signatures from
+    clients, provided one's server implementation actually provides for doing
+    so.
+
+    - This includes basic support for sending ``MSG_EXT_INFO`` (containing
+      ``server-sig-algs`` only) to clients advertising ``ext-info-c`` in their
+      key exchange list.
+
+  In order to implement the above, the following API additions were made:
+
+  - `PKey.sign_ssh_data <paramiko.pkey.PKey>`: Grew an extra, optional
+    ``algorithm`` keyword argument (defaulting to ``None`` for most subclasses,
+    and to ``"ssh-rsa"`` for `~paramiko.rsakey.RSAKey`).
+  - A new `~paramiko.ssh_exception.SSHException` subclass was added,
+    `~paramiko.ssh_exception.IncompatiblePeer`, and is raised in all spots
+    where key exchange aborts due to algorithmic incompatibility.
+
+    - Like all other exceptions in that module, it inherits from
+    ``SSHException``, and as we did not change anything else about the raising
+    (i.e. the attributes and message text are the same) this change is
+    backwards compatible.
+
+  - `~paramiko.transport.Transport` grew a ``_preferred_pubkeys`` attribute and
+    matching ``preferred_pubkeys`` property to match the other, kex-focused,
+    such members. This allows client pubkey authentication to honor the
+    ``disabled_algorithms`` feature.
+
+  Thanks to Krisztián Kovács for the report and an early stab at a patch, as
+  well as the numerous users who submitted feedback on the issue, including but
+  not limited to: Christopher Rabotin, Sam Bull, and Manfred Kaiser.
+
 - :release:`2.8.1 <2021-11-28>`
 - :bug:`985` (via :issue:`992`) Fix listdir failure when server uses a locale.
   Now on Python 2.7 `SFTPAttributes <paramiko.sftp_attr.SFTPAttributes>` will
diff --git a/tests/test_kex.py b/tests/test_kex.py
index c251611a..b73989c2 100644
--- a/tests/test_kex.py
+++ b/tests/test_kex.py
@@ -76,7 +76,7 @@ class FakeKey(object):
     def asbytes(self):
         return b"fake-key"
 
-    def sign_ssh_data(self, H):
+    def sign_ssh_data(self, H, algorithm):
         return b"fake-sig"
 
 
@@ -93,6 +93,7 @@ class FakeTransport(object):
     remote_version = "SSH-2.0-lame"
     local_kex_init = "local-kex-init"
     remote_kex_init = "remote-kex-init"
+    host_key_type = "fake-key"
 
     def _send_message(self, m):
         self._message = m
diff --git a/tests/test_pkey.py b/tests/test_pkey.py
index 94b2492b..0cc20133 100644
--- a/tests/test_pkey.py
+++ b/tests/test_pkey.py
@@ -63,6 +63,8 @@ FINGER_ECDSA_256 = "256 25:19:eb:55:e6:a1:47:ff:4f:38:d2:75:6f:a5:d5:60"
 FINGER_ECDSA_384 = "384 c1:8d:a0:59:09:47:41:8e:a8:a6:07:01:29:23:b4:65"
 FINGER_ECDSA_521 = "521 44:58:22:52:12:33:16:0e:ce:0e:be:2c:7c:7e:cc:1e"
 SIGNED_RSA = "20:d7:8a:31:21:cb:f7:92:12:f2:a4:89:37:f5:78:af:e6:16:b6:25:b9:97:3d:a2:cd:5f:ca:20:21:73:4c:ad:34:73:8f:20:77:28:e2:94:15:08:d8:91:40:7a:85:83:bf:18:37:95:dc:54:1a:9b:88:29:6c:73:ca:38:b4:04:f1:56:b9:f2:42:9d:52:1b:29:29:b4:4f:fd:c9:2d:af:47:d2:40:76:30:f3:63:45:0c:d9:1d:43:86:0f:1c:70:e2:93:12:34:f3:ac:c5:0a:2f:14:50:66:59:f1:88:ee:c1:4a:e9:d1:9c:4e:46:f0:0e:47:6f:38:74:f1:44:a8"  # noqa
+SIGNED_RSA_256 = "cc:6:60:e0:0:2c:ac:9e:26:bc:d5:68:64:3f:9f:a7:e5:aa:41:eb:88:4a:25:5:9c:93:84:66:ef:ef:60:f4:34:fb:f4:c8:3d:55:33:6a:77:bd:b2:ee:83:f:71:27:41:7e:f5:7:5:0:a9:4c:7:80:6f:be:76:67:cb:58:35:b9:2b:f3:c2:d3:3c:ee:e1:3f:59:e0:fa:e4:5c:92:ed:ae:74:de:d:d6:27:16:8f:84:a3:86:68:c:94:90:7d:6e:cc:81:12:d8:b6:ad:aa:31:a8:13:3d:63:81:3e:bb:5:b6:38:4d:2:d:1b:5b:70:de:83:cc:3a:cb:31"  # noqa
+SIGNED_RSA_512 = "87:46:8b:75:92:33:78:a0:22:35:32:39:23:c6:ab:e1:6:92:ad:bc:7f:6e:ab:19:32:e4:78:b2:2c:8f:1d:c:65:da:fc:a5:7:ca:b6:55:55:31:83:b1:a0:af:d1:95:c5:2e:af:56:ba:f5:41:64:f:39:9d:af:82:43:22:8f:90:52:9d:89:e7:45:97:df:f3:f2:bc:7b:3a:db:89:e:34:fd:18:62:25:1b:ef:77:aa:c6:6c:99:36:3a:84:d6:9c:2a:34:8c:7f:f4:bb:c9:a5:9a:6c:11:f2:cf:da:51:5e:1e:7f:90:27:34:de:b2:f3:15:4f:db:47:32:6b:a7"  # noqa
 FINGER_RSA_2K_OPENSSH = "2048 68:d1:72:01:bf:c0:0c:66:97:78:df:ce:75:74:46:d6"
 FINGER_DSS_1K_OPENSSH = "1024 cf:1d:eb:d7:61:d3:12:94:c6:c0:c6:54:35:35:b0:82"
 FINGER_EC_384_OPENSSH = "384 72:14:df:c1:9a:c3:e6:0e:11:29:d6:32:18:7b:ea:9b"
@@ -238,21 +240,29 @@ class KeyTest(unittest.TestCase):
         self.assertTrue(not pub.can_sign())
         self.assertEqual(key, pub)
 
-    def test_sign_rsa(self):
-        # verify that the rsa private key can sign and verify
+    def _sign_and_verify_rsa(self, algorithm, saved_sig):
         key = RSAKey.from_private_key_file(_support("test_rsa.key"))
-        msg = key.sign_ssh_data(b"ice weasels")
-        self.assertTrue(type(msg) is Message)
+        msg = key.sign_ssh_data(b"ice weasels", algorithm)
+        assert isinstance(msg, Message)
         msg.rewind()
-        self.assertEqual("ssh-rsa", msg.get_text())
-        sig = bytes().join(
-            [byte_chr(int(x, 16)) for x in SIGNED_RSA.split(":")]
+        assert msg.get_text() == algorithm
+        expected = bytes().join(
+            [byte_chr(int(x, 16)) for x in saved_sig.split(":")]
         )
-        self.assertEqual(sig, msg.get_binary())
+        assert msg.get_binary() == expected
         msg.rewind()
         pub = RSAKey(data=key.asbytes())
         self.assertTrue(pub.verify_ssh_sig(b"ice weasels", msg))
 
+    def test_sign_and_verify_ssh_rsa(self):
+        self._sign_and_verify_rsa("ssh-rsa", SIGNED_RSA)
+
+    def test_sign_and_verify_rsa_sha2_512(self):
+        self._sign_and_verify_rsa("rsa-sha2-512", SIGNED_RSA_512)
+
+    def test_sign_and_verify_rsa_sha2_256(self):
+        self._sign_and_verify_rsa("rsa-sha2-256", SIGNED_RSA_256)
+
     def test_sign_dss(self):
         # verify that the dss private key can sign and verify
         key = DSSKey.from_private_key_file(_support("test_dss.key"))
diff --git a/tests/test_transport.py b/tests/test_transport.py
index e1e37e47..6145e5cb 100644
--- a/tests/test_transport.py
+++ b/tests/test_transport.py
@@ -23,6 +23,7 @@ Some unit tests for the ssh2 protocol in Transport.
 from __future__ import with_statement
 
 from binascii import hexlify
+from contextlib import contextmanager
 import select
 import socket
 import time
@@ -38,6 +39,8 @@ from paramiko import (
     Packetizer,
     RSAKey,
     SSHException,
+    AuthenticationException,
+    IncompatiblePeer,
     SecurityOptions,
     ServerInterface,
     Transport,
@@ -80,6 +83,9 @@ class NullServer(ServerInterface):
     paranoid_did_public_key = False
     paranoid_key = DSSKey.from_private_key_file(_support("test_dss.key"))
 
+    def __init__(self, allowed_keys=None):
+        self.allowed_keys = allowed_keys if allowed_keys is not None else []
+
     def get_allowed_auths(self, username):
         if username == "slowdive":
             return "publickey,password"
@@ -90,6 +96,11 @@ class NullServer(ServerInterface):
             return AUTH_SUCCESSFUL
         return AUTH_FAILED
 
+    def check_auth_publickey(self, username, key):
+        if key in self.allowed_keys:
+            return AUTH_SUCCESSFUL
+        return AUTH_FAILED
+
     def check_channel_request(self, kind, chanid):
         if kind == "bogus":
             return OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED
@@ -154,6 +165,7 @@ class TransportTest(unittest.TestCase):
         self.socks.close()
         self.sockc.close()
 
+    # TODO: unify with newer contextmanager
     def setup_test_server(
         self, client_options=None, server_options=None, connect_kwargs=None
     ):
@@ -1169,3 +1181,260 @@ class AlgorithmDisablingTests(unittest.TestCase):
         assert "ssh-dss" not in server_keys
         assert "diffie-hellman-group14-sha256" not in kexen
         assert "zlib" not in compressions
+
+
+@contextmanager
+def server(
+    hostkey=None,
+    init=None,
+    server_init=None,
+    client_init=None,
+    connect=None,
+    pubkeys=None,
+    catch_error=False,
+):
+    """
+    SSH server contextmanager for testing.
+
+    :param hostkey:
+        Host key to use for the server; if None, loads
+        ``test_rsa.key``.
+    :param init:
+        Default `Transport` constructor kwargs to use for both sides.
+    :param server_init:
+        Extends and/or overrides ``init`` for server transport only.
+    :param client_init:
+        Extends and/or overrides ``init`` for client transport only.
+    :param connect:
+        Kwargs to use for ``connect()`` on the client.
+    :param pubkeys:
+        List of public keys for auth.
+    :param catch_error:
+        Whether to capture connection errors & yield from contextmanager.
+        Necessary for connection_time exception testing.
+    """
+    if init is None:
+        init = {}
+    if server_init is None:
+        server_init = {}
+    if client_init is None:
+        client_init = {}
+    if connect is None:
+        connect = dict(username="slowdive", password="pygmalion")
+    socks = LoopSocket()
+    sockc = LoopSocket()
+    sockc.link(socks)
+    tc = Transport(sockc, **dict(init, **client_init))
+    ts = Transport(socks, **dict(init, **server_init))
+
+    if hostkey is None:
+        hostkey = RSAKey.from_private_key_file(_support("test_rsa.key"))
+    ts.add_server_key(hostkey)
+    event = threading.Event()
+    server = NullServer(allowed_keys=pubkeys)
+    assert not event.is_set()
+    assert not ts.is_active()
+    assert tc.get_username() is None
+    assert ts.get_username() is None
+    assert not tc.is_authenticated()
+    assert not ts.is_authenticated()
+
+    err = None
+    # Trap errors and yield instead of raising right away;  otherwise callers
+    # cannot usefully deal with problems at connect time which stem from errors
+    # in the server side.
+    try:
+        ts.start_server(event, server)
+        tc.connect(**connect)
+
+        event.wait(1.0)
+        assert event.is_set()
+        assert ts.is_active()
+        assert tc.is_active()
+
+    except Exception as e:
+        if not catch_error:
+            raise
+        err = e
+
+    yield (tc, ts, err) if catch_error else (tc, ts)
+
+    tc.close()
+    ts.close()
+    socks.close()
+    sockc.close()
+
+
+_disable_sha2 = dict(
+    disabled_algorithms=dict(keys=["rsa-sha2-256", "rsa-sha2-512"])
+)
+_disable_sha1 = dict(disabled_algorithms=dict(keys=["ssh-rsa"]))
+_disable_sha2_pubkey = dict(
+    disabled_algorithms=dict(pubkeys=["rsa-sha2-256", "rsa-sha2-512"])
+)
+_disable_sha1_pubkey = dict(disabled_algorithms=dict(pubkeys=["ssh-rsa"]))
+
+
+class TestSHA2SignatureKeyExchange(unittest.TestCase):
+    # NOTE: these all rely on the default server() hostkey being RSA
+    # NOTE: these rely on both sides being properly implemented re: agreed-upon
+    # hostkey during kex being what's actually used. Truly proving that eg
+    # SHA512 was used, is quite difficult w/o super gross hacks. However, there
+    # are new tests in test_pkey.py which use known signature blobs to prove
+    # the SHA2 family was in fact used!
+
+    def test_base_case_ssh_rsa_still_used_as_fallback(self):
+        # Prove that ssh-rsa is used if either, or both, participants have SHA2
+        # algorithms disabled
+        for which in ("init", "client_init", "server_init"):
+            with server(**{which: _disable_sha2}) as (tc, _):
+                assert tc.host_key_type == "ssh-rsa"
+
+    def test_kex_with_sha2_512(self):
+        # It's the default!
+        with server() as (tc, _):
+            assert tc.host_key_type == "rsa-sha2-512"
+
+    def test_kex_with_sha2_256(self):
+        # No 512 -> you get 256
+        with server(
+            init=dict(disabled_algorithms=dict(keys=["rsa-sha2-512"]))
+        ) as (tc, _):
+            assert tc.host_key_type == "rsa-sha2-256"
+
+    def _incompatible_peers(self, client_init, server_init):
+        with server(
+            client_init=client_init, server_init=server_init, catch_error=True
+        ) as (tc, ts, err):
+            # If neither side blew up then that's bad!
+            assert err is not None
+            # If client side blew up first, it'll be straightforward
+            if isinstance(err, IncompatiblePeer):
+                pass
+            # If server side blew up first, client sees EOF & we need to check
+            # the server transport for its saved error (otherwise it can only
+            # appear in log output)
+            elif isinstance(err, EOFError):
+                assert ts.saved_exception is not None
+                assert isinstance(ts.saved_exception, IncompatiblePeer)
+            # If it was something else, welp
+            else:
+                raise err
+
+    def test_client_sha2_disabled_server_sha1_disabled_no_match(self):
+        self._incompatible_peers(
+            client_init=_disable_sha2, server_init=_disable_sha1
+        )
+
+    def test_client_sha1_disabled_server_sha2_disabled_no_match(self):
+        self._incompatible_peers(
+            client_init=_disable_sha1, server_init=_disable_sha2
+        )
+
+    def test_explicit_client_hostkey_not_limited(self):
+        # Be very explicit about the hostkey on BOTH ends,
+        # and ensure it still ends up choosing sha2-512.
+        # (This is a regression test vs previous implementation which overwrote
+        # the entire preferred-hostkeys structure when given an explicit key as
+        # a client.)
+        hostkey = RSAKey.from_private_key_file(_support("test_rsa.key"))
+        with server(hostkey=hostkey, connect=dict(hostkey=hostkey)) as (tc, _):
+            assert tc.host_key_type == "rsa-sha2-512"
+
+
+class TestExtInfo(unittest.TestCase):
+    def test_ext_info_handshake(self):
+        with server() as (tc, _):
+            kex = tc._get_latest_kex_init()
+            assert kex["kex_algo_list"][-1] == "ext-info-c"
+            assert tc.server_extensions == {
+                "server-sig-algs": b"ssh-ed25519,ecdsa-sha2-nistp256,ecdsa-sha2-nistp384,ecdsa-sha2-nistp521,rsa-sha2-512,rsa-sha2-256,ssh-rsa,ssh-dss"  # noqa
+            }
+
+    def test_client_uses_server_sig_algs_for_pubkey_auth(self):
+        privkey = RSAKey.from_private_key_file(_support("test_rsa.key"))
+        with server(
+            pubkeys=[privkey],
+            connect=dict(pkey=privkey),
+            server_init=dict(
+                disabled_algorithms=dict(pubkeys=["rsa-sha2-512"])
+            ),
+        ) as (tc, _):
+            assert tc.is_authenticated()
+            # Client settled on 256 despite itself not having 512 disabled
+            assert tc._agreed_pubkey_algorithm == "rsa-sha2-256"
+
+
+# TODO: these could move into test_auth.py but that badly needs refactoring
+# with this module anyways...
+class TestSHA2SignaturePubkeys(unittest.TestCase):
+    def test_pubkey_auth_honors_disabled_algorithms(self):
+        privkey = RSAKey.from_private_key_file(_support("test_rsa.key"))
+        with server(
+            pubkeys=[privkey],
+            connect=dict(pkey=privkey),
+            init=dict(
+                disabled_algorithms=dict(
+                    pubkeys=["ssh-rsa", "rsa-sha2-256", "rsa-sha2-512"]
+                )
+            ),
+            catch_error=True,
+        ) as (_, _, err):
+            assert isinstance(err, SSHException)
+            assert "no RSA pubkey algorithms" in str(err)
+
+    def test_client_sha2_disabled_server_sha1_disabled_no_match(self):
+        privkey = RSAKey.from_private_key_file(_support("test_rsa.key"))
+        with server(
+            pubkeys=[privkey],
+            connect=dict(pkey=privkey),
+            client_init=_disable_sha2_pubkey,
+            server_init=_disable_sha1_pubkey,
+            catch_error=True,
+        ) as (tc, ts, err):
+            assert isinstance(err, AuthenticationException)
+
+    def test_client_sha1_disabled_server_sha2_disabled_no_match(self):
+        privkey = RSAKey.from_private_key_file(_support("test_rsa.key"))
+        with server(
+            pubkeys=[privkey],
+            connect=dict(pkey=privkey),
+            client_init=_disable_sha1_pubkey,
+            server_init=_disable_sha2_pubkey,
+            catch_error=True,
+        ) as (tc, ts, err):
+            assert isinstance(err, AuthenticationException)
+
+    def test_ssh_rsa_still_used_when_sha2_disabled(self):
+        privkey = RSAKey.from_private_key_file(_support("test_rsa.key"))
+        # NOTE: this works because key obj comparison uses public bytes
+        # TODO: would be nice for PKey to grow a legit "give me another obj of
+        # same class but just the public bits" using asbytes()
+        with server(
+            pubkeys=[privkey], connect=dict(pkey=privkey), init=_disable_sha2
+        ) as (tc, _):
+            assert tc.is_authenticated()
+
+    def test_sha2_512(self):
+        privkey = RSAKey.from_private_key_file(_support("test_rsa.key"))
+        with server(
+            pubkeys=[privkey],
+            connect=dict(pkey=privkey),
+            init=dict(
+                disabled_algorithms=dict(pubkeys=["ssh-rsa", "rsa-sha2-256"])
+            ),
+        ) as (tc, ts):
+            assert tc.is_authenticated()
+            assert tc._agreed_pubkey_algorithm == "rsa-sha2-512"
+
+    def test_sha2_256(self):
+        privkey = RSAKey.from_private_key_file(_support("test_rsa.key"))
+        with server(
+            pubkeys=[privkey],
+            connect=dict(pkey=privkey),
+            init=dict(
+                disabled_algorithms=dict(pubkeys=["ssh-rsa", "rsa-sha2-512"])
+            ),
+        ) as (tc, ts):
+            assert tc.is_authenticated()
+            assert tc._agreed_pubkey_algorithm == "rsa-sha2-256"