[collect] Abstract transport protocol from SoSNode

Since its addition to sos, collect has assumed the use of a system
installation of SSH in order to connect to the nodes identified for
collection. However, there may be use cases and desires to use other
transport protocols.

As such, provide an abstraction for these protocols in the form of the
new `RemoteTransport` class that `SoSNode` will now leverage. So far an
abstraction for the currently used SSH ControlPersist function is
provided, along with a psuedo abstraction for local execution so that
SoSNode does not directly need to make more "if local then foo" checks
than are absolutely necessary.

Related: #2668

Signed-off-by: Jake Hunsaker <jhunsake@redhat.com>

13 files changed, 705 insertions, 344 deletions

diff --git a/setup.py b/setup.py
index 7653b59d..25e87a71 100644
--- a/setup.py
+++ b/setup.py
@@ -101,8 +101,8 @@ setup(
         'sos.policies.distros', 'sos.policies.runtimes',
         'sos.policies.package_managers', 'sos.policies.init_systems',
         'sos.report', 'sos.report.plugins', 'sos.collector',
-        'sos.collector.clusters', 'sos.cleaner', 'sos.cleaner.mappings',
-        'sos.cleaner.parsers', 'sos.cleaner.archives'
+        'sos.collector.clusters', 'sos.collector.transports', 'sos.cleaner',
+        'sos.cleaner.mappings', 'sos.cleaner.parsers', 'sos.cleaner.archives'
     ],
     cmdclass=cmdclass,
     command_options=command_options,
diff --git a/sos/collector/__init__.py b/sos/collector/__init__.py
index b2a07f37..da912655 100644
--- a/sos/collector/__init__.py
+++ b/sos/collector/__init__.py
@@ -17,7 +17,6 @@ import re
 import string
 import socket
 import shutil
-import subprocess
 import sys
 
 from datetime import datetime
@@ -28,7 +27,6 @@ from pipes import quote
 from textwrap import fill
 from sos.cleaner import SoSCleaner
 from sos.collector.sosnode import SosNode
-from sos.collector.exceptions import ControlPersistUnsupportedException
 from sos.options import ClusterOption
 from sos.component import SoSComponent
 from sos import __version__
@@ -154,7 +152,6 @@ class SoSCollector(SoSComponent):
             try:
                 self.parse_node_strings()
                 self.parse_cluster_options()
-                self._check_for_control_persist()
                 self.log_debug('Executing %s' % ' '.join(s for s in sys.argv))
                 self.log_debug("Found cluster profiles: %s"
                                % self.clusters.keys())
@@ -437,33 +434,6 @@ class SoSCollector(SoSComponent):
                                  action='extend',
                                  help='List of usernames to obfuscate')
 
-    def _check_for_control_persist(self):
-        """Checks to see if the local system supported SSH ControlPersist.
-
-        ControlPersist allows OpenSSH to keep a single open connection to a
-        remote host rather than building a new session each time. This is the
-        same feature that Ansible uses in place of paramiko, which we have a
-        need to drop in sos-collector.
-
-        This check relies on feedback from the ssh binary. The command being
-        run should always generate stderr output, but depending on what that
-        output reads we can determine if ControlPersist is supported or not.
-
-        For our purposes, a host that does not support ControlPersist is not
-        able to run sos-collector.
-
-        Returns
-            True if ControlPersist is supported, else raise Exception.
-        """
-        ssh_cmd = ['ssh', '-o', 'ControlPersist']
-        cmd = subprocess.Popen(ssh_cmd, stdout=subprocess.PIPE,
-                               stderr=subprocess.PIPE)
-        out, err = cmd.communicate()
-        err = err.decode('utf-8')
-        if 'Bad configuration option' in err or 'Usage:' in err:
-            raise ControlPersistUnsupportedException
-        return True
-
     def exit(self, msg, error=1):
         """Used to safely terminate if sos-collector encounters an error"""
         self.log_error(msg)
@@ -482,7 +452,7 @@ class SoSCollector(SoSComponent):
             'cmdlineopts': self.opts,
             'need_sudo': True if self.opts.ssh_user != 'root' else False,
             'tmpdir': self.tmpdir,
-            'hostlen': len(self.opts.primary) or len(self.hostname),
+            'hostlen': max(len(self.opts.primary), len(self.hostname)),
             'policy': self.policy
         }
 
@@ -1047,9 +1017,10 @@ class SoSCollector(SoSComponent):
             self.node_list.append(self.hostname)
         self.reduce_node_list()
         try:
-            self.commons['hostlen'] = len(max(self.node_list, key=len))
+            _node_max = len(max(self.node_list, key=len))
+            self.commons['hostlen'] = max(_node_max, self.commons['hostlen'])
         except (TypeError, ValueError):
