gh-160520: Apply asyncio ssl_shutdown_timeout after peer EOF (GH-160521)

Keep the SSL shutdown timer running until connection_lost(), and start
it when the peer shuts down first, so that a peer which sends EOF but
does not read cannot keep a closed transport open indefinitely.

ANT-2026-81HUBBJ7

Author: gpshead

Lib/asyncio/sslproto.py

@@ -669,13 +669,21 @@ def _do_shutdown(self):
             self._on_shutdown_complete(None)
 
     def _on_shutdown_complete(self, shutdown_exc):
-        if self._shutdown_timeout_handle is not None:
-            self._shutdown_timeout_handle.cancel()
-            self._shutdown_timeout_handle = None
-
         if shutdown_exc:
+            if self._shutdown_timeout_handle is not None:
+                self._shutdown_timeout_handle.cancel()
+                self._shutdown_timeout_handle = None
             self._fatal_error(shutdown_exc)
         else:
+            # transport.close() flushes its write buffer first, which a peer
+            # that does not read can delay forever.  Keep the timeout (or
+            # start it, if the peer initiated the shutdown) until
+            # connection_lost().
+            if self._shutdown_timeout_handle is None:
+                self._shutdown_timeout_handle = self._loop.call_later(
+                    self._ssl_shutdown_timeout,
+                    self._check_shutdown_timeout
+                )
             self._loop.call_soon(self._transport.close)
 
     def _abort(self, exc):

Lib/test/test_asyncio/test_ssl.py

@@ -1736,6 +1736,74 @@ async def client(addr):
         with self.tcp_server(server) as srv:
             loop.run_until_complete(client(srv.addr))
 
+    def test_shutdown_timeout_after_eof_from_not_reading_peer(self):
+        # ssl_shutdown_timeout also applies when the peer has sent EOF,
+        # before or after close(), but does not read what is still queued.
+        loop = self.loop
+        filled = threading.Event()
+        done = threading.Event()
+
+        def server(sock):
+            sslctx = self._create_server_ssl_context(
+                test_utils.ONLYCERT,
+                test_utils.ONLYKEY
+            )
+            sock = sslctx.wrap_socket(sock, server_side=True)
+            filled.wait(support.SHORT_TIMEOUT)
+            # sending EOF here, and never reading
+            sock.shutdown(socket.SHUT_WR)
+            done.wait(support.SHORT_TIMEOUT)
+            sock.close()
+
+        class Protocol(asyncio.Protocol):
+            def __init__(self):
+                self.fut = asyncio.Future(loop=loop)
+                self.paused = False
+
+            def pause_writing(self):
+                self.paused = True
+
+            def resume_writing(self):
+                self.paused = False
+
+            def connection_lost(self, exc):
+                if not self.fut.done():
+                    self.fut.set_result(exc)
+
+        async def client(addr, close):
+            ctx = self._create_client_ssl_context()
+            tr, pr = await loop.create_connection(
+                Protocol, *addr, ssl=ctx,
+                ssl_shutdown_timeout=0.1)
+            try:
+                # Write until the transport cannot pass any more on to the
+                # peer; how much the kernel accepts varies by platform.
+                chunk = bytes(support.SOCK_MAX_SIZE)
+                for _ in range(16):
+                    if pr.paused:
+                        break
+                    tr.write(chunk)
+                    await asyncio.sleep(0)
+                else:
+                    self.fail('the transport never paused writing')
+                if close:
+                    tr.close()
+                filled.set()
+                async with asyncio.timeout(support.SHORT_TIMEOUT):
+                    exc = await pr.fut
+            finally:
+                filled.set()
+                done.set()
+                tr.abort()
+            self.assertIsInstance(exc, TimeoutError)
+
+        for close in (True, False):
+            with self.subTest(close_before_eof=close):
+                filled.clear()
+                done.clear()
+                with self.tcp_server(server) as srv:
+                    loop.run_until_complete(client(srv.addr, close))
+
 
 ###############################################################################
 # Socket Testing Utilities

Misc/NEWS.d/next/Library/2026-10-22-18-30-00.gh-issue-160520.sSdTmo.rst

@@ -0,0 +1,4 @@
+Fix :mod:`asyncio` SSL transports not applying *ssl_shutdown_timeout* when
+the peer has sent EOF but does not read the data still queued for it.
+Closing such a connection could wait indefinitely; it is now aborted with
+:exc:`TimeoutError` once the timeout expires.