Getting frequent AssertionError exceptions when forwarding TCP

[misa-hase]

I'm running asyncssh 2.21.0 on my server as a reverse SSH listener - using asyncssh.listen_reverse - which forwards 2 TCP ports to my local host using forward_local_port.
It's been working great but as the number of connections to the asyncssh server have increased I've started getting frequent forwarding-related AssertionErrors on connection_lost callback:

 Traceback (most recent call last):
  File "asyncssh/connection.py", line 1529, in _recv_data
  File "asyncssh/connection.py", line 1689, in _recv_packet
  File "asyncssh/packet.py", line 238, in process_packet
  File "asyncssh/channel.py", line 628, in _process_eof
  File "asyncssh/channel.py", line 357, in _flush_recv_buf
  File "asyncssh/forward.py", line 155, in eof_received
  File "asyncssh/forward.py", line 94, in write_eof
AssertionError

The assertion itself seems to be assert self._transport is not None from write_eof in asyncssh/forward.py.
For whatever reason my self._transport ends up becoming None?

[ronf]

One way this can happen is if you make calls on an AsyncSSH SSHForwarder after calling close() on it or implicitly closing it when using it as a context manager and exiting the block. Are you doing anything like that in your code as you try to clean things up?

[misa-hase]

Thanks for the reply!

In the acceptor function that I pass to the asyncssh.listen_reverse, I do the following:

1. I create 1x forwarding Unix socket using forward_local_path_to_port and 2x TCP ones using forward_local_port
2. I also create an asyncio task that occasionally sends files via SFTP
3. Then I call await conn.wait_closed() in a try-finally block to wait until connection is torn down. (I don't actually close the connection on either side unless the application is being shutdown.)
4. In the finally block of conn.wait_closed I call close and wait_closed on every listener.

I don't really do much in my own SSHClient class I pass as client_factory either, I only allow 1 connection at a time so if there's more than 1 client connecting I drop them with abort on connection_made. I don't really think this could be the issue as connection_made is called on SSH connections not forwarding ones IIRC.

Other than that I don't really interact with the connection at all. I have enabled keepalive, but I expect connections closed because of keepalive to be closed with with the exception asyncssh.misc.ConnectionLost: Server not responding to keepalive.

Not sure if it's related but the Unix socket gets most of the connections and they are long lasting Websocket connections.

Here's a very simplified view of my SSH acceptor:

async def ssh_acceptor(conn: asyncssh.SSHClientConnection) -> None:
	listeners = []
	listeners.append(await conn.forward_local_port(...))
	listeners.append(await conn.forward_local_port(...)) 
	listeners.append(await conn.forward_local_path_to_port(...))

	try:
		# Create task that occasionally sends files
		sftp_task = asyncio.create_task(...)

		await conn.wait_closed()
	finally:
		sftp_task.cancel()
		await sftp_task

		for listener in listeners:
			listener.close()
		 	await listener.wait_closed()

[misa-hase]

Seems like this issue is exacerbated when SFTP transfers are happening frequently.

[ronf]

One thing I noticed here is that you are cleaning up the listeners for the forwarded ports, but not any existing opened tunnels on those ports. When the SSH connection closes, AsyncSSH should do this automatically for you, as well as automatically cleaning up any listeners which are still active, but there could be race conditions in there if both AsyncSSH and your code is trying to clean things up in parallel. If you comment out the explicit calls to listener.close(), does that make a difference?

Also, note that you can get multiple connections to the forwarded ports, and each will open its own tunnel over the SSH connection. Do you want that, or do you want to limit it to one of each of these tunnels per SSH connection? You could use an accept_handler() to limit things to one per connection, and potentially even shut down the listener after the first connection of each type arrives.

[misa-hase]

One thing I noticed here is that you are cleaning up the listeners for the forwarded ports, but not any existing opened tunnels on those ports. When the SSH connection closes, AsyncSSH should do this automatically for you, as well as automatically cleaning up any listeners which are still active, but there could be race conditions in there if both AsyncSSH and your code is trying to clean things up in parallel. If you comment out the explicit calls to listener.close(), does that make a difference?

Ah, I wasn't aware of that. I did comment out the listener.close() calls but still got the AssertionError after a while

Also, note that you can get multiple connections to the forwarded ports, and each will open its own tunnel over the SSH connection. Do you want that, or do you want to limit it to one of each of these tunnels per SSH connection? You could use an accept_handler() to limit things to one per connection, and potentially even shut down the listener after the first connection of each type arrives.

I do want to get multiple connections to the forwarded ports. Is there a limit to the number of connections that can be forwarded over a single SSH connection?
My code is basically a fancy reverse proxy where the backend connects to the asyncssh-based reverse listener and the reverse listener forwards the ports I'd mentioned.

Also, I tried using a TCP socket instead of Unix one and not performing any SFTP operations while TCP forwarding but to no avail, still got the AssertionError on write_eof.

Why should EOF be sent anyway? Could something be wrong on the backend side?
All backend-side does is do a connect_reverse and then wait_closed.
The SSHServer passed as server_factory does some authentication stuff and returns True on connection_requested

[misa-hase]

I've noticed a few malfunctioning clients practically spamming the forwarded ports with connections.
It might be due to the large number of connections causing the asyncssh process to hit the 'Max open files' soft limit.

I don't remember ever seeing a EMFILE (Too many files) OSError so far but I've raised the limit anyway. I'll let you know if that fixes the issue or not.

[ronf]

There are two SSHForwarder objects associated with each tunnel. One is associated with the TCP (or UNIX domain) connection and the other is associated with the SSH session opened to tunnel the traffic. Whenever the other end of those connections get closed, an eof_received() callback should be triggered. If there's still a peer for that SSHForwarder when it gets eof_received(), it should trigger write_eof(). These are "half" closes, indicating that no more data will be arriving in one of the two directions, but once both directions are closed, the SSHForwarders should be cleaned up.

Looking at how the EOF handling has evolved on the main SSH connection, I think maybe I need to allow eof_received() to arrive even after a close(), in case the TCP/UNIX socket queues up an EOF callback at the same time as a result of the SSHConnection closing and the automatic cleanup kicking in.

To work around this for now, you could try changing the assert() in write_eof() to an "if" that just returns if self._transport isn't set, without calling write_eof() on the transport.

If sockets are not being closed properly, it might be possible to run out of fds, but there's no evidence of that here so far. In fact, seeing _transport becoming unset is a good indication that the sockets are in fact getting cleaned up properly.

[misa-hase]

Yup, still got the same AssertionError exceptions after raising the FD count limit.
I'll try the mentioned workaround next.

[misa-hase]

I'll try the mentioned workaround next.

With the workaround applied I no longer get the AssertionError as expected, and I have not noticed any side effects occurring as a result of the workaround.

[ronf]

Thanks for reporting and testing this! I'll look into making this change for SSHForwarder, mirroring what is already being done in SSHConnection.

[ronf]

This is now available in the "develop" branch as commit 3d31c4c.

[ronf]

This fix is now available in AsyncSSH 2.21.1.

[jabbera]

This fix is now available in AsyncSSH 2.21.1.

@ronf, unsure if you want me to start a new issue but I am now seeing this issue occasionally during forwarding initiation with 2.21.1

'2025-10-14 17:35:40 - group.arch.rh.ssh_proxy - INFO - [965550d2-7f39-420d-add3-88dbe2e66e6a User: snip] Forwarding TCP connection to ('jupyter-snip.jhub-user-pod.jhub.svc.cluster.local', 22)'
'2025-10-14 17:35:40 - asyncio - ERROR - Fatal error: protocol.data_received() call failed.
protocol: <asyncssh.forward.SSHForwarder object at 0x7f330bea66d0>
transport: <_SelectorSocketTransport fd=31 read=polling write=<idle, bufsize=0>>'
Traceback (most recent call last):
File "/usr/local/lib/python3.13/asyncio/selector_events.py", line 1019, in _read_ready__data_received
self._protocol.data_received(data)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^
File "/usr/local/lib/python3.13/site-packages/asyncssh/forward.py", line 146, in data_received
self._peer.write(data)
~~~~~~~~~~~~~~~~^^^^^^
File "/usr/local/lib/python3.13/site-packages/asyncssh/forward.py", line 88, in write
assert self._transport is not None
^^^^^^^^^^^^^^^^^^^^^^^^^^^
AssertionError

[ronf]

It looks like you're seeing this issue on both write() and write_eof(). The fixed mentioned here only applies to write_eof(). I should be able to make similar changes to write().

That said, this suggests that this isn't just a race where both ends of a connection are closed simultaneously. It seems like one side might be closing while the other side is not yet finished sending data, and so there's data in flight when the close happens. Is this something you are expecting in your application? I wonder if there might be an earlier error which could be triggering the mid-stream close.

Either way, the assert here is definitely worth fixing. I'll have an additional patch to try shortly.