-            self.commons['hostlen'] = len(self.opts.primary)
+            pass
 
     def _connect_to_node(self, node):
         """Try to connect to the node, and if we can add to the client list to
@@ -1068,7 +1039,7 @@ class SoSCollector(SoSComponent):
                 client.set_node_manifest(getattr(self.collect_md.nodes,
                                                  node[0]))
             else:
-                client.close_ssh_session()
+                client.disconnect()
         except Exception:
             pass
 
@@ -1077,12 +1048,11 @@ class SoSCollector(SoSComponent):
         provided on the command line
         """
         disclaimer = ("""\
-This utility is used to collect sosreports from multiple \
-nodes simultaneously. It uses OpenSSH's ControlPersist feature \
-to connect to nodes and run commands remotely. If your system \
-installation of OpenSSH is older than 5.6, please upgrade.
+This utility is used to collect sos reports from multiple \
+nodes simultaneously. Remote connections are made and/or maintained \
+to those nodes via well-known transport protocols such as SSH.
 
-An archive of sosreport tarballs collected from the nodes will be \
+An archive of sos report tarballs collected from the nodes will be \
 generated in %s and may be provided to an appropriate support representative.
 
 The generated archive may contain data considered sensitive \
@@ -1230,10 +1200,10 @@ this utility or remote systems that it connects to.
             self.log_error("Error running sosreport: %s" % err)
 
     def close_all_connections(self):
-        """Close all ssh sessions for nodes"""
+        """Close all sessions for nodes"""
         for client in self.client_list:
-            self.log_debug('Closing SSH connection to %s' % client.address)
-            client.close_ssh_session()
+            self.log_debug('Closing connection to %s' % client.address)
+            client.disconnect()
 
     def create_cluster_archive(self):
         """Calls for creation of tar archive then cleans up the temporary
diff --git a/sos/collector/clusters/__init__.py b/sos/collector/clusters/__init__.py
index 2b5d7018..64ac2a44 100644
--- a/sos/collector/clusters/__init__.py
+++ b/sos/collector/clusters/__init__.py
@@ -183,8 +183,8 @@ class Cluster():
         :rtype: ``dict``
         """
         res = self.primary.run_command(cmd, get_pty=True, need_root=need_root)
-        if res['stdout']:
-            res['stdout'] = res['stdout'].replace('Password:', '')
+        if res['output']:
+            res['output'] = res['output'].replace('Password:', '')
         return res
 
     def setup(self):
diff --git a/sos/collector/clusters/jbon.py b/sos/collector/clusters/jbon.py
index 488fbd16..8f083ac6 100644
--- a/sos/collector/clusters/jbon.py
+++ b/sos/collector/clusters/jbon.py
@@ -28,3 +28,5 @@ class jbon(Cluster):
         # This should never be called, but as insurance explicitly never
         # allow this to be enabled via the determine_cluster() path
         return False
+
+# vim: set et ts=4 sw=4 :
diff --git a/sos/collector/clusters/kubernetes.py b/sos/collector/clusters/kubernetes.py
index cdbf8861..99f788dc 100644
--- a/sos/collector/clusters/kubernetes.py
+++ b/sos/collector/clusters/kubernetes.py
@@ -34,7 +34,7 @@ class kubernetes(Cluster):
         if res['status'] == 0:
             nodes = []
             roles = [x for x in self.get_option('role').split(',') if x]
-            for nodeln in res['stdout'].splitlines()[1:]:
+            for nodeln in res['output'].splitlines()[1:]:
                 node = nodeln.split()
                 if not roles:
                     nodes.append(node[0])
@@ -44,3 +44,5 @@ class kubernetes(Cluster):
             return nodes
         else:
             raise Exception('Node enumeration did not return usable output')
+
+# vim: set et ts=4 sw=4 :
diff --git a/sos/collector/clusters/ocp.py b/sos/collector/clusters/ocp.py
index 5479417d..ad97587f 100644
--- a/sos/collector/clusters/ocp.py
+++ b/sos/collector/clusters/ocp.py
@@ -93,7 +93,7 @@ class ocp(Cluster):
         res = self.exec_primary_cmd(self.fmt_oc_cmd(cmd))
         if res['status'] == 0:
             roles = [r for r in self.get_option('role').split(':')]
-            self.node_dict = self._build_dict(res['stdout'].splitlines())
+            self.node_dict = self._build_dict(res['output'].splitlines())
             for node in self.node_dict:
                 if roles:
                     for role in roles:
@@ -103,7 +103,7 @@ class ocp(Cluster):
                     nodes.append(node)
         else:
             msg = "'oc' command failed"
-            if 'Missing or incomplete' in res['stdout']:
+            if 'Missing or incomplete' in res['output']:
                 msg = ("'oc' failed due to missing kubeconfig on primary node."
                        " Specify one via '-c ocp.kubeconfig=<path>'")
             raise Exception(msg)
@@ -168,3 +168,5 @@ class ocp(Cluster):
     def set_node_options(self, node):
         # don't attempt OC API collections on non-primary nodes
         node.plugin_options.append('openshift.no-oc=on')
+
+# vim: set et ts=4 sw=4 :
diff --git a/sos/collector/clusters/ovirt.py b/sos/collector/clusters/ovirt.py
index 079a122e..bd2d0c74 100644
--- a/sos/collector/clusters/ovirt.py
+++ b/sos/collector/clusters/ovirt.py
@@ -98,7 +98,7 @@ class ovirt(Cluster):
             return []
         res = self._run_db_query(self.dbquery)
         if res['status'] == 0:
-            nodes = res['stdout'].splitlines()[2:-1]
+            nodes = res['output'].splitlines()[2:-1]
             return [n.split('(')[0].strip() for n in nodes]
         else:
             raise Exception('database query failed, return code: %s'
@@ -114,7 +114,7 @@ class ovirt(Cluster):
         engconf = '/etc/ovirt-engine/engine.conf.d/10-setup-database.conf'
         res = self.exec_primary_cmd('cat %s' % engconf, need_root=True)
         if res['status'] == 0:
-            config = res['stdout'].splitlines()
+            config = res['output'].splitlines()
             for line in config:
                 try:
                     k = str(line.split('=')[0])
@@ -141,7 +141,7 @@ class ovirt(Cluster):
                '--batch -o postgresql {}'
                ).format(self.conf['ENGINE_DB_PASSWORD'], sos_opt)
         db_sos = self.exec_primary_cmd(cmd, need_root=True)
-        for line in db_sos['stdout'].splitlines():
+        for line in db_sos['output'].splitlines():
             if fnmatch.fnmatch(line, '*sosreport-*tar*'):
                 _pg_dump = line.strip()
                 self.primary.manifest.add_field('postgresql_dump',
@@ -180,5 +180,7 @@ class rhhi_virt(rhv):
         ret = self._run_db_query('SELECT count(server_id) FROM gluster_server')
         if ret['status'] == 0:
             # if there are any entries in this table, RHHI-V is in use
-            return ret['stdout'].splitlines()[2].strip() != '0'
+            return ret['output'].splitlines()[2].strip() != '0'
         return False
+
+# vim: set et ts=4 sw=4 :
diff --git a/sos/collector/clusters/pacemaker.py b/sos/collector/clusters/pacemaker.py
index 034f3f3e..55024314 100644
--- a/sos/collector/clusters/pacemaker.py
+++ b/sos/collector/clusters/pacemaker.py
@@ -27,7 +27,7 @@ class pacemaker(Cluster):
             self.log_error('Cluster status could not be determined. Is the '
                            'cluster running on this node?')
             return []
-        if 'node names do not match' in self.res['stdout']:
+        if 'node names do not match' in self.res['output']:
             self.log_warn('Warning: node name mismatch reported. Attempts to '
                           'connect to some nodes may fail.\n')
         return self.parse_pcs_output()
@@ -41,17 +41,19 @@ class pacemaker(Cluster):
         return nodes
 
     def get_online_nodes(self):
-        for line in self.res['stdout'].splitlines():
+        for line in self.res['output'].splitlines():
             if line.startswith('Online:'):
                 nodes = line.split('[')[1].split(']')[0]
                 return [n for n in nodes.split(' ') if n]
 
     def get_offline_nodes(self):
         offline = []
-        for line in self.res['stdout'].splitlines():
+        for line in self.res['output'].splitlines():
             if line.startswith('Node') and line.endswith('(offline)'):
                 offline.append(line.split()[1].replace(':', ''))
             if line.startswith('OFFLINE:'):
                 nodes = line.split('[')[1].split(']')[0]
                 offline.extend([n for n in nodes.split(' ') if n])
         return offline
+
+# vim: set et ts=4 sw=4 :
diff --git a/sos/collector/clusters/satellite.py b/sos/collector/clusters/satellite.py
index e123c8a3..7c21e553 100644
--- a/sos/collector/clusters/satellite.py
+++ b/sos/collector/clusters/satellite.py
@@ -28,7 +28,7 @@ class satellite(Cluster):
         res = self.exec_primary_cmd(cmd, need_root=True)
         if res['status'] == 0:
             nodes = [
-                n.strip() for n in res['stdout'].splitlines()
+                n.strip() for n in res['output'].splitlines()
                 if 'could not change directory' not in n
             ]
             return nodes
@@ -38,3 +38,5 @@ class satellite(Cluster):
         if node.address == self.primary.address:
             return 'satellite'
         return 'capsule'
+
+# vim: set et ts=4 sw=4 :
diff --git a/sos/collector/sosnode.py b/sos/collector/sosnode.py
index 4b1ee109..f79bd5ff 100644
--- a/sos/collector/sosnode.py
+++ b/sos/collector/sosnode.py
@@ -12,22 +12,16 @@ import fnmatch
 import inspect
 import logging
 import os
-import pexpect
 import re
-import shutil
 
 from distutils.version import LooseVersion
 from pipes import quote
 from sos.policies import load
 from sos.policies.init_systems import InitSystem
-from sos.collector.exceptions import (InvalidPasswordException,
-                                      TimeoutPasswordAuthException,
-                                      PasswordRequestException,
-                                      AuthPermissionDeniedException,
+from sos.collector.transports.control_persist import SSHControlPersist
+from sos.collector.transports.local import LocalTransport
+from sos.collector.exceptions import (CommandTimeoutException,
                                       ConnectionException,
-                                      CommandTimeoutException,
-                                      ConnectionTimeoutException,
-                                      ControlSocketMissingException,
                                       UnsupportedHostException)
 
 
@@ -67,34 +61,25 @@ class SosNode():
             'sos_cmd': commons['sos_cmd']
         }
         self.sos_bin = 'sosreport'
-        filt = ['localhost', '127.0.0.1']
         self.soslog = logging.getLogger('sos')
         self.ui_log = logging.getLogger('sos_ui')
-        self.control_path = ("%s/.sos-collector-%s"
-                             % (self.tmpdir, self.address))
-        self.ssh_cmd = self._create_ssh_command()
-        if self.address not in filt:
-            try:
-                self.connected = self._create_ssh_session()
-            except Exception as err:
-                self.log_error('Unable to open SSH session: %s' % err)
-                raise
-        else:
-            self.connected = True
-            self.local = True
-            self.need_sudo = os.getuid() != 0
+        self._transport = self._load_remote_transport(commons)
+        try:
+            self._transport.connect(self._password)
+        except Exception as err:
+            self.log_error('Unable to open remote session: %s' % err)
+            raise
         # load the host policy now, even if we don't want to load further
         # host information. This is necessary if we're running locally on the
         # cluster primary but do not want a local report as we still need to do
         # package checks in that instance
         self.host = self.determine_host_policy()
-        self.get_hostname()
+        self.hostname = self._transport.hostname
         if self.local and self.opts.no_local:
             load_facts = False
         if self.connected and load_facts:
             if not self.host:
-                self.connected = False
-                self.close_ssh_session()
+                self._transport.disconnect()
                 return None
             if self.local:
                 if self.check_in_container():
@@ -103,11 +88,26 @@ class SosNode():
                 self.create_sos_container()
             self._load_sos_info()
 
-    def _create_ssh_command(self):
-        """Build the complete ssh command for this node"""
-        cmd = "ssh -oControlPath=%s " % self.control_path
-        cmd += "%s@%s " % (self.opts.ssh_user, self.address)
-        return cmd
+    @property
+    def connected(self):
+        if self._transport:
+            return self._transport.connected
+        # if no transport, we're running locally
+        return True
+
+    def disconnect(self):
+        """Wrapper to close the remote session via our transport agent
+        """
+        self._transport.disconnect()
+
+    def _load_remote_transport(self, commons):
+        """Determine the type of remote transport to load for this node, then
+        return an instantiated instance of that transport
+        """
+        if self.address in ['localhost', '127.0.0.1']:
+            self.local = True
+            return LocalTransport(self.address, commons)
+        return SSHControlPersist(self.address, commons)
 
     def _fmt_msg(self, msg):
         return '{:<{}} : {}'.format(self._hostname, self.hostlen + 1, msg)
@@ -135,6 +135,7 @@ class SosNode():
         self.manifest.add_field('policy', self.host.distro)
         self.manifest.add_field('sos_version', self.sos_info['version'])
         self.manifest.add_field('final_sos_command', '')
+        self.manifest.add_field('transport', self._transport.name)
 
     def check_in_container(self):
         """
@@ -160,13 +161,13 @@ class SosNode():
             res = self.run_command(cmd, need_root=True)
             if res['status'] in [0, 125]:
                 if res['status'] == 125:
-                    if 'unable to retrieve auth token' in res['stdout']:
+                    if 'unable to retrieve auth token' in res['output']:
                         self.log_error(
                             "Could not pull image. Provide either a username "
                             "and password or authfile"
                         )
                         raise Exception
-                    elif 'unknown: Not found' in res['stdout']:
+                    elif 'unknown: Not found' in res['output']:
                         self.log_error('Specified image not found on registry')
                         raise Exception
                     # 'name exists' with code 125 means the container was
@@ -181,11 +182,11 @@ class SosNode():
                     return True
                 else:
                     self.log_error("Could not start container after create: %s"
-                                   % ret['stdout'])
+                                   % ret['output'])
                     raise Exception
             else:
                 self.log_error("Could not create container on host: %s"
-                               % res['stdout'])
+                               % res['output'])
                 raise Exception
 
     def get_container_auth(self):
