Fix start_tls() loses buffer data and hangs when upgrading the connection

[fizzi01]

This PR fixes a critical issue where buffered data in StreamReader is lost when upgrading a connection to TLS using StreamWriter.start_tls() or loop.start_tls(), causing the call to hang indefinitely. The issue exists in both asyncio (see python/cpython#142352) and uvloop.

The main focus is to verify and guarantee that any data buffered before upgrading a connection to TLS is not lost and is correctly transferred. These changes are effectively a port of python/cpython#142354 with absolutely no difference, except for the test that creates a TCP server, establishes a connection, buffers data, performs a TLS upgrade, and verifies that all messages (both pre-upgrade and post-upgrade) are received correctly.

Fixes #707

[fizzi01]

The current fix breaks the Test:test_start_tls because it uses protocol classes without the _stream_reader attribute.
A workaround could be to fallback to the old behavior when that attribute is missing:

async def start_tls(self, transport, protocol, ...):
        # Transfer buffered data from the old protocol to the new one.
        stream_buff = None
        if hasattr(protocol, '_stream_reader'):
            stream_reader = protocol._stream_reader
            if stream_reader is not None and hasattr(stream_reader, '_buffer'):
                stream_buff = stream_reader._buffer

        if stream_buff is not None:
            ssl_protocol._incoming.write(stream_buff)
            stream_buff.clear()

or to modify the base protocol class implementation in `test_context.py:

class _BaseProtocol(asyncio.BaseProtocol):
    def __init__(self, cvar, *, loop=None):
        ...
        self._stream_reader: asyncio.StreamReader = asyncio.StreamReader(loop=loop)

Do we always have a protocol with a _stream_reader attribute?

[fantix]

Thanks for the PR!

Do we always have a protocol with a _stream_reader attribute?

No I don't think so. Though, I don't understand the test failure - looks like you already have if not hasattr(protocol, '_stream_reader'): return?

[fizzi01]

Though, I don't understand the test failure - looks like you already have if not hasattr(protocol, '_stream_reader'): return?

The function start_tls() should either return a transport or raise an error. Since we don't consistently handle a buffer, we can just skip the new buffer logic and run as before in that cases.

transport = await self.loop.start_tls(
                    proto.transport, proto, client_sslctx,
                    server_hostname='127.0.0.1',
                )