Twiddle RSA signature algorithm fallback behavior in AuthOnlyHandler

Also includes more rearranging of AuthHandler related tests

3 files changed, 189 insertions, 135 deletions

diff --git a/paramiko/auth_handler.py b/paramiko/auth_handler.py
index d9e3e07c..539bf973 100644
--- a/paramiko/auth_handler.py
+++ b/paramiko/auth_handler.py
@@ -1075,3 +1075,16 @@ class AuthOnlyHandler(AuthHandler):
             m.add_string(submethods)
 
         return self.send_auth_request(username, "keyboard-interactive", finish)
+
+    # NOTE: not strictly 'auth only' related, but allows users to opt-in.
+    def _choose_fallback_pubkey_algorithm(self, key_type, my_algos):
+        msg = "Server did not send a server-sig-algs list; defaulting to something in our preferred algorithms list"  # noqa
+        self._log(DEBUG, msg)
+        if key_type in my_algos:
+            msg = f"Current key type, {key_type!r}, is in our preferred list; using that"  # noqa
+            algo = key_type
+        else:
+            algo = my_algos[0]
+            msg = f"{key_type!r} not in our list - trying first list item instead, {algo!r}"  # noqa
+        self._log(DEBUG, msg)
+        return algo
diff --git a/tests/auth.py b/tests/auth.py
index 08de6148..a12aa5fe 100644
--- a/tests/auth.py
+++ b/tests/auth.py
@@ -1,38 +1,39 @@
-# Copyright (C) 2008  Robey Pointer <robeypointer@gmail.com>
-#
-# This file is part of paramiko.
-#
-# Paramiko is free software; you can redistribute it and/or modify it under the
-# terms of the GNU Lesser General Public License as published by the Free
-# Software Foundation; either version 2.1 of the License, or (at your option)
-# any later version.
-#
-# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY
-# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
-# A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
-# details.
-#
-# You should have received a copy of the GNU Lesser General Public License
-# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
-# 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA.
-
 """
-Some unit tests for authenticating over a Transport.
+Tests focusing primarily on the authentication step.
+
+Thus, they concern AuthHandler, with a side of Transport.
 """
 
-import unittest
 from pytest import raises
 
 from paramiko import (
+    RSAKey,
     DSSKey,
     BadAuthenticationType,
     AuthenticationException,
+    SSHException,
+    ServiceRequestingTransport,
 )
 
-from ._util import _support, server, unicodey
+from ._util import (
+    _support,
+    server,
+    unicodey,
+    requires_sha1_signing,
+    _disable_sha2,
+    _disable_sha2_pubkey,
+    _disable_sha1_pubkey,
+)
 
 
 class AuthHandler_:
+    """
+    Most of these tests are explicit about the auth method they call.
+
+    This is because not too many other tests do so (they rely on the implicit
+    auth trigger of various connect() kwargs).
+    """
+
     def bad_auth_type(self):
         """
         verify that we get the right exception when an unsupported auth
@@ -134,3 +135,157 @@ class AuthHandler_:
             tc.auth_timeout = 1  # 1 second, to speed up test
             tc.auth_password("unresponsive-server", "hello")
             assert "Authentication timeout" in str(info.value)
+
+
+class AuthOnlyHandler_:
+    def _server(self, *args, **kwargs):
+        kwargs.setdefault("transport_factory", ServiceRequestingTransport)
+        return server(*args, **kwargs)
+
+    class fallback_pubkey_algorithm:
+        @requires_sha1_signing
+        def key_type_algo_selected_when_no_server_sig_algs(self):
+            privkey = RSAKey.from_private_key_file(_support("rsa.key"))
+            # Server pretending to be an apparently common setup:
+            # - doesn't support (or have enabled) sha2
+            # - also doesn't support (or have enabled) server-sig-algs/ext-info
+            # This is the scenario in which Paramiko has to guess-the-algo, and
+            # where servers that don't support sha2 or server-sig-algs can give
+            # us trouble.
+            server_init = dict(_disable_sha2_pubkey, server_sig_algs=False)
+            with self._server(
+                pubkeys=[privkey],
+                connect=dict(pkey=privkey),
+                server_init=server_init,
+                catch_error=True,
+            ) as (tc, ts, err):
+                # Auth did work
+                assert tc.is_authenticated()
+                # Selected ssh-rsa, instead of first-in-the-list (rsa-sha2-512)
+                assert tc._agreed_pubkey_algorithm == "ssh-rsa"
+
+        @requires_sha1_signing
+        def uses_first_preferred_algo_if_key_type_not_in_list(self):
+            # This is functionally the same as legacy AuthHandler, just
+            # arriving at the same place in a different manner.
+            privkey = RSAKey.from_private_key_file(_support("rsa.key"))
+            server_init = dict(_disable_sha2_pubkey, server_sig_algs=False)
+            with self._server(
+                pubkeys=[privkey],
+                connect=dict(pkey=privkey),
+                server_init=server_init,
+                client_init=_disable_sha1_pubkey,  # no ssh-rsa
+                catch_error=True,
+            ) as (tc, ts, err):
+                assert not tc.is_authenticated()
+                assert isinstance(err, AuthenticationException)
+                assert tc._agreed_pubkey_algorithm == "rsa-sha2-512"
+
+
+class SHA2SignaturePubkeys:
+    def pubkey_auth_honors_disabled_algorithms(self):
+        privkey = RSAKey.from_private_key_file(_support("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 client_sha2_disabled_server_sha1_disabled_no_match(self):
+        privkey = RSAKey.from_private_key_file(_support("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 client_sha1_disabled_server_sha2_disabled_no_match(self):
+        privkey = RSAKey.from_private_key_file(_support("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)
+
+    @requires_sha1_signing
+    def ssh_rsa_still_used_when_sha2_disabled(self):
+        privkey = RSAKey.from_private_key_file(_support("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()
+
+    @requires_sha1_signing
+    def first_client_preferred_algo_used_when_no_server_sig_algs(self):
+        privkey = RSAKey.from_private_key_file(_support("rsa.key"))
+        # Server pretending to be an apparently common setup:
+        # - doesn't support (or have enabled) sha2
+        # - also doesn't support (or have enabled) server-sig-algs/ext-info
+        # This is the scenario in which Paramiko has to guess-the-algo, and
+        # where servers that don't support sha2 or server-sig-algs give us
+        # trouble.
+        server_init = dict(_disable_sha2_pubkey, server_sig_algs=False)
+        with server(
+            pubkeys=[privkey],
+            connect=dict(username="slowdive", pkey=privkey),
+            server_init=server_init,
+            catch_error=True,
+        ) as (tc, ts, err):
+            assert not tc.is_authenticated()
+            assert isinstance(err, AuthenticationException)
+            # Oh no! this isn't ssh-rsa, and our server doesn't support sha2!
+            assert tc._agreed_pubkey_algorithm == "rsa-sha2-512"
+
+    def sha2_512(self):
+        privkey = RSAKey.from_private_key_file(_support("rsa.key"))
+        with server(
+            pubkeys=[privkey],
+            # TODO: why is this passing without a username?
+            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 sha2_256(self):
+        privkey = RSAKey.from_private_key_file(_support("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"
+
+    def sha2_256_when_client_only_enables_256(self):
+        privkey = RSAKey.from_private_key_file(_support("rsa.key"))
+        with server(
+            pubkeys=[privkey],
+            connect=dict(pkey=privkey),
+            # Client-side only; server still accepts all 3.
+            client_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"
diff --git a/tests/test_transport.py b/tests/test_transport.py
index ee00830a..d8b6cb99 100644
--- a/tests/test_transport.py
+++ b/tests/test_transport.py
@@ -36,7 +36,6 @@ from paramiko import (
     Packetizer,
     RSAKey,
     SSHException,
-    AuthenticationException,
     IncompatiblePeer,
     SecurityOptions,
     Transport,
@@ -64,8 +63,6 @@ from ._util import (
     server,
     _disable_sha2,
     _disable_sha1,
-    _disable_sha2_pubkey,
-    _disable_sha1_pubkey,
     TestServer as NullServer,
 )
 from ._loop import LoopSocket
@@ -1216,114 +1213,3 @@ class TestExtInfo(unittest.TestCase):
             # Client settled on 256 despite itself not having 512 disabled (and
             # otherwise, 512 would have been earlier in the preferred list)
             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("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("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("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)
-
-    @requires_sha1_signing
-    def test_ssh_rsa_still_used_when_sha2_disabled(self):
-        privkey = RSAKey.from_private_key_file(_support("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()
-
-    @requires_sha1_signing
-    def test_first_client_preferred_algo_used_when_no_server_sig_algs(self):
-        privkey = RSAKey.from_private_key_file(_support("rsa.key"))
-        # Server pretending to be an apparently common setup:
-        # - doesn't support (or have enabled) sha2
-        # - also doesn't support (or have enabled) server-sig-algs/ext-info
-        # This is the scenario in which Paramiko has to guess-the-algo, and
-        # where servers that don't support sha2 or server-sig-algs give us
-        # trouble.
-        server_init = dict(_disable_sha2_pubkey, server_sig_algs=False)
-        with server(
-            pubkeys=[privkey],
-            connect=dict(username="slowdive", pkey=privkey),
-            server_init=server_init,
-            catch_error=True,
-        ) as (tc, ts, err):
-            assert not tc.is_authenticated()
-            assert isinstance(err, AuthenticationException)
-            # Oh no! this isn't ssh-rsa, and our server doesn't support sha2!
-            assert tc._agreed_pubkey_algorithm == "rsa-sha2-512"
-
-    def test_sha2_512(self):
-        privkey = RSAKey.from_private_key_file(_support("rsa.key"))
-        with server(
-            pubkeys=[privkey],
-            # TODO: why is this passing without a username?
-            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("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"
-
-    def test_sha2_256_when_client_only_enables_256(self):
-        privkey = RSAKey.from_private_key_file(_support("rsa.key"))
-        with server(
-            pubkeys=[privkey],
-            connect=dict(pkey=privkey),
-            # Client-side only; server still accepts all 3.
-            client_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"