@@ -204,18 +205,11 @@ class SosNode():
 
     def file_exists(self, fname, need_root=False):
         """Checks for the presence of fname on the remote node"""
-        if not self.local:
-            try:
-                res = self.run_command("stat %s" % fname, need_root=need_root)
-                return res['status'] == 0
-            except Exception:
-                return False
-        else:
-            try:
-                os.stat(fname)
-                return True
-            except Exception:
-                return False
+        try:
+            res = self.run_command("stat %s" % fname, need_root=need_root)
+            return res['status'] == 0
+        except Exception:
+            return False
 
     @property
     def _hostname(self):
@@ -223,18 +217,6 @@ class SosNode():
             return self.hostname
         return self.address
 
-    @property
-    def control_socket_exists(self):
-        """Check if the SSH control socket exists
-
-        The control socket is automatically removed by the SSH daemon in the
-        event that the last connection to the node was greater than the timeout
-        set by the ControlPersist option. This can happen for us if we are
-        collecting from a large number of nodes, and the timeout expires before
-        we start collection.
-        """
-        return os.path.exists(self.control_path)
-
     def _sanitize_log_msg(self, msg):
         """Attempts to obfuscate sensitive information in log messages such as
         passwords"""
@@ -264,12 +246,6 @@ class SosNode():
         msg = '[%s:%s] %s' % (self._hostname, caller, msg)
         self.soslog.debug(msg)
 
