From 0a670b90953b69ebeb05089a7b936a7edb439ac7 Mon Sep 17 00:00:00 2001 From: Alex MKX Date: Tue, 23 Jun 2026 11:12:07 +0300 Subject: [PATCH 01/13] feat(forward): SSHForwardTracker class hierarchy + tracker_factory Replaces the experimental ForwardTracker Protocol (v2.23.0+forward-tracker.1) with a class hierarchy and per-connection factory pattern, per upstream review in ronf/asyncssh#807. - SSHForwardTracker: parent class with no-op default hooks (connection_lost, forward_local_bytes, forward_remote_bytes). - SSHPortForwardTracker(SSHForwardTracker): adds connection_made(forwarder, orig_host, orig_port) for TCP local forwards. - SSHPathForwardTracker(SSHForwardTracker): adds connection_made(forwarder) for UNIX-domain local forwards. - New tracker_factory kw on forward_local_port, forward_local_port_to_path, forward_local_path, forward_local_path_to_port. Factory is invoked once per accepted connection. - Byte hooks fire once per SSH message (observer-only; return values ignored). Buggy hooks/factories are isolated by per-call try/except. - Docs: dedicated 'Forward Tracker Classes' section in api.rst with factory pattern, example, and tracker-aware method list. No changes.rst entry (per maintainer request). - 12 new tests in tests/test_forward.py covering both subclasses, byte hooks, factory exception swallow, and hook exception swallow. Scope: local forwarding only. Remote-forward methods and forward_socks left for a follow-up; the design accommodates them without redesign. --- asyncssh/__init__.py | 6 +- asyncssh/connection.py | 77 +++++++++++++----- asyncssh/forward.py | 177 ++++++++++++++++++++++++++++++++++++++++- asyncssh/listener.py | 18 +++-- docs/api.rst | 64 +++++++++++++++ tests/test_forward.py | 143 +++++++++++++++++++++++++++++++++ 6 files changed, 453 insertions(+), 32 deletions(-) diff --git a/asyncssh/__init__.py b/asyncssh/__init__.py index fb316e0e..b2b49fc7 100644 --- a/asyncssh/__init__.py +++ b/asyncssh/__init__.py @@ -40,7 +40,8 @@ from .config import ConfigParseError -from .forward import SSHForwarder +from .forward import SSHForwarder, SSHForwardTracker +from .forward import SSHPortForwardTracker, SSHPathForwardTracker from .connection import SSHAcceptor, SSHClientConnection, SSHServerConnection from .connection import SSHClientConnectionOptions, SSHServerConnectionOptions @@ -147,7 +148,8 @@ 'SSHAgentKeyPair', 'SSHAuthorizedKeys', 'SSHCertificate', 'SSHClient', 'SSHClientChannel', 'SSHClientConnection', 'SSHClientConnectionOptions', 'SSHClientProcess', 'SSHClientSession', 'SSHCompletedProcess', - 'SSHForwarder', 'SSHKey', 'SSHKeyPair', 'SSHKnownHosts', + 'SSHForwarder', 'SSHForwardTracker', 'SSHPortForwardTracker', + 'SSHPathForwardTracker', 'SSHKey', 'SSHKeyPair', 'SSHKnownHosts', 'SSHLineEditorChannel', 'SSHListener', 'SSHReader', 'SSHServer', 'SSHServerChannel', 'SSHServerConnection', 'SSHServerConnectionOptions', 'SSHServerProcess', diff --git a/asyncssh/connection.py b/asyncssh/connection.py index 89fdb165..64fb61f4 100644 --- a/asyncssh/connection.py +++ b/asyncssh/connection.py @@ -85,7 +85,7 @@ from .encryption import encryption_needs_mac from .encryption import get_encryption_params, get_encryption -from .forward import SSHForwarder +from .forward import SSHForwarder, SSHForwardTrackerFactory from .gss import GSSBase, GSSClient, GSSServer, GSSError @@ -3210,7 +3210,9 @@ async def forward_unix_connection(self, dest_path: str) -> SSHForwarder: async def forward_local_port( self, listen_host: str, listen_port: int, dest_host: str, dest_port: int, - accept_handler: Optional[SSHAcceptHandler] = None) -> SSHListener: + accept_handler: Optional[SSHAcceptHandler] = None, + tracker_factory: + Optional[SSHForwardTrackerFactory] = None) -> SSHListener: """Set up local port forwarding This method is a coroutine which attempts to set up port @@ -3233,11 +3235,18 @@ async def forward_local_port( or not to allow connection forwarding, returning `True` to accept the connection and begin forwarding or `False` to reject and close it. + :param tracker_factory: + An optional callable invoked once per accepted connection + which returns a new :class:`SSHPortForwardTracker` (or + :class:`SSHForwardTracker` subclass) for observing that + connection's lifecycle. `None` (default) disables tracking + with no overhead. :type listen_host: `str` :type listen_port: `int` :type dest_host: `str` :type dest_port: `int` :type accept_handler: `callable` or coroutine + :type tracker_factory: `callable` or `None` :returns: :class:`SSHListener` @@ -3278,10 +3287,9 @@ async def tunnel_connection( (dest_host, dest_port)) try: - listener = await create_tcp_forward_listener(self, self._loop, - tunnel_connection, - listen_host, - listen_port) + listener = await create_tcp_forward_listener( + self, self._loop, tunnel_connection, listen_host, listen_port, + tracker_factory=tracker_factory) except OSError as exc: self.logger.debug1('Failed to create local TCP listener: %s', exc) raise @@ -3297,8 +3305,10 @@ async def tunnel_connection( return listener @async_context_manager - async def forward_local_path(self, listen_path: str, - dest_path: str) -> SSHListener: + async def forward_local_path( + self, listen_path: str, dest_path: str, + tracker_factory: + Optional[SSHForwardTrackerFactory] = None) -> SSHListener: """Set up local UNIX domain socket forwarding This method is a coroutine which attempts to set up UNIX domain @@ -3311,8 +3321,15 @@ async def forward_local_path(self, listen_path: str, The path on the local host to listen on :param dest_path: The path on the remote host to forward the connections to + :param tracker_factory: + An optional callable invoked once per accepted connection + which returns a new :class:`SSHPathForwardTracker` (or + :class:`SSHForwardTracker` subclass) for observing that + connection's lifecycle. `None` (default) disables tracking + with no overhead. :type listen_path: `str` :type dest_path: `str` + :type tracker_factory: `callable` or `None` :returns: :class:`SSHListener` @@ -3332,9 +3349,9 @@ async def tunnel_connection( listen_path, dest_path) try: - listener = await create_unix_forward_listener(self, self._loop, - tunnel_connection, - listen_path) + listener = await create_unix_forward_listener( + self, self._loop, tunnel_connection, listen_path, + tracker_factory=tracker_factory) except OSError as exc: self.logger.debug1('Failed to create local UNIX listener: %s', exc) raise @@ -5304,7 +5321,9 @@ async def open_tap(self, *args: object, **kwargs: object) -> \ @async_context_manager async def forward_local_port_to_path( self, listen_host: str, listen_port: int, dest_path: str, - accept_handler: Optional[SSHAcceptHandler] = None) -> SSHListener: + accept_handler: Optional[SSHAcceptHandler] = None, + tracker_factory: + Optional[SSHForwardTrackerFactory] = None) -> SSHListener: """Set up local TCP port forwarding to a remote UNIX domain socket This method is a coroutine which attempts to set up port @@ -5325,10 +5344,17 @@ async def forward_local_port_to_path( or not to allow connection forwarding, returning `True` to accept the connection and begin forwarding or `False` to reject and close it. + :param tracker_factory: + An optional callable invoked once per accepted connection + which returns a new :class:`SSHPortForwardTracker` (or + :class:`SSHForwardTracker` subclass) for observing that + connection's lifecycle. `None` (default) disables tracking + with no overhead. :type listen_host: `str` :type listen_port: `int` :type dest_path: `str` :type accept_handler: `callable` or coroutine + :type tracker_factory: `callable` or `None` :returns: :class:`SSHListener` @@ -5362,10 +5388,9 @@ async def tunnel_connection( (listen_host, listen_port), dest_path) try: - listener = await create_tcp_forward_listener(self, self._loop, - tunnel_connection, - listen_host, - listen_port) + listener = await create_tcp_forward_listener( + self, self._loop, tunnel_connection, listen_host, listen_port, + tracker_factory=tracker_factory) except OSError as exc: self.logger.debug1('Failed to create local TCP listener: %s', exc) raise @@ -5378,9 +5403,10 @@ async def tunnel_connection( return listener @async_context_manager - async def forward_local_path_to_port(self, listen_path: str, - dest_host: str, - dest_port: int) -> SSHListener: + async def forward_local_path_to_port( + self, listen_path: str, dest_host: str, dest_port: int, + tracker_factory: + Optional[SSHForwardTrackerFactory] = None) -> SSHListener: """Set up local UNIX domain socket forwarding to a remote TCP port This method is a coroutine which attempts to set up UNIX domain @@ -5395,9 +5421,16 @@ async def forward_local_path_to_port(self, listen_path: str, The hostname or address to forward the connections to :param dest_port: The port number to forward the connections to + :param tracker_factory: + An optional callable invoked once per accepted connection + which returns a new :class:`SSHPathForwardTracker` (or + :class:`SSHForwardTracker` subclass) for observing that + connection's lifecycle. `None` (default) disables tracking + with no overhead. :type listen_path: `str` :type dest_host: `str` :type dest_port: `int` + :type tracker_factory: `callable` or `None` :returns: :class:`SSHListener` @@ -5417,9 +5450,9 @@ async def tunnel_connection( listen_path, (dest_host, dest_port)) try: - listener = await create_unix_forward_listener(self, self._loop, - tunnel_connection, - listen_path) + listener = await create_unix_forward_listener( + self, self._loop, tunnel_connection, listen_path, + tracker_factory=tracker_factory) except OSError as exc: self.logger.debug1('Failed to create local UNIX listener: %s', exc) raise diff --git a/asyncssh/forward.py b/asyncssh/forward.py index 8470c000..7428c658 100644 --- a/asyncssh/forward.py +++ b/asyncssh/forward.py @@ -38,6 +38,113 @@ SSHForwarderCoro = Callable[..., Awaitable] +class SSHForwardTracker: + """Base class for observing the lifecycle of a forwarded connection + + A tracker observes a single forwarded connection on a local + listener. A `tracker_factory` passed to one of the + :meth:`forward_local_port() ` + family of methods is called once per accepted connection and must + return a new tracker instance, on which asyncssh then calls the + hooks below for the life of that connection. + + All hooks run inside the asyncio event loop and **must not block** + (no I/O, no sleeps). They are pure observers: return values are + ignored and the forwarded data is never altered. Each hook has a + no-op default, so a subclass need only override the ones it cares + about. Exceptions raised by a hook are caught and discarded, so a + buggy tracker can never break forwarding. + + This base class defines the hooks shared by all forward types. + Use :class:`SSHPortForwardTracker` for TCP local forwards and + :class:`SSHPathForwardTracker` for UNIX domain socket local + forwards; they differ only in the signature of `connection_made`. + + """ + + def connection_lost(self, exc: Optional[Exception]) -> None: + """Called when the forwarded connection has closed + + :param exc: + The exception which caused the connection to close, or + `None` if the connection closed cleanly. + :type exc: :class:`Exception` or `None` + + """ + + def forward_local_bytes(self, data: bytes) -> None: + """Called for data forwarded from the local side into the tunnel + + :param data: + A block of bytes received on the local connection and + about to be sent over the SSH connection. This is called + once per received block, not once per byte. + :type data: `bytes` + + """ + + def forward_remote_bytes(self, data: bytes) -> None: + """Called for data forwarded from the tunnel to the local side + + :param data: + A block of bytes received over the SSH connection and + about to be written to the local connection. This is + called once per received block, not once per byte. + :type data: `bytes` + + """ + + +class SSHPortForwardTracker(SSHForwardTracker): + """Tracker for local TCP port forwards + + Used with + :meth:`forward_local_port() ` + and :meth:`forward_local_port_to_path() + `. + + """ + + def connection_made(self, forwarder: 'SSHForwarder', + orig_host: str, orig_port: int) -> None: + """Called when a new TCP connection is accepted on the listener + + :param forwarder: + The forwarder handling this connection. + :param orig_host: + The originating client host. + :param orig_port: + The originating client port. + :type forwarder: :class:`SSHForwarder` + :type orig_host: `str` + :type orig_port: `int` + + """ + + +class SSHPathForwardTracker(SSHForwardTracker): + """Tracker for local UNIX domain socket forwards + + Used with + :meth:`forward_local_path() ` + and :meth:`forward_local_path_to_port() + `. + + """ + + def connection_made(self, forwarder: 'SSHForwarder') -> None: + """Called when a new UNIX domain connection is accepted + + :param forwarder: + The forwarder handling this connection. + :type forwarder: :class:`SSHForwarder` + + """ + + +SSHForwardTrackerFactory = Callable[[], SSHForwardTracker] + + class SSHForwarder(asyncio.BaseProtocol): """SSH port forwarding connection handler""" @@ -192,10 +299,59 @@ def close(self) -> None: class SSHLocalForwarder(SSHForwarder): """Local forwarding connection handler""" - def __init__(self, conn: 'SSHConnection', coro: SSHForwarderCoro): + def __init__(self, conn: 'SSHConnection', coro: SSHForwarderCoro, + tracker_factory: Optional[SSHForwardTrackerFactory] = None): super().__init__() self._conn = conn self._coro = coro + self._tracker_factory = tracker_factory + self._tracker: Optional[SSHForwardTracker] = None + + def _create_tracker(self) -> None: + """Instantiate this connection's tracker from the factory, if any""" + + if self._tracker_factory is None: + return + + try: + self._tracker = self._tracker_factory() + except Exception: # pylint: disable=broad-exception-caught + # A buggy factory must not break forwarding. + self._tracker = None + + def data_received(self, data: bytes, + datatype: Optional[int] = None) -> None: + """Handle incoming data from the local transport""" + + if self._tracker is not None: + try: + self._tracker.forward_local_bytes(data) + except Exception: # pylint: disable=broad-exception-caught + pass + + super().data_received(data, datatype) + + def write(self, data: bytes) -> None: + """Write tunnel data out to the local transport""" + + if self._tracker is not None: + try: + self._tracker.forward_remote_bytes(data) + except Exception: # pylint: disable=broad-exception-caught + pass + + super().write(data) + + def connection_lost(self, exc: Optional[Exception]) -> None: + """Handle a closed local connection""" + + if self._tracker is not None: + try: + self._tracker.connection_lost(exc) + except Exception: # pylint: disable=broad-exception-caught + pass + + super().connection_lost(exc) async def _forward(self, *args: object) -> None: """Begin local forwarding""" @@ -234,11 +390,21 @@ def connection_made(self, transport: asyncio.BaseTransport) -> None: super().connection_made(transport) + orig_host, orig_port = '', 0 peername = cast(SockAddr, transport.get_extra_info('peername')) if peername: # pragma: no branch orig_host, orig_port = peername[:2] + self._create_tracker() + + if self._tracker is not None: + try: + cast(SSHPortForwardTracker, self._tracker).connection_made( + self, orig_host, orig_port) + except Exception: # pylint: disable=broad-exception-caught + pass + self.forward(orig_host, orig_port) @@ -249,4 +415,13 @@ def connection_made(self, transport: asyncio.BaseTransport) -> None: """Handle a newly opened connection""" super().connection_made(transport) + + self._create_tracker() + + if self._tracker is not None: + try: + cast(SSHPathForwardTracker, self._tracker).connection_made(self) + except Exception: # pylint: disable=broad-exception-caught + pass + self.forward() diff --git a/asyncssh/listener.py b/asyncssh/listener.py index e9cc475b..b77db809 100644 --- a/asyncssh/listener.py +++ b/asyncssh/listener.py @@ -28,7 +28,7 @@ from typing import Sequence, Set, Tuple, Type, Union from typing_extensions import Self -from .forward import SSHForwarderCoro +from .forward import SSHForwarderCoro, SSHForwardTrackerFactory from .forward import SSHLocalPortForwarder, SSHLocalPathForwarder from .misc import HostPort, MaybeAwait from .session import SSHTCPSession, SSHUNIXSession @@ -345,14 +345,16 @@ async def create_tcp_local_listener( async def create_tcp_forward_listener(conn: 'SSHConnection', loop: asyncio.AbstractEventLoop, coro: SSHForwarderCoro, listen_host: str, - listen_port: int) -> \ - 'SSHForwardListener': + listen_port: int, + tracker_factory: + Optional[SSHForwardTrackerFactory] = + None) -> 'SSHForwardListener': """Create a listener to forward traffic from a local TCP port over SSH""" def protocol_factory() -> asyncio.BaseProtocol: """Start a port forwarder for each new local connection""" - return SSHLocalPortForwarder(conn, coro) + return SSHLocalPortForwarder(conn, coro, tracker_factory) return await create_tcp_local_listener(conn, loop, protocol_factory, listen_host, listen_port) @@ -361,14 +363,16 @@ def protocol_factory() -> asyncio.BaseProtocol: async def create_unix_forward_listener(conn: 'SSHConnection', loop: asyncio.AbstractEventLoop, coro: SSHForwarderCoro, - listen_path: str) -> \ - 'SSHForwardListener': + listen_path: str, + tracker_factory: + Optional[SSHForwardTrackerFactory] = + None) -> 'SSHForwardListener': """Create a listener to forward a local UNIX domain socket over SSH""" def protocol_factory() -> asyncio.BaseProtocol: """Start a path forwarder for each new local connection""" - return SSHLocalPathForwarder(conn, coro) + return SSHLocalPathForwarder(conn, coro, tracker_factory) server = await loop.create_unix_server(protocol_factory, listen_path) diff --git a/docs/api.rst b/docs/api.rst index 046245b3..f3f08d51 100644 --- a/docs/api.rst +++ b/docs/api.rst @@ -1009,6 +1009,70 @@ Forwarder Classes ============================== = +Forward Tracker Classes +======================= + +The ``forward_local_*`` methods on :class:`SSHClientConnection` accept an +optional ``tracker_factory`` argument: a zero-argument callable invoked +once per accepted connection which returns a new +:class:`SSHForwardTracker` instance. asyncssh then calls that instance's +hooks for the life of the connection, giving applications a passive view +of per-connection lifecycle and byte flow -- useful for idle-based +auto-shutdown, connection counting, or traffic metrics. + +The hooks are pure observers: they run inside the asyncio event loop, +must not block, and never alter the forwarded data (return values are +ignored). Every hook has a no-op default, so a subclass overrides only +what it needs, and exceptions raised by a hook or factory are caught and +discarded so a buggy tracker cannot break forwarding. + +Use :class:`SSHPortForwardTracker` with the TCP-listener methods +(:meth:`forward_local_port() ` and +:meth:`forward_local_port_to_path() +`) and +:class:`SSHPathForwardTracker` with the UNIX-domain-listener methods +(:meth:`forward_local_path() ` and +:meth:`forward_local_path_to_port() +`). The two differ only in +the signature of ``connection_made``. + + .. code-block:: python + + class ConnCounter(asyncssh.SSHPortForwardTracker): + def __init__(self, counter): + self._counter = counter + + def connection_made(self, forwarder, orig_host, orig_port): + self._counter.active += 1 + + def connection_lost(self, exc): + self._counter.active -= 1 + + listener = await conn.forward_local_port( + '', 0, 'remote-host', 80, + tracker_factory=lambda: ConnCounter(counter)) + +.. autoclass:: SSHForwardTracker() + + ============================== = + .. automethod:: connection_lost + .. automethod:: forward_local_bytes + .. automethod:: forward_remote_bytes + ============================== = + +.. autoclass:: SSHPortForwardTracker() + + ============================== = + .. automethod:: connection_made + ============================== = + +.. autoclass:: SSHPathForwardTracker() + + ============================== = + .. automethod:: connection_made + ============================== = + + Listener Classes ================ diff --git a/tests/test_forward.py b/tests/test_forward.py index dbfd792e..465cbd24 100644 --- a/tests/test_forward.py +++ b/tests/test_forward.py @@ -697,6 +697,120 @@ async def accept_handler(_orig_host: str, _orig_port: int) -> bool: writer.close() await maybe_wait_closed(writer) + @asynctest + async def test_forward_local_port_tracker_factory_fires_made_and_lost(self): + """A port tracker sees connection_made and connection_lost""" + + events = [] + + class _RecordingTracker(asyncssh.SSHPortForwardTracker): + def connection_made(self, forwarder, orig_host, orig_port): + events.append(('made', forwarder, orig_host, orig_port)) + + def connection_lost(self, exc): + events.append(('lost', exc)) + + async with self.connect() as conn: + async with conn.forward_local_port( + '', 0, '', 7, + tracker_factory=_RecordingTracker) as listener: + await self._check_local_connection(listener.get_port()) + await asyncio.sleep(0.1) + + kinds = [event[0] for event in events] + self.assertIn('made', kinds) + self.assertIn('lost', kinds) + + made = next(event for event in events if event[0] == 'made') + self.assertIsInstance(made[1], asyncssh.SSHForwarder) + self.assertEqual(made[2], '127.0.0.1') + self.assertIsInstance(made[3], int) + + @asynctest + async def test_forward_local_port_tracker_factory_per_connection(self): + """The factory is called once per accepted connection""" + + trackers = [] + + class _CountingTracker(asyncssh.SSHPortForwardTracker): + def __init__(self): + trackers.append(self) + + def factory(): + return _CountingTracker() + + async with self.connect() as conn: + async with conn.forward_local_port( + '', 0, '', 7, tracker_factory=factory) as listener: + listen_port = listener.get_port() + await self._check_local_connection(listen_port) + await self._check_local_connection(listen_port) + await asyncio.sleep(0.1) + + self.assertEqual(len(trackers), 2) + + @asynctest + async def test_forward_local_port_tracker_byte_hooks(self): + """Byte hooks observe both forwarding directions""" + + local_bytes = bytearray() + remote_bytes = bytearray() + + class _ByteTracker(asyncssh.SSHPortForwardTracker): + def forward_local_bytes(self, data): + local_bytes.extend(data) + + def forward_remote_bytes(self, data): + remote_bytes.extend(data) + + async with self.connect() as conn: + async with conn.forward_local_port( + '', 0, '', 7, + tracker_factory=_ByteTracker) as listener: + await self._check_local_connection(listener.get_port()) + await asyncio.sleep(0.1) + + line = (str(id(self)) + '\n').encode('utf-8') + self.assertEqual(bytes(local_bytes), line) + self.assertEqual(bytes(remote_bytes), line) + + @asynctest + async def test_forward_local_port_tracker_factory_exception_swallowed(self): + """A factory that raises does not break forwarding""" + + def factory(): + raise RuntimeError('factory boom') + + async with self.connect() as conn: + async with conn.forward_local_port( + '', 0, '', 7, tracker_factory=factory) as listener: + await self._check_local_connection(listener.get_port(), + delay=0.1) + + @asynctest + async def test_forward_local_port_tracker_hook_exception_swallowed(self): + """A tracker whose hooks raise does not break forwarding""" + + class _BuggyTracker(asyncssh.SSHPortForwardTracker): + def connection_made(self, forwarder, orig_host, orig_port): + raise RuntimeError('made boom') + + def connection_lost(self, exc): + raise RuntimeError('lost boom') + + def forward_local_bytes(self, data): + raise RuntimeError('local boom') + + def forward_remote_bytes(self, data): + raise RuntimeError('remote boom') + + async with self.connect() as conn: + async with conn.forward_local_port( + '', 0, '', 7, + tracker_factory=_BuggyTracker) as listener: + await self._check_local_connection(listener.get_port(), + delay=0.1) + @unittest.skipIf(sys.platform == 'win32', 'skip UNIX domain socket tests on Windows') @asynctest @@ -1149,6 +1263,35 @@ async def test_forward_local_path(self): try_remove('local') + @asynctest + async def test_forward_local_path_tracker_factory(self): + """A path tracker sees connection_made (no addr) and connection_lost""" + + events = [] + + class _RecordingTracker(asyncssh.SSHPathForwardTracker): + def connection_made(self, forwarder): + events.append(('made', forwarder)) + + def connection_lost(self, exc): + events.append(('lost', exc)) + + async with self.connect() as conn: + async with conn.forward_local_path( + 'local', '/echo', + tracker_factory=_RecordingTracker): + await self._check_local_unix_connection('local') + await asyncio.sleep(0.1) + + try_remove('local') + + kinds = [event[0] for event in events] + self.assertIn('made', kinds) + self.assertIn('lost', kinds) + + made = next(event for event in events if event[0] == 'made') + self.assertIsInstance(made[1], asyncssh.SSHForwarder) + @asynctest async def test_forward_local_port_to_path_accept_handler(self): """Test forwarding of port to UNIX path with accept handler""" From 938b88cbda588ff2a258d3b43082904259ee0279 Mon Sep 17 00:00:00 2001 From: Alex MKX Date: Tue, 23 Jun 2026 11:36:37 +0300 Subject: [PATCH 02/13] fix(forward): make tracker connection_lost fire exactly once The local-forward _forward() coroutine calls connection_lost(exc) manually on channel-open failure, and the local transport's later close fires a second connection_lost(None) on the protocol via asyncio. Clearing self._tracker on the first call makes the hook fire exactly once. Also tightens the patch: - Drop redundant self._tracker = None in _create_tracker's except (the __init__ default already covers a raising factory). - New regression test asserting the tracker's connection_lost fires exactly once on a denied-by-accept_handler forward. - Replace asyncio.sleep(0.1) await-for-lost in existing tests with a per-test asyncio.Event resolved by the hook itself. --- asyncssh/forward.py | 20 ++++++++++++---- tests/test_forward.py | 54 +++++++++++++++++++++++++++++++++++++++---- 2 files changed, 64 insertions(+), 10 deletions(-) diff --git a/asyncssh/forward.py b/asyncssh/forward.py index 7428c658..653c1914 100644 --- a/asyncssh/forward.py +++ b/asyncssh/forward.py @@ -316,8 +316,9 @@ def _create_tracker(self) -> None: try: self._tracker = self._tracker_factory() except Exception: # pylint: disable=broad-exception-caught - # A buggy factory must not break forwarding. - self._tracker = None + # A buggy factory must not break forwarding; + # self._tracker remains the __init__ default of None. + pass def data_received(self, data: bytes, datatype: Optional[int] = None) -> None: @@ -343,11 +344,20 @@ def write(self, data: bytes) -> None: super().write(data) def connection_lost(self, exc: Optional[Exception]) -> None: - """Handle a closed local connection""" + """Handle a closed local connection - if self._tracker is not None: + This is also called manually from `_forward()` on a channel + open failure, so the local transport's eventual close fires + a second `connection_lost(None)` on the protocol. The tracker + reference is cleared on the first call so the hook fires + exactly once per connection. + """ + + tracker, self._tracker = self._tracker, None + + if tracker is not None: try: - self._tracker.connection_lost(exc) + tracker.connection_lost(exc) except Exception: # pylint: disable=broad-exception-caught pass diff --git a/tests/test_forward.py b/tests/test_forward.py index 465cbd24..6e21fb56 100644 --- a/tests/test_forward.py +++ b/tests/test_forward.py @@ -702,6 +702,7 @@ async def test_forward_local_port_tracker_factory_fires_made_and_lost(self): """A port tracker sees connection_made and connection_lost""" events = [] + lost = asyncio.Event() class _RecordingTracker(asyncssh.SSHPortForwardTracker): def connection_made(self, forwarder, orig_host, orig_port): @@ -709,13 +710,14 @@ def connection_made(self, forwarder, orig_host, orig_port): def connection_lost(self, exc): events.append(('lost', exc)) + lost.set() async with self.connect() as conn: async with conn.forward_local_port( '', 0, '', 7, tracker_factory=_RecordingTracker) as listener: await self._check_local_connection(listener.get_port()) - await asyncio.sleep(0.1) + await asyncio.wait_for(lost.wait(), timeout=1.0) kinds = [event[0] for event in events] self.assertIn('made', kinds) @@ -731,10 +733,15 @@ async def test_forward_local_port_tracker_factory_per_connection(self): """The factory is called once per accepted connection""" trackers = [] + lost_events = [] class _CountingTracker(asyncssh.SSHPortForwardTracker): def __init__(self): trackers.append(self) + lost_events.append(asyncio.Event()) + + def connection_lost(self, exc): + lost_events[trackers.index(self)].set() def factory(): return _CountingTracker() @@ -745,7 +752,9 @@ def factory(): listen_port = listener.get_port() await self._check_local_connection(listen_port) await self._check_local_connection(listen_port) - await asyncio.sleep(0.1) + await asyncio.wait_for( + asyncio.gather(*(e.wait() for e in lost_events)), + timeout=1.0) self.assertEqual(len(trackers), 2) @@ -755,6 +764,7 @@ async def test_forward_local_port_tracker_byte_hooks(self): local_bytes = bytearray() remote_bytes = bytearray() + lost = asyncio.Event() class _ByteTracker(asyncssh.SSHPortForwardTracker): def forward_local_bytes(self, data): @@ -763,12 +773,15 @@ def forward_local_bytes(self, data): def forward_remote_bytes(self, data): remote_bytes.extend(data) + def connection_lost(self, exc): + lost.set() + async with self.connect() as conn: async with conn.forward_local_port( '', 0, '', 7, tracker_factory=_ByteTracker) as listener: await self._check_local_connection(listener.get_port()) - await asyncio.sleep(0.1) + await asyncio.wait_for(lost.wait(), timeout=1.0) line = (str(id(self)) + '\n').encode('utf-8') self.assertEqual(bytes(local_bytes), line) @@ -809,7 +822,36 @@ def forward_remote_bytes(self, data): '', 0, '', 7, tracker_factory=_BuggyTracker) as listener: await self._check_local_connection(listener.get_port(), - delay=0.1) + delay=0.1) + + @asynctest + async def test_forward_local_port_tracker_lost_fires_exactly_once(self): + """connection_lost fires once even when ChannelOpenError triggers + a manual notify in _forward() followed by the asyncio close path.""" + + lost_count = 0 + + class _Counting(asyncssh.SSHPortForwardTracker): + def connection_lost(self, exc): + nonlocal lost_count + lost_count += 1 + + async def deny(_h, _p): + return False + + async with self.connect() as conn: + async with conn.forward_local_port( + '', 0, '', 7, + accept_handler=deny, + tracker_factory=_Counting) as listener: + reader, writer = await asyncio.open_connection( + '127.0.0.1', listener.get_port()) + self.assertEqual((await reader.read()), b'') + writer.close() + await maybe_wait_closed(writer) + await asyncio.sleep(0.1) # bounded: upper-bound for any spurious duplicate + + self.assertEqual(lost_count, 1) @unittest.skipIf(sys.platform == 'win32', 'skip UNIX domain socket tests on Windows') @@ -1268,6 +1310,7 @@ async def test_forward_local_path_tracker_factory(self): """A path tracker sees connection_made (no addr) and connection_lost""" events = [] + lost = asyncio.Event() class _RecordingTracker(asyncssh.SSHPathForwardTracker): def connection_made(self, forwarder): @@ -1275,13 +1318,14 @@ def connection_made(self, forwarder): def connection_lost(self, exc): events.append(('lost', exc)) + lost.set() async with self.connect() as conn: async with conn.forward_local_path( 'local', '/echo', tracker_factory=_RecordingTracker): await self._check_local_unix_connection('local') - await asyncio.sleep(0.1) + await asyncio.wait_for(lost.wait(), timeout=1.0) try_remove('local') From 5217feefae2c52c035d8834f71b0320324710a39 Mon Sep 17 00:00:00 2001 From: Alex MKX Date: Fri, 10 Jul 2026 21:02:00 +0300 Subject: [PATCH 03/13] review: address ronf feedback on PR #807 - Do not export SSHForwardTracker publicly (Port/Path only) - Split factory type into SSHPortForwardTrackerFactory / SSHPathForwardTrackerFactory - Drop None from tracker_factory :type: docstrings; name the factory type - Pass tracker_factory positionally into the listener helper - Rename per-connection factory test to describe behavior - broad-exception-caught -> broad-except (older pylint compat) - Add docstrings to test tracker classes; fix short var names - Cover connection_made exception swallow for the Path case --- asyncssh/__init__.py | 4 ++-- asyncssh/connection.py | 27 ++++++++++++++------------- asyncssh/forward.py | 21 ++++++++++++--------- asyncssh/listener.py | 7 ++++--- tests/test_forward.py | 39 ++++++++++++++++++++++++++++++++++++--- 5 files changed, 68 insertions(+), 30 deletions(-) diff --git a/asyncssh/__init__.py b/asyncssh/__init__.py index b2b49fc7..a7d999a2 100644 --- a/asyncssh/__init__.py +++ b/asyncssh/__init__.py @@ -40,7 +40,7 @@ from .config import ConfigParseError -from .forward import SSHForwarder, SSHForwardTracker +from .forward import SSHForwarder from .forward import SSHPortForwardTracker, SSHPathForwardTracker from .connection import SSHAcceptor, SSHClientConnection, SSHServerConnection @@ -148,7 +148,7 @@ 'SSHAgentKeyPair', 'SSHAuthorizedKeys', 'SSHCertificate', 'SSHClient', 'SSHClientChannel', 'SSHClientConnection', 'SSHClientConnectionOptions', 'SSHClientProcess', 'SSHClientSession', 'SSHCompletedProcess', - 'SSHForwarder', 'SSHForwardTracker', 'SSHPortForwardTracker', + 'SSHForwarder', 'SSHPortForwardTracker', 'SSHPathForwardTracker', 'SSHKey', 'SSHKeyPair', 'SSHKnownHosts', 'SSHLineEditorChannel', 'SSHListener', 'SSHReader', 'SSHServer', 'SSHServerChannel', 'SSHServerConnection', diff --git a/asyncssh/connection.py b/asyncssh/connection.py index 64fb61f4..96bd048e 100644 --- a/asyncssh/connection.py +++ b/asyncssh/connection.py @@ -85,7 +85,8 @@ from .encryption import encryption_needs_mac from .encryption import get_encryption_params, get_encryption -from .forward import SSHForwarder, SSHForwardTrackerFactory +from .forward import SSHForwarder +from .forward import SSHPortForwardTrackerFactory, SSHPathForwardTrackerFactory from .gss import GSSBase, GSSClient, GSSServer, GSSError @@ -3212,7 +3213,7 @@ async def forward_local_port( dest_host: str, dest_port: int, accept_handler: Optional[SSHAcceptHandler] = None, tracker_factory: - Optional[SSHForwardTrackerFactory] = None) -> SSHListener: + Optional[SSHPortForwardTrackerFactory] = None) -> SSHListener: """Set up local port forwarding This method is a coroutine which attempts to set up port @@ -3246,7 +3247,7 @@ async def forward_local_port( :type dest_host: `str` :type dest_port: `int` :type accept_handler: `callable` or coroutine - :type tracker_factory: `callable` or `None` + :type tracker_factory: :class:`SSHPortForwardTrackerFactory` :returns: :class:`SSHListener` @@ -3289,7 +3290,7 @@ async def tunnel_connection( try: listener = await create_tcp_forward_listener( self, self._loop, tunnel_connection, listen_host, listen_port, - tracker_factory=tracker_factory) + tracker_factory) except OSError as exc: self.logger.debug1('Failed to create local TCP listener: %s', exc) raise @@ -3308,7 +3309,7 @@ async def tunnel_connection( async def forward_local_path( self, listen_path: str, dest_path: str, tracker_factory: - Optional[SSHForwardTrackerFactory] = None) -> SSHListener: + Optional[SSHPathForwardTrackerFactory] = None) -> SSHListener: """Set up local UNIX domain socket forwarding This method is a coroutine which attempts to set up UNIX domain @@ -3329,7 +3330,7 @@ async def forward_local_path( with no overhead. :type listen_path: `str` :type dest_path: `str` - :type tracker_factory: `callable` or `None` + :type tracker_factory: :class:`SSHPathForwardTrackerFactory` :returns: :class:`SSHListener` @@ -3351,7 +3352,7 @@ async def tunnel_connection( try: listener = await create_unix_forward_listener( self, self._loop, tunnel_connection, listen_path, - tracker_factory=tracker_factory) + tracker_factory) except OSError as exc: self.logger.debug1('Failed to create local UNIX listener: %s', exc) raise @@ -5323,7 +5324,7 @@ async def forward_local_port_to_path( self, listen_host: str, listen_port: int, dest_path: str, accept_handler: Optional[SSHAcceptHandler] = None, tracker_factory: - Optional[SSHForwardTrackerFactory] = None) -> SSHListener: + Optional[SSHPortForwardTrackerFactory] = None) -> SSHListener: """Set up local TCP port forwarding to a remote UNIX domain socket This method is a coroutine which attempts to set up port @@ -5354,7 +5355,7 @@ async def forward_local_port_to_path( :type listen_port: `int` :type dest_path: `str` :type accept_handler: `callable` or coroutine - :type tracker_factory: `callable` or `None` + :type tracker_factory: :class:`SSHPortForwardTrackerFactory` :returns: :class:`SSHListener` @@ -5390,7 +5391,7 @@ async def tunnel_connection( try: listener = await create_tcp_forward_listener( self, self._loop, tunnel_connection, listen_host, listen_port, - tracker_factory=tracker_factory) + tracker_factory) except OSError as exc: self.logger.debug1('Failed to create local TCP listener: %s', exc) raise @@ -5406,7 +5407,7 @@ async def tunnel_connection( async def forward_local_path_to_port( self, listen_path: str, dest_host: str, dest_port: int, tracker_factory: - Optional[SSHForwardTrackerFactory] = None) -> SSHListener: + Optional[SSHPathForwardTrackerFactory] = None) -> SSHListener: """Set up local UNIX domain socket forwarding to a remote TCP port This method is a coroutine which attempts to set up UNIX domain @@ -5430,7 +5431,7 @@ async def forward_local_path_to_port( :type listen_path: `str` :type dest_host: `str` :type dest_port: `int` - :type tracker_factory: `callable` or `None` + :type tracker_factory: :class:`SSHPathForwardTrackerFactory` :returns: :class:`SSHListener` @@ -5452,7 +5453,7 @@ async def tunnel_connection( try: listener = await create_unix_forward_listener( self, self._loop, tunnel_connection, listen_path, - tracker_factory=tracker_factory) + tracker_factory) except OSError as exc: self.logger.debug1('Failed to create local UNIX listener: %s', exc) raise diff --git a/asyncssh/forward.py b/asyncssh/forward.py index 653c1914..e3955976 100644 --- a/asyncssh/forward.py +++ b/asyncssh/forward.py @@ -24,7 +24,7 @@ import socket from types import TracebackType from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, Optional -from typing import Type, cast +from typing import Type, Union, cast from typing_extensions import Self from .misc import ChannelOpenError, SockAddr @@ -142,7 +142,8 @@ def connection_made(self, forwarder: 'SSHForwarder') -> None: """ -SSHForwardTrackerFactory = Callable[[], SSHForwardTracker] +SSHPortForwardTrackerFactory = Callable[[], SSHPortForwardTracker] +SSHPathForwardTrackerFactory = Callable[[], SSHPathForwardTracker] class SSHForwarder(asyncio.BaseProtocol): @@ -300,7 +301,9 @@ class SSHLocalForwarder(SSHForwarder): """Local forwarding connection handler""" def __init__(self, conn: 'SSHConnection', coro: SSHForwarderCoro, - tracker_factory: Optional[SSHForwardTrackerFactory] = None): + tracker_factory: + Optional[Union[SSHPortForwardTrackerFactory, + SSHPathForwardTrackerFactory]] = None): super().__init__() self._conn = conn self._coro = coro @@ -315,7 +318,7 @@ def _create_tracker(self) -> None: try: self._tracker = self._tracker_factory() - except Exception: # pylint: disable=broad-exception-caught + except Exception: # pylint: disable=broad-except # A buggy factory must not break forwarding; # self._tracker remains the __init__ default of None. pass @@ -327,7 +330,7 @@ def data_received(self, data: bytes, if self._tracker is not None: try: self._tracker.forward_local_bytes(data) - except Exception: # pylint: disable=broad-exception-caught + except Exception: # pylint: disable=broad-except pass super().data_received(data, datatype) @@ -338,7 +341,7 @@ def write(self, data: bytes) -> None: if self._tracker is not None: try: self._tracker.forward_remote_bytes(data) - except Exception: # pylint: disable=broad-exception-caught + except Exception: # pylint: disable=broad-except pass super().write(data) @@ -358,7 +361,7 @@ def connection_lost(self, exc: Optional[Exception]) -> None: if tracker is not None: try: tracker.connection_lost(exc) - except Exception: # pylint: disable=broad-exception-caught + except Exception: # pylint: disable=broad-except pass super().connection_lost(exc) @@ -412,7 +415,7 @@ def connection_made(self, transport: asyncio.BaseTransport) -> None: try: cast(SSHPortForwardTracker, self._tracker).connection_made( self, orig_host, orig_port) - except Exception: # pylint: disable=broad-exception-caught + except Exception: # pylint: disable=broad-except pass self.forward(orig_host, orig_port) @@ -431,7 +434,7 @@ def connection_made(self, transport: asyncio.BaseTransport) -> None: if self._tracker is not None: try: cast(SSHPathForwardTracker, self._tracker).connection_made(self) - except Exception: # pylint: disable=broad-exception-caught + except Exception: # pylint: disable=broad-except pass self.forward() diff --git a/asyncssh/listener.py b/asyncssh/listener.py index b77db809..c9d6e483 100644 --- a/asyncssh/listener.py +++ b/asyncssh/listener.py @@ -28,7 +28,8 @@ from typing import Sequence, Set, Tuple, Type, Union from typing_extensions import Self -from .forward import SSHForwarderCoro, SSHForwardTrackerFactory +from .forward import SSHForwarderCoro +from .forward import SSHPortForwardTrackerFactory, SSHPathForwardTrackerFactory from .forward import SSHLocalPortForwarder, SSHLocalPathForwarder from .misc import HostPort, MaybeAwait from .session import SSHTCPSession, SSHUNIXSession @@ -347,7 +348,7 @@ async def create_tcp_forward_listener(conn: 'SSHConnection', coro: SSHForwarderCoro, listen_host: str, listen_port: int, tracker_factory: - Optional[SSHForwardTrackerFactory] = + Optional[SSHPortForwardTrackerFactory] = None) -> 'SSHForwardListener': """Create a listener to forward traffic from a local TCP port over SSH""" @@ -365,7 +366,7 @@ async def create_unix_forward_listener(conn: 'SSHConnection', coro: SSHForwarderCoro, listen_path: str, tracker_factory: - Optional[SSHForwardTrackerFactory] = + Optional[SSHPathForwardTrackerFactory] = None) -> 'SSHForwardListener': """Create a listener to forward a local UNIX domain socket over SSH""" diff --git a/tests/test_forward.py b/tests/test_forward.py index 6e21fb56..3a31fc2d 100644 --- a/tests/test_forward.py +++ b/tests/test_forward.py @@ -705,6 +705,8 @@ async def test_forward_local_port_tracker_factory_fires_made_and_lost(self): lost = asyncio.Event() class _RecordingTracker(asyncssh.SSHPortForwardTracker): + """Tracker which records connection_made and connection_lost""" + def connection_made(self, forwarder, orig_host, orig_port): events.append(('made', forwarder, orig_host, orig_port)) @@ -729,13 +731,16 @@ def connection_lost(self, exc): self.assertIsInstance(made[3], int) @asynctest - async def test_forward_local_port_tracker_factory_per_connection(self): - """The factory is called once per accepted connection""" + async def test_tracker_factory_invoked_once_per_connection(self): + """A distinct tracker instance is created for each accepted + connection, and each instance's connection_lost fires once""" trackers = [] lost_events = [] class _CountingTracker(asyncssh.SSHPortForwardTracker): + """Tracker which records each instance created by the factory""" + def __init__(self): trackers.append(self) lost_events.append(asyncio.Event()) @@ -767,6 +772,8 @@ async def test_forward_local_port_tracker_byte_hooks(self): lost = asyncio.Event() class _ByteTracker(asyncssh.SSHPortForwardTracker): + """Tracker which records bytes seen in both forwarding directions""" + def forward_local_bytes(self, data): local_bytes.extend(data) @@ -805,6 +812,8 @@ async def test_forward_local_port_tracker_hook_exception_swallowed(self): """A tracker whose hooks raise does not break forwarding""" class _BuggyTracker(asyncssh.SSHPortForwardTracker): + """Tracker whose hooks all raise, to verify they're swallowed""" + def connection_made(self, forwarder, orig_host, orig_port): raise RuntimeError('made boom') @@ -832,11 +841,13 @@ async def test_forward_local_port_tracker_lost_fires_exactly_once(self): lost_count = 0 class _Counting(asyncssh.SSHPortForwardTracker): + """Tracker which counts how many times connection_lost fires""" + def connection_lost(self, exc): nonlocal lost_count lost_count += 1 - async def deny(_h, _p): + async def deny(_orig_host, _orig_port): return False async with self.connect() as conn: @@ -1313,6 +1324,8 @@ async def test_forward_local_path_tracker_factory(self): lost = asyncio.Event() class _RecordingTracker(asyncssh.SSHPathForwardTracker): + """Tracker which records connection_made and connection_lost""" + def connection_made(self, forwarder): events.append(('made', forwarder)) @@ -1336,6 +1349,26 @@ def connection_lost(self, exc): made = next(event for event in events if event[0] == 'made') self.assertIsInstance(made[1], asyncssh.SSHForwarder) + @asynctest + async def test_forward_local_path_tracker_hook_exception_swallowed(self): + """A path tracker whose connection_made raises does not break + forwarding""" + + class _BuggyTracker(asyncssh.SSHPathForwardTracker): + """Tracker whose connection_made hook raises, to verify it's + swallowed""" + + def connection_made(self, forwarder): + raise RuntimeError('made boom') + + async with self.connect() as conn: + async with conn.forward_local_path( + 'local', '/echo', + tracker_factory=_BuggyTracker): + await self._check_local_unix_connection('local') + + try_remove('local') + @asynctest async def test_forward_local_port_to_path_accept_handler(self): """Test forwarding of port to UNIX path with accept handler""" From 8ca835dc1cbe3f494f82e2c209ba3ec1125c9305 Mon Sep 17 00:00:00 2001 From: Alex MKX Date: Fri, 10 Jul 2026 21:13:10 +0300 Subject: [PATCH 04/13] docs(api): document Port/Path trackers directly, drop non-public base autoclass SSHForwardTracker is no longer exported from the package root, so the standalone autoclass directive would break the Sphinx build. Per the PR #807 review, document SSHPortForwardTracker and SSHPathForwardTracker directly, listing each callback under both (connection_made signatures differ; connection_lost / forward_local_bytes / forward_remote_bytes are inherited from the base). --- docs/api.rst | 37 +++++++++++++++++++------------------ 1 file changed, 19 insertions(+), 18 deletions(-) diff --git a/docs/api.rst b/docs/api.rst index f3f08d51..a752a876 100644 --- a/docs/api.rst +++ b/docs/api.rst @@ -1014,11 +1014,13 @@ Forward Tracker Classes The ``forward_local_*`` methods on :class:`SSHClientConnection` accept an optional ``tracker_factory`` argument: a zero-argument callable invoked -once per accepted connection which returns a new -:class:`SSHForwardTracker` instance. asyncssh then calls that instance's -hooks for the life of the connection, giving applications a passive view -of per-connection lifecycle and byte flow -- useful for idle-based -auto-shutdown, connection counting, or traffic metrics. +once per accepted connection which returns a tracker instance -- +:class:`SSHPortForwardTracker` for TCP listeners or +:class:`SSHPathForwardTracker` for UNIX domain listeners. asyncssh then +calls that instance's hooks for the life of the connection, giving +applications a passive view of per-connection lifecycle and byte flow -- +useful for idle-based auto-shutdown, connection counting, or traffic +metrics. The hooks are pure observers: they run inside the asyncio event loop, must not block, and never alter the forwarded data (return values are @@ -1033,8 +1035,9 @@ Use :class:`SSHPortForwardTracker` with the TCP-listener methods :class:`SSHPathForwardTracker` with the UNIX-domain-listener methods (:meth:`forward_local_path() ` and :meth:`forward_local_path_to_port() -`). The two differ only in -the signature of ``connection_made``. +`). The two classes share +the same set of hooks and differ only in the signature of +``connection_made``. .. code-block:: python @@ -1052,25 +1055,23 @@ the signature of ``connection_made``. '', 0, 'remote-host', 80, tracker_factory=lambda: ConnCounter(counter)) -.. autoclass:: SSHForwardTracker() +.. autoclass:: SSHPortForwardTracker() - ============================== = + ==================================== = + .. automethod:: connection_made .. automethod:: connection_lost .. automethod:: forward_local_bytes .. automethod:: forward_remote_bytes - ============================== = - -.. autoclass:: SSHPortForwardTracker() - - ============================== = - .. automethod:: connection_made - ============================== = + ==================================== = .. autoclass:: SSHPathForwardTracker() - ============================== = + ==================================== = .. automethod:: connection_made - ============================== = + .. automethod:: connection_lost + .. automethod:: forward_local_bytes + .. automethod:: forward_remote_bytes + ==================================== = Listener Classes From fb9ea52b353984a1938dd9a3a592364384005824 Mon Sep 17 00:00:00 2001 From: Alex MKX Date: Sun, 12 Jul 2026 11:50:51 +0300 Subject: [PATCH 05/13] review: address ronf 2026-07-12 follow-up on PR #807 - __all__: order tracker classes case-sensitive alphabetically (SSHPathForwardTracker before SSHPortForwardTracker, after SSHListener), kept compact on one line - connection.py: drop the '(or SSHForwardTracker subclass)' clause from the four tracker_factory param docs; SSHForwardTracker is non-public - no change needed to :type tracker_factory: factory-alias refs; sphinx -W build is already clean (nitpicky mode off in docs/conf.py) --- asyncssh/__init__.py | 8 ++++---- asyncssh/connection.py | 20 ++++++++------------ 2 files changed, 12 insertions(+), 16 deletions(-) diff --git a/asyncssh/__init__.py b/asyncssh/__init__.py index a7d999a2..a1f589dd 100644 --- a/asyncssh/__init__.py +++ b/asyncssh/__init__.py @@ -41,7 +41,7 @@ from .config import ConfigParseError from .forward import SSHForwarder -from .forward import SSHPortForwardTracker, SSHPathForwardTracker +from .forward import SSHPathForwardTracker, SSHPortForwardTracker from .connection import SSHAcceptor, SSHClientConnection, SSHServerConnection from .connection import SSHClientConnectionOptions, SSHServerConnectionOptions @@ -148,9 +148,9 @@ 'SSHAgentKeyPair', 'SSHAuthorizedKeys', 'SSHCertificate', 'SSHClient', 'SSHClientChannel', 'SSHClientConnection', 'SSHClientConnectionOptions', 'SSHClientProcess', 'SSHClientSession', 'SSHCompletedProcess', - 'SSHForwarder', 'SSHPortForwardTracker', - 'SSHPathForwardTracker', 'SSHKey', 'SSHKeyPair', 'SSHKnownHosts', - 'SSHLineEditorChannel', 'SSHListener', 'SSHReader', 'SSHServer', + 'SSHForwarder', 'SSHKey', 'SSHKeyPair', 'SSHKnownHosts', + 'SSHLineEditorChannel', 'SSHListener', 'SSHPathForwardTracker', + 'SSHPortForwardTracker', 'SSHReader', 'SSHServer', 'SSHServerChannel', 'SSHServerConnection', 'SSHServerConnectionOptions', 'SSHServerProcess', 'SSHServerProcessFactory', 'SSHServerSession', diff --git a/asyncssh/connection.py b/asyncssh/connection.py index 96bd048e..2fc783b9 100644 --- a/asyncssh/connection.py +++ b/asyncssh/connection.py @@ -3238,9 +3238,8 @@ async def forward_local_port( reject and close it. :param tracker_factory: An optional callable invoked once per accepted connection - which returns a new :class:`SSHPortForwardTracker` (or - :class:`SSHForwardTracker` subclass) for observing that - connection's lifecycle. `None` (default) disables tracking + which returns a new :class:`SSHPortForwardTracker` for observing + that connection's lifecycle. `None` (default) disables tracking with no overhead. :type listen_host: `str` :type listen_port: `int` @@ -3324,9 +3323,8 @@ async def forward_local_path( The path on the remote host to forward the connections to :param tracker_factory: An optional callable invoked once per accepted connection - which returns a new :class:`SSHPathForwardTracker` (or - :class:`SSHForwardTracker` subclass) for observing that - connection's lifecycle. `None` (default) disables tracking + which returns a new :class:`SSHPathForwardTracker` for observing + that connection's lifecycle. `None` (default) disables tracking with no overhead. :type listen_path: `str` :type dest_path: `str` @@ -5347,9 +5345,8 @@ async def forward_local_port_to_path( reject and close it. :param tracker_factory: An optional callable invoked once per accepted connection - which returns a new :class:`SSHPortForwardTracker` (or - :class:`SSHForwardTracker` subclass) for observing that - connection's lifecycle. `None` (default) disables tracking + which returns a new :class:`SSHPortForwardTracker` for observing + that connection's lifecycle. `None` (default) disables tracking with no overhead. :type listen_host: `str` :type listen_port: `int` @@ -5424,9 +5421,8 @@ async def forward_local_path_to_port( The port number to forward the connections to :param tracker_factory: An optional callable invoked once per accepted connection - which returns a new :class:`SSHPathForwardTracker` (or - :class:`SSHForwardTracker` subclass) for observing that - connection's lifecycle. `None` (default) disables tracking + which returns a new :class:`SSHPathForwardTracker` for observing + that connection's lifecycle. `None` (default) disables tracking with no overhead. :type listen_path: `str` :type dest_host: `str` From 5f43209e33303cca3632e2a7e536fb42ac319c80 Mon Sep 17 00:00:00 2001 From: Alex MKX Date: Sat, 18 Jul 2026 10:05:49 +0300 Subject: [PATCH 06/13] review: address ronf follow-up on PR #807 (tracker doc, _create_tracker arg, else-pragma, test name) --- asyncssh/forward.py | 43 ++++++++++++++++++++----------------------- tests/test_forward.py | 2 +- 2 files changed, 21 insertions(+), 24 deletions(-) diff --git a/asyncssh/forward.py b/asyncssh/forward.py index e3955976..0495361d 100644 --- a/asyncssh/forward.py +++ b/asyncssh/forward.py @@ -23,8 +23,8 @@ import asyncio import socket from types import TracebackType -from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, Optional -from typing import Type, Union, cast +from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, Generic +from typing import Optional, Type, TypeVar, cast from typing_extensions import Self from .misc import ChannelOpenError, SockAddr @@ -41,8 +41,8 @@ class SSHForwardTracker: """Base class for observing the lifecycle of a forwarded connection - A tracker observes a single forwarded connection on a local - listener. A `tracker_factory` passed to one of the + A tracker observes a single forwarded connection. A + `tracker_factory` passed to one of the :meth:`forward_local_port() ` family of methods is called once per accepted connection and must return a new tracker instance, on which asyncssh then calls the @@ -145,6 +145,8 @@ def connection_made(self, forwarder: 'SSHForwarder') -> None: SSHPortForwardTrackerFactory = Callable[[], SSHPortForwardTracker] SSHPathForwardTrackerFactory = Callable[[], SSHPathForwardTracker] +_Tracker = TypeVar('_Tracker', bound=SSHForwardTracker) + class SSHForwarder(asyncio.BaseProtocol): """SSH port forwarding connection handler""" @@ -297,27 +299,26 @@ def close(self) -> None: peer.close() -class SSHLocalForwarder(SSHForwarder): +class SSHLocalForwarder(SSHForwarder, Generic[_Tracker]): """Local forwarding connection handler""" def __init__(self, conn: 'SSHConnection', coro: SSHForwarderCoro, - tracker_factory: - Optional[Union[SSHPortForwardTrackerFactory, - SSHPathForwardTrackerFactory]] = None): + tracker_factory: Optional[Callable[[], _Tracker]] = None): super().__init__() self._conn = conn self._coro = coro - self._tracker_factory = tracker_factory - self._tracker: Optional[SSHForwardTracker] = None + self._tracker: Optional[_Tracker] = None + self._create_tracker(tracker_factory) - def _create_tracker(self) -> None: + def _create_tracker( + self, tracker_factory: Optional[Callable[[], _Tracker]]) -> None: """Instantiate this connection's tracker from the factory, if any""" - if self._tracker_factory is None: + if tracker_factory is None: return try: - self._tracker = self._tracker_factory() + self._tracker = tracker_factory() except Exception: # pylint: disable=broad-except # A buggy factory must not break forwarding; # self._tracker remains the __init__ default of None. @@ -395,7 +396,7 @@ def forward(self, *args: object) -> None: self._conn.create_task(self._forward(*args)) -class SSHLocalPortForwarder(SSHLocalForwarder): +class SSHLocalPortForwarder(SSHLocalForwarder[SSHPortForwardTracker]): """Local TCP port forwarding connection handler""" def connection_made(self, transport: asyncio.BaseTransport) -> None: @@ -403,25 +404,23 @@ def connection_made(self, transport: asyncio.BaseTransport) -> None: super().connection_made(transport) - orig_host, orig_port = '', 0 peername = cast(SockAddr, transport.get_extra_info('peername')) if peername: # pragma: no branch orig_host, orig_port = peername[:2] - - self._create_tracker() + else: # pragma: no cover + orig_host, orig_port = '', 0 if self._tracker is not None: try: - cast(SSHPortForwardTracker, self._tracker).connection_made( - self, orig_host, orig_port) + self._tracker.connection_made(self, orig_host, orig_port) except Exception: # pylint: disable=broad-except pass self.forward(orig_host, orig_port) -class SSHLocalPathForwarder(SSHLocalForwarder): +class SSHLocalPathForwarder(SSHLocalForwarder[SSHPathForwardTracker]): """Local UNIX domain socket forwarding connection handler""" def connection_made(self, transport: asyncio.BaseTransport) -> None: @@ -429,11 +428,9 @@ def connection_made(self, transport: asyncio.BaseTransport) -> None: super().connection_made(transport) - self._create_tracker() - if self._tracker is not None: try: - cast(SSHPathForwardTracker, self._tracker).connection_made(self) + self._tracker.connection_made(self) except Exception: # pylint: disable=broad-except pass diff --git a/tests/test_forward.py b/tests/test_forward.py index 3a31fc2d..d3f4d417 100644 --- a/tests/test_forward.py +++ b/tests/test_forward.py @@ -698,7 +698,7 @@ async def accept_handler(_orig_host: str, _orig_port: int) -> bool: await maybe_wait_closed(writer) @asynctest - async def test_forward_local_port_tracker_factory_fires_made_and_lost(self): + async def test_port_tracker_made_and_lost(self): """A port tracker sees connection_made and connection_lost""" events = [] From 236ca293d6d6c617d21eabc4b5eda909263a0f02 Mon Sep 17 00:00:00 2001 From: Alex MKX Date: Sat, 18 Jul 2026 11:01:31 +0300 Subject: [PATCH 07/13] review: extract tracker-hook exception guard into a helper (PR #807) Addresses ronf's suggestion to DRY the repeated try/except-on-Exception around tracker hook calls into a single utility. --- asyncssh/forward.py | 45 ++++++++++++++++++++------------------------- 1 file changed, 20 insertions(+), 25 deletions(-) diff --git a/asyncssh/forward.py b/asyncssh/forward.py index 0495361d..bedff956 100644 --- a/asyncssh/forward.py +++ b/asyncssh/forward.py @@ -324,26 +324,30 @@ def _create_tracker( # self._tracker remains the __init__ default of None. pass - def data_received(self, data: bytes, - datatype: Optional[int] = None) -> None: - """Handle incoming data from the local transport""" + def _notify_tracker(self, tracker: Optional[_Tracker], + notify: Callable[[_Tracker], None]) -> None: + """Invoke a tracker hook, swallowing exceptions from buggy trackers""" - if self._tracker is not None: + if tracker is not None: try: - self._tracker.forward_local_bytes(data) + notify(tracker) except Exception: # pylint: disable=broad-except pass + def data_received(self, data: bytes, + datatype: Optional[int] = None) -> None: + """Handle incoming data from the local transport""" + + self._notify_tracker( + self._tracker, lambda t: t.forward_local_bytes(data)) + super().data_received(data, datatype) def write(self, data: bytes) -> None: """Write tunnel data out to the local transport""" - if self._tracker is not None: - try: - self._tracker.forward_remote_bytes(data) - except Exception: # pylint: disable=broad-except - pass + self._notify_tracker( + self._tracker, lambda t: t.forward_remote_bytes(data)) super().write(data) @@ -359,11 +363,7 @@ def connection_lost(self, exc: Optional[Exception]) -> None: tracker, self._tracker = self._tracker, None - if tracker is not None: - try: - tracker.connection_lost(exc) - except Exception: # pylint: disable=broad-except - pass + self._notify_tracker(tracker, lambda t: t.connection_lost(exc)) super().connection_lost(exc) @@ -411,11 +411,9 @@ def connection_made(self, transport: asyncio.BaseTransport) -> None: else: # pragma: no cover orig_host, orig_port = '', 0 - if self._tracker is not None: - try: - self._tracker.connection_made(self, orig_host, orig_port) - except Exception: # pylint: disable=broad-except - pass + self._notify_tracker( + self._tracker, + lambda t: t.connection_made(self, orig_host, orig_port)) self.forward(orig_host, orig_port) @@ -428,10 +426,7 @@ def connection_made(self, transport: asyncio.BaseTransport) -> None: super().connection_made(transport) - if self._tracker is not None: - try: - self._tracker.connection_made(self) - except Exception: # pylint: disable=broad-except - pass + self._notify_tracker( + self._tracker, lambda t: t.connection_made(self)) self.forward() From 02a6f3038f918d7eb2763737d2ba912827b7dd5f Mon Sep 17 00:00:00 2001 From: Alex MKX Date: Fri, 31 Jul 2026 10:11:36 +0300 Subject: [PATCH 08/13] review: address ronf 2026-07-23 feedback on PR #807 Replace tracker-notify lambdas with nested functions, make _notify_tracker a staticmethod, and shorten tracker test names. --- asyncssh/forward.py | 43 ++++++++++++++++++++++++++++++++----------- tests/test_forward.py | 14 +++++++------- 2 files changed, 39 insertions(+), 18 deletions(-) diff --git a/asyncssh/forward.py b/asyncssh/forward.py index bedff956..02ce7f70 100644 --- a/asyncssh/forward.py +++ b/asyncssh/forward.py @@ -324,7 +324,8 @@ def _create_tracker( # self._tracker remains the __init__ default of None. pass - def _notify_tracker(self, tracker: Optional[_Tracker], + @staticmethod + def _notify_tracker(tracker: Optional[_Tracker], notify: Callable[[_Tracker], None]) -> None: """Invoke a tracker hook, swallowing exceptions from buggy trackers""" @@ -338,16 +339,24 @@ def data_received(self, data: bytes, datatype: Optional[int] = None) -> None: """Handle incoming data from the local transport""" - self._notify_tracker( - self._tracker, lambda t: t.forward_local_bytes(data)) + def notify(tracker: _Tracker) -> None: + """Report locally forwarded bytes to the tracker""" + + tracker.forward_local_bytes(data) + + self._notify_tracker(self._tracker, notify) super().data_received(data, datatype) def write(self, data: bytes) -> None: """Write tunnel data out to the local transport""" - self._notify_tracker( - self._tracker, lambda t: t.forward_remote_bytes(data)) + def notify(tracker: _Tracker) -> None: + """Report remotely forwarded bytes to the tracker""" + + tracker.forward_remote_bytes(data) + + self._notify_tracker(self._tracker, notify) super().write(data) @@ -363,7 +372,12 @@ def connection_lost(self, exc: Optional[Exception]) -> None: tracker, self._tracker = self._tracker, None - self._notify_tracker(tracker, lambda t: t.connection_lost(exc)) + def notify(tracker: _Tracker) -> None: + """Report the closed connection to the tracker""" + + tracker.connection_lost(exc) + + self._notify_tracker(tracker, notify) super().connection_lost(exc) @@ -411,9 +425,12 @@ def connection_made(self, transport: asyncio.BaseTransport) -> None: else: # pragma: no cover orig_host, orig_port = '', 0 - self._notify_tracker( - self._tracker, - lambda t: t.connection_made(self, orig_host, orig_port)) + def notify(tracker: SSHPortForwardTracker) -> None: + """Report the new connection to the tracker""" + + tracker.connection_made(self, orig_host, orig_port) + + self._notify_tracker(self._tracker, notify) self.forward(orig_host, orig_port) @@ -426,7 +443,11 @@ def connection_made(self, transport: asyncio.BaseTransport) -> None: super().connection_made(transport) - self._notify_tracker( - self._tracker, lambda t: t.connection_made(self)) + def notify(tracker: SSHPathForwardTracker) -> None: + """Report the new connection to the tracker""" + + tracker.connection_made(self) + + self._notify_tracker(self._tracker, notify) self.forward() diff --git a/tests/test_forward.py b/tests/test_forward.py index d3f4d417..0d18ce5b 100644 --- a/tests/test_forward.py +++ b/tests/test_forward.py @@ -731,7 +731,7 @@ def connection_lost(self, exc): self.assertIsInstance(made[3], int) @asynctest - async def test_tracker_factory_invoked_once_per_connection(self): + async def test_tracker_factory_per_connection(self): """A distinct tracker instance is created for each accepted connection, and each instance's connection_lost fires once""" @@ -764,7 +764,7 @@ def factory(): self.assertEqual(len(trackers), 2) @asynctest - async def test_forward_local_port_tracker_byte_hooks(self): + async def test_port_tracker_byte_hooks(self): """Byte hooks observe both forwarding directions""" local_bytes = bytearray() @@ -795,7 +795,7 @@ def connection_lost(self, exc): self.assertEqual(bytes(remote_bytes), line) @asynctest - async def test_forward_local_port_tracker_factory_exception_swallowed(self): + async def test_port_tracker_factory_exception_swallowed(self): """A factory that raises does not break forwarding""" def factory(): @@ -808,7 +808,7 @@ def factory(): delay=0.1) @asynctest - async def test_forward_local_port_tracker_hook_exception_swallowed(self): + async def test_port_tracker_hook_exception_swallowed(self): """A tracker whose hooks raise does not break forwarding""" class _BuggyTracker(asyncssh.SSHPortForwardTracker): @@ -834,7 +834,7 @@ def forward_remote_bytes(self, data): delay=0.1) @asynctest - async def test_forward_local_port_tracker_lost_fires_exactly_once(self): + async def test_port_tracker_lost_fires_once(self): """connection_lost fires once even when ChannelOpenError triggers a manual notify in _forward() followed by the asyncio close path.""" @@ -1317,7 +1317,7 @@ async def test_forward_local_path(self): try_remove('local') @asynctest - async def test_forward_local_path_tracker_factory(self): + async def test_path_tracker_made_and_lost(self): """A path tracker sees connection_made (no addr) and connection_lost""" events = [] @@ -1350,7 +1350,7 @@ def connection_lost(self, exc): self.assertIsInstance(made[1], asyncssh.SSHForwarder) @asynctest - async def test_forward_local_path_tracker_hook_exception_swallowed(self): + async def test_path_tracker_hook_exception_swallowed(self): """A path tracker whose connection_made raises does not break forwarding""" From 2bf4164d65b25ed4226f44f210c443207fb372b3 Mon Sep 17 00:00:00 2001 From: Alex MKX Date: Fri, 31 Jul 2026 10:15:45 +0300 Subject: [PATCH 09/13] review: fix continuation alignment in _notify_tracker signature --- asyncssh/forward.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/asyncssh/forward.py b/asyncssh/forward.py index 02ce7f70..4bdbb197 100644 --- a/asyncssh/forward.py +++ b/asyncssh/forward.py @@ -326,7 +326,7 @@ def _create_tracker( @staticmethod def _notify_tracker(tracker: Optional[_Tracker], - notify: Callable[[_Tracker], None]) -> None: + notify: Callable[[_Tracker], None]) -> None: """Invoke a tracker hook, swallowing exceptions from buggy trackers""" if tracker is not None: From bc198571d3de626a9c5c3ba63235bf95f839cca3 Mon Sep 17 00:00:00 2001 From: Alex MKX Date: Thu, 6 Aug 2026 12:04:29 +0300 Subject: [PATCH 10/13] refactor(forward): share tracker plumbing between local and remote forwarders Lift the tracker ownership, the factory and hook exception guard and the exactly-once connection_lost logic out of SSHLocalForwarder into a shared SSHTrackedForwarder base, and add the remote counterparts beside the local ones. SSHLocalForwarder keeps its own byte hook mapping; the remote forwarders use the mirrored one, since the hook names say where the bytes were generated rather than which method carried them. Factor the socket-opening bodies of forward_connection() and forward_unix_connection() into private helpers which pair a caller supplied forwarder with the newly opened local destination connection. The forwarder is created before that connection is opened, so a tracked remote forwarder gets exactly one tracker per connection accepted on the remote listener and can be told the connection was lost when the local destination can't be reached. Also drop the remaining local/remote wording from the tracker class docstrings; the distinction there is TCP versus UNIX domain socket listeners, not local versus remote forwarding. --- asyncssh/connection.py | 95 ++++++++++++++++----- asyncssh/forward.py | 182 ++++++++++++++++++++++++++++++++--------- 2 files changed, 219 insertions(+), 58 deletions(-) diff --git a/asyncssh/connection.py b/asyncssh/connection.py index 2fc783b9..65b80e71 100644 --- a/asyncssh/connection.py +++ b/asyncssh/connection.py @@ -3152,6 +3152,79 @@ async def create_unix_connection( raise NotImplementedError + async def _forward_tcp_connection( + self, forwarder_factory: Callable[[], SSHForwarder], + dest_host: str, dest_port: int) -> SSHForwarder: + """Pair a new forwarder with a local TCP destination connection + + The forwarder returned by `forwarder_factory` becomes the SSH + side of the tunnel and is paired with a plain + :class:`SSHForwarder` on the newly opened local connection. + + The forwarder is created before the local connection is opened + so that a tracked forwarder reports exactly one tracker per + connection accepted on the listener, even when the local + destination turns out to be unreachable. In that case the + forwarder is told the connection was lost before the + :exc:`ChannelOpenError` is raised. + + """ + + forwarder = forwarder_factory() + + try: + _, peer = await self._loop.create_connection(SSHForwarder, + dest_host, dest_port) + + self.logger.info(' Forwarding TCP connection to %s', + (dest_host, dest_port)) + except OSError as exc: + open_error = ChannelOpenError(OPEN_CONNECT_FAILED, str(exc)) + + forwarder.connection_lost(open_error) + + raise open_error from None + + dest_forwarder = cast(SSHForwarder, peer) + + forwarder.set_peer(dest_forwarder) + dest_forwarder.set_peer(forwarder) + + return forwarder + + async def _forward_unix_connection( + self, forwarder_factory: Callable[[], SSHForwarder], + dest_path: str) -> SSHForwarder: + """Pair a new forwarder with a local UNIX destination connection + + This is the UNIX domain socket equivalent of + :meth:`_forward_tcp_connection`, with the same ordering + between creating the forwarder and opening the local + destination connection. + + """ + + forwarder = forwarder_factory() + + try: + _, peer = \ + await self._loop.create_unix_connection(SSHForwarder, dest_path) + + self.logger.info(' Forwarding UNIX connection to %s', dest_path) + except OSError as exc: + open_error = ChannelOpenError(OPEN_CONNECT_FAILED, str(exc)) + + forwarder.connection_lost(open_error) + + raise open_error from None + + dest_forwarder = cast(SSHForwarder, peer) + + forwarder.set_peer(dest_forwarder) + dest_forwarder.set_peer(forwarder) + + return forwarder + async def forward_connection( self, dest_host: str, dest_port: int) -> SSHForwarder: """Forward a tunneled TCP connection @@ -3171,16 +3244,8 @@ async def forward_connection( """ - try: - _, peer = await self._loop.create_connection(SSHForwarder, - dest_host, dest_port) - - self.logger.info(' Forwarding TCP connection to %s', - (dest_host, dest_port)) - except OSError as exc: - raise ChannelOpenError(OPEN_CONNECT_FAILED, str(exc)) from None - - return SSHForwarder(cast(SSHForwarder, peer)) + return await self._forward_tcp_connection(SSHForwarder, dest_host, + dest_port) async def forward_unix_connection(self, dest_path: str) -> SSHForwarder: """Forward a tunneled UNIX domain socket connection @@ -3197,15 +3262,7 @@ async def forward_unix_connection(self, dest_path: str) -> SSHForwarder: """ - try: - _, peer = \ - await self._loop.create_unix_connection(SSHForwarder, dest_path) - - self.logger.info(' Forwarding UNIX connection to %s', dest_path) - except OSError as exc: - raise ChannelOpenError(OPEN_CONNECT_FAILED, str(exc)) from None - - return SSHForwarder(cast(SSHForwarder, peer)) + return await self._forward_unix_connection(SSHForwarder, dest_path) @async_context_manager async def forward_local_port( diff --git a/asyncssh/forward.py b/asyncssh/forward.py index 4bdbb197..a2da9257 100644 --- a/asyncssh/forward.py +++ b/asyncssh/forward.py @@ -44,7 +44,9 @@ class SSHForwardTracker: A tracker observes a single forwarded connection. A `tracker_factory` passed to one of the :meth:`forward_local_port() ` - family of methods is called once per accepted connection and must + or :meth:`forward_remote_port() + ` family of methods is + called once per connection accepted on that listener and must return a new tracker instance, on which asyncssh then calls the hooks below for the life of that connection. @@ -56,9 +58,9 @@ class SSHForwardTracker: buggy tracker can never break forwarding. This base class defines the hooks shared by all forward types. - Use :class:`SSHPortForwardTracker` for TCP local forwards and - :class:`SSHPathForwardTracker` for UNIX domain socket local - forwards; they differ only in the signature of `connection_made`. + Use :class:`SSHPortForwardTracker` when the listener is a TCP port + and :class:`SSHPathForwardTracker` when it is a UNIX domain socket; + they differ only in the signature of `connection_made`. """ @@ -96,12 +98,16 @@ def forward_remote_bytes(self, data: bytes) -> None: class SSHPortForwardTracker(SSHForwardTracker): - """Tracker for local TCP port forwards + """Tracker for forwards with a TCP port listener Used with - :meth:`forward_local_port() ` - and :meth:`forward_local_port_to_path() - `. + :meth:`forward_local_port() `, + :meth:`forward_local_port_to_path() + `, + :meth:`forward_remote_port() + `, and + :meth:`forward_remote_port_to_path() + `. """ @@ -123,12 +129,16 @@ def connection_made(self, forwarder: 'SSHForwarder', class SSHPathForwardTracker(SSHForwardTracker): - """Tracker for local UNIX domain socket forwards + """Tracker for forwards with a UNIX domain socket listener Used with - :meth:`forward_local_path() ` - and :meth:`forward_local_path_to_port() - `. + :meth:`forward_local_path() `, + :meth:`forward_local_path_to_port() + `, + :meth:`forward_remote_path() + `, and + :meth:`forward_remote_path_to_port() + `. """ @@ -299,14 +309,24 @@ def close(self) -> None: peer.close() -class SSHLocalForwarder(SSHForwarder, Generic[_Tracker]): - """Local forwarding connection handler""" +class SSHTrackedForwarder(SSHForwarder, Generic[_Tracker]): + """Forwarding connection handler which reports to a tracker - def __init__(self, conn: 'SSHConnection', coro: SSHForwarderCoro, - tracker_factory: Optional[Callable[[], _Tracker]] = None): + This is the shared base for the forwarders which sit at the local + end of a tracked connection, whether that connection was accepted + on a local or on a remote listener. It owns the per-connection + tracker, guards every hook call against exceptions raised by a + buggy tracker, and reports the closed connection exactly once. + + Subclasses decide which of their methods report which byte hook, + since that depends on which end of the tunnel they sit on, and + report `connection_made` once they know their listener's arguments. + + """ + + def __init__( + self, tracker_factory: Optional[Callable[[], _Tracker]] = None): super().__init__() - self._conn = conn - self._coro = coro self._tracker: Optional[_Tracker] = None self._create_tracker(tracker_factory) @@ -335,6 +355,39 @@ def _notify_tracker(tracker: Optional[_Tracker], except Exception: # pylint: disable=broad-except pass + def connection_lost(self, exc: Optional[Exception]) -> None: + """Handle a closed connection + + This is also called manually when the connection could not be + fully set up -- on a channel open failure for a local forward + and on a local destination open failure for a remote one -- so + the transport's eventual close fires a second + `connection_lost(None)` on the protocol. The tracker reference + is cleared on the first call so the hook fires exactly once + per connection. + """ + + tracker, self._tracker = self._tracker, None + + def notify(tracker: _Tracker) -> None: + """Report the closed connection to the tracker""" + + tracker.connection_lost(exc) + + self._notify_tracker(tracker, notify) + + super().connection_lost(exc) + + +class SSHLocalForwarder(SSHTrackedForwarder[_Tracker]): + """Local forwarding connection handler""" + + def __init__(self, conn: 'SSHConnection', coro: SSHForwarderCoro, + tracker_factory: Optional[Callable[[], _Tracker]] = None): + super().__init__(tracker_factory) + self._conn = conn + self._coro = coro + def data_received(self, data: bytes, datatype: Optional[int] = None) -> None: """Handle incoming data from the local transport""" @@ -360,27 +413,6 @@ def notify(tracker: _Tracker) -> None: super().write(data) - def connection_lost(self, exc: Optional[Exception]) -> None: - """Handle a closed local connection - - This is also called manually from `_forward()` on a channel - open failure, so the local transport's eventual close fires - a second `connection_lost(None)` on the protocol. The tracker - reference is cleared on the first call so the hook fires - exactly once per connection. - """ - - tracker, self._tracker = self._tracker, None - - def notify(tracker: _Tracker) -> None: - """Report the closed connection to the tracker""" - - tracker.connection_lost(exc) - - self._notify_tracker(tracker, notify) - - super().connection_lost(exc) - async def _forward(self, *args: object) -> None: """Begin local forwarding""" @@ -451,3 +483,75 @@ def notify(tracker: SSHPathForwardTracker) -> None: self._notify_tracker(self._tracker, notify) self.forward() + + +class SSHRemoteForwarder(SSHTrackedForwarder[_Tracker]): + """Remote forwarding connection handler + + This handles the SSH channel opened when the remote listener + accepts a connection, paired with a plain :class:`SSHForwarder` + on the local destination connection. + + Its byte hooks are the mirror image of :class:`SSHLocalForwarder`, + because the hook names say where the bytes were generated rather + than which method carried them. Data delivered here by the SSH + channel was generated on the remote host, and data written here on + behalf of the local destination connection was generated locally. + + """ + + def data_received(self, data: bytes, + datatype: Optional[int] = None) -> None: + """Handle incoming data from the SSH channel""" + + def notify(tracker: _Tracker) -> None: + """Report remotely forwarded bytes to the tracker""" + + tracker.forward_remote_bytes(data) + + self._notify_tracker(self._tracker, notify) + + super().data_received(data, datatype) + + def write(self, data: bytes) -> None: + """Write local destination data out to the SSH channel""" + + def notify(tracker: _Tracker) -> None: + """Report locally forwarded bytes to the tracker""" + + tracker.forward_local_bytes(data) + + self._notify_tracker(self._tracker, notify) + + super().write(data) + + +class SSHRemotePortForwarder(SSHRemoteForwarder[SSHPortForwardTracker]): + """Remote TCP port forwarding connection handler""" + + def __init__(self, + tracker_factory: Optional[SSHPortForwardTrackerFactory], + orig_host: str, orig_port: int): + super().__init__(tracker_factory) + + def notify(tracker: SSHPortForwardTracker) -> None: + """Report the new connection to the tracker""" + + tracker.connection_made(self, orig_host, orig_port) + + self._notify_tracker(self._tracker, notify) + + +class SSHRemotePathForwarder(SSHRemoteForwarder[SSHPathForwardTracker]): + """Remote UNIX domain socket forwarding connection handler""" + + def __init__(self, + tracker_factory: Optional[SSHPathForwardTrackerFactory]): + super().__init__(tracker_factory) + + def notify(tracker: SSHPathForwardTracker) -> None: + """Report the new connection to the tracker""" + + tracker.connection_made(self) + + self._notify_tracker(self._tracker, notify) From 2054a2465fe2a4ac60028903857fa34a4eb10a80 Mon Sep 17 00:00:00 2001 From: Alex MKX Date: Thu, 6 Aug 2026 12:05:08 +0300 Subject: [PATCH 11/13] feat(forward): add tracker_factory to the forward_remote_* methods Add an optional tracker_factory to forward_remote_port(), forward_remote_path(), forward_remote_port_to_path() and forward_remote_path_to_port(). The tracker kind follows the listener endpoint, matching the local methods: the TCP-listener methods take an SSHPortForwardTrackerFactory and report the forwarded-tcpip origin to connection_made, and the UNIX-listener methods take an SSHPathForwardTrackerFactory. Cover both remote listener kinds with tests mirroring the local ones, plus new byte hook tests -- local and remote -- which send a request and a deliberately different reply, so a swapped pair of byte hooks fails instead of passing on an echo. Raise pylint's max-module-lines, which connection.py now exceeds. --- asyncssh/connection.py | 99 ++++++-- asyncssh/forward.py | 10 +- pylintrc | 2 +- tests/test_forward.py | 525 +++++++++++++++++++++++++++++++++++++++++ 4 files changed, 611 insertions(+), 25 deletions(-) diff --git a/asyncssh/connection.py b/asyncssh/connection.py index 65b80e71..d32b4224 100644 --- a/asyncssh/connection.py +++ b/asyncssh/connection.py @@ -87,6 +87,7 @@ from .forward import SSHForwarder from .forward import SSHPortForwardTrackerFactory, SSHPathForwardTrackerFactory +from .forward import SSHRemotePathForwarder, SSHRemotePortForwarder from .gss import GSSBase, GSSClient, GSSServer, GSSError @@ -5516,9 +5517,11 @@ async def tunnel_connection( return listener @async_context_manager - async def forward_remote_port(self, listen_host: str, - listen_port: int, dest_host: str, - dest_port: int) -> SSHListener: + async def forward_remote_port( + self, listen_host: str, listen_port: int, + dest_host: str, dest_port: int, + tracker_factory: + Optional[SSHPortForwardTrackerFactory] = None) -> SSHListener: """Set up remote port forwarding This method is a coroutine which attempts to set up port @@ -5536,10 +5539,17 @@ async def forward_remote_port(self, listen_host: str, The hostname or address to forward connections to :param dest_port: The port number to forward connections to + :param tracker_factory: + An optional callable invoked once per connection accepted + on the remote listener which returns a new + :class:`SSHPortForwardTracker` for observing that + connection's lifecycle. `None` (default) disables tracking + with no overhead. :type listen_host: `str` :type listen_port: `int` :type dest_host: `str` :type dest_port: `int` + :type tracker_factory: :class:`SSHPortForwardTrackerFactory` :returns: :class:`SSHListener` @@ -5547,12 +5557,19 @@ async def forward_remote_port(self, listen_host: str, """ - def session_factory(_orig_host: str, - _orig_port: int) -> Awaitable[SSHTCPSession]: + def session_factory(orig_host: str, + orig_port: int) -> Awaitable[SSHTCPSession]: """Return an SSHTCPSession used to do remote port forwarding""" + def forwarder_factory() -> SSHForwarder: + """Return a forwarder tracking this remote connection""" + + return SSHRemotePortForwarder(tracker_factory, orig_host, + orig_port) + return cast(Awaitable[SSHTCPSession], - self.forward_connection(dest_host, dest_port)) + self._forward_tcp_connection(forwarder_factory, + dest_host, dest_port)) self.logger.info('Creating remote TCP forwarder from %s to %s', (listen_host, listen_port), (dest_host, dest_port)) @@ -5561,8 +5578,10 @@ def session_factory(_orig_host: str, listen_port) @async_context_manager - async def forward_remote_path(self, listen_path: str, - dest_path: str) -> SSHListener: + async def forward_remote_path( + self, listen_path: str, dest_path: str, + tracker_factory: + Optional[SSHPathForwardTrackerFactory] = None) -> SSHListener: """Set up remote UNIX domain socket forwarding This method is a coroutine which attempts to set up UNIX domain @@ -5576,8 +5595,15 @@ async def forward_remote_path(self, listen_path: str, The path on the remote host to listen on :param dest_path: The path on the local host to forward connections to + :param tracker_factory: + An optional callable invoked once per connection accepted + on the remote listener which returns a new + :class:`SSHPathForwardTracker` for observing that + connection's lifecycle. `None` (default) disables tracking + with no overhead. :type listen_path: `str` :type dest_path: `str` + :type tracker_factory: :class:`SSHPathForwardTrackerFactory` :returns: :class:`SSHListener` @@ -5588,8 +5614,14 @@ async def forward_remote_path(self, listen_path: str, def session_factory() -> Awaitable[SSHUNIXSession[bytes]]: """Return an SSHUNIXSession used to do remote path forwarding""" + def forwarder_factory() -> SSHForwarder: + """Return a forwarder tracking this remote connection""" + + return SSHRemotePathForwarder(tracker_factory) + return cast(Awaitable[SSHUNIXSession[bytes]], - self.forward_unix_connection(dest_path)) + self._forward_unix_connection(forwarder_factory, + dest_path)) self.logger.info('Creating remote UNIX forwarder from %s to %s', listen_path, dest_path) @@ -5597,9 +5629,10 @@ def session_factory() -> Awaitable[SSHUNIXSession[bytes]]: return await self.create_unix_server(session_factory, listen_path) @async_context_manager - async def forward_remote_port_to_path(self, listen_host: str, - listen_port: int, - dest_path: str) -> SSHListener: + async def forward_remote_port_to_path( + self, listen_host: str, listen_port: int, dest_path: str, + tracker_factory: + Optional[SSHPortForwardTrackerFactory] = None) -> SSHListener: """Set up remote TCP port forwarding to a local UNIX domain socket This method is a coroutine which attempts to set up port @@ -5615,9 +5648,16 @@ async def forward_remote_port_to_path(self, listen_host: str, The port number on the remote host to listen on :param dest_path: The path on the local host to forward connections to + :param tracker_factory: + An optional callable invoked once per connection accepted + on the remote listener which returns a new + :class:`SSHPortForwardTracker` for observing that + connection's lifecycle. `None` (default) disables tracking + with no overhead. :type listen_host: `str` :type listen_port: `int` :type dest_path: `str` + :type tracker_factory: :class:`SSHPortForwardTrackerFactory` :returns: :class:`SSHListener` @@ -5625,12 +5665,19 @@ async def forward_remote_port_to_path(self, listen_host: str, """ - def session_factory(_orig_host: str, - _orig_port: int) -> Awaitable[SSHUNIXSession]: + def session_factory(orig_host: str, + orig_port: int) -> Awaitable[SSHUNIXSession]: """Return an SSHTCPSession used to do remote port forwarding""" + def forwarder_factory() -> SSHForwarder: + """Return a forwarder tracking this remote connection""" + + return SSHRemotePortForwarder(tracker_factory, orig_host, + orig_port) + return cast(Awaitable[SSHUNIXSession], - self.forward_unix_connection(dest_path)) + self._forward_unix_connection(forwarder_factory, + dest_path)) self.logger.info('Creating remote TCP forwarder from %s to %s', (listen_host, listen_port), dest_path) @@ -5639,9 +5686,10 @@ def session_factory(_orig_host: str, listen_port) @async_context_manager - async def forward_remote_path_to_port(self, listen_path: str, - dest_host: str, - dest_port: int) -> SSHListener: + async def forward_remote_path_to_port( + self, listen_path: str, dest_host: str, dest_port: int, + tracker_factory: + Optional[SSHPathForwardTrackerFactory] = None) -> SSHListener: """Set up remote UNIX domain socket forwarding to a local TCP port This method is a coroutine which attempts to set up UNIX domain @@ -5657,9 +5705,16 @@ async def forward_remote_path_to_port(self, listen_path: str, The hostname or address to forward connections to :param dest_port: The port number to forward connections to + :param tracker_factory: + An optional callable invoked once per connection accepted + on the remote listener which returns a new + :class:`SSHPathForwardTracker` for observing that + connection's lifecycle. `None` (default) disables tracking + with no overhead. :type listen_path: `str` :type dest_host: `str` :type dest_port: `int` + :type tracker_factory: :class:`SSHPathForwardTrackerFactory` :returns: :class:`SSHListener` @@ -5670,8 +5725,14 @@ async def forward_remote_path_to_port(self, listen_path: str, def session_factory() -> Awaitable[SSHTCPSession[bytes]]: """Return an SSHUNIXSession used to do remote path forwarding""" + def forwarder_factory() -> SSHForwarder: + """Return a forwarder tracking this remote connection""" + + return SSHRemotePathForwarder(tracker_factory) + return cast(Awaitable[SSHTCPSession[bytes]], - self.forward_connection(dest_host, dest_port)) + self._forward_tcp_connection(forwarder_factory, + dest_host, dest_port)) self.logger.info('Creating remote UNIX forwarder from %s to %s', listen_path, (dest_host, dest_port)) diff --git a/asyncssh/forward.py b/asyncssh/forward.py index a2da9257..ccca18f0 100644 --- a/asyncssh/forward.py +++ b/asyncssh/forward.py @@ -529,9 +529,9 @@ def notify(tracker: _Tracker) -> None: class SSHRemotePortForwarder(SSHRemoteForwarder[SSHPortForwardTracker]): """Remote TCP port forwarding connection handler""" - def __init__(self, - tracker_factory: Optional[SSHPortForwardTrackerFactory], - orig_host: str, orig_port: int): + def __init__( + self, tracker_factory: Optional[SSHPortForwardTrackerFactory], + orig_host: str, orig_port: int): super().__init__(tracker_factory) def notify(tracker: SSHPortForwardTracker) -> None: @@ -545,8 +545,8 @@ def notify(tracker: SSHPortForwardTracker) -> None: class SSHRemotePathForwarder(SSHRemoteForwarder[SSHPathForwardTracker]): """Remote UNIX domain socket forwarding connection handler""" - def __init__(self, - tracker_factory: Optional[SSHPathForwardTrackerFactory]): + def __init__( + self, tracker_factory: Optional[SSHPathForwardTrackerFactory]): super().__init__(tracker_factory) def notify(tracker: SSHPathForwardTracker) -> None: diff --git a/pylintrc b/pylintrc index 4a2e8716..787592e3 100644 --- a/pylintrc +++ b/pylintrc @@ -192,7 +192,7 @@ single-line-if-stmt=no no-space-check=trailing-comma,dict-separator # Maximum number of lines in a module -max-module-lines=10000 +max-module-lines=11000 # String used as indentation unit. This is usually " " (4 spaces) or "\t" (1 # tab). diff --git a/tests/test_forward.py b/tests/test_forward.py index 0d18ce5b..d0030f77 100644 --- a/tests/test_forward.py +++ b/tests/test_forward.py @@ -71,6 +71,29 @@ def _unix_listener_non_async(): return _echo_non_async +_REQUEST = b'request\n' +_RESPONSE = b'a distinctly different response\n' + + +async def _distinct_reply(reader, writer): + """Answer a request with a response which is not an echo of it + + Tracker byte hooks are named after the host where the bytes were + generated, so an echo destination would pass even if the two hooks + were wired backwards. This destination makes the two directions + carry distinguishable data. + + """ + + await reader.readline() + + writer.write(_RESPONSE) + await writer.drain() + + writer.close() + await maybe_wait_closed(writer) + + async def _pause(reader, writer): """Sleep to allow buffered data to build up and trigger a pause""" @@ -159,6 +182,8 @@ def connection_requested(self, dest_host, dest_port, orig_host, orig_port): return (self._conn.create_tcp_channel(), echo) elif dest_port == 10: return _async_runtime_error + elif dest_port == 11: + return _distinct_reply else: return True @@ -297,6 +322,20 @@ async def _check_local_connection(self, listen_port, delay=None): await self._check_echo_line(reader, writer, delay=delay) + async def _check_distinct_reply(self, listen_port): + """Open a local connection and check the non-echoed reply to it""" + + reader, writer = await asyncio.open_connection('127.0.0.1', + listen_port) + + writer.write(_REQUEST) + await writer.drain() + + self.assertEqual((await reader.readline()), _RESPONSE) + + writer.close() + await maybe_wait_closed(writer) + async def _check_local_unix_connection(self, listen_path): """Open a local connection and test if an input line is echoed back""" @@ -794,6 +833,42 @@ def connection_lost(self, exc): self.assertEqual(bytes(local_bytes), line) self.assertEqual(bytes(remote_bytes), line) + @asynctest + async def test_port_tracker_byte_hook_direction(self): + """Byte hooks report where locally forwarded bytes were generated + + The request is generated by the client of the local listener + and the reply by the remote destination, and the two are + deliberately different, so this fails if the hooks are swapped. + + """ + + local_bytes = bytearray() + remote_bytes = bytearray() + lost = asyncio.Event() + + class _ByteTracker(asyncssh.SSHPortForwardTracker): + """Tracker recording bytes seen in each forwarding direction""" + + def forward_local_bytes(self, data): + local_bytes.extend(data) + + def forward_remote_bytes(self, data): + remote_bytes.extend(data) + + def connection_lost(self, exc): + lost.set() + + async with self.connect() as conn: + async with conn.forward_local_port( + '', 0, '', 11, + tracker_factory=_ByteTracker) as listener: + await self._check_distinct_reply(listener.get_port()) + await asyncio.wait_for(lost.wait(), timeout=1.0) + + self.assertEqual(bytes(local_bytes), _REQUEST) + self.assertEqual(bytes(remote_bytes), _RESPONSE) + @asynctest async def test_port_tracker_factory_exception_swallowed(self): """A factory that raises does not break forwarding""" @@ -1024,6 +1099,279 @@ async def test_forward_remote_port_to_path(self): try_remove('local') + @asynctest + async def test_remote_port_tracker_made_and_lost(self): + """A remote port tracker sees connection_made and connection_lost""" + + events = [] + lost = asyncio.Event() + + class _RecordingTracker(asyncssh.SSHPortForwardTracker): + """Tracker which records connection_made and connection_lost""" + + def connection_made(self, forwarder, orig_host, orig_port): + events.append(('made', forwarder, orig_host, orig_port)) + + def connection_lost(self, exc): + events.append(('lost', exc)) + lost.set() + + server = await asyncio.start_server(echo, None, 0, + family=socket.AF_INET) + server_port = server.sockets[0].getsockname()[1] + + async with self.connect() as conn: + async with conn.forward_remote_port( + '', 0, '127.0.0.1', server_port, + tracker_factory=_RecordingTracker) as listener: + await self._check_local_connection(listener.get_port()) + await asyncio.wait_for(lost.wait(), timeout=1.0) + + server.close() + await server.wait_closed() + + kinds = [event[0] for event in events] + self.assertIn('made', kinds) + self.assertIn('lost', kinds) + + made = next(event for event in events if event[0] == 'made') + self.assertIsInstance(made[1], asyncssh.SSHForwarder) + self.assertEqual(made[2], '127.0.0.1') + self.assertIsInstance(made[3], int) + + @asynctest + async def test_remote_port_tracker_factory_per_connection(self): + """A distinct tracker instance is created for each connection + accepted on the remote listener""" + + trackers = [] + lost_events = [] + + class _CountingTracker(asyncssh.SSHPortForwardTracker): + """Tracker which records each instance created by the factory""" + + def __init__(self): + trackers.append(self) + lost_events.append(asyncio.Event()) + + def connection_lost(self, exc): + lost_events[trackers.index(self)].set() + + def factory(): + """Return a new counting tracker""" + + return _CountingTracker() + + server = await asyncio.start_server(echo, None, 0, + family=socket.AF_INET) + server_port = server.sockets[0].getsockname()[1] + + async with self.connect() as conn: + async with conn.forward_remote_port( + '', 0, '127.0.0.1', server_port, + tracker_factory=factory) as listener: + listen_port = listener.get_port() + await self._check_local_connection(listen_port) + await self._check_local_connection(listen_port) + await asyncio.wait_for( + asyncio.gather(*(e.wait() for e in lost_events)), + timeout=1.0) + + server.close() + await server.wait_closed() + + self.assertEqual(len(trackers), 2) + + @asynctest + async def test_remote_port_tracker_byte_hooks(self): + """Byte hooks report where remotely forwarded bytes were generated + + The request is generated by the client of the remote listener + and the reply by the local destination, and the two are + deliberately different, so this fails if the hooks are swapped. + + """ + + local_bytes = bytearray() + remote_bytes = bytearray() + lost = asyncio.Event() + + class _ByteTracker(asyncssh.SSHPortForwardTracker): + """Tracker recording bytes seen in each forwarding direction""" + + def forward_local_bytes(self, data): + local_bytes.extend(data) + + def forward_remote_bytes(self, data): + remote_bytes.extend(data) + + def connection_lost(self, exc): + lost.set() + + server = await asyncio.start_server(_distinct_reply, None, 0, + family=socket.AF_INET) + server_port = server.sockets[0].getsockname()[1] + + async with self.connect() as conn: + async with conn.forward_remote_port( + '', 0, '127.0.0.1', server_port, + tracker_factory=_ByteTracker) as listener: + await self._check_distinct_reply(listener.get_port()) + await asyncio.wait_for(lost.wait(), timeout=1.0) + + server.close() + await server.wait_closed() + + self.assertEqual(bytes(remote_bytes), _REQUEST) + self.assertEqual(bytes(local_bytes), _RESPONSE) + + @asynctest + async def test_remote_port_tracker_factory_exception_swallowed(self): + """A remote factory that raises does not break forwarding""" + + def factory(): + """Fail to return a tracker""" + + raise RuntimeError('factory boom') + + server = await asyncio.start_server(echo, None, 0, + family=socket.AF_INET) + server_port = server.sockets[0].getsockname()[1] + + async with self.connect() as conn: + async with conn.forward_remote_port( + '', 0, '127.0.0.1', server_port, + tracker_factory=factory) as listener: + await self._check_local_connection(listener.get_port()) + + server.close() + await server.wait_closed() + + @asynctest + async def test_remote_port_tracker_hook_exception_swallowed(self): + """A remote tracker whose hooks raise does not break forwarding""" + + class _BuggyTracker(asyncssh.SSHPortForwardTracker): + """Tracker whose hooks all raise, to verify they're swallowed""" + + def connection_made(self, forwarder, orig_host, orig_port): + raise RuntimeError('made boom') + + def connection_lost(self, exc): + raise RuntimeError('lost boom') + + def forward_local_bytes(self, data): + raise RuntimeError('local boom') + + def forward_remote_bytes(self, data): + raise RuntimeError('remote boom') + + server = await asyncio.start_server(_distinct_reply, None, 0, + family=socket.AF_INET) + server_port = server.sockets[0].getsockname()[1] + + async with self.connect() as conn: + async with conn.forward_remote_port( + '', 0, '127.0.0.1', server_port, + tracker_factory=_BuggyTracker) as listener: + await self._check_distinct_reply(listener.get_port()) + + server.close() + await server.wait_closed() + + @asynctest + async def test_remote_port_tracker_lost_fires_once(self): + """connection_lost fires once on a remote forward, including when + the local destination connection can't be opened""" + + lost_count = 0 + + class _Counting(asyncssh.SSHPortForwardTracker): + """Tracker which counts how many times connection_lost fires""" + + def connection_lost(self, exc): + nonlocal lost_count + lost_count += 1 + + server = await asyncio.start_server(echo, None, 0, + family=socket.AF_INET) + server_port = server.sockets[0].getsockname()[1] + + async with self.connect() as conn: + async with conn.forward_remote_port( + '', 0, '127.0.0.1', server_port, + tracker_factory=_Counting) as listener: + await self._check_local_connection(listener.get_port()) + await asyncio.sleep(0.1) + + server.close() + await server.wait_closed() + + self.assertEqual(lost_count, 1) + + # A destination which refuses the connection must still produce + # exactly one tracker, made and then immediately lost. + + lost_count = 0 + + sock = socket.socket() + sock.bind(('127.0.0.1', 0)) + closed_port = sock.getsockname()[1] + sock.close() + + async with self.connect() as conn: + async with conn.forward_remote_port( + '', 0, '127.0.0.1', closed_port, + tracker_factory=_Counting) as listener: + reader, writer = await asyncio.open_connection( + '127.0.0.1', listener.get_port()) + + self.assertEqual((await reader.read()), b'') + + writer.close() + await maybe_wait_closed(writer) + await asyncio.sleep(0.1) + + self.assertEqual(lost_count, 1) + + @unittest.skipIf(sys.platform == 'win32', + 'skip UNIX domain socket tests on Windows') + @asynctest + async def test_forward_remote_port_to_path_tracker(self): + """A remote TCP listener to a local path uses a port tracker""" + + events = [] + lost = asyncio.Event() + + class _RecordingTracker(asyncssh.SSHPortForwardTracker): + """Tracker which records connection_made and connection_lost""" + + def connection_made(self, forwarder, orig_host, orig_port): + events.append(('made', forwarder, orig_host, orig_port)) + + def connection_lost(self, exc): + events.append(('lost', exc)) + lost.set() + + server = await asyncio.start_unix_server(echo, 'local') + + async with self.connect() as conn: + async with conn.forward_remote_port_to_path( + '', 0, 'local', + tracker_factory=_RecordingTracker) as listener: + await self._check_local_connection(listener.get_port()) + await asyncio.wait_for(lost.wait(), timeout=1.0) + + server.close() + await server.wait_closed() + + try_remove('local') + + made = next(event for event in events if event[0] == 'made') + self.assertIsInstance(made[1], asyncssh.SSHForwarder) + self.assertEqual(made[2], '127.0.0.1') + self.assertIsInstance(made[3], int) + @asynctest async def test_forward_remote_specific_port(self): """Test forwarding of a specific remote port""" @@ -1468,6 +1816,183 @@ async def test_forward_remote_path_to_port(self): try_remove('echo') + @asynctest + async def test_remote_path_tracker_made_and_lost(self): + """A remote path tracker sees connection_made (no addr) and + connection_lost""" + + events = [] + lost = asyncio.Event() + + class _RecordingTracker(asyncssh.SSHPathForwardTracker): + """Tracker which records connection_made and connection_lost""" + + def connection_made(self, forwarder): + events.append(('made', forwarder)) + + def connection_lost(self, exc): + events.append(('lost', exc)) + lost.set() + + # pylint doesn't think start_unix_server exists + # pylint: disable=no-member + server = await asyncio.start_unix_server(echo, 'local') + # pylint: enable=no-member + + path = os.path.abspath('echo') + + async with self.connect() as conn: + async with conn.forward_remote_path( + path, 'local', tracker_factory=_RecordingTracker): + await self._check_local_unix_connection('echo') + await asyncio.wait_for(lost.wait(), timeout=1.0) + + server.close() + await server.wait_closed() + + try_remove('echo') + try_remove('local') + + kinds = [event[0] for event in events] + self.assertIn('made', kinds) + self.assertIn('lost', kinds) + + made = next(event for event in events if event[0] == 'made') + self.assertIsInstance(made[1], asyncssh.SSHForwarder) + + @asynctest + async def test_remote_path_tracker_byte_hooks(self): + """Remote path byte hooks report where the bytes were generated""" + + local_bytes = bytearray() + remote_bytes = bytearray() + lost = asyncio.Event() + + class _ByteTracker(asyncssh.SSHPathForwardTracker): + """Tracker recording bytes seen in each forwarding direction""" + + def forward_local_bytes(self, data): + local_bytes.extend(data) + + def forward_remote_bytes(self, data): + remote_bytes.extend(data) + + def connection_lost(self, exc): + lost.set() + + # pylint doesn't think start_unix_server exists + # pylint: disable=no-member + server = await asyncio.start_unix_server(_distinct_reply, 'local') + # pylint: enable=no-member + + path = os.path.abspath('echo') + + async with self.connect() as conn: + async with conn.forward_remote_path( + path, 'local', tracker_factory=_ByteTracker): + # pylint: disable=no-member + reader, writer = await asyncio.open_unix_connection('echo') + # pylint: enable=no-member + + writer.write(_REQUEST) + await writer.drain() + + self.assertEqual((await reader.readline()), _RESPONSE) + + writer.close() + await maybe_wait_closed(writer) + + await asyncio.wait_for(lost.wait(), timeout=1.0) + + server.close() + await server.wait_closed() + + try_remove('echo') + try_remove('local') + + self.assertEqual(bytes(remote_bytes), _REQUEST) + self.assertEqual(bytes(local_bytes), _RESPONSE) + + @asynctest + async def test_remote_path_tracker_hook_exception_swallowed(self): + """A remote path tracker whose hooks raise does not break + forwarding""" + + class _BuggyTracker(asyncssh.SSHPathForwardTracker): + """Tracker whose hooks all raise, to verify they're swallowed""" + + def connection_made(self, forwarder): + raise RuntimeError('made boom') + + def connection_lost(self, exc): + raise RuntimeError('lost boom') + + def forward_local_bytes(self, data): + raise RuntimeError('local boom') + + def forward_remote_bytes(self, data): + raise RuntimeError('remote boom') + + # pylint doesn't think start_unix_server exists + # pylint: disable=no-member + server = await asyncio.start_unix_server(echo, 'local') + # pylint: enable=no-member + + path = os.path.abspath('echo') + + async with self.connect() as conn: + async with conn.forward_remote_path( + path, 'local', tracker_factory=_BuggyTracker): + await self._check_local_unix_connection('echo') + + server.close() + await server.wait_closed() + + try_remove('echo') + try_remove('local') + + @asynctest + async def test_forward_remote_path_to_port_tracker(self): + """A remote path listener to a local TCP port uses a path tracker""" + + events = [] + lost = asyncio.Event() + + class _RecordingTracker(asyncssh.SSHPathForwardTracker): + """Tracker which records connection_made and connection_lost""" + + def connection_made(self, forwarder): + events.append(('made', forwarder)) + + def connection_lost(self, exc): + events.append(('lost', exc)) + lost.set() + + server = await asyncio.start_server(echo, None, 0, + family=socket.AF_INET) + server_port = server.sockets[0].getsockname()[1] + + path = os.path.abspath('echo') + + async with self.connect() as conn: + async with conn.forward_remote_path_to_port( + path, '127.0.0.1', server_port, + tracker_factory=_RecordingTracker): + await self._check_local_unix_connection('echo') + await asyncio.wait_for(lost.wait(), timeout=1.0) + + server.close() + await server.wait_closed() + + try_remove('echo') + + kinds = [event[0] for event in events] + self.assertIn('made', kinds) + self.assertIn('lost', kinds) + + made = next(event for event in events if event[0] == 'made') + self.assertIsInstance(made[1], asyncssh.SSHForwarder) + @asynctest async def test_forward_remote_path_failure(self): """Test failure of forwarding a remote UNIX domain path""" From 78e5e7ee5dbc0c90f1b3fef1e05ee014c7de5418 Mon Sep 17 00:00:00 2001 From: Alex MKX Date: Thu, 6 Aug 2026 12:05:08 +0300 Subject: [PATCH 12/13] docs(api): document tracker support for the forward_remote_* methods Widen the tracker guide from forward_local_* to both method families, state the listener-endpoint rule which picks the tracker class, spell out that the byte hooks name the host where the bytes were generated rather than the direction the connection was set up in, and show the same tracker class used with a remote listener. --- docs/api.rst | 46 ++++++++++++++++++++++++++++++++++++---------- 1 file changed, 36 insertions(+), 10 deletions(-) diff --git a/docs/api.rst b/docs/api.rst index a752a876..96017376 100644 --- a/docs/api.rst +++ b/docs/api.rst @@ -1012,9 +1012,10 @@ Forwarder Classes Forward Tracker Classes ======================= -The ``forward_local_*`` methods on :class:`SSHClientConnection` accept an -optional ``tracker_factory`` argument: a zero-argument callable invoked -once per accepted connection which returns a tracker instance -- +The ``forward_local_*`` and ``forward_remote_*`` methods on +:class:`SSHClientConnection` accept an optional ``tracker_factory`` +argument: a zero-argument callable invoked once per connection accepted +on that listener which returns a tracker instance -- :class:`SSHPortForwardTracker` for TCP listeners or :class:`SSHPathForwardTracker` for UNIX domain listeners. asyncssh then calls that instance's hooks for the life of the connection, giving @@ -1028,17 +1029,34 @@ ignored). Every hook has a no-op default, so a subclass overrides only what it needs, and exceptions raised by a hook or factory are caught and discarded so a buggy tracker cannot break forwarding. -Use :class:`SSHPortForwardTracker` with the TCP-listener methods -(:meth:`forward_local_port() ` and +Which tracker class to use is decided by the listener endpoint, not by +the destination. Use :class:`SSHPortForwardTracker` with the methods +which listen on a TCP port (:meth:`forward_local_port() +`, :meth:`forward_local_port_to_path() -`) and -:class:`SSHPathForwardTracker` with the UNIX-domain-listener methods -(:meth:`forward_local_path() ` and +`, +:meth:`forward_remote_port() `, +and :meth:`forward_remote_port_to_path() +`) and +:class:`SSHPathForwardTracker` with the methods which listen on a UNIX +domain socket (:meth:`forward_local_path() +`, :meth:`forward_local_path_to_port() -`). The two classes share -the same set of hooks and differ only in the signature of +`, +:meth:`forward_remote_path() `, +and :meth:`forward_remote_path_to_port() +`). The two classes +share the same set of hooks and differ only in the signature of ``connection_made``. +The two byte hooks are named after the host where the bytes were +generated, not after the direction the connection was set up in. For a +local forward, ``forward_local_bytes`` sees what the client of the local +listener sent and ``forward_remote_bytes`` sees what came back over the +SSH connection. For a remote forward, ``forward_remote_bytes`` sees what +the client of the remote listener sent and ``forward_local_bytes`` sees +what the local destination sent back. + .. code-block:: python class ConnCounter(asyncssh.SSHPortForwardTracker): @@ -1055,6 +1073,14 @@ the same set of hooks and differ only in the signature of '', 0, 'remote-host', 80, tracker_factory=lambda: ConnCounter(counter)) + # The same tracker class works for a remote TCP listener, where + # connection_made reports the client which connected to the + # listening port opened on the SSH server + + listener = await conn.forward_remote_port( + '', 8080, 'localhost', 80, + tracker_factory=lambda: ConnCounter(counter)) + .. autoclass:: SSHPortForwardTracker() ==================================== = From 96624f4dea5d96556f8bf8c0a82055cc922e8712 Mon Sep 17 00:00:00 2001 From: Alex MKX Date: Thu, 6 Aug 2026 12:28:39 +0300 Subject: [PATCH 13/13] fix(forward): report connection_lost when a destination open is cancelled The remote forwarding helpers create the forwarder (and therefore fire the tracker's connection_made) before opening the local destination socket, but only mapped OSError. A cancellation, or any other exception raised while opening that socket, left the tracker having seen connection_made with no matching connection_lost, breaking the documented "exactly one tracker lifecycle per accepted connection" contract. Deliver connection_lost on every exceptional exit, passing None rather than a BaseException such as CancelledError, which is not an Exception. The OSError to ChannelOpenError mapping is unchanged. Also strengthen the tracker tests: the refused-destination cases now assert the tracker count and the ordered made/lost sequence including the ChannelOpenError, and the buggy-tracker tests now record entry into each hook and assert all four ran, so they fail if a hook dispatch is dropped. Use a named factory instead of a lambda in the docs example. --- asyncssh/connection.py | 10 ++ docs/api.rst | 7 +- tests/test_forward.py | 201 +++++++++++++++++++++++++++++++++++++++-- 3 files changed, 206 insertions(+), 12 deletions(-) diff --git a/asyncssh/connection.py b/asyncssh/connection.py index d32b4224..39061378 100644 --- a/asyncssh/connection.py +++ b/asyncssh/connection.py @@ -3185,6 +3185,11 @@ async def _forward_tcp_connection( forwarder.connection_lost(open_error) raise open_error from None + except BaseException as exc: + forwarder.connection_lost( + exc if isinstance(exc, Exception) else None) + + raise dest_forwarder = cast(SSHForwarder, peer) @@ -3218,6 +3223,11 @@ async def _forward_unix_connection( forwarder.connection_lost(open_error) raise open_error from None + except BaseException as exc: + forwarder.connection_lost( + exc if isinstance(exc, Exception) else None) + + raise dest_forwarder = cast(SSHForwarder, peer) diff --git a/docs/api.rst b/docs/api.rst index 96017376..1801963b 100644 --- a/docs/api.rst +++ b/docs/api.rst @@ -1069,9 +1069,12 @@ what the local destination sent back. def connection_lost(self, exc): self._counter.active -= 1 + def tracker_factory(): + return ConnCounter(counter) + listener = await conn.forward_local_port( '', 0, 'remote-host', 80, - tracker_factory=lambda: ConnCounter(counter)) + tracker_factory=tracker_factory) # The same tracker class works for a remote TCP listener, where # connection_made reports the client which connected to the @@ -1079,7 +1082,7 @@ what the local destination sent back. listener = await conn.forward_remote_port( '', 8080, 'localhost', 80, - tracker_factory=lambda: ConnCounter(counter)) + tracker_factory=tracker_factory) .. autoclass:: SSHPortForwardTracker() diff --git a/tests/test_forward.py b/tests/test_forward.py index d0030f77..568d3266 100644 --- a/tests/test_forward.py +++ b/tests/test_forward.py @@ -30,6 +30,8 @@ from unittest.mock import patch import asyncssh +from asyncssh.constants import OPEN_CONNECT_FAILED +from asyncssh.forward import SSHRemotePathForwarder, SSHRemotePortForwarder from asyncssh.misc import maybe_wait_closed, write_file from asyncssh.packet import String, UInt32 from asyncssh.public_key import CERT_TYPE_USER @@ -935,7 +937,8 @@ async def deny(_orig_host, _orig_port): self.assertEqual((await reader.read()), b'') writer.close() await maybe_wait_closed(writer) - await asyncio.sleep(0.1) # bounded: upper-bound for any spurious duplicate + # bounded wait, to catch any spurious duplicate + await asyncio.sleep(0.1) self.assertEqual(lost_count, 1) @@ -1251,19 +1254,27 @@ def factory(): async def test_remote_port_tracker_hook_exception_swallowed(self): """A remote tracker whose hooks raise does not break forwarding""" + hooks = set() + lost = asyncio.Event() + class _BuggyTracker(asyncssh.SSHPortForwardTracker): """Tracker whose hooks all raise, to verify they're swallowed""" def connection_made(self, forwarder, orig_host, orig_port): + hooks.add('connection_made') raise RuntimeError('made boom') def connection_lost(self, exc): + hooks.add('connection_lost') + lost.set() raise RuntimeError('lost boom') def forward_local_bytes(self, data): + hooks.add('forward_local_bytes') raise RuntimeError('local boom') def forward_remote_bytes(self, data): + hooks.add('forward_remote_bytes') raise RuntimeError('remote boom') server = await asyncio.start_server(_distinct_reply, None, 0, @@ -1275,23 +1286,38 @@ def forward_remote_bytes(self, data): '', 0, '127.0.0.1', server_port, tracker_factory=_BuggyTracker) as listener: await self._check_distinct_reply(listener.get_port()) + await asyncio.wait_for(lost.wait(), timeout=1.0) server.close() await server.wait_closed() + self.assertEqual(hooks, {'connection_made', 'connection_lost', + 'forward_local_bytes', + 'forward_remote_bytes'}) + @asynctest async def test_remote_port_tracker_lost_fires_once(self): """connection_lost fires once on a remote forward, including when the local destination connection can't be opened""" - lost_count = 0 + trackers = [] + events = [] class _Counting(asyncssh.SSHPortForwardTracker): - """Tracker which counts how many times connection_lost fires""" + """Tracker which records a connection lifecycle""" + + def connection_made(self, forwarder, orig_host, orig_port): + events.append(('made', forwarder)) def connection_lost(self, exc): - nonlocal lost_count - lost_count += 1 + events.append(('lost', exc)) + + def tracker_factory(): + """Create and record a tracker for each accepted connection""" + + tracker = _Counting() + trackers.append(tracker) + return tracker server = await asyncio.start_server(echo, None, 0, family=socket.AF_INET) @@ -1300,19 +1326,21 @@ def connection_lost(self, exc): async with self.connect() as conn: async with conn.forward_remote_port( '', 0, '127.0.0.1', server_port, - tracker_factory=_Counting) as listener: + tracker_factory=tracker_factory) as listener: await self._check_local_connection(listener.get_port()) await asyncio.sleep(0.1) server.close() await server.wait_closed() - self.assertEqual(lost_count, 1) + self.assertEqual(len(trackers), 1) + self.assertEqual([event[0] for event in events], ['made', 'lost']) # A destination which refuses the connection must still produce # exactly one tracker, made and then immediately lost. - lost_count = 0 + trackers.clear() + events.clear() sock = socket.socket() sock.bind(('127.0.0.1', 0)) @@ -1322,7 +1350,7 @@ def connection_lost(self, exc): async with self.connect() as conn: async with conn.forward_remote_port( '', 0, '127.0.0.1', closed_port, - tracker_factory=_Counting) as listener: + tracker_factory=tracker_factory) as listener: reader, writer = await asyncio.open_connection( '127.0.0.1', listener.get_port()) @@ -1332,7 +1360,54 @@ def connection_lost(self, exc): await maybe_wait_closed(writer) await asyncio.sleep(0.1) - self.assertEqual(lost_count, 1) + self.assertEqual(len(trackers), 1) + self.assertEqual([event[0] for event in events], ['made', 'lost']) + self.assertIsInstance(events[1][1], asyncssh.ChannelOpenError) + self.assertEqual(events[1][1].code, OPEN_CONNECT_FAILED) + + @asynctest + async def test_remote_port_tracker_cancelled_destination(self): + """A cancelled TCP destination connection closes its tracker""" + + trackers = [] + events = [] + + class _RecordingTracker(asyncssh.SSHPortForwardTracker): + """Tracker which records a connection lifecycle""" + + def connection_made(self, forwarder, orig_host, orig_port): + events.append('made') + + def connection_lost(self, exc): + events.append(('lost', exc)) + + def tracker_factory(): + """Create and record a tracker for each accepted connection""" + + tracker = _RecordingTracker() + trackers.append(tracker) + return tracker + + def forwarder_factory(): + """Create a tracker-enabled remote TCP forwarder""" + + return SSHRemotePortForwarder(tracker_factory, 'orig', 1) + + async def cancelled(*args, **kwargs): + """Cancel the local destination connection""" + + raise asyncio.CancelledError + + async with self.connect() as conn: + # pylint: disable=protected-access + with patch.object(conn._loop, 'create_connection', cancelled): + with self.assertRaises(asyncio.CancelledError): + await conn._forward_tcp_connection(forwarder_factory, + 'dest', 1) + # pylint: enable=protected-access + + self.assertEqual(len(trackers), 1) + self.assertEqual(events, ['made', ('lost', None)]) @unittest.skipIf(sys.platform == 'win32', 'skip UNIX domain socket tests on Windows') @@ -1918,19 +1993,27 @@ async def test_remote_path_tracker_hook_exception_swallowed(self): """A remote path tracker whose hooks raise does not break forwarding""" + hooks = set() + lost = asyncio.Event() + class _BuggyTracker(asyncssh.SSHPathForwardTracker): """Tracker whose hooks all raise, to verify they're swallowed""" def connection_made(self, forwarder): + hooks.add('connection_made') raise RuntimeError('made boom') def connection_lost(self, exc): + hooks.add('connection_lost') + lost.set() raise RuntimeError('lost boom') def forward_local_bytes(self, data): + hooks.add('forward_local_bytes') raise RuntimeError('local boom') def forward_remote_bytes(self, data): + hooks.add('forward_remote_bytes') raise RuntimeError('remote boom') # pylint doesn't think start_unix_server exists @@ -1944,6 +2027,7 @@ def forward_remote_bytes(self, data): async with conn.forward_remote_path( path, 'local', tracker_factory=_BuggyTracker): await self._check_local_unix_connection('echo') + await asyncio.wait_for(lost.wait(), timeout=1.0) server.close() await server.wait_closed() @@ -1951,6 +2035,103 @@ def forward_remote_bytes(self, data): try_remove('echo') try_remove('local') + self.assertEqual(hooks, {'connection_made', 'connection_lost', + 'forward_local_bytes', + 'forward_remote_bytes'}) + + @asynctest + async def test_remote_path_tracker_lost_on_refused_destination(self): + """A refused UNIX destination reports a complete tracker lifecycle""" + + trackers = [] + events = [] + lost = asyncio.Event() + + class _RecordingTracker(asyncssh.SSHPathForwardTracker): + """Tracker which records a connection lifecycle""" + + def connection_made(self, forwarder): + events.append(('made', forwarder)) + + def connection_lost(self, exc): + events.append(('lost', exc)) + lost.set() + + def tracker_factory(): + """Create and record a tracker for each accepted connection""" + + tracker = _RecordingTracker() + trackers.append(tracker) + return tracker + + path = os.path.abspath('echo') + try_remove('echo') + try_remove('missing') + + async with self.connect() as conn: + async with conn.forward_remote_path( + path, 'missing', tracker_factory=tracker_factory): + # pylint: disable=no-member + reader, writer = await asyncio.open_unix_connection('echo') + # pylint: enable=no-member + + self.assertEqual((await reader.read()), b'') + + writer.close() + await maybe_wait_closed(writer) + await asyncio.wait_for(lost.wait(), timeout=1.0) + + try_remove('echo') + + self.assertEqual(len(trackers), 1) + self.assertEqual([event[0] for event in events], ['made', 'lost']) + self.assertIsInstance(events[1][1], asyncssh.ChannelOpenError) + self.assertEqual(events[1][1].code, OPEN_CONNECT_FAILED) + + @asynctest + async def test_remote_path_tracker_cancelled_destination(self): + """A cancelled UNIX destination connection closes its tracker""" + + trackers = [] + events = [] + + class _RecordingTracker(asyncssh.SSHPathForwardTracker): + """Tracker which records a connection lifecycle""" + + def connection_made(self, forwarder): + events.append('made') + + def connection_lost(self, exc): + events.append(('lost', exc)) + + def tracker_factory(): + """Create and record a tracker for each accepted connection""" + + tracker = _RecordingTracker() + trackers.append(tracker) + return tracker + + def forwarder_factory(): + """Create a tracker-enabled remote UNIX forwarder""" + + return SSHRemotePathForwarder(tracker_factory) + + async def cancelled(*args, **kwargs): + """Cancel the local destination connection""" + + raise asyncio.CancelledError + + async with self.connect() as conn: + # pylint: disable=protected-access + with patch.object(conn._loop, 'create_unix_connection', cancelled): + with self.assertRaises(asyncio.CancelledError): + await conn._forward_unix_connection(forwarder_factory, + 'dest') + # pylint: enable=protected-access + + self.assertEqual(len(trackers), 1) + self.assertEqual(events, ['made', ('lost', None)]) + @asynctest async def test_forward_remote_path_to_port_tracker(self): """A remote path listener to a local TCP port uses a path tracker"""