diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index a4c5d2232e..474ce77cc8 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -29,6 +29,11 @@ Fixed
 
 * Fixed a bug in the example nginx HA template declared headers twice (bug fix) #4966
   Contributed by @punkrokk
+* Fixed a bug in the ``paramiko_ssh`` runner where SSH sockets were not getting cleaned
+  up correctly, specifically when specifying a bastion host / jump box. (bug fix) #4973
+
+  Contributed by Nick Maludy (@nmaludy Encore Technologies)
+  
 
 3.2.0 - April 27, 2020
 ----------------------
diff --git a/st2actions/tests/unit/test_paramiko_ssh.py b/st2actions/tests/unit/test_paramiko_ssh.py
index c6da6e3058..e7e136c792 100644
--- a/st2actions/tests/unit/test_paramiko_ssh.py
+++ b/st2actions/tests/unit/test_paramiko_ssh.py
@@ -814,20 +814,46 @@ def test_socket_closed(self):
 
         # Make sure .close() doesn't actually call anything real
         ssh_client.client = Mock()
-        ssh_client.sftp_client = None
-        ssh_client.bastion_client = None
+        ssh_client.sftp_client = Mock()
+        ssh_client.bastion_client = Mock()
 
         ssh_client.socket = Mock()
+        ssh_client.bastion_socket = Mock()
 
         # Make sure we havent called any close methods at this point
         # TODO: Replace these with .assert_not_called() once it's Python 3.6+ only
-        self.assertEqual(ssh_client.socket.process.kill.call_count, 0)
-        self.assertEqual(ssh_client.socket.process.poll.call_count, 0)
+        self.assertEqual(ssh_client.socket.close.call_count, 0)
+        self.assertEqual(ssh_client.client.close.call_count, 0)
+        self.assertEqual(ssh_client.sftp_client.close.call_count, 0)
+        self.assertEqual(ssh_client.bastion_socket.close.call_count, 0)
+        self.assertEqual(ssh_client.bastion_client.close.call_count, 0)
 
         # Call the function that has changed
         ssh_client.close()
 
-        # Make sure we have called kill and poll
         # TODO: Replace these with .assert_called_once() once it's Python 3.6+ only
-        self.assertEqual(ssh_client.socket.process.kill.call_count, 1)
-        self.assertEqual(ssh_client.socket.process.poll.call_count, 1)
+        self.assertEqual(ssh_client.socket.close.call_count, 1)
+        self.assertEqual(ssh_client.client.close.call_count, 1)
+        self.assertEqual(ssh_client.sftp_client.close.call_count, 1)
+        self.assertEqual(ssh_client.bastion_socket.close.call_count, 1)
+        self.assertEqual(ssh_client.bastion_client.close.call_count, 1)
+
+    @patch.object(ParamikoSSHClient, '_is_key_file_needs_passphrase',
+                  MagicMock(return_value=False))
+    def test_socket_not_closed_if_none(self):
+        conn_params = {'hostname': 'dummy.host.org',
+                       'username': 'ubuntu',
+                       'password': 'pass',
+                       'timeout': '600'}
+        ssh_client = ParamikoSSHClient(**conn_params)
+
+        # Make sure .close() doesn't actually call anything real
+        ssh_client.client = None
+        ssh_client.sftp_client = None
+        ssh_client.bastion_client = None
+
+        ssh_client.socket = None
+        ssh_client.bastion_socket = None
+
+        # Call the function, this should not throw an exception
+        ssh_client.close()
diff --git a/st2common/st2common/runners/paramiko_ssh.py b/st2common/st2common/runners/paramiko_ssh.py
index 2e5154db9f..b96098a89e 100644
--- a/st2common/st2common/runners/paramiko_ssh.py
+++ b/st2common/st2common/runners/paramiko_ssh.py
@@ -452,19 +452,18 @@ def run(self, cmd, timeout=None, quote=False, call_line_handler_func=False):
         return [stdout, stderr, status]
 
     def close(self):
-        self.logger.debug('Closing server connection')
-
-        self.client.close()
-
         if self.socket:
-            self.logger.debug('Closing proxycommand socket connection')
-            # https://github.com/paramiko/paramiko/issues/789  Avoid zombie ssh processes
-            self.socket.process.kill()
-            self.socket.process.poll()
+            self.socket.close()
+
+        if self.client:
+            self.client.close()
 
         if self.sftp_client:
             self.sftp_client.close()
 
+        if self.bastion_socket:
+            self.bastion_socket.close()
+
         if self.bastion_client:
             self.bastion_client.close()