-    def get_hostname(self):
-        """Get the node's hostname"""
-        sout = self.run_command('hostname')
-        self.hostname = sout['stdout'].strip()
-        self.log_info('Hostname set to %s' % self.hostname)
-
     def _format_cmd(self, cmd):
         """If we need to provide a sudo or root password to a command, then
         here we prefix the command with the correct bits
@@ -280,19 +256,6 @@ class SosNode():
             return "sudo -S %s" % cmd
         return cmd
 
-    def _fmt_output(self, output=None, rc=0):
-        """Formats the returned output from a command into a dict"""
-        if rc == 0:
-            stdout = output
-            stderr = ''
-        else:
-            stdout = ''
-            stderr = output
-        res = {'status': rc,
-               'stdout': stdout,
-               'stderr': stderr}
-        return res
-
     def _load_sos_info(self):
         """Queries the node for information about the installed version of sos
         """
@@ -306,7 +269,7 @@ class SosNode():
             pkgs = self.run_command(self.host.container_version_command,
                                     use_container=True, need_root=True)
             if pkgs['status'] == 0:
-                ver = pkgs['stdout'].strip().split('-')[1]
+                ver = pkgs['output'].strip().split('-')[1]
                 if ver:
                     self.sos_info['version'] = ver
             else:
@@ -321,18 +284,21 @@ class SosNode():
                 self.log_error('sos is not installed on this node')
             self.connected = False
             return False
-        cmd = 'sosreport -l'
+        # sos-4.0 changes the binary
+        if self.check_sos_version('4.0'):
+            self.sos_bin = 'sos report'
+        cmd = "%s -l" % self.sos_bin
         sosinfo = self.run_command(cmd, use_container=True, need_root=True)
         if sosinfo['status'] == 0:
-            self._load_sos_plugins(sosinfo['stdout'])
+            self._load_sos_plugins(sosinfo['output'])
         if self.check_sos_version('3.6'):
             self._load_sos_presets()
 
     def _load_sos_presets(self):
-        cmd = 'sosreport --list-presets'
+        cmd = '%s --list-presets' % self.sos_bin
         res = self.run_command(cmd, use_container=True, need_root=True)
         if res['status'] == 0:
-            for line in res['stdout'].splitlines():
+            for line in res['output'].splitlines():
                 if line.strip().startswith('name:'):
                     pname = line.split('name:')[1].strip()
                     self.sos_info['presets'].append(pname)
@@ -372,21 +338,7 @@ class SosNode():
         """Reads the specified file and returns the contents"""
         try:
             self.log_info("Reading file %s" % to_read)
-            if not self.local:
-                res = self.run_command("cat %s" % to_read, timeout=5)
-                if res['status'] == 0:
-                    return res['stdout']
-                else:
-                    if 'No such file' in res['stdout']:
-                        self.log_debug("File %s does not exist on node"
-                                       % to_read)
-                    else:
-                        self.log_error("Error reading %s: %s" %
-                                       (to_read, res['stdout'].split(':')[1:]))
-                    return ''
-            else:
-                with open(to_read, 'r') as rfile:
-                    return rfile.read()
+            return self._transport.read_file(to_read)
         except Exception as err:
             self.log_error("Exception while reading %s: %s" % (to_read, err))
             return ''
@@ -400,7 +352,8 @@ class SosNode():
                           % self.commons['policy'].distro)
             return self.commons['policy']
         host = load(cache={}, sysroot=self.opts.sysroot, init=InitSystem(),
-                    probe_runtime=True, remote_exec=self.ssh_cmd,
+                    probe_runtime=True,
+                    remote_exec=self._transport.remote_exec,
                     remote_check=self.read_file('/etc/os-release'))
         if host:
             self.log_info("loaded policy %s for host" % host.distro)
@@ -422,7 +375,7 @@ class SosNode():
         return self.host.package_manager.pkg_by_name(pkg) is not None
 
     def run_command(self, cmd, timeout=180, get_pty=False, need_root=False,
-                    force_local=False, use_container=False, env=None):
+                    use_container=False, env=None):
         """Runs a given cmd, either via the SSH session or locally
 
         Arguments:
@@ -433,58 +386,37 @@ class SosNode():
             need_root - if a command requires root privileges, setting this to
                         True tells sos-collector to format the command with
                         sudo or su - as appropriate and to input the password
