[SOLVED] How do I enable TLS on an already connected Python asyncio stream?

How do I enable TLS on an already connected Python asyncio stream?

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?

Solution

As of python 3.11, this is as simple as calling StreamWriter.start_tls https://docs.python.org/3/library/asyncio-stream.html#asyncio.StreamWriter.start_tls

E.g. for a client:

import asyncio
import ssl
reader, writer = await asyncio.open_connection('your_server_uri_here', port)
context = ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH)
# Adjust the ssl context for your particular needs (e.g. self signed certificate or what not)
await writer.start_tls(context)

and for a server, you would add the upgrade code to your handler

import asyncio
from asyncio.streams import StreamReader, StreamWriter
import ssl

# Modify to match your needs
context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
context.load_cert_chain("path/to/cert", "path/to/key")

def handler(reader: StreamReader, writer: StreamWriter):
  await writer.start_tls(context)
  # Do stuff

await asyncio.start_server(handler, "server hostname", 8000)