Issue in handle_async_request when using Proxy - Connection not being closed when await stream.start_tls(kwargs) raises Error

[RubensMachado-Dev]

The issue occurs in the code block async with Trace("start_tls", logger, request, kwargs). When a request is made to a server without a valid SSL certificate, the **stream = await stream.start_tls(kwargs) line raises an error. Consequently, the connection remains active and isn't closed. The httpx library relies on the httpcore connection pool, doesn't recognize that this connection should be cleaned and reused. Eventually, when all connections are active and unreused, httpx raises a httpx.pooltimeout error. This error is triggered by the **stream = await stream.start_tls(kwargs) line.

Note: The proxy must be set to trigger this error.

To reproduce this issue, set up a self-signed SSL certificate, create an HTTP server, and generate your SSL certificate. The error can be reproduced with the following script (Note: Not all proxies trigger <class 'httpx.ConnectTimeout'> or <class 'httpx.ConnectError'>):

import asyncio
import http.server
import ssl

import httpx
from httpx import Limits, Timeout


async def main():
    max_connections = 2
    session = httpx.AsyncClient(
        timeout=Timeout(10),
        limits=Limits(max_connections=max_connections, max_keepalive_connections=2, keepalive_expiry=2),
        proxy="http://0.0.0.0:8899")
    server_address = ("localhost", 4443)
    httpd = http.server.HTTPServer(server_address, http.server.SimpleHTTPRequestHandler)

    httpd.socket = ssl.wrap_socket(httpd.socket, keyfile="key.pem", certfile="cert.pem", server_side=True)

    print("Serving on https://localhost:4443")
    for _ in range(0,max_connections+1):
        try:
            await session.get('https://localhost:4443')
        except Exception as e:
            print(e.__class__)
            continue

if __name__ == "__main__":
    asyncio.run(main())

For setting a local proxy. It can be handled by using proxy.py library

[janosik47]

Placing my raw story here - same as above

In my case I am observing the following behavior. From time to time ConnectTimeout is happening. When the number of ConnectTimeouts reaches the connection pool's max_connection then any subsequent requests fails with PoolTimeout. The system ends up in an unrecoverable limbo state.

I did some debugging and looks to me the AsyncTunnelHTTPConnection is not handling properly exceptions.
If the code between lines 299 and 326 raises an exception then the self._connection is an ACTIVE proxy tunnel. However, the exception is only caught by the connection pool handler (here), which cannot 'clean it up'.
From the pool perspective the connection object stays ACTIVE so cannot be reused nor closed, its essentially a leak.

If that repeated as many times as many connections are allowed the pool can no create any new one and will always throw PoolTimeout.

It is trivial to reproduce when debugging with low connect timeout setting. Place breakpoint in here and wait longer then the timeout value. That will raise ConnectTimeout.

The AsyncTunnelHTTPConnection is using the network stream to establish TLS, which by itself correctly handles exceptions, however it is an internal object of the tunnel’s connection (here) and closing it does not close (change state) of the tunnel itself.

My proposal is to try/catch the lines 299 and 326 and close tunnel, e.g.

          try:
                stream = connect_response.extensions["network_stream"]

                # Upgrade the stream to SSL
                ssl_context = (
                    default_ssl_context()
                    if self._ssl_context is None
                    else self._ssl_context
                )
                alpn_protocols = ["http/1.1", "h2"] if self._http2 else ["http/1.1"]
                ssl_context.set_alpn_protocols(alpn_protocols)

                kwargs = {
                    "ssl_context": ssl_context,
                    "server_hostname": self._remote_origin.host.decode("ascii"),
                    "timeout": timeout,
                }
                async with Trace("start_tls", logger, request, kwargs) as trace:
                    stream = await stream.start_tls(**kwargs)
                    trace.return_value = stream

                # Determine if we should be using HTTP/1.1 or HTTP/2
                ssl_object = stream.get_extra_info("ssl_object")
                http2_negotiated = (
                    ssl_object is not None
                    and ssl_object.selected_alpn_protocol() == "h2"
                )
          except Exception as ex:
                await self._connection.aclose()
                raise ex
          else:
                # Create the HTTP/1.1 or HTTP/2 connection
                if http2_negotiated or (self._http2 and not self._http1):
                    from .http2 import AsyncHTTP2Connection

                    self._connection = AsyncHTTP2Connection(
                        origin=self._remote_origin,
                        stream=stream,
                        keepalive_expiry=self._keepalive_expiry,
                    )
                else:
                    self._connection = AsyncHTTP11Connection(
                        origin=self._remote_origin,
                        stream=stream,
                        keepalive_expiry=self._keepalive_expiry,
                    )

                self._connected = True