-            force_local - force a command to run locally. Mainly used for scp.
             use_container - Run this command in a container *IF* the host is
                             containerized
         """
-        if not self.control_socket_exists and not self.local:
-            self.log_debug('Control socket does not exist, attempting to '
-                           're-create')
+        if not self.connected and not self.local:
+            self.log_debug('Node is disconnected, attempting to reconnect')
             try:
-                _sock = self._create_ssh_session()
-                if not _sock:
-                    self.log_debug('Failed to re-create control socket')
-                    raise ControlSocketMissingException
+                reconnected = self._transport.reconnect(self._password)
+                if not reconnected:
+                    self.log_debug('Failed to reconnect to node')
+                    raise ConnectionException
             except Exception as err:
-                self.log_error('Cannot run command: control socket does not '
-                               'exist')
-                self.log_debug("Error while trying to create new SSH control "
-                               "socket: %s" % err)
+                self.log_debug("Error while trying to reconnect: %s" % err)
                 raise
         if use_container and self.host.containerized:
             cmd = self.host.format_container_command(cmd)
         if need_root:
-            get_pty = True
             cmd = self._format_cmd(cmd)
-        self.log_debug('Running command %s' % cmd)
+
         if 'atomic' in cmd:
             get_pty = True
-        if not self.local and not force_local:
-            cmd = "%s %s" % (self.ssh_cmd, quote(cmd))
-        else:
-            if get_pty:
-                cmd = "/bin/bash -c %s" % quote(cmd)
+
+        if get_pty:
+            cmd = "/bin/bash -c %s" % quote(cmd)
+
         if env:
             _cmd_env = self.env_vars
             env = _cmd_env.update(env)
-        res = pexpect.spawn(cmd, encoding='utf-8', env=env)
-        if need_root:
-            if self.need_sudo:
-                res.sendline(self.opts.sudo_pw)
-            if self.opts.become_root:
-                res.sendline(self.opts.root_password)
-        output = res.expect([pexpect.EOF, pexpect.TIMEOUT],
-                            timeout=timeout)
-        if output == 0:
-            out = res.before
-            res.close()
-            rc = res.exitstatus
-            return {'status': rc, 'stdout': out}
-        elif output == 1:
-            raise CommandTimeoutException(cmd)
+        return self._transport.run_command(cmd, timeout, need_root, env)
 
     def sosreport(self):
-        """Run a sosreport on the node, then collect it"""
+        """Run an sos report on the node, then collect it"""
         try:
             path = self.execute_sos_command()
             if path:
@@ -497,109 +429,6 @@ class SosNode():
             pass
         self.cleanup()
 
-    def _create_ssh_session(self):
-        """
-        Using ControlPersist, create the initial connection to the node.
-
-        This will generate an OpenSSH ControlPersist socket within the tmp
-        directory created or specified for sos-collector to use.
-
-        At most, we will wait 30 seconds for a connection. This involves a 15
-        second wait for the initial connection attempt, and a subsequent 15
-        second wait for a response when we supply a password.
-
-        Since we connect to nodes in parallel (using the --threads value), this
-        means that the time between 'Connecting to nodes...' and 'Beginning
-        collection of sosreports' that users see can be up to an amount of time
-        equal to 30*(num_nodes/threads) seconds.
-
-        Returns
-            True if session is successfully opened, else raise Exception
-        """
-        # Don't use self.ssh_cmd here as we need to add a few additional
-        # parameters to establish the initial connection
-        self.log_info('Opening SSH session to create control socket')
-        connected = False
-        ssh_key = ''
-        ssh_port = ''
-        if self.opts.ssh_port != 22:
-            ssh_port = "-p%s " % self.opts.ssh_port
-        if self.opts.ssh_key:
-            ssh_key = "-i%s" % self.opts.ssh_key
-        cmd = ("ssh %s %s -oControlPersist=600 -oControlMaster=auto "
-               "-oStrictHostKeyChecking=no -oControlPath=%s %s@%s "
-               "\"echo Connected\"" % (ssh_key,
-                                       ssh_port,
-                                       self.control_path,
-                                       self.opts.ssh_user,
-                                       self.address))
-        res = pexpect.spawn(cmd, encoding='utf-8')
-
-        connect_expects = [
-            u'Connected',
-            u'password:',
-            u'.*Permission denied.*',
-            u'.* port .*: No route to host',
-            u'.*Could not resolve hostname.*',
-            pexpect.TIMEOUT
-        ]
-
-        index = res.expect(connect_expects, timeout=15)
-
-        if index == 0:
-            connected = True
-        elif index == 1:
-            if self._password:
-                pass_expects = [
-                    u'Connected',
-                    u'Permission denied, please try again.',
-                    pexpect.TIMEOUT
-                ]
-                res.sendline(self._password)
-                pass_index = res.expect(pass_expects, timeout=15)
-                if pass_index == 0:
-                    connected = True
-                elif pass_index == 1:
-                    # Note that we do not get an exitstatus here, so matching
-                    # this line means an invalid password will be reported for
-                    # both invalid passwords and invalid user names
-                    raise InvalidPasswordException
-                elif pass_index == 2:
-                    raise TimeoutPasswordAuthException
-            else:
-                raise PasswordRequestException
-        elif index == 2:
-            raise AuthPermissionDeniedException
-        elif index == 3:
-            raise ConnectionException(self.address, self.opts.ssh_port)
-        elif index == 4:
-            raise ConnectionException(self.address)
-        elif index == 5:
-            raise ConnectionTimeoutException
-        else:
-            raise Exception("Unknown error, client returned %s" % res.before)
-        if connected:
-            self.log_debug("Successfully created control socket at %s"
-                           % self.control_path)
-            return True
-        return False
-
-    def close_ssh_session(self):
-        """Remove the control socket to effectively terminate the session"""
-        if self.local:
-            return True
-        try:
-            res = self.run_command("rm -f %s" % self.control_path,
-                                   force_local=True)
-            if res['status'] == 0:
-                return True
-            self.log_error("Could not remove ControlPath %s: %s"
-                           % (self.control_path, res['stdout']))
-            return False
-        except Exception as e:
-            self.log_error('Error closing SSH session: %s' % e)
-            return False
-
     def _preset_exists(self, preset):
         """Verifies if the given preset exists on the node"""
         return preset in self.sos_info['presets']
@@ -646,8 +475,8 @@ class SosNode():
         self.cluster = cluster
 
     def update_cmd_from_cluster(self):
-        """This is used to modify the sosreport command run on the nodes.
-        By default, sosreport is run without any options, using this will
+        """This is used to modify the sos report command run on the nodes.
+        By default, sos report is run without any options, using this will
         allow the profile to specify what plugins to run or not and what
         options to use.
 
@@ -727,10 +556,6 @@ class SosNode():
             if self.opts.since:
                 sos_opts.append('--since=%s' % quote(self.opts.since))
 
-        # sos-4.0 changes the binary
-        if self.check_sos_version('4.0'):
-            self.sos_bin = 'sos report'
-
         if self.check_sos_version('4.1'):
             if self.opts.skip_commands:
                 sos_opts.append(
@@ -811,7 +636,7 @@ class SosNode():
         self.manifest.add_field('final_sos_command', self.sos_cmd)
 
     def determine_sos_label(self):
-        """Determine what, if any, label should be added to the sosreport"""
+        """Determine what, if any, label should be added to the sos report"""
         label = ''
         label += self.cluster.get_node_label(self)
 
@@ -822,7 +647,7 @@ class SosNode():
         if not label:
             return None
 
-        self.log_debug('Label for sosreport set to %s' % label)
+        self.log_debug('Label for sos report set to %s' % label)
         if self.check_sos_version('3.6'):
             lcmd = '--label'
         else:
@@ -844,20 +669,20 @@ class SosNode():
 
     def determine_sos_error(self, rc, stdout):
         if rc == -1:
-            return 'sosreport process received SIGKILL on node'
+            return 'sos report process received SIGKILL on node'
         if rc == 1:
             if 'sudo' in stdout:
                 return 'sudo attempt failed'
         if rc == 127:
-            return 'sosreport terminated unexpectedly. Check disk space'
+            return 'sos report terminated unexpectedly. Check disk space'
         if len(stdout) > 0:
             return stdout.split('\n')[0:1]
         else:
             return 'sos exited with code %s' % rc
 
     def execute_sos_command(self):
-        """Run sosreport and capture the resulting file path"""
-        self.ui_msg('Generating sosreport...')
+        """Run sos report and capture the resulting file path"""
+        self.ui_msg('Generating sos report...')
         try:
             path = False
             checksum = False
@@ -867,7 +692,7 @@ class SosNode():
                                    use_container=True,
                                    env=self.sos_env_vars)
             if res['status'] == 0:
-                for line in res['stdout'].splitlines():
+                for line in res['output'].splitlines():
                     if fnmatch.fnmatch(line, '*sosreport-*tar*'):
                         path = line.strip()
                     if line.startswith((" sha256\t", " md5\t")):
@@ -884,44 +709,31 @@ class SosNode():
                     else:
                         self.manifest.add_field('checksum_type', 'unknown')
             else:
-                err = self.determine_sos_error(res['status'], res['stdout'])
-                self.log_debug("Error running sosreport. rc = %s msg = %s"
-                               % (res['status'], res['stdout'] or
-                                  res['stderr']))
+                err = self.determine_sos_error(res['status'], res['output'])
+                self.log_debug("Error running sos report. rc = %s msg = %s"
+                               % (res['status'], res['output']))
                 raise Exception(err)
             return path
         except CommandTimeoutException:
             self.log_error('Timeout exceeded')
             raise
         except Exception as e:
-            self.log_error('Error running sosreport: %s' % e)
+            self.log_error('Error running sos report: %s' % e)
             raise
 
     def retrieve_file(self, path):
         """Copies the specified file from the host to our temp dir"""
         destdir = self.tmpdir + '/'
-        dest = destdir + path.split('/')[-1]
+        dest = os.path.join(destdir, path.split('/')[-1])
         try:
-            if not self.local:
-                if self.file_exists(path):
-                    self.log_info("Copying remote %s to local %s" %
-                                  (path, destdir))
-                    cmd = "/usr/bin/scp -oControlPath=%s %s@%s:%s %s" % (
-                        self.control_path,
-                        self.opts.ssh_user,
-                        self.address,
-                        path,
-                        destdir
-                    )
-                    res = self.run_command(cmd, force_local=True)
-                    return res['status'] == 0
-                else:
-                    self.log_debug("Attempting to copy remote file %s, but it "
-                                   "does not exist on filesystem" % path)
-                    return False
+            if self.file_exists(path):
+                self.log_info("Copying remote %s to local %s" %
+                              (path, destdir))
+                self._transport.retrieve_file(path, dest)
             else:
-                self.log_debug("Moving %s to %s" % (path, destdir))
-                shutil.copy(path, dest)
+                self.log_debug("Attempting to copy remote file %s, but it "
+                               "does not exist on filesystem" % path)
+                return False
             return True
         except Exception as err:
             self.log_debug("Failed to retrieve %s: %s" % (path, err))
@@ -933,7 +745,7 @@ class SosNode():
         """
         path = ''.join(path.split())
         try:
-            if len(path) <= 2:  # ensure we have a non '/' path
+            if len(path.split('/')) <= 2:  # ensure we have a non '/' path
                 self.log_debug("Refusing to remove path %s: appears to be "
                                "incorrect and possibly dangerous" % path)
                 return False
@@ -959,14 +771,14 @@ class SosNode():
                 except Exception:
                     self.log_error('Failed to make archive readable')
                     return False
-            self.soslog.info('Retrieving sosreport from %s' % self.address)
-            self.ui_msg('Retrieving sosreport...')
+            self.soslog.info('Retrieving sos report from %s' % self.address)
+            self.ui_msg('Retrieving sos report...')
             ret = self.retrieve_file(self.sos_path)
             if ret:
