gh-160499: Pause asyncio TLS reads when SSL replies pile up

SSLProtocol read flow control only looked at the incoming BIO, so records
OpenSSL writes on its own while reading a non-reading peer (renegotiation
alerts, KeyUpdate replies) grew the outgoing BIO without bound.  Count that
non-application output toward the read high-water mark and re-check it when
writing resumes.

ANT-2026-RNWRQ22W

Author: gpshead

Lib/asyncio/sslproto.py

@@ -308,6 +308,8 @@ def __init__(self, loop, app_protocol, sslcontext, waiter,
         # App data write buffering
         self._write_backlog = collections.deque()
         self._write_buffer_size = 0
+        # How much of self._outgoing is encrypted app data
+        self._outgoing_app_size = 0
 
         self._waiter = waiter
         self._loop = loop
@@ -710,6 +712,7 @@ def _write_appdata(self, list_of_data):
             self._fatal_error(ex, 'Fatal error on SSL protocol')
 
     def _do_write(self):
+        pending = self._outgoing.pending
         try:
             while self._write_backlog:
                 data = self._write_backlog[0]
@@ -723,11 +726,13 @@ def _do_write(self):
                     self._write_buffer_size -= data_len
         except SSLAgainErrors:
             pass
+        self._outgoing_app_size += self._outgoing.pending - pending
         self._process_outgoing()
 
     def _process_outgoing(self):
         if not self._ssl_writing_paused:
             data = self._outgoing.read()
+            self._outgoing_app_size = 0
             if len(data):
                 self._transport.write(data)
         self._control_app_writing()
@@ -894,7 +899,14 @@ def resume():
     # Flow control for reads from SSL socket
 
     def _control_ssl_reading(self):
-        size = self._get_read_buffer_size()
+        # Also count the records OpenSSL writes on its own while reading
+        # (alerts, KeyUpdate replies, ...) which the peer does not read:
+        # the app is not involved in producing them, so pausing it does
+        # not bound them; only pausing reads from the peer does.  App data
+        # is left out so that two peers which both write before they read
+        # cannot deadlock.
+        size = (self._get_read_buffer_size() +
+                self._outgoing.pending - self._outgoing_app_size)
         if size >= self._incoming_high_water and not self._ssl_reading_paused:
             self._ssl_reading_paused = True
             self._transport.pause_reading()
@@ -927,6 +939,7 @@ def resume_writing(self):
         assert self._ssl_writing_paused
         self._ssl_writing_paused = False
         self._process_outgoing()
+        self._control_ssl_reading()
 
     def _fatal_error(self, exc, message='Fatal error on transport'):
         if self._transport:

Lib/test/test_asyncio/test_sslproto.py

@@ -265,6 +265,46 @@ def test_write_after_closing(self):
         # should not raise
         self.assertIsNone(transp.write(b'data'))
 
+    def test_pause_reading_if_ssl_replies_are_not_read(self):
+        # Output which OpenSSL produces on its own while reading (alerts,
+        # KeyUpdate, ...) must not pile up if the peer does not read it.
+        ssl_proto = self.ssl_protocol()
+        transport = self.connection_made(ssl_proto, do_handshake=mock.Mock())
+        reply = b'x' * (ssl_proto._incoming_high_water // 2)
+
+        def read(*args):
+            ssl_proto._outgoing.write(reply)
+            raise ssl.SSLWantReadError
+        ssl_proto._sslobj.read.side_effect = read
+
+        ssl_proto.pause_writing()
+        ssl_proto.buffer_updated(0)
+        transport.pause_reading.assert_not_called()
+        ssl_proto.buffer_updated(0)
+        transport.pause_reading.assert_called_once_with()
+        transport.resume_reading.assert_not_called()
+        transport.write.assert_not_called()
+
+        ssl_proto.resume_writing()
+        transport.write.assert_called_once_with(reply * 2)
+        transport.resume_reading.assert_called_once_with()
+
+    def test_no_pause_reading_if_app_data_is_not_read(self):
+        # Peers which both write a lot before they read must not deadlock.
+        ssl_proto = self.ssl_protocol()
+        transport = self.connection_made(ssl_proto, do_handshake=mock.Mock())
+        ssl_proto._sslobj.write.side_effect = ssl_proto._outgoing.write
+        data = b'x' * ssl_proto._incoming_high_water
+
+        ssl_proto.pause_writing()
+        ssl_proto._app_transport.write(data)
+        ssl_proto.buffer_updated(0)
+        transport.pause_reading.assert_not_called()
+
+        ssl_proto.resume_writing()
+        transport.write.assert_called_once_with(data)
+        transport.resume_reading.assert_not_called()
+
 
 ##############################################################################
 # Start TLS Tests

Misc/NEWS.d/next/Security/2026-10-22-08-08-57.gh-issue-160499.pf9H9g.rst

@@ -0,0 +1,5 @@
+Fix unbounded memory growth in :mod:`asyncio` TLS connections when a peer
+stops reading but keeps sending records that make OpenSSL generate its own
+replies (such as TLS 1.2 renegotiation alerts).  Reading from the peer is now
+paused when this output accumulates, instead of only when application data
+does.