I have a Python asyncio server written using the high-level Streams API. I want to enable TLS on an already established connection, as in STARTTLS in the SMTP and IMAP protocols. The asyncio event loop has a start_tls() function (added in Python 3.7), but it takes a protocol and a transport rather than a stream. The streams API does let you get the transport via StreamWriter.transport. I don't see a way to change the transport, which would be required after calling start_tls().

Is it possible to use start_tls() with the streams API?

Looking at the code for the streams API you'll notice that StreamReader and StreamWriter both store their transport in an internal _transport variable. It turns out that if you call start_tls() and then store the new transport in those variables it works just fine. All the usual caveats with using an internal API apply of course. Here's what this looks like for a server. On a client I think you can just drop the load_cert_chain and server_side bits.

transport = writer.transport
protocol = transport.get_protocol()
loop = asyncio.get_event_loop()
ssl_context = ssl.SSLContext()
ssl_context.load_cert_chain("/path/to/certchain", "/path/to/key")
new_transport = await loop.start_tls(transport, protocol, ssl_context, server_side=True)
writer._transport = new_transport
reader._transport = new_transport