tcpclient: close the SSLIOStream on TLS handshake timeout so the socket is released

tornado/iostream.py

@@ -1256,6 +1256,14 @@ def start_tls(
         ssl_stream._ssl_connect_future = future
         ssl_stream.max_buffer_size = self.max_buffer_size
         ssl_stream.read_chunk_size = self.read_chunk_size
+        # Stash a back-reference to the new stream on the returned future
+        # so callers that race the handshake against a timeout (e.g.
+        # TCPClient.connect with both ssl_options and timeout) can close
+        # the underlying socket if the handshake doesn't complete. Without
+        # this, gen.with_timeout leaves the SSLIOStream registered on the
+        # IOLoop with no reachable reference, and the socket file
+        # descriptor is leaked forever.
+        future._ssl_stream = ssl_stream  # type: ignore[attr-defined]
         return future
 
     def _handle_connect(self) -> None:

tornado/tcpclient.py

@@ -273,12 +273,21 @@ async def connect(
         # the same host. (http://tools.ietf.org/html/rfc6555#section-4.2)
         if ssl_options is not None:
             if timeout is not None:
-                stream = await gen.with_timeout(
-                    timeout,
-                    stream.start_tls(
-                        False, ssl_options=ssl_options, server_hostname=host
-                    ),
+                tls_future = stream.start_tls(
+                    False, ssl_options=ssl_options, server_hostname=host
                 )
+                try:
+                    stream = await gen.with_timeout(timeout, tls_future)
+                except gen.TimeoutError:
+                    # gen.with_timeout does not cancel its inner future, so
+                    # the SSLIOStream is still registered on the IOLoop with
+                    # no reachable reference. IOStream.start_tls stashes a
+                    # back-reference on the returned future so we can close
+                    # the stream here and release the underlying socket.
+                    ssl_stream = getattr(tls_future, "_ssl_stream", None)
+                    if ssl_stream is not None:
+                        ssl_stream.close()
+                    raise
             else:
                 stream = await stream.start_tls(
                     False, ssl_options=ssl_options, server_hostname=host

tornado/test/tcpclient_test.py

@@ -16,17 +16,20 @@
 import socket
 import typing
 import unittest
+import os
+import ssl
 from contextlib import closing
 
 from tornado.concurrent import Future
+from tornado import gen
 from tornado.gen import TimeoutError
 from tornado.iostream import IOStream
 from tornado.netutil import Resolver, bind_sockets
 from tornado.queues import Queue
 from tornado.tcpclient import TCPClient, _Connector
 from tornado.tcpserver import TCPServer
 from tornado.test.util import refusing_port, skipIfNoIPv6, skipIfNonUnix
-from tornado.testing import AsyncTestCase, gen_test
+from tornado.testing import AsyncTestCase, bind_unused_port, gen_test
 
 # Fake address families for testing.  Used in place of AF_INET
 # and AF_INET6 because some installations do not have AF_INET6.
@@ -173,6 +176,76 @@ def resolve(self, *args, **kwargs):
                 "1.2.3.4", 12345, timeout=timeout
             )
 
+    @gen_test
+    def test_ssl_handshake_timeout_does_not_leak_socket(self):
+        # Regression test for #3614: when TCPClient.connect is called with
+        # both ssl_options and timeout, a TLS handshake timeout used to leak
+        # the underlying socket. gen.with_timeout doesn't cancel its inner
+        # future, so the SSLIOStream (which holds the real socket) was
+        # left registered on the IOLoop with no reachable reference.
+        # A server that accepts the TCP connection but never speaks TLS.
+        # The client will connect, kick off the TLS handshake, then wait
+        # forever for a response that never comes.
+        from tornado.netutil import bind_sockets
+
+        sockets = bind_sockets(0, "127.0.0.1", family=socket.AF_INET)
+        port = sockets[0].getsockname()[1]
+
+        def _accept(fd, events):
+            for sock in sockets:
+                try:
+                    conn, _ = sock.accept()
+                    # Hold the raw accepted socket open so it can be
+                    # counted separately, but never hand it to an SSL
+                    # server. The client's TLS handshake will hang
+                    # waiting for a ServerHello that never arrives.
+                    self._held_server_sockets.append(conn)
+                except BlockingIOError:
+                    pass
+
+        self._held_server_sockets: list[socket.socket] = []
+        for sock in sockets:
+            self.io_loop.add_handler(
+                sock.fileno(), _accept, self.io_loop.READ
+            )
+
+        try:
+            fd_dir = "/proc/self/fd"
+            before = len(os.listdir(fd_dir))
+
+            ssl_ctx = ssl.create_default_context()
+            ssl_ctx.check_hostname = False
+            ssl_ctx.verify_mode = ssl.CERT_NONE
+            with self.assertRaises(TimeoutError):
+                yield TCPClient().connect(
+                    "127.0.0.1",
+                    port,
+                    ssl_options=ssl_ctx,
+                    timeout=0.05,
+                )
+
+            # The IOLoop is responsible for closing the SSLIOStream's
+            # socket. Run a few iterations to give it the chance.
+            for _ in range(5):
+                yield gen.sleep(0.01)
+
+            # The raw accepted server sockets are still open; subtract
+            # those out so we're really measuring the client-side leak.
+            held = len(self._held_server_sockets)
+            after = len(os.listdir(fd_dir))
+            # Each accepted server connection accounts for one fd
+            # on the server side, and the TCPClient's connect did not
+            # open any new fds on the client side once the handshake
+            # is unwound on timeout. So the only delta should be the
+            # accepted connections.
+            self.assertEqual(after, before + held)
+        finally:
+            for sock in sockets:
+                self.io_loop.remove_handler(sock.fileno())
+                sock.close()
+            for sock in self._held_server_sockets:
+                sock.close()
+
 
 class TestConnectorSplit(unittest.TestCase):
     def test_one_family(self):