Connection lost before connect to HTTP Proxy #738
-
|
Good morning, I have the following code that makes a tunnel with a HTTP Proxy class HTTPProxyTunnel:
"""HTTP Proxy Tunner.
Attributes:
__proxy_data: Proxy Data
"""
def __init__(self, proxy_data: ProxyData) -> None:
"""HTTP Proxy Tunner.
Args:
proxy_data: Proxy Data
"""
self.__proxy_data: ProxyData = proxy_data
async def create_connection(
self,
protocol_factory: type[asyncio.BaseProtocol],
remote_host: str,
remote_port: int,
) -> tuple[asyncio.BaseTransport, asyncio.BaseProtocol]:
"""Create Connection for http proxy.
Args:
protocol_factory: Session Factory from create tunnel
remote_host: Host Address to connect
remote_port: Port to connect
Returns:
Connection with HTTP Proxy
Raises:
ConnectionError: Connection Error
"""
reader: asyncio.StreamReader
writer: asyncio.StreamWriter
print("CREATE CONNECTION - GO")
reader, writer = await asyncio.open_connection(self.__proxy_data.host, self.__proxy_data.port)
print("CREATE CONNECTION - DONE")
print("WRITE HTTP COMMAND - GO")
writer.write(
(
f"CONNECT {remote_host}:{remote_port} HTTP/1.1\r\n"
f"Host: {remote_host}:{remote_port}\r\n"
"Accept: */*\r\n"
"\r\n"
).encode("ascii")
)
print("WRITE HTTP COMMAND - DONE")
print("READ LINE - GO")
first_line: bytes = await reader.readline()
print(f"READ LINE - DONE {first_line!r}")
if not first_line.startswith(b"HTTP/1.1 200 "):
print("NO 200")
raise ConnectionError(f"Unexpected response: {first_line.decode('utf-8', errors='ignore')}") # noqa: TRY003
async for line in reader:
print(f"READ LINE - {line!r}")
if line == b"\r\n":
break
print("RETURN - GO")
transport: asyncio.WriteTransport = writer.transport
protocol: asyncio.BaseProtocol = protocol_factory()
transport.set_protocol(protocol)
protocol.connection_made(transport)
print("RETURN - DONE")
return transport, protocolWhen I run it, it fails, but it is after returning the asyncssh control. If I remove the break when the reader finds it \r\n, I see that it connects to the server. Have there been any changes that need to be applied to the tunnel? Or why is it disconnected? I can't find the problem |
Beta Was this translation helpful? Give feedback.
Replies: 5 comments 20 replies
-
|
I think there may be a race condition here, where the SSH version line is sometimes being delivered to the HTTP reader rather than to the new AsyncSSH protocol object. You might try adding a couple of lines such as: print("RETURN - GO")
transport: asyncio.WriteTransport = writer.transport
protocol: asyncio.BaseProtocol = protocol_factory()
transport.set_protocol(protocol)
buffered_data = await reader.read(1024)
protocol.connection_made(transport)
protocol.data_received(buffered_data)
print("RETURN - DONE")
return transport, protocol |
Beta Was this translation helpful? Give feedback.
-
|
This reads any buffered data (generally the SSL version line from the remote host) and makes it available to the data_received() callback after switch the transport over to point at the new SSH protocol object. |
Beta Was this translation helpful? Give feedback.
-
|
I just tried it, it seems to stay in RETURN - GO so I guess it does not read anything and is waiting to read something. In my case the proxy is http not https. |
Beta Was this translation helpful? Give feedback.
-
|
I was testing with an HTTP proxy here as well (no SSL involved). If the connection to the remote host succeeds (which it should have if you got back a 200 OK response) and you're connecting to an SSH server, it should be writing its version string when you connect. So, even if there was no data buffered yet, something should be coming. Perhaps try a different order here: print("RETURN - GO")
transport: asyncio.WriteTransport = writer.transport
protocol: asyncio.BaseProtocol = protocol_factory()
buffered_data = await reader.read(1024)
transport.set_protocol(protocol)
protocol.connection_made(transport)
protocol.data_received(buffered_data)
print("RETURN - DONE")
return transport, protocolThat way the read will definitely complete before switching the transport over to delivering new incoming bytes to the SSH protocol object. |
Beta Was this translation helpful? Give feedback.
-
|
Yes, this is the code print("RETURN - GO")
transport: asyncio.WriteTransport = writer.transport
protocol: asyncio.BaseProtocol = protocol_factory()
buffered_data: bytes = await reader.read(1024)
print(f"BUFFER DATA - {buffered_data!r}")
transport.set_protocol(protocol)
protocol.connection_made(transport)
protocol.data_received(buffered_data)
print("RETURN - DONE")
return transport, protocol |
Beta Was this translation helpful? Give feedback.
-
|
If that line got delivered to AsyncSSH, it should respond with a KEXINIT message. What do you see in the debug log? To enable that, put the following at the top of the file: import logging
logging.basicConfig(level='DEBUG')
asyncssh.set_debug_level(3) |
Beta Was this translation helpful? Give feedback.
-
|
Sure!!! This is all I get: |
Beta Was this translation helpful? Give feedback.
-
|
Ok - so the KEXINIT is going out, but the peer's KEXINIT is either not being sent or not being delivered. This suggests the set_protocol() may not be working like it did in older Python versions. If you try running with an older Python, does anything change in the behavior? |
Beta Was this translation helpful? Give feedback.
-
|
Sure! I test with various versions:
It seems that from python 3.11 onwards it does not work. |
Beta Was this translation helpful? Give feedback.
-
|
Great! I was seeing slightly different behavior myself in 3.8, but I haven't had a chance to dig into that yet. Also, I found a problem with the second ordering I suggested. You might want to try the following instead: print("RETURN - GO")
buffered_data = await reader.read(1024)
transport: asyncio.WriteTransport = writer.transport
protocol: asyncio.BaseProtocol = protocol_factory()
transport.set_protocol(protocol)
protocol.connection_made(transport)
protocol.data_received(buffered_data)
print("RETURN - DONE")
return transport, protocolThis makes sure all reads are done before switch to the new protocol object, and the set_protocol happens before the protocol object attempts to process any input or write any output. |
Beta Was this translation helpful? Give feedback.
-
|
Sure!! I have applied this change, although the behavior is the same (I imagine that one thing is when to read and another that the set_protocol works well). |
Beta Was this translation helpful? Give feedback.
-
|
Does this mean you're still seeing a problem on Python 3.11 and later, but not on earlier versions? With the latest change above, it seems to be working better for me on both older and newer Python versions, and it works both when I break data up into multiple reads and when all the data arrives in a single read (both the 200 OK and the SSL version string). My test doesn't carry through to seeing if the KEXINIT is accepted after that, though, or the handshake fully completes. I need to install an HTTP proxy that can tunnel data inside the HTTP CONNECT tunnel for this to be a proper test. |
Beta Was this translation helpful? Give feedback.
-
|
Yes, I still see the same results with the above mentioned python versions. |
Beta Was this translation helpful? Give feedback.
-
|
I was able to reproduce it using this container with docker to create an http proxy on local (the only thing is HTTP/1.0 200 instead of HTTP/1.1 200). docker run --rm -d \
-p "3128:3128/tcp" \
-p "1080:1080/tcp" \
ghcr.io/tarampampam/3proxy:1Whit this code import asyncio
from pprint import pprint
import asyncssh
class HTTPProxyTunnel:
"""HTTP Proxy Tunner."""
def __init__(self) -> None:
"""HTTP Proxy Tunner."""
self.__proxy_host: str = "localhost"
self.__proxy_port: int = 3128
async def create_connection(
self,
protocol_factory,
remote_host,
remote_port,
):
"""Create Connection for http proxy.
Args:
protocol_factory: Session Factory from create tunnel
remote_host: Host Address to connect
remote_port: Port to connect
Returns:
Connection with HTTP Proxy
Raises:
ConnectionError: Connection Error
"""
print("CREATE CONNECTION - GO")
reader, writer = await asyncio.open_connection(self.__proxy_host, self.__proxy_port)
print("CREATE CONNECTION - DONE")
print("WRITE HTTP COMMAND - GO")
writer.write(
(
f"CONNECT {remote_host}:{remote_port} HTTP/1.1\r\n"
f"Host: {remote_host}:{remote_port}\r\n"
"\r\n"
).encode("ascii")
)
print("WRITE HTTP COMMAND - DONE")
print("READ LINE - GO")
first_line = await reader.readline()
print(f"READ LINE - DONE {first_line!r}")
if not first_line.startswith(b"HTTP/1.0 200 "):
print("NO 200")
raise ConnectionError(f"Unexpected response: {first_line.decode('utf-8', errors='ignore')}") # noqa: TRY003
async for line in reader:
print(f"READ LINE - {line!r}")
if line == b"\r\n":
break
print("RETURN - GO")
buffered_data = await reader.read(1024)
print(f"BUFFER DATA - {buffered_data!r}")
transport = writer.transport
protocol = protocol_factory()
transport.set_protocol(protocol)
protocol.connection_made(transport)
protocol.data_received(buffered_data)
print("RETURN - DONE")
return transport, protocol
async def main():
executor = await asyncssh.connect(
host="HOST",
username="USER",
password="PASS",
tunnel=HTTPProxyTunnel(),
known_hosts=None,
)
result = await executor.run("uptime")
pprint(result)
executor.close()
await executor.wait_closed()
import logging
logging.basicConfig(level='DEBUG')
asyncssh.set_debug_level(3)
asyncio.run(main()) |
Beta Was this translation helpful? Give feedback.
-
|
I have checked with pycharm debugging and I can't see any difference. I see that both the protocol is put in the transport and the transport in the protocol, so I understand that these 2 lines work (Or at least the assignment of properties, I do not know if there will be some different behavior later on) transport.set_protocol(protocol)
protocol.connection_made(transport) |
Beta Was this translation helpful? Give feedback.
-
|
I've been able to reproduce this here, with a simple HTTP CONNECT proxy I wrote in Python (so I can more easily see what it receives before the connection fails). It works properly for me for Python 3.7-3.10 and fails for Python 3.11 and later. In the case where it fails, it seems to send out the version line and the KEXINIT, which means the call to connection_made() on the new protocol must be getting made, as is the call to data_received() on the buffered data (incoming SSH version line). However, after the new transport and new protocol are returned, somehow the connection_lost() method is immediately called on the protocol. I'm still trying to figure out what the differences are between the 3.10 and 3.11 versions of asyncio that could explain this. |
Beta Was this translation helpful? Give feedback.
-
|
Ok - I think I see what's going on. It's related to the destructor being called on the stream writer which is used to send the HTTP CONNECT. When the create_connection() function exits, this variable goes out of scope and the destructor is called. Since this writer still has a pointer to the transport, calling the destructor ends up calling close() on the transport (and generates a warning about it being unclosed). This code is new to Python 3.11. Unfortunately, I don't really see a good way to clear the transport from this stream writer when moving it over to the new protocol object. A workaround for this issue is to just keep a pointer to the writer as a member variable of the HTTPProxyTunnel class, to prevent the writer from being closed until the tunnel object goes out of scope. I think you'd still see warnings about an unclosed transport in that case when the connection finally did close, but it wouldn't cause the close to happen while we're trying to open the tunnel. Another workaround would be to do a The other option would be to build the HTTPProxyTunnel as a Protocol and use callbacks instead of asynchronous calls. That way, you have more control when you switch protocol objects, avoiding the close and the associated warning. It would be a lot uglier written that way, though. |
Beta Was this translation helpful? Give feedback.
-
|
Wow, you are the best!!! I was able to fix it this way: import asyncio
from pprint import pprint
import asyncssh
class HTTPProxyTunnel:
"""HTTP Proxy Tunner."""
def __init__(self) -> None:
"""HTTP Proxy Tunner."""
self.__proxy_host: str = "localhost"
self.__proxy_port: int = 3128
self.__reader: asyncio.StreamReader
self.__writer: asyncio.StreamWriter
async def create_connection(
self,
protocol_factory,
remote_host,
remote_port,
):
"""Create Connection for http proxy.
Args:
protocol_factory: Session Factory from create tunnel
remote_host: Host Address to connect
remote_port: Port to connect
Returns:
Connection with HTTP Proxy
Raises:
ConnectionError: Connection Error
"""
print("CREATE CONNECTION - GO")
reader, writer = await asyncio.open_connection(self.__proxy_host, self.__proxy_port)
self.__reader = reader
self.__writer = writer
print("CREATE CONNECTION - DONE")
print("WRITE HTTP COMMAND - GO")
writer.write(
(
f"CONNECT {remote_host}:{remote_port} HTTP/1.1\r\n"
f"Host: {remote_host}:{remote_port}\r\n"
"\r\n"
).encode("ascii")
)
print("WRITE HTTP COMMAND - DONE")
print("READ LINE - GO")
first_line = await reader.readline()
print(f"READ LINE - DONE {first_line!r}")
if not first_line.startswith(b"HTTP/1.0 200 "):
print("NO 200")
raise ConnectionError(f"Unexpected response: {first_line.decode('utf-8', errors='ignore')}") # noqa: TRY003
async for line in reader:
print(f"READ LINE - {line!r}")
if line == b"\r\n":
break
print("RETURN - GO")
buffered_data = await reader.read(1024)
print(f"BUFFER DATA - {buffered_data!r}")
transport = writer.transport
protocol = protocol_factory()
transport.set_protocol(protocol)
protocol.connection_made(transport)
protocol.data_received(buffered_data)
print("RETURN - DONE")
return transport, protocol
async def main():
executor = await asyncssh.connect(
host="HOST",
username="USER",
password="PASS",
tunnel=HTTPProxyTunnel(),
known_hosts=None,
)
result = await executor.run("uptime")
pprint(result)
executor.close()
await executor.wait_closed()
import logging
logging.basicConfig(level='DEBUG')
asyncssh.set_debug_level(3)
asyncio.run(main())If I create them and assign them as a property of each instance of HTTPProxyTunnel, I make them to be there even if the execution of connection_made finishes, this way there close will not be called. Thank you very much for your help!!!! You are the best! |
Beta Was this translation helpful? Give feedback.
-
|
Happy to help! This looks like it should work well. The only thing to be aware of is that you may get an "unclosed transport" warning when the HTTPProxyTunnel object goes out of scope. If you want to avoid that, you could add a |
Beta Was this translation helpful? Give feedback.
-
|
Great! From what I have understood, asyncssh to close connection calls a close() method, as well as to create it calls create_connetion(). So I implement the close() to close the writer when the connection is closed. My code would look like this: import asyncio
import re
from pprint import pprint
import asyncssh
class HTTPProxyTunnel:
"""HTTP Proxy Tunner."""
def __init__(self, proxy_host: str, proxy_port: int) -> None:
"""HTTP Proxy Tunner.
Args:
proxy_host: Proxy Host
proxy_port: Proxy Port
"""
self.__proxy_host: str = proxy_host
self.__proxy_port: int = proxy_port
self.__reader: asyncio.StreamReader | None = None
self.__writer: asyncio.StreamWriter | None = None
async def create_connection(
self,
protocol_factory: Callable[[], asyncio.Protocol],
remote_host: str,
remote_port: int,
) -> tuple[asyncio.BaseTransport, asyncio.Protocol]:
"""Create Connection for http proxy.
Args:
protocol_factory: Session Factory from create tunnel
remote_host: Host Address to connect
remote_port: Port to connect
Returns:
Connection with HTTP Proxy
Raises:
ConnectionError: Connection Error
"""
self.__reader, self.__writer = await asyncio.open_connection(self.__proxy_host, self.__proxy_port)
self.__writer.write(
f"CONNECT {remote_host}:{remote_port} HTTP/1.1\r\nHost: {remote_host}:{remote_port}\r\n\r\n".encode("ascii")
)
first_line: bytes = await self.__reader.readline()
if not re.fullmatch(rb"HTTP/1.\d 200 .*\n", first_line):
raise ConnectionError(f"Unexpected response: {first_line.decode('utf-8', errors='ignore')}") # noqa: TRY003 EM102
async for line in self.__reader:
if line == b"\r\n":
break
buffered_data: bytes = await self.__reader.read(1024)
transport: asyncio.WriteTransport = self.__writer.transport
protocol: asyncio.Protocol = protocol_factory()
transport.set_protocol(protocol)
protocol.connection_made(transport)
protocol.data_received(buffered_data)
return transport, protocol
def close(self) -> None:
"""Close Connection for HTTP Proxy."""
if self.__writer:
self.__writer.close()
async def main():
tunnel = HTTPProxyTunnel("localhost", 3128)
executor = await asyncssh.connect(
host="HOST",
username="ROOT",
password="PASS",
tunnel=tunnel,
known_hosts=None,
)
result = await executor.run("uptime")
pprint(result)
executor.close()
await executor.wait_closed()
import logging
logging.basicConfig(level='DEBUG')
asyncssh.set_debug_level(3)
asyncio.run(main())I understand that this would be fine, right? Or am I missing something? |
Beta Was this translation helpful? Give feedback.
-
|
That's right - that looks like it should avoid the "unclosed transport" warning. |
Beta Was this translation helpful? Give feedback.
-
|
I leave here a message to make it easier to find. @ronf has discovered that as of python 3.11 the asyncio.StreamWriter closes the connection when destroyed. Here is an example of the implementation in case it is useful for someone in the future: import asyncio
import re
from pprint import pprint
import asyncssh
class HTTPProxyTunnel:
"""HTTP Proxy Tunner."""
def __init__(self, proxy_host: str, proxy_port: int) -> None:
"""HTTP Proxy Tunner.
Args:
proxy_host: Proxy Host
proxy_port: Proxy Port
"""
self.__proxy_host: str = proxy_host
self.__proxy_port: int = proxy_port
self.__reader: asyncio.StreamReader | None = None
self.__writer: asyncio.StreamWriter | None = None
async def create_connection(
self,
protocol_factory: Callable[[], asyncio.Protocol],
remote_host: str,
remote_port: int,
) -> tuple[asyncio.BaseTransport, asyncio.Protocol]:
"""Create Connection for http proxy.
Args:
protocol_factory: Session Factory from create tunnel
remote_host: Host Address to connect
remote_port: Port to connect
Returns:
Connection with HTTP Proxy
Raises:
ConnectionError: Connection Error
"""
self.__reader, self.__writer = await asyncio.open_connection(self.__proxy_host, self.__proxy_port)
self.__writer.write(
f"CONNECT {remote_host}:{remote_port} HTTP/1.1\r\nHost: {remote_host}:{remote_port}\r\n\r\n".encode("ascii")
)
first_line: bytes = await self.__reader.readline()
if not re.fullmatch(rb"HTTP/1.\d 200 .*\n", first_line):
raise ConnectionError(f"Unexpected response: {first_line.decode('utf-8', errors='ignore')}") # noqa: TRY003 EM102
async for line in self.__reader:
if line == b"\r\n":
break
buffered_data: bytes = await self.__reader.read(1024)
transport: asyncio.WriteTransport = self.__writer.transport
protocol: asyncio.Protocol = protocol_factory()
transport.set_protocol(protocol)
protocol.connection_made(transport)
protocol.data_received(buffered_data)
return transport, protocol
def close(self) -> None:
"""Close Connection for HTTP Proxy."""
if self.__writer:
self.__writer.close()
async def main():
tunnel = HTTPProxyTunnel("localhost", 3128)
executor = await asyncssh.connect(
host="HOST",
username="ROOT",
password="PASS",
tunnel=tunnel,
known_hosts=None,
)
result = await executor.run("uptime")
pprint(result)
executor.close()
await executor.wait_closed()
import logging
logging.basicConfig(level='DEBUG')
asyncssh.set_debug_level(3)
asyncio.run(main())The differences with the code in the question are:
Thank you so much @ronf, without your help this wouldn't have been possible!!!! |
Beta Was this translation helpful? Give feedback.
-
|
No problem! Thanks for summarizing the fix... |
Beta Was this translation helpful? Give feedback.
I leave here a message to make it easier to find.
@ronf has discovered that as of python 3.11 the asyncio.StreamWriter closes the connection when destroyed.
So you have to change the implementation to make it work.
Here is an example of the implementation in case it is useful for someone in the future: