Opt-in overhaul to how MSG_SERVICE_REQUEST is done

- New subclass(es) for opt-in use. Most below messages refer to them,
  not parent classes.
- In parent classes, make handler tables instance attributes for easier
  subclass twiddling.
- Refactor Transport-level session check
- Refactor Transport-level auth handler instantiation (but keep behavior
  the same, for now)
- Add service-request handler to Transport subclass, and remove from
  AuthHandler subclass
- Remove manual event injection from the handful of Transport auth
  methods which supported it. Suspect unused, don't need the extra
  complexity, and wasn't consistent anyways - can add back smarter later
  if anyone needs it.
- Not bothering with gssapi at all for now as I cannot easily test it
- Primarily tested against the new AuthStrategy architecture

5 files changed, 376 insertions, 34 deletions

diff --git a/paramiko/__init__.py b/paramiko/__init__.py
index 3537f1d0..9a987e1a 100644
--- a/paramiko/__init__.py
+++ b/paramiko/__init__.py
@@ -19,7 +19,11 @@
 # flake8: noqa
 import sys
 from paramiko._version import __version__, __version_info__
-from paramiko.transport import SecurityOptions, Transport
+from paramiko.transport import (
+    SecurityOptions,
+    Transport,
+    ServiceRequestingTransport,
+)
 from paramiko.client import (
     SSHClient,
     MissingHostKeyPolicy,
diff --git a/paramiko/auth_handler.py b/paramiko/auth_handler.py
index 22f506c1..aba4b1c5 100644
--- a/paramiko/auth_handler.py
+++ b/paramiko/auth_handler.py
@@ -21,6 +21,7 @@
 """
 
 import weakref
+import threading
 import time
 import re
 
@@ -241,7 +242,9 @@ class AuthHandler:
             if not self.transport.is_active():
                 e = self.transport.get_exception()
                 if (e is None) or issubclass(e.__class__, EOFError):
-                    e = AuthenticationException("Authentication failed.")
+                    e = AuthenticationException(
+                        "Authentication failed: transport shut down or saw EOF"
+                    )
                 raise e
             if event.is_set():
                 break
@@ -254,6 +257,7 @@ class AuthHandler:
                 e = AuthenticationException("Authentication failed.")
             # this is horrible.  Python Exception isn't yet descended from
             # object, so type(e) won't work. :(
+            # TODO 4.0: lol. just lmao.
             if issubclass(e.__class__, PartialAuthentication):
                 return e.allowed_types
             raise e
@@ -367,9 +371,6 @@ class AuthHandler:
     def _parse_service_accept(self, m):
         service = m.get_text()
         if service == "ssh-userauth":
-            # TODO 4.0: this message sucks ass. change it to something more
-            # obvious. it always appears to mean "we already authed" but no! it
-            # just means "we are allowed to TRY authing!"
             self._log(DEBUG, "userauth is OK")
             m = Message()
             m.add_byte(cMSG_USERAUTH_REQUEST)
@@ -725,6 +726,9 @@ Error Message: {}
 
     def _parse_userauth_failure(self, m):
         authlist = m.get_list()
+        # TODO 4.0: we aren't giving callers access to authlist _unless_ it's
+        # partial authentication, so eg authtype=none can't work unless we
+        # tweak this.
         partial = m.get_boolean()
         if partial:
             self._log(INFO, "Authentication continues...")
@@ -805,7 +809,6 @@ Error Message: {}
             self.auth_event.set()
         return
 
-    # TODO: do the same to the other tables, in Transport.
     # TODO 4.0: MAY make sense to make these tables into actual
     # classes/instances that can be fed a mode bool or whatever. Or,
     # alternately (both?) make the message types small classes or enums that
@@ -814,20 +817,27 @@ Error Message: {}
     # TODO: if we do that, also expose 'em publicly.
 
     # Messages which should be handled _by_ servers (sent by clients)
-    _server_handler_table = {
-        MSG_SERVICE_REQUEST: _parse_service_request,
-        MSG_USERAUTH_REQUEST: _parse_userauth_request,
-        MSG_USERAUTH_INFO_RESPONSE: _parse_userauth_info_response,
-    }
+    @property
+    def _server_handler_table(self):
+        return {
+            # TODO 4.0: MSG_SERVICE_REQUEST ought to eventually move into
+            # Transport's server mode like the client side did, just for
+            # consistency.
+            MSG_SERVICE_REQUEST: self._parse_service_request,
+            MSG_USERAUTH_REQUEST: self._parse_userauth_request,
+            MSG_USERAUTH_INFO_RESPONSE: self._parse_userauth_info_response,
+        }
 
     # Messages which should be handled _by_ clients (sent by servers)
-    _client_handler_table = {
-        MSG_SERVICE_ACCEPT: _parse_service_accept,
-        MSG_USERAUTH_SUCCESS: _parse_userauth_success,
-        MSG_USERAUTH_FAILURE: _parse_userauth_failure,
-        MSG_USERAUTH_BANNER: _parse_userauth_banner,
-        MSG_USERAUTH_INFO_REQUEST: _parse_userauth_info_request,
-    }
+    @property
+    def _client_handler_table(self):
+        return {
+            MSG_SERVICE_ACCEPT: self._parse_service_accept,
+            MSG_USERAUTH_SUCCESS: self._parse_userauth_success,
+            MSG_USERAUTH_FAILURE: self._parse_userauth_failure,
+            MSG_USERAUTH_BANNER: self._parse_userauth_banner,
+            MSG_USERAUTH_INFO_REQUEST: self._parse_userauth_info_request,
+        }
 
     # NOTE: prior to the fix for #1283, this was a static dict instead of a
     # property. Should be backwards compatible in most/all cases.
@@ -945,3 +955,115 @@ class GssapiWithMicAuthHandler:
         # TODO: determine if we can cut this up like we did for the primary
         # AuthHandler class.
         return self.__handler_table
+
+
+class AuthOnlyHandler(AuthHandler):
+    """
+    AuthHandler, and just auth, no service requests!
+
+    .. versionadded:: 3.2
+    """
+
+    # NOTE: this purposefully duplicates some of the parent class in order to
+    # modernize, refactor, etc. The intent is that eventually we will collapse
+    # this one onto the parent in a backwards incompatible release.
+
+    @property
+    def _client_handler_table(self):
+        my_table = super()._client_handler_table.copy()
+        del my_table[MSG_SERVICE_ACCEPT]
+        return my_table
+
+    def send_auth_request(self, username, method, finish_message=None):
+        """
+        Submit a userauth request message & wait for response.
+
+        Performs the transport message send call, sets self.auth_event, and
+        will lock-n-block as necessary to both send, and wait for response to,
+        the USERAUTH_REQUEST.
+
+        Most callers will want to supply a callback to ``finish_message``,
+        which accepts a Message ``m`` and may call mutator methods on it to add
+        more fields.
+        """
+        # Store a few things for reference in handlers, including auth failure
+        # handler (which needs to know if we were using a bad method, etc)
+        self.auth_method = method
+        self.username = username
+        # Generic userauth request fields
+        m = Message()
+        m.add_byte(cMSG_USERAUTH_REQUEST)
+        m.add_string(username)
+        m.add_string("ssh-connection")
+        m.add_string(method)
+        # Caller usually has more to say, such as injecting password, key etc
+        finish_message(m)
+        # TODO 4.0: seems odd to have the client handle the lock and not
+        # Transport; that _may_ have been an artifact of allowing user
+        # threading event injection? Regardless, we don't want to move _this_
+        # locking into Transport._send_message now, because lots of other
+        # untouched code also uses that method and we might end up
+        # double-locking (?) but 4.0 would be a good time to revisit.
+        with self.transport.lock:
+            self.transport._send_message(m)
+        # We have cut out the higher level event args, but self.auth_event is
+        # still required for self.wait_for_response to function correctly (it's
+        # the mechanism used by the auth success/failure handlers, the abort
+        # handler, and a few other spots like in gssapi.
+        # TODO: interestingly, wait_for_response itself doesn't actually
+        # enforce that its event argument and self.auth_event are the same...
+        self.auth_event = threading.Event()
+        return self.wait_for_response(self.auth_event)
+
+    def auth_none(self, username):
+        return self.send_auth_request(username, "none")
+
+    def auth_publickey(self, username, key):
+        key_type, bits = self._get_key_type_and_bits(key)
+        algorithm = self._finalize_pubkey_algorithm(key_type)
+        blob = self._get_session_blob(
+            key,
+            "ssh-connection",
+            username,
+            algorithm,
+        )
+
+        def finish(m):
+            # This field doesn't appear to be named, but is False when querying
+            # for permission (ie knowing whether to even prompt a user for
+            # passphrase, etc) or True when just going for it. Paramiko has
+            # never bothered with the former type of message, apparently.
+            m.add_boolean(True)
+            m.add_string(algorithm)
+            m.add_string(bits)
+            m.add_string(key.sign_ssh_data(blob, algorithm))
+
+        return self.send_auth_request(username, "publickey", finish)
+
+    def auth_password(self, username, password):
+        def finish(m):
+            # Unnamed field that equates to "I am changing my password", which
+            # Paramiko clientside never supported and serverside only sort of
+            # supported.
+            m.add_boolean(False)
+            m.add_string(b(password))
+
+        return self.send_auth_request(username, "password", finish)
+
+    def auth_interactive(self, username, handler, submethods=""):
+        """
+        response_list = handler(title, instructions, prompt_list)
+        """
+        # Unlike most siblings, this auth method _does_ require other
+        # superclass handlers (eg userauth info request) to understand
+        # what's going on, so we still set some self attributes.
+        self.auth_method = "keyboard_interactive"
+        self.interactive_handler = handler
+
+        def finish(m):
+            # Empty string for deprecated language tag field, per RFC 4256:
+            # https://www.rfc-editor.org/rfc/rfc4256#section-3.1
+            m.add_string("")
+            m.add_string(submethods)
+
+        return self.send_auth_request(username, "keyboard-interactive", finish)
diff --git a/paramiko/transport.py b/paramiko/transport.py
index c4d7efd1..875d50c2 100644
--- a/paramiko/transport.py
+++ b/paramiko/transport.py
@@ -34,7 +34,7 @@ from cryptography.hazmat.primitives.ciphers import algorithms, Cipher, modes
 
 import paramiko
 from paramiko import util
-from paramiko.auth_handler import AuthHandler
+from paramiko.auth_handler import AuthHandler, AuthOnlyHandler
 from paramiko.ssh_gss import GSSAuth
 from paramiko.channel import Channel
 from paramiko.common import (
@@ -64,6 +64,8 @@ from paramiko.common import (
     MSG_GLOBAL_REQUEST,
     MSG_REQUEST_SUCCESS,
     MSG_REQUEST_FAILURE,
+    cMSG_SERVICE_REQUEST,
+    MSG_SERVICE_ACCEPT,
     MSG_CHANNEL_OPEN_SUCCESS,
     MSG_CHANNEL_OPEN_FAILURE,
     MSG_CHANNEL_OPEN,
@@ -531,6 +533,20 @@ class Transport(threading.Thread, ClosingContextManager):
         self.server_accept_cv = threading.Condition(self.lock)
         self.subsystem_table = {}
 
+        # Handler table, now set at init time for easier per-instance
+        # manipulation and subclass twiddling.
+        self._handler_table = {
+            MSG_EXT_INFO: self._parse_ext_info,
+            MSG_NEWKEYS: self._parse_newkeys,
+            MSG_GLOBAL_REQUEST: self._parse_global_request,
+            MSG_REQUEST_SUCCESS: self._parse_request_success,
+            MSG_REQUEST_FAILURE: self._parse_request_failure,
+            MSG_CHANNEL_OPEN_SUCCESS: self._parse_channel_open_success,
+            MSG_CHANNEL_OPEN_FAILURE: self._parse_channel_open_failure,
+            MSG_CHANNEL_OPEN: self._parse_channel_open,
+            MSG_KEXINIT: self._negotiate_keys,
+        }
+
     def _filter_algorithm(self, type_):
         default = getattr(self, "_preferred_{}".format(type_))
         return tuple(
@@ -2140,7 +2156,7 @@ class Transport(threading.Thread, ClosingContextManager):
                         if error_msg:
                             self._send_message(error_msg)
                         else:
-                            self._handler_table[ptype](self, m)
+                            self._handler_table[ptype](m)
                     elif ptype in self._channel_handler_table:
                         chanid = m.get_int()
                         chan = self._channels.get(chanid)
@@ -2166,7 +2182,7 @@ class Transport(threading.Thread, ClosingContextManager):
                         and ptype in self.auth_handler._handler_table
                     ):
                         handler = self.auth_handler._handler_table[ptype]
-                        handler(self.auth_handler, m)
+                        handler(m)
                         if len(self._expected_packet) > 0:
                             continue
                     else:
@@ -2987,18 +3003,6 @@ class Transport(threading.Thread, ClosingContextManager):
         finally:
             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,
-        MSG_REQUEST_FAILURE: _parse_request_failure,
-        MSG_CHANNEL_OPEN_SUCCESS: _parse_channel_open_success,
-        MSG_CHANNEL_OPEN_FAILURE: _parse_channel_open_failure,
-        MSG_CHANNEL_OPEN: _parse_channel_open,
-        MSG_KEXINIT: _negotiate_keys,
-    }
-
     _channel_handler_table = {
         MSG_CHANNEL_SUCCESS: Channel._request_success,
         MSG_CHANNEL_FAILURE: Channel._request_failed,
@@ -3138,3 +3142,161 @@ class ChannelMap:
             return len(self._map)
         finally:
             self._lock.release()
+
+
+class ServiceRequestingTransport(Transport):
+    """
+    Transport, but also handling service requests, like it oughtta!
+
+    .. versionadded:: 3.2
+    """
+
+    # NOTE: this purposefully duplicates some of the parent class in order to
+    # modernize, refactor, etc. The intent is that eventually we will collapse
+    # this one onto the parent in a backwards incompatible release.
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._service_userauth_accepted = False
+        self._handler_table[MSG_SERVICE_ACCEPT] = self._parse_service_accept
+
+    def _parse_service_accept(self, m):
+        service = m.get_text()
+        # Short-circuit for any service name not ssh-userauth.
+        # NOTE: it's technically possible for 'service name' in
+        # SERVICE_REQUEST/ACCEPT messages to be "ssh-connection" --
+        # but I don't see evidence of Paramiko ever initiating or expecting to
+        # receive one of these. We /do/ see the 'service name' field in
+        # MSG_USERAUTH_REQUEST/ACCEPT/FAILURE set to this string, but that is a
+        # different set of handlers, so...!
+        if service != "ssh-userauth":
+            # TODO 4.0: consider erroring here (with an ability to opt out?)
+            # instead as it probably means something went Very Wrong.
+            self._log(
+                DEBUG, 'Service request "{}" accepted (?)'.format(service)
+            )
+            return
+        # Record that we saw a service-userauth acceptance, meaning we are free
+        # to submit auth requests.
+        self._service_userauth_accepted = True
+        self._log(DEBUG, "MSG_SERVICE_ACCEPT received; auth may begin")
+
+    def ensure_session(self):
+        # Make sure we're not trying to auth on a not-yet-open or
+        # already-closed transport session; that's our responsibility, not that
+        # of AuthHandler.
+        if (not self.active) or (not self.initial_kex_done):
+            # TODO: better error message? this can happen in many places, eg
+            # user error (authing before connecting) or developer error (some
+            # improperly handled pre/mid auth shutdown didn't become fatal
+            # enough). The latter is much more common & should ideally be fixed
+            # by terminating things harder?
+            raise SSHException("No existing session")
+        # Also make sure we've actually been told we are allowed to auth.
+        if self._service_userauth_accepted:
+            return
+        # Or request to do so, otherwise.
+        m = Message()
+        m.add_byte(cMSG_SERVICE_REQUEST)
+        m.add_string("ssh-userauth")
+        self._log(DEBUG, "Sending MSG_SERVICE_REQUEST: ssh-userauth")
+        self._send_message(m)
+        # Now we wait to hear back; the user is expecting a blocking-style auth
+        # request so there's no point giving control back anywhere.
+        while not self._service_userauth_accepted:
+            # TODO: feels like we're missing an AuthHandler Event like
+            # 'self.auth_event' which is set when AuthHandler shuts down in
+            # ways good AND bad. Transport only seems to have completion_event
+            # which is unclear re: intent, eg it's set by newkeys which always
+            # happens on connection, so it'll always be set by the time we get
+            # here.
+            # NOTE: this copies the timing of event.wait() in
+            # AuthHandler.wait_for_response, re: 1/10 of a second. Could
+            # presumably be smaller, but seems unlikely this period is going to
+            # be "too long" for any code doing ssh networking...
+            time.sleep(0.1)
+        self.auth_handler = self.get_auth_handler()
+
+    def get_auth_handler(self):
+        # NOTE: using new sibling subclass instead of classic AuthHandler
+        return AuthOnlyHandler(self)
+
+    def auth_none(self, username):
+        # TODO 4.0: merge to parent, preserving (most of) docstring
+        self.ensure_session()
+        return self.auth_handler.auth_none(username)
+
+    def auth_password(self, username, password, fallback=True):
+        # TODO 4.0: merge to parent, preserving (most of) docstring
+        self.ensure_session()
+        try:
+            return self.auth_handler.auth_password(username, password)
+        except BadAuthenticationType as e:
+            # if password auth isn't allowed, but keyboard-interactive *is*,
+            # try to fudge it
+            if not fallback or ("keyboard-interactive" not in e.allowed_types):
+                raise
+            try:
+
+                def handler(title, instructions, fields):
+                    if len(fields) > 1:
+                        raise SSHException("Fallback authentication failed.")
+                    if len(fields) == 0:
+                        # for some reason, at least on os x, a 2nd request will
+                        # be made with zero fields requested.  maybe it's just
+                        # to try to fake out automated scripting of the exact
+                        # type we're doing here.  *shrug* :)
+                        return []
+                    return [password]
+
+                return self.auth_interactive(username, handler)
+            except SSHException:
+                # attempt to fudge failed; just raise the original exception
+                raise e
+
+    def auth_publickey(self, username, key):
+        # TODO 4.0: merge to parent, preserving (most of) docstring
+        self.ensure_session()
+        return self.auth_handler.auth_publickey(username, key)
+
+    def auth_interactive(self, username, handler, submethods=""):
+        # TODO 4.0: merge to parent, preserving (most of) docstring
+        self.ensure_session()
+        return self.auth_handler.auth_interactive(
+            username, handler, submethods
+        )
+
+    def auth_interactive_dumb(self, username, handler=None, submethods=""):
+        # TODO 4.0: merge to parent, preserving (most of) docstring
+        # NOTE: legacy impl omitted equiv of ensure_session since it just wraps
+        # another call to an auth method. however we reinstate it for
+        # consistency reasons.
+        self.ensure_session()
+        if not handler:
+
+            def handler(title, instructions, prompt_list):
+                answers = []
+                if title:
+                    print(title.strip())
+                if instructions:
+                    print(instructions.strip())
+                for prompt, show_input in prompt_list:
+                    print(prompt.strip(), end=" ")
+                    answers.append(input())
+                return answers
+
+        return self.auth_interactive(username, handler, submethods)
+
+    def auth_gssapi_with_mic(self, username, gss_host, gss_deleg_creds):
+        # TODO 4.0: merge to parent, preserving (most of) docstring
+        self.ensure_session()
+        self.auth_handler = self.get_auth_handler()
+        return self.auth_handler.auth_gssapi_with_mic(
+            username, gss_host, gss_deleg_creds
+        )
+
+    def auth_gssapi_keyex(self, username):
+        # TODO 4.0: merge to parent, preserving (most of) docstring
+        self.ensure_session()
+        self.auth_handler = self.get_auth_handler()
+        return self.auth_handler.auth_gssapi_keyex(username)
diff --git a/sites/www/changelog.rst b/sites/www/changelog.rst
index 39034a1c..b1de893e 100644
--- a/sites/www/changelog.rst
+++ b/sites/www/changelog.rst
@@ -2,6 +2,59 @@
 Changelog
 =========
 
+- :bug:`23 major` Since its inception, Paramiko has (for reasons lost to time)
+  implemented authentication as a side effect of handling affirmative replies
+  to ``MSG_SERVICE_REQUEST`` protocol messages. What this means is Paramiko
+  makes one such request before every ``MSG_USERAUTH_REQUEST``, i.e. every auth
+  attempt.
+
+  OpenSSH doesn't care if clients send multiple service requests, but other
+  server implementations are often stricter in what they accept after an
+  initial service request (due to the RFCs not being clear). This can result in
+  odd behavior when a user doesn't authenticate successfully on the very first
+  try (for example, when the right key for a target host is the third in one's
+  ssh-agent).
+
+  This version of Paramiko now contains an opt-in
+  `~paramiko.transport.Transport` subclass,
+  `~paramiko.transport.ServiceRequestingTransport`, which more-correctly
+  implements service request handling in the Transport, and uses an
+  auth-handler subclass internally which has been similarly adapted. Users
+  wanting to try this new experimental code path may hand this class to
+  `SSHClient.connect <paramiko.client.SSHClient.connect>` as its
+  ``transport_factory`` kwarg.
+
+  .. warning::
+      This feature is **EXPERIMENTAL** and its code may be subject to change.
+
+      In addition:
+        - minor backwards incompatible changes exist in the new code paths,
+          most notably the removal of the (inconsistently applied and rarely
+          used) ``event`` arguments to the ``auth_xxx`` methods.
+        - GSSAPI support has only been partially implemented, and is untested.
+
+  .. note::
+      Some minor backwards-_compatible_ changes were made to the **existing**
+      Transport and AuthHandler classes to facilitate the new code. For
+      example, ``Transport._handler_table`` and
+      ``AuthHandler._client_handler_table`` are now propertes instead of raw
+      attributes.
+
+- :feature:`387` Users of `~paramiko.client.SSHClient` can now configure the
+  authentication logic Paramiko uses when connecting to servers; this
+  functionality is intended for advanced users and higher-level libraries such
+  as `Fabric <https://fabfile.org>`_. See :ref:`the conceptual API docs
+  <auth-flow>` for details.
+
+  Fabric's co-temporal release includes a proof-of-concept use of this feature,
+  implementing an auth flow much closer to that of the OpenSSH client (versus
+  Paramiko's legacy behavior). It is **strongly recommended** that if this
+  interests you, investigate replacing any direct use of ``SSHClient`` with
+  Fabric's ``Connection``.
+
+  .. warning::
+      This feature is **EXPERIMENTAL**; please see its docs for details.
+
 - :feature:`-` Enhanced `~paramiko.agent.AgentKey` with new attributes, such
   as:
 
diff --git a/tests/test_transport.py b/tests/test_transport.py
index 4062d767..6feccf1d 100644
--- a/tests/test_transport.py
+++ b/tests/test_transport.py
@@ -1099,7 +1099,8 @@ class TransportTest(unittest.TestCase):
     def test_server_transports_reject_client_message_types(self):
         # TODO: handle Transport's own tables too, not just its inner auth
         # handler's table. See TODOs in auth_handler.py
-        for message_type in AuthHandler._client_handler_table:
+        some_handler = AuthHandler(self.tc)  # kludge to get _some_ AH instance
+        for message_type in some_handler._client_handler_table:
             self._send_client_message(message_type)
             self._expect_unimplemented()
             # Reset for rest of loop