-                self.ui_msg('Successfully collected sosreport')
+                self.ui_msg('Successfully collected sos report')
                 self.file_list.append(self.sos_path.split('/')[-1])
             else:
-                self.log_error('Failed to retrieve sosreport')
+                self.log_error('Failed to retrieve sos report')
                 raise SystemExit
             return True
         else:
@@ -976,8 +788,8 @@ class SosNode():
             else:
                 e = [x.strip() for x in self.stdout.readlines() if x.strip][-1]
             self.soslog.error(
-                'Failed to run sosreport on %s: %s' % (self.address, e))
-            self.log_error('Failed to run sosreport. %s' % e)
+                'Failed to run sos report on %s: %s' % (self.address, e))
+            self.log_error('Failed to run sos report. %s' % e)
             return False
 
     def remove_sos_archive(self):
@@ -986,20 +798,20 @@ class SosNode():
         if self.sos_path is None:
             return
         if 'sosreport' not in self.sos_path:
-            self.log_debug("Node sosreport path %s looks incorrect. Not "
+            self.log_debug("Node sos report path %s looks incorrect. Not "
                            "attempting to remove path" % self.sos_path)
             return
         removed = self.remove_file(self.sos_path)
         if not removed:
-            self.log_error('Failed to remove sosreport')
+            self.log_error('Failed to remove sos report')
 
     def cleanup(self):
         """Remove the sos archive from the node once we have it locally"""
         self.remove_sos_archive()
         if self.sos_path:
             for ext in ['.sha256', '.md5']:
-                if os.path.isfile(self.sos_path + ext):
-                    self.remove_file(self.sos_path + ext)
+                if self.remove_file(self.sos_path + ext):
+                    break
         cleanup = self.host.set_cleanup_cmd()
         if cleanup:
             self.run_command(cleanup, need_root=True)
@@ -1040,3 +852,5 @@ class SosNode():
             msg = "Exception while making %s readable. Return code was %s"
             self.log_error(msg % (filepath, res['status']))
             raise Exception
+
+# vim: set et ts=4 sw=4 :
diff --git a/sos/collector/transports/__init__.py b/sos/collector/transports/__init__.py
new file mode 100644
index 00000000..5be7dc6d
--- /dev/null
+++ b/sos/collector/transports/__init__.py
@@ -0,0 +1,317 @@
+# Copyright Red Hat 2021, Jake Hunsaker <jhunsake@redhat.com>
+
+# This file is part of the sos project: https://github.com/sosreport/sos
+#
+# This copyrighted material is made available to anyone wishing to use,
+# modify, copy, or redistribute it subject to the terms and conditions of
+# version 2 of the GNU General Public License.
+#
+# See the LICENSE file in the source distribution for further information.
+
+import inspect
+import logging
+import pexpect
+import re
+
+from pipes import quote
+from sos.collector.exceptions import (ConnectionException,
+                                      CommandTimeoutException)
+
+
+class RemoteTransport():
+    """The base class used for defining supported remote transports to connect
+    to remote nodes in conjunction with `sos collect`.
+
+    This abstraction is used to manage the backend connections to nodes so that
+    SoSNode() objects can be leveraged generically to connect to nodes, inspect
+    those nodes, and run commands on them.
+    """
+
+    name = 'undefined'
+
+    def __init__(self, address, commons):
+        self.address = address
+        self.opts = commons['cmdlineopts']
+        self.tmpdir = commons['tmpdir']
+        self.need_sudo = commons['need_sudo']
+        self._hostname = None
+        self.soslog = logging.getLogger('sos')
+        self.ui_log = logging.getLogger('sos_ui')
+
+    def _sanitize_log_msg(self, msg):
+        """Attempts to obfuscate sensitive information in log messages such as
+        passwords"""
+        reg = r'(?P<var>(pass|key|secret|PASS|KEY|SECRET).*?=)(?P<value>.*?\s)'
+        return re.sub(reg, r'\g<var>****** ', msg)
+
+    def log_info(self, msg):
+        """Used to print and log info messages"""
+        caller = inspect.stack()[1][3]
+        lmsg = '[%s:%s] %s' % (self.hostname, caller, msg)
+        self.soslog.info(lmsg)
+
+    def log_error(self, msg):
+        """Used to print and log error messages"""
+        caller = inspect.stack()[1][3]
+        lmsg = '[%s:%s] %s' % (self.hostname, caller, msg)
+        self.soslog.error(lmsg)
+
+    def log_debug(self, msg):
+        """Used to print and log debug messages"""
+        msg = self._sanitize_log_msg(msg)
+        caller = inspect.stack()[1][3]
+        msg = '[%s:%s] %s' % (self.hostname, caller, msg)
+        self.soslog.debug(msg)
+
+    @property
+    def hostname(self):
+        if self._hostname and 'localhost' not in self._hostname:
+            return self._hostname
+        return self.address
+
+    @property
+    def connected(self):
+        """Is the transport __currently__ connected to the node, or otherwise
+        capable of seamlessly running a command or similar on the node?
+        """
+        return False
+
+    @property
+    def remote_exec(self):
+        """This is the command string needed to leverage the remote transport
+        when executing commands. For example, for an SSH transport this would
+        be the `ssh <options>` string prepended to any command so that the
+        command is executed by the ssh binary.
+
+        This is also referenced by the `remote_exec` parameter for policies
+        when loading a policy for a remote node
+        """
+        return None
+
+    def connect(self, password):
+        """Perform the connection steps in order to ensure that we are able to
+        connect to the node for all future operations. Note that this should
+        not provide an interactive shell at this time.
+        """
+        if self._connect(password):
+            if not self._hostname:
+                self._get_hostname()
+            return True
+        return False
+
+    def _connect(self, password):
+        """Actually perform the connection requirements. Should be overridden
+        by specific transports that subclass RemoteTransport
+        """
+        raise NotImplementedError("Transport %s does not define connect"
+                                  % self.name)
+
+    def reconnect(self, password):
+        """Attempts to reconnect to the node using the standard connect()
+        but does not do so indefinitely. This imposes a strict number of retry
+        attempts before failing out
+        """
+        attempts = 1
+        last_err = 'unknown'
+        while attempts < 5:
+            self.log_debug("Attempting reconnect (#%s) to node" % attempts)
+            try:
+                if self.connect(password):
+                    return True
+            except Exception as err:
+                self.log_debug("Attempt #%s exception: %s" % (attempts, err))
+                last_err = err
+            attempts += 1
+        self.log_error("Unable to reconnect to node after 5 attempts, "
+                       "aborting.")
+        raise ConnectionException("last exception from transport: %s"
+                                  % last_err)
+
+    def disconnect(self):
+        """Perform whatever steps are necessary, if any, to terminate any
+        connection to the node
+        """
+        try:
+            if self._disconnect():
+                self.log_debug("Successfully disconnected from node")
+            else:
+                self.log_error("Unable to successfully disconnect, see log for"
+                               " more details")
+        except Exception as err:
+            self.log_error("Failed to disconnect: %s" % err)
+
+    def _disconnect(self):
+        raise NotImplementedError("Transport %s does not define disconnect"
+                                  % self.name)
+
+    def run_command(self, cmd, timeout=180, need_root=False, env=None):
+        """Run a command on the node, returning its output and exit code.
+        This should return the exit code of the command being executed, not the
+        exit code of whatever mechanism the transport uses to execute that
+        command
+
+        :param cmd:         The command to run
+        :type cmd:          ``str``
+
+        :param timeout:     The maximum time in seconds to allow the cmd to run
+        :type timeout:      ``int``
+
+        :param get_pty:     Does ``cmd`` require a pty?
+        :type get_pty:      ``bool``
+
+        :param need_root:   Does ``cmd`` require root privileges?
+        :type neeed_root:   ``bool``
+
+        :param env:         Specify env vars to be passed to the ``cmd``
+        :type env:          ``dict``
+
+        :returns:           Output of ``cmd`` and the exit code
+        :rtype:             ``dict`` with keys ``output`` and ``status``
+        """
+        self.log_debug('Running command %s' % cmd)
+        # currently we only use/support the use of pexpect for handling the
+        # execution of these commands, as opposed to directly invoking
+        # subprocess.Popen() in conjunction with tools like sshpass.
+        # If that changes in the future, we'll add decision making logic here
+        # to route to the appropriate handler, but for now we just go straight
+        # to using pexpect
+        return self._run_command_with_pexpect(cmd, timeout, need_root, env)
+
+    def _format_cmd_for_exec(self, cmd):
+        """Format the command in the way needed for the remote transport to
+        successfully execute it as one would when manually executing it
+
+        :param cmd:     The command being executed, as formatted by SoSNode
+        :type cmd:      ``str``
+
+
+        :returns:       The command further formatted as needed by this
+                        transport
+        :rtype:         ``str``
+        """
+        cmd = "%s %s" % (self.remote_exec, quote(cmd))
+        cmd = cmd.lstrip()
+        return cmd
+
+    def _run_command_with_pexpect(self, cmd, timeout, need_root, env):
+        """Execute the command using pexpect, which allows us to more easily
+        handle prompts and timeouts compared to directly leveraging the
+        subprocess.Popen() method.
+
+        :param cmd:     The command to execute. This will be automatically
+                        formatted to use the transport.
+        :type cmd:      ``str``
+
+        :param timeout: The maximum time in seconds to run ``cmd``
+        :type timeout:  ``int``
+
+        :param need_root:   Does ``cmd`` need to run as root or with sudo?
+        :type need_root:    ``bool``
+
+        :param env:     Any env vars that ``cmd`` should be run with
+        :type env:      ``dict``
+        """
+        cmd = self._format_cmd_for_exec(cmd)
+        result = pexpect.spawn(cmd, encoding='utf-8', env=env)
+
+        _expects = [pexpect.EOF, pexpect.TIMEOUT]
+        if need_root and self.opts.ssh_user != 'root':
+            _expects.extend([
+                '\\[sudo\\] password for .*:',
+                'Password:'
+            ])
+
+        index = result.expect(_expects, timeout=timeout)
+
+        if index in [2, 3]:
+            self._send_pexpect_password(index, result)
+            index = result.expect(_expects, timeout=timeout)
+
+        if index == 0:
+            out = result.before
+            result.close()
+            return {'status': result.exitstatus, 'output': out}
+        elif index == 1:
+            raise CommandTimeoutException(cmd)
+
+    def _send_pexpect_password(self, index, result):
+        """Handle password prompts for sudo and su usage for non-root SSH users
+
+        :param index:       The index pexpect.spawn returned to match against
+                            either a sudo or su prompt
+        :type index:        ``int``
+
+        :param result:      The spawn running the command
+        :type result:       ``pexpect.spawn``
+        """
+        if index == 2:
+            if not self.opts.sudo_pw and not self.opt.nopasswd_sudo:
+                msg = ("Unable to run command: sudo password "
+                       "required but not provided")
+                self.log_error(msg)
+                raise Exception(msg)
+            result.sendline(self.opts.sudo_pw)
+        elif index == 3:
+            if not self.opts.root_password:
+                msg = ("Unable to run command as root: no root password given")
+                self.log_error(msg)
+                raise Exception(msg)
+            result.sendline(self.opts.root_password)
+
+    def _get_hostname(self):
+        """Determine the hostname of the node and set that for future reference
+        and logging
+
+        :returns:   The hostname of the system, per the `hostname` command
+        :rtype:     ``str``
+        """
+        _out = self.run_command('hostname')
+        if _out['status'] == 0:
+            self._hostname = _out['output'].strip()
+        self.log_info("Hostname set to %s" % self._hostname)
+        return self._hostname
+
+    def retrieve_file(self, fname, dest):
+        """Copy a remote file, fname, to dest on the local node
+
+        :param fname:   The name of the file to retrieve
+        :type fname:    ``str``
+
+        :param dest:    Where to save the file to locally
+        :type dest:     ``str``
+
+        :returns:   True if file was successfully copied from remote, or False
+        :rtype:     ``bool``
+        """
+        return self._retrieve_file(fname, dest)
+
+    def _retrieve_file(self, fname, dest):
+        raise NotImplementedError("Transport %s does not support file copying"
+                                  % self.name)
+
+    def read_file(self, fname):
+        """Read the given file fname and return its contents
+
+        :param fname:   The name of the file to read
+        :type fname:    ``str``
+
+        :returns:   The content of the file
+        :rtype:     ``str``
+        """
+        self.log_debug("Reading file %s" % fname)
+        return self._read_file(fname)
+
+    def _read_file(self, fname):
+        res = self.run_command("cat %s" % fname, timeout=5)
+        if res['status'] == 0:
+            return res['output']
+        else:
+            if 'No such file' in res['output']:
+                self.log_debug("File %s does not exist on node"
+                               % fname)
+            else:
+                self.log_error("Error reading %s: %s" %
+                               (fname, res['output'].split(':')[1:]))
+            return ''
+
+# vim: set et ts=4 sw=4 :
diff --git a/sos/collector/transports/control_persist.py b/sos/collector/transports/control_persist.py
new file mode 100644
index 00000000..3e848b41
--- /dev/null
+++ b/sos/collector/transports/control_persist.py
@@ -0,0 +1,199 @@
+# Copyright Red Hat 2021, Jake Hunsaker <jhunsake@redhat.com>
+
+# This file is part of the sos project: https://github.com/sosreport/sos
+#
+# This copyrighted material is made available to anyone wishing to use,
+# modify, copy, or redistribute it subject to the terms and conditions of
+# version 2 of the GNU General Public License.
+#
+# See the LICENSE file in the source distribution for further information.
+
+
+import os
+import pexpect
+import subprocess
+
+from sos.collector.transports import RemoteTransport
+from sos.collector.exceptions import (InvalidPasswordException,
+                                      TimeoutPasswordAuthException,
+                                      PasswordRequestException,
+                                      AuthPermissionDeniedException,
+                                      ConnectionException,
+                                      ConnectionTimeoutException,
+                                      ControlSocketMissingException,
+                                      ControlPersistUnsupportedException)
+from sos.utilities import sos_get_command_output
+
+
+class SSHControlPersist(RemoteTransport):
+    """A transport for collect that leverages OpenSSH's Control Persist
+    functionality which uses control sockets to transparently keep a connection
+    open to the remote host without needing to rebuild the SSH connection for
+    each and every command executed on the node
+    """
+
+    name = 'control_persist'
+
+    def _check_for_control_persist(self):
+        """Checks to see if the local system supported SSH ControlPersist.
+
+        ControlPersist allows OpenSSH to keep a single open connection to a
+        remote host rather than building a new session each time. This is the
+        same feature that Ansible uses in place of paramiko, which we have a
+        need to drop in sos-collector.
+
+        This check relies on feedback from the ssh binary. The command being
+        run should always generate stderr output, but depending on what that
+        output reads we can determine if ControlPersist is supported or not.
+
+        For our purposes, a host that does not support ControlPersist is not
+        able to run sos-collector.
+
+        Returns
+            True if ControlPersist is supported, else raise Exception.
+        """
+        ssh_cmd = ['ssh', '-o', 'ControlPersist']
+        cmd = subprocess.Popen(ssh_cmd, stdout=subprocess.PIPE,
+                               stderr=subprocess.PIPE)
+        out, err = cmd.communicate()
+        err = err.decode('utf-8')
+        if 'Bad configuration option' in err or 'Usage:' in err:
+            raise ControlPersistUnsupportedException
+        return True
+
+    def _connect(self, password=''):
+        """
+        Using ControlPersist, create the initial connection to the node.
+
+        This will generate an OpenSSH ControlPersist socket within the tmp
+        directory created or specified for sos-collector to use.
+
+        At most, we will wait 30 seconds for a connection. This involves a 15
+        second wait for the initial connection attempt, and a subsequent 15
+        second wait for a response when we supply a password.
+
+        Since we connect to nodes in parallel (using the --threads value), this
+        means that the time between 'Connecting to nodes...' and 'Beginning
+        collection of sosreports' that users see can be up to an amount of time
+        equal to 30*(num_nodes/threads) seconds.
+
+        Returns
+            True if session is successfully opened, else raise Exception
+        """
+        try:
+            self._check_for_control_persist()
+        except ControlPersistUnsupportedException:
+            self.log_error("OpenSSH ControlPersist is not locally supported. "
+                           "Please update your OpenSSH installation.")
+            raise
+        self.log_info('Opening SSH session to create control socket')
+        self.control_path = ("%s/.sos-collector-%s" % (self.tmpdir,
+                                                       self.address))
+        self.ssh_cmd = ''
+        connected = False
+        ssh_key = ''
+        ssh_port = ''
+        if self.opts.ssh_port != 22:
+            ssh_port = "-p%s " % self.opts.ssh_port
+        if self.opts.ssh_key:
+            ssh_key = "-i%s" % self.opts.ssh_key
+
+        cmd = ("ssh %s %s -oControlPersist=600 -oControlMaster=auto "
+               "-oStrictHostKeyChecking=no -oControlPath=%s %s@%s "
+               "\"echo Connected\"" % (ssh_key,
+                                       ssh_port,
+                                       self.control_path,
+                                       self.opts.ssh_user,
+                                       self.address))
+        res = pexpect.spawn(cmd, encoding='utf-8')
+
+        connect_expects = [
+            u'Connected',
+            u'password:',
+            u'.*Permission denied.*',
+            u'.* port .*: No route to host',
+            u'.*Could not resolve hostname.*',
+            pexpect.TIMEOUT
+        ]
+
+        index = res.expect(connect_expects, timeout=15)
+
+        if index == 0:
+            connected = True
+        elif index == 1:
+            if password:
+                pass_expects = [
+                    u'Connected',
+                    u'Permission denied, please try again.',
+                    pexpect.TIMEOUT
+                ]
+                res.sendline(password)
+                pass_index = res.expect(pass_expects, timeout=15)
+                if pass_index == 0:
+                    connected = True
+                elif pass_index == 1:
+                    # Note that we do not get an exitstatus here, so matching
+                    # this line means an invalid password will be reported for
+                    # both invalid passwords and invalid user names
+                    raise InvalidPasswordException
+                elif pass_index == 2:
+                    raise TimeoutPasswordAuthException
+            else:
+                raise PasswordRequestException
+        elif index == 2:
+            raise AuthPermissionDeniedException
+        elif index == 3:
+            raise ConnectionException(self.address, self.opts.ssh_port)
+        elif index == 4:
+            raise ConnectionException(self.address)
+        elif index == 5:
+            raise ConnectionTimeoutException
+        else:
+            raise Exception("Unknown error, client returned %s" % res.before)
+        if connected:
+            if not os.path.exists(self.control_path):
+                raise ControlSocketMissingException
+            self.log_debug("Successfully created control socket at %s"
+                           % self.control_path)
+            return True
+        return False
+
+    def _disconnect(self):
+        if os.path.exists(self.control_path):
+            try:
+                os.remove(self.control_path)
+                return True
+            except Exception as err:
+                self.log_debug("Could not disconnect properly: %s" % err)
+                return False
+        self.log_debug("Control socket not present when attempting to "
+                       "terminate session")
+
+    @property
+    def connected(self):
+        """Check if the SSH control socket exists
+
+        The control socket is automatically removed by the SSH daemon in the
+        event that the last connection to the node was greater than the timeout
+        set by the ControlPersist option. This can happen for us if we are
+        collecting from a large number of nodes, and the timeout expires before
+        we start collection.
+        """
+        return os.path.exists(self.control_path)
+
+    @property
+    def remote_exec(self):
+        if not self.ssh_cmd:
+            self.ssh_cmd = "ssh -oControlPath=%s %s@%s" % (
+                self.control_path, self.opts.ssh_user, self.address
+            )
+        return self.ssh_cmd
+
+    def _retrieve_file(self, fname, dest):
+        cmd = "/usr/bin/scp -oControlPath=%s %s@%s:%s %s" % (
+            self.control_path, self.opts.ssh_user, self.address, fname, dest
+        )
+        res = sos_get_command_output(cmd)
+        return res['status'] == 0
+
+# vim: set et ts=4 sw=4 :
diff --git a/sos/collector/transports/local.py b/sos/collector/transports/local.py
new file mode 100644
index 00000000..a4897f19
--- /dev/null
+++ b/sos/collector/transports/local.py
@@ -0,0 +1,49 @@
+# Copyright Red Hat 2021, Jake Hunsaker <jhunsake@redhat.com>
+
+# This file is part of the sos project: https://github.com/sosreport/sos
+#
+# This copyrighted material is made available to anyone wishing to use,
+# modify, copy, or redistribute it subject to the terms and conditions of
+# version 2 of the GNU General Public License.
+#
+# See the LICENSE file in the source distribution for further information.
+
+import os
+import shutil
+
+from sos.collector.transports import RemoteTransport
+
+
+class LocalTransport(RemoteTransport):
+    """A 'transport' to represent a local node. This allows us to more easily
+    extend SoSNode() without having a ton of 'if local' or similar checks in
+    more places than we actually need them
+    """
+
+    name = 'local_node'
+
+    def _connect(self, password):
+        return True
+
+    def _disconnect(self):
+        return True
+
+    @property
+    def connected(self):
+        return True
+
+    def _retrieve_file(self, fname, dest):
+        self.log_debug("Moving %s to %s" % (fname, dest))
+        shutil.copy(fname, dest)
+
+    def _format_cmd_for_exec(self, cmd):
+        return cmd
+
+    def _read_file(self, fname):
+        if os.path.exists(fname):
+            with open(fname, 'r') as rfile:
+                return rfile.read()
+        self.log_debug("No such file: %s" % fname)
+        return ''
+
+# vim: set et ts=4 sw=4 :