pipelining fxies (#78111)

Moved check to connection as it should be the final decider
Added property to become plugins to indicate support
Also removed hardcoded su exception
Added tty detection logic for ssh (pipelining won't work if tty is needed or forced)

Co-authored-by: Sloane Hertel <19572925+s-hertel@users.noreply.github.com>

changelogs/fragments/pipelining_refactor.yml

@@ -0,0 +1,5 @@
+minor_changes:
+  - pipelining logic has mostly moved to connection plugins so they can decide/override settings.
+  - ssh connection plugin now overrides pipelining when a tty is requested.
+  - become plugins get new property 'pipelining' to show support or lack there of for the feature.
+  - removed harcoding of su plugin as it now works with pipelining.

lib/ansible/plugins/action/__init__.py

@@ -358,35 +358,11 @@ class ActionBase(ABC):
 
         return getattr(self, 'TRANSFERS_FILES', False)
 
-    def _is_pipelining_enabled(self, module_style, wrap_async=False):
+    def _is_pipelining_enabled(self, module_style: str, wrap_async: bool = False) -> bool:
         """
-        Determines if we are required and can do pipelining
+        Determines if we are required and can do pipelining, only 'new' style modules can support pipelining
         """
-
-        try:
-            is_enabled = self._connection.get_option('pipelining')
-        except (KeyError, AttributeError, ValueError):
-            is_enabled = self._play_context.pipelining
-
-        # winrm supports async pipeline
-        # TODO: make other class property 'has_async_pipelining' to separate cases
-        always_pipeline = self._connection.always_pipeline_modules
-
-        # su does not work with pipelining
-        # TODO: add has_pipelining class prop to become plugins
-        become_exception = (self._connection.become.name if self._connection.become else '') != 'su'
-
-        # any of these require a true
-        conditions = [
-            self._connection.has_pipelining,    # connection class supports it
-            is_enabled or always_pipeline,      # enabled via config or forced via connection (eg winrm)
-            module_style == "new",              # old style modules do not support pipelining
-            not C.DEFAULT_KEEP_REMOTE_FILES,    # user wants remote files
-            not wrap_async or always_pipeline,  # async does not normally support pipelining unless it does (eg winrm)
-            become_exception,
-        ]
-
-        return all(conditions)
+        return bool(module_style == 'new' and self._connection.is_pipelining_enabled(wrap_async))
 
     def _get_admin_users(self):
         """

lib/ansible/plugins/become/__init__.py

@@ -34,6 +34,9 @@ class BecomeBase(AnsiblePlugin):
     # plugin requires a tty, i.e su
     require_tty = False
 
+    # plugin allows for pipelining executio
+    pipelining = True
+
     # prompt to match
     prompt = ''
 

lib/ansible/plugins/become/su.py

@@ -101,6 +101,8 @@ class BecomeModule(BecomeBase):
 
     name = 'su'
 
+    pipelining = False
+
     # messages for detecting prompted password issues
     fail = ('Authentication failure',)
 

lib/ansible/plugins/connection/__init__.py

@@ -296,6 +296,23 @@ class ConnectionBase(AnsiblePlugin):
 
         return var_options
 
+    def is_pipelining_enabled(self, wrap_async: bool = False) -> bool:
+
+        is_enabled = False
+        if self.has_pipelining and (not self.become or self.become.pipelining):
+            try:
+                is_enabled = self.get_option('pipelining')
+            except KeyError:
+                is_enabled = getattr(self._play_context, 'pipelining', False)
+
+        # TODO: deprecate always_pipeline_modules and has_native_async in favor for each plugin overriding this function
+        conditions = [
+            is_enabled or self.always_pipeline_modules,       # enabled via config or forced via connection (eg winrm)
+            not C.DEFAULT_KEEP_REMOTE_FILES,                  # user wants remote files
+            not wrap_async or self.has_native_async,          # async does not normally support pipelining unless it does (eg winrm)
+        ]
+        return all(conditions)
+
 
 class NetworkConnectionBase(ConnectionBase):
     """

lib/ansible/plugins/connection/ssh.py

@@ -382,6 +382,7 @@ DOCUMENTATION = """
 """
 
 import collections.abc as c
+import argparse
 import errno
 import contextlib
 import fcntl
@@ -632,6 +633,11 @@ class Connection(ConnectionBase):
             self.module_implementation_preferences = ('.ps1', '.exe', '')
             self.allow_executable = False
 
+        # parser to discover 'passed options', used later on for pipelining resolution
+        self._tty_parser = argparse.ArgumentParser()
+        self._tty_parser.add_argument('-t', action='count')
+        self._tty_parser.add_argument('-o', action='append')
+
     # The connection is created by running ssh/scp/sftp from the exec_command,
     # put_file, and fetch_file methods, so we don't need to do any connection
     # management here.
@@ -1489,3 +1495,41 @@ class Connection(ConnectionBase):
 
     def close(self) -> None:
         self._connected = False
+
+    @property
+    def has_tty(self):
+        return self._is_tty_requested()
+
+    def _is_tty_requested(self):
+
+        # check if we require tty (only from our args, cannot see options in configuration files)
+        opts = []
+        for opt in ('ssh_args', 'ssh_common_args', 'ssh_extra_args'):
+            attr = self.get_option(opt)
+            if attr is not None:
+                opts.extend(self._split_ssh_args(attr))
+
+        args, dummy = self._tty_parser.parse_known_args(opts)
+
+        if args.t:
+            return True
+
+        for arg in args.o or []:
+            if '=' in arg:
+                val = arg.split('=', 1)
+            else:
+                val = arg.split(maxsplit=1)
+
+            if val[0].lower().strip() == 'requesttty':
+                if val[1].lower().strip() in ('yes', 'force'):
+                    return True
+
+        return False
+
+    def is_pipelining_enabled(self, wrap_async=False):
+        """ override parent method and ensure we don't request a tty """
+
+        if self._is_tty_requested():
+            return False
+        else:
+            return super(Connection, self).is_pipelining_enabled(wrap_async)