Add ssh -L / ssh.connect_remote() workaround when AllowTcpForwarding is disabled

CHANGELOG.md

@@ -83,6 +83,7 @@ The table below shows which release corresponds to each branch, and what date th
 - [#2527][2527] Allow setting debugger path via `context.gdb_binary`
 - [#2530][2530] Do NOT error when passing directory arguments in `checksec` commandline tool.
 - [#2529][2529] Add LoongArch64 support
+- [#2538][2538] Add `ssh -L` / `ssh.connect_remote()` workaround when `AllowTcpForwarding` is disabled
 
 [2519]: https://github.com/Gallopsled/pwntools/pull/2519
 [2507]: https://github.com/Gallopsled/pwntools/pull/2507
@@ -93,6 +94,7 @@ The table below shows which release corresponds to each branch, and what date th
 [2527]: https://github.com/Gallopsled/pwntools/pull/2527
 [2530]: https://github.com/Gallopsled/pwntools/pull/2530
 [2529]: https://github.com/Gallopsled/pwntools/pull/2529
+[2538]: https://github.com/Gallopsled/pwntools/pull/2538
 
 ## 4.15.0 (`beta`)
 

pwnlib/tubes/ssh.py

@@ -441,14 +441,34 @@ def __init__(self, parent, host, port, *a, **kw):
         # keep the parent from being garbage collected in some cases
         self.parent = parent
 
+        # keep reference to tunnel process to avoid garbage collection
+        self.tunnel = None
+
         self.host  = parent.host
         self.rhost = host
         self.rport = port
 
+        import paramiko.ssh_exception
         msg = 'Connecting to %s:%d via SSH to %s' % (self.rhost, self.rport, self.host)
         with self.waitfor(msg) as h:
             try:
                 self.sock = parent.transport.open_channel('direct-tcpip', (host, port), ('127.0.0.1', 0))
+            except paramiko.ssh_exception.ChannelException as e:
+                # Workaround AllowTcpForwarding no in sshd_config
+                if e.args != (1, 'Administratively prohibited'):
+                    self.exception(str(e))
+                    raise e
+
+                self.debug('Failed to open channel, trying to connect to remote port manually using netcat.')
+                if parent.which('nc'):
+                    ncat = 'nc'
+                elif parent.which('ncat'):
+                    ncat = 'ncat'
+                else:
+                    self.exception('Could not find ncat or nc on remote. Cannot connect to remote port.')
+                    raise
+                self.tunnel = parent.process([ncat, host, str(port)])
+                self.sock = self.tunnel.sock
             except Exception as e:
                 self.exception(str(e))
                 raise
@@ -949,6 +969,7 @@ def process(self, argv=None, executable=None, tty=True, cwd=None, env=None, igno
             self.upload_data(script, tmpfile)
             return tmpfile
 
+        executable = executable or argv[0]
         if self.isEnabledFor(logging.DEBUG):
             execve_repr = "execve(%r, %s, %s)" % (executable,
                                                   argv,