Add support for forwarding incoming sessions over an SSH tunnel

This commit makes it easier for applications to accept incoming tunneled
session, connection, or TUN/TAP requests on an SSHServerConnection and
forward them over an upstream SSHClientConnection. The methods on
SSHServer to accept these requests can now return a SSHClientConnection
object to forward the traffic over, instead of having to accept the
request, open a corresponding upstream session, and then relay data
between the two SSH sessions.

Author: ronf

asyncssh/connection.py

@@ -6305,6 +6305,9 @@ def _process_session_open(self, packet: SSHPacket) -> \
             if not result:
                 raise ChannelOpenError(OPEN_CONNECT_FAILED, 'Session refused')
 
+            if isinstance(result, SSHClientConnection):
+                result = self.forward_tunneled_session(result)
+
             if isinstance(result, tuple):
                 chan, result = result
             else:
@@ -6360,6 +6363,10 @@ def _process_direct_tcpip_open(self, packet: SSHPacket) -> \
         if result is True:
             result = cast(SSHTCPSession[bytes],
                           self.forward_connection(dest_host, dest_port))
+        elif isinstance(result, SSHClientConnection):
+            result = cast(Awaitable[SSHTCPSession[bytes]],
+                          self.forward_tunneled_connection(
+                              result, dest_host, dest_port))
 
         if isinstance(result, tuple):
             chan, result = result
@@ -6506,6 +6513,10 @@ def _process_direct_streamlocal_at_openssh_dot_com_open(
         if result is True:
             result = cast(SSHUNIXSession[bytes],
                           self.forward_unix_connection(dest_path))
+        elif isinstance(result, SSHClientConnection):
+            result = cast(Awaitable[SSHUNIXSession[bytes]],
+                          self.forward_tunneled_unix_connection(
+                              result, dest_path))
 
         if isinstance(result, tuple):
             chan, result = result
@@ -6621,10 +6632,14 @@ def _process_tun_at_openssh_dot_com_open(
             result = False
 
         if not result:
-            raise ChannelOpenError(OPEN_CONNECT_FAILED, 'Connection refused')
+            raise ChannelOpenError(OPEN_CONNECT_FAILED,
+                                   'TUN/TAP request refused')
 
         if result is True:
             result = cast(SSHTunTapSession, self.forward_tuntap(mode, unit))
+        elif isinstance(result, SSHClientConnection):
+            result = cast(Awaitable[SSHTunTapSession],
+                          self.forward_tunneled_tuntap(result, mode, unit))
 
         if isinstance(result, tuple):
             chan, result = result
@@ -7179,6 +7194,76 @@ async def open_agent_connection(self) -> \
 
         return SSHReader[bytes](session, chan), SSHWriter[bytes](session, chan)
 
+    async def forward_tunneled_session(
+            self, conn: SSHClientConnection) -> SSHServerProcess:
+        """Forward a tunneled session between SSH connections"""
+
+        async def process_factory(process: SSHServerProcess) -> None:
+            """Return an upstream process used to forward the session"""
+
+            encoding, errors = process.channel.get_encoding()
+
+            upstream_process: SSHClientProcess = await conn.create_process(
+                command=process.command, subsystem=process.subsystem,
+                env=process.env, term_type=process.term_type,
+                term_size=process.term_size, term_modes=process.term_modes,
+                encoding=encoding, errors=errors, stdin=process.stdin,
+                stdout=process.stdout, stderr=process.stderr)
+
+            await upstream_process.wait_closed()
+
+        self.logger.info('  Forwarding session via SSH tunnel')
+
+        return SSHServerProcess(process_factory, None, MIN_SFTP_VERSION, False)
+
+    async def forward_tunneled_connection(
+            self, conn: SSHClientConnection,
+            dest_host: str, dest_port: int) -> SSHForwarder:
+        """Forward a tunneled TCP connection between SSH connections"""
+
+        _, peer = await conn.create_connection(
+            cast(SSHTCPSessionFactory[bytes], SSHForwarder),
+            dest_host, dest_port)
+
+        self.logger.info('  Forwarding TCP connection to %s via SSH tunnel',
+                         (dest_host, dest_port))
+
+        return SSHForwarder(cast(SSHForwarder, peer))
+
+    async def forward_tunneled_unix_connection(
+            self, conn: SSHClientConnection,
+            dest_path: str) -> SSHForwarder:
+        """Forward a tunneled UNIX connection between SSH connections"""
+
+        _, peer = await conn.create_unix_connection(
+            cast(SSHUNIXSessionFactory[bytes], SSHForwarder), dest_path)
+
+        self.logger.info('  Forwarding UNIX connection to %s via SSH tunnel',
+                         dest_path)
+
+        return SSHForwarder(cast(SSHForwarder, peer))
+
+    async def forward_tunneled_tuntap(
+            self, conn: SSHClientConnection,
+            mode: int, unit: Optional[int]) -> SSHForwarder:
+        """Forward a TUN/TAP connection between SSH connections"""
+
+        if mode == SSH_TUN_MODE_POINTTOPOINT:
+            create_func = conn.create_tun
+            layer = 3
+        else:
+            create_func = conn.create_tap
+            layer = 2
+
+        transport, peer = await create_func(
+            cast(SSHTunTapSessionFactory, SSHForwarder), unit)
+        interface = transport.get_extra_info('interface')
+
+        self.logger.info('  Forwarding layer %d traffic to %s via SSH tunnel',
+                         layer, interface)
+
+        return SSHForwarder(cast(SSHForwarder, peer))
+
 
 class SSHConnectionOptions(Options, Generic[_Options]):
     """SSH connection options"""

asyncssh/server.py

@@ -31,30 +31,36 @@
 
 if TYPE_CHECKING:
     # pylint: disable=cyclic-import
-    from .connection import SSHServerConnection, SSHAcceptHandler
+    from .connection import SSHClientConnection, SSHServerConnection
+    from .connection import SSHAcceptHandler
     from .channel import SSHServerChannel, SSHTCPChannel, SSHUNIXChannel
     from .channel import SSHTunTapChannel
     from .session import SSHServerSession, SSHTCPSession, SSHUNIXSession
     from .session import SSHTunTapSession
 
 
-_NewSession = \
-    Union[bool, MaybeAwait['SSHServerSession'], SSHServerSessionFactory,
+_NewSession = Union[
+    bool, 'SSHClientConnection',
+    MaybeAwait['SSHServerSession'], SSHServerSessionFactory,
     Tuple['SSHServerChannel', MaybeAwait['SSHServerSession']],
     Tuple['SSHServerChannel', SSHServerSessionFactory]]
-_NewTCPSession = \
-    Union[bool, MaybeAwait['SSHTCPSession'], SSHSocketSessionFactory,
+_NewTCPSession = Union[
+    bool, 'SSHClientConnection',
+    MaybeAwait['SSHTCPSession'], SSHSocketSessionFactory,
     Tuple['SSHTCPChannel', MaybeAwait['SSHTCPSession']],
     Tuple['SSHTCPChannel', SSHSocketSessionFactory]]
-_NewUNIXSession = \
-    Union[bool, MaybeAwait['SSHUNIXSession'], SSHSocketSessionFactory,
+_NewUNIXSession = Union[
+    bool, 'SSHClientConnection',
+    MaybeAwait['SSHUNIXSession'], SSHSocketSessionFactory,
     Tuple['SSHUNIXChannel', MaybeAwait['SSHUNIXSession']],
     Tuple['SSHUNIXChannel', SSHSocketSessionFactory]]
-_NewTunTapSession = \
-    Union[bool, MaybeAwait['SSHTunTapSession'], SSHSocketSessionFactory,
+_NewTunTapSession = Union[
+    bool, 'SSHClientConnection',
+    MaybeAwait['SSHTunTapSession'], SSHSocketSessionFactory,
     Tuple['SSHTunTapChannel', MaybeAwait['SSHTunTapSession']],
     Tuple['SSHTunTapChannel', SSHSocketSessionFactory]]
-_NewListener = Union[bool, 'SSHAcceptHandler', MaybeAwait[SSHListener]]
+_NewTCPListener = Union[bool, 'SSHAcceptHandler', MaybeAwait[SSHListener]]
+_NewUNIXListener = Union[bool, MaybeAwait[SSHListener]]
 
 
 class SSHServer:
@@ -749,6 +755,11 @@ def connection_requested(self, dest_host: str, dest_port: int,
            :exc:`ChannelOpenError` exception with the reason for
            the failure.
 
+           If the application wishes to tunnel the connection over
+           another SSH connection, this method should return an
+           :class:`SSHClientConnection` connected to the desired
+           tunnel host.
+
            If the application wishes to process the data on the
            connection itself, this method should return either an
            :class:`SSHTCPSession` object which can be used to process the
@@ -802,7 +813,7 @@ def connection_requested(self, dest_host: str, dest_port: int,
         return False # pragma: no cover
 
     def server_requested(self, listen_host: str,
-                         listen_port: int) -> MaybeAwait[_NewListener]:
+                         listen_port: int) -> MaybeAwait[_NewTCPListener]:
         """Handle a request to listen on a TCP/IP address and port
 
            This method is called when a client makes a request to
@@ -864,6 +875,11 @@ def unix_connection_requested(self, dest_path: str) -> _NewUNIXSession:
            :exc:`ChannelOpenError` exception with the reason for
            the failure.
 
+           If the application wishes to tunnel the connection over
+           another SSH connection, this method should return an
+           :class:`SSHClientConnection` connected to the desired
+           tunnel host.
+
            If the application wishes to process the data on the
            connection itself, this method should return either an
            :class:`SSHUNIXSession` object which can be used to process the
@@ -908,7 +924,7 @@ def unix_connection_requested(self, dest_path: str) -> _NewUNIXSession:
         return False # pragma: no cover
 
     def unix_server_requested(self, listen_path: str) -> \
-            MaybeAwait[_NewListener]:
+            MaybeAwait[_NewUNIXListener]:
         """Handle a request to listen on a UNIX domain socket
 
            This method is called when a client makes a request to
@@ -958,14 +974,19 @@ def tun_requested(self, unit: Optional[int]) -> _NewTunTapSession:
            by the server. Applications wishing to accept such tunnels must
            override this method.
 
-           To allow standard path forwarding of data on the connection to the
+           To allow standard forwarding of data on the connection to the
            requested TUN device, this method should return `True`.
 
            To reject this request, this method should return `False`
            to send back a "Connection refused" response or raise an
            :exc:`ChannelOpenError` exception with the reason for
            the failure.
 
+           If the application wishes to tunnel the data over another
+           SSH connection, this method should return an
+           :class:`SSHClientConnection` connected to the desired
+           tunnel host.
+
            If the application wishes to process the data on the
            connection itself, this method should return either an
            :class:`SSHTunTapSession` object which can be used to process the
@@ -1016,14 +1037,19 @@ def tap_requested(self, unit: Optional[int]) -> _NewTunTapSession:
            by the server. Applications wishing to accept such tunnels must
            override this method.
 
-           To allow standard path forwarding of data on the connection to the
-           requested TUN device, this method should return `True`.
+           To allow standard forwarding of data on the connection to the
+           requested TAP device, this method should return `True`.
 
            To reject this request, this method should return `False`
            to send back a "Connection refused" response or raise an
            :exc:`ChannelOpenError` exception with the reason for
            the failure.
 
+           If the application wishes to tunnel the data over another
+           SSH connection, this method should return an
+           :class:`SSHClientConnection` connected to the desired
+           tunnel host.
+
            If the application wishes to process the data on the
            connection itself, this method should return either an
            :class:`SSHTunTapSession` object which can be used to process the

tests/test_forward.py

@@ -229,6 +229,25 @@ async def unix_server_requested(self, listen_path):
             return listen_path != 'fail'
 
 
+class _UpstreamForwardingServer(Server):
+    """Server for testing forwarding between SSH connections"""
+
+    def __init__(self, upstream_conn):
+        super().__init__()
+
+        self._upstream_conn = upstream_conn
+
+    def connection_requested(self, dest_host, dest_port, orig_host, orig_port):
+        """Handle a request to create a new connection"""
+
+        return self._upstream_conn
+
+    def unix_connection_requested(self, dest_path):
+        """Handle a request to create a new UNIX domain connection"""
+
+        return self._upstream_conn
+
+
 class _CheckForwarding(ServerTestCase):
     """Utility functions for AsyncSSH forwarding unit tests"""
 
@@ -296,8 +315,8 @@ class _TestTCPForwarding(_CheckForwarding):
     async def start_server(cls):
         """Start an SSH server which supports TCP connection forwarding"""
 
-        return (await cls.create_server(
-            _TCPConnectionServer, authorized_client_keys='authorized_keys'))
+        return await cls.create_server(
+            _TCPConnectionServer, authorized_client_keys='authorized_keys')
 
     async def _check_connection(self, conn, dest_host='',
                                 dest_port=7, **kwargs):
@@ -876,6 +895,25 @@ async def test_cancel_forward_remote_port_invalid_unicode(self):
 
                 self.assertEqual(pkttype, asyncssh.MSG_REQUEST_FAILURE)
 
+    @asynctest
+    async def test_upstream_forward_local_port(self):
+        """Test upstream forwarding of a local port"""
+
+        def upstream_server():
+            """Return a server capable of forwarding between SSH connections"""
+
+            return _UpstreamForwardingServer(upstream_conn)
+
+        async with self.connect() as upstream_conn:
+            upstream_listener = await self.create_server(upstream_server)
+            upstream_port = upstream_listener.get_port()
+
+            async with self.connect('127.0.0.1', upstream_port) as conn:
+                async with conn.forward_local_port('', 0, '', 7) as listener:
+                    await self._check_local_connection(listener.get_port())
+
+            upstream_listener.close()
+
     @asynctest
     async def test_add_channel_after_close(self):
         """Test opening a connection after a close"""
@@ -963,8 +1001,8 @@ class _TestUNIXForwarding(_CheckForwarding):
     async def start_server(cls):
         """Start an SSH server which supports UNIX connection forwarding"""
 
-        return (await cls.create_server(
-            _UNIXConnectionServer, authorized_client_keys='authorized_keys'))
+        return await cls.create_server(
+            _UNIXConnectionServer, authorized_client_keys='authorized_keys')
 
     async def _check_unix_connection(self, conn, dest_path='/echo', **kwargs):
         """Open a UNIX connection and test if an input line is echoed back"""
@@ -1233,6 +1271,25 @@ async def test_cancel_forward_remote_path_invalid_unicode(self):
 
                 self.assertEqual(pkttype, asyncssh.MSG_REQUEST_FAILURE)
 
+    @asynctest
+    async def test_upstream_forward_local_path(self):
+        """Test upstream forwarding of a local path"""
+
+        def upstream_server():
+            """Return a server capable of forwarding between SSH connections"""
+
+            return _UpstreamForwardingServer(upstream_conn)
+
+        async with self.connect() as upstream_conn:
+            upstream_listener = await self.create_server(upstream_server)
+            upstream_port = upstream_listener.get_port()
+
+            async with self.connect('127.0.0.1', upstream_port) as conn:
+                async with conn.forward_local_path('local', '/echo'):
+                    await self._check_local_unix_connection('local')
+
+            upstream_listener.close()
+
 
 class _TestAsyncUNIXForwarding(_TestUNIXForwarding):
     """Unit tests for AsyncSSH UNIX connection forwarding with async return"""
@@ -1253,8 +1310,8 @@ class _TestSOCKSForwarding(_CheckForwarding):
     async def start_server(cls):
         """Start an SSH server which supports TCP connection forwarding"""
 
-        return (await cls.create_server(
-            _TCPConnectionServer, authorized_client_keys='authorized_keys'))
+        return await cls.create_server(
+            _TCPConnectionServer, authorized_client_keys='authorized_keys')
 
     async def _check_early_error(self, reader, writer, data):
         """Check errors in the initial SOCKS message"""