如何在已连接的 Python 异步流上启用 TLS？

Question

我有一个使用高级Streams API编写的 Python 异步服务器。我想在已建立的连接上启用 TLS，如 SMTP 和 IMAP 协议中的 STARTTLS。asyncio 事件循环有一个start_tls()函数（在 Python 3.7 中添加），但它需要一个协议和一个传输而不是一个流。流 API 确实允许您通过StreamWriter.transport获取传输。我没有看到改变传输的方法，这在调用 start_tls() 后是必需的。

是否可以将 start_tls() 与流 API 一起使用？

Answer 1

从 python 3.11 开始，这就像调用StreamWriter.start_tls https://docs.python.org/3/library/asyncio-stream.html#asyncio.StreamWriter.start_tls一样简单

例如对于客户：

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)

对于服务器，您可以将升级代码添加到您的处理程序中

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)