如何创建异步 ssl 数据报端点?
Posted
技术标签:
【中文标题】如何创建异步 ssl 数据报端点?【英文标题】:How to create asyncio ssl datagram endpoint? 【发布时间】:2021-04-05 21:06:38 【问题描述】:我不明白在什么时候以及以什么方式建立连接 ssl。 context.wrap_socket(sock, server_side=True) 和 ssock.accept 在 connection_made 或 loop.start_tls...
import socket
import ssl
import asyncio
class MyProtocol(asyncio.DatagramProtocol):
def connection_made(self, transport):
pass
def datagram_received(self, data, client_address):
pass
if __name__ == '__main__':
loop = asyncio.get_event_loop()
address = ('', 11111)
ssl_context = ssl.create_default_context()
ssl_context.load_cert_chain(certfile='iotivitycloud.crt', keyfile='iotivitycloud.key')
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
sock.bind(address)
transport, protocol = loop.run_until_complete(
loop.create_datagram_endpoint(
lambda: MyProtocol(), sock=sock))
loop.run_forever()
【问题讨论】:
【参考方案1】:要启用 TLS over UDP,您需要安装 DTLS 库 python3-dtls
。
import asyncio
import logging
import ssl
from socket import AF_INET, IPPROTO_UDP, SOCK_DGRAM, socket
from ssl import SSLSocket
from typing import Optional, Tuple
from dtls import do_patch
do_patch()
logger = logging.getLogger(__name__)
class MyProtocol(asyncio.DatagramProtocol):
def __init__(self, on_con_lost: asyncio.Future) -> None:
self.on_con_lost = on_con_lost
def connection_made(self, transport: asyncio.BaseTransport) -> None:
logger.debug("Connection opened")
self.transport = transport
def datagram_received(
self, data: bytes, addr: Optional[Tuple[str, int]]
) -> None:
logger.debug("Received: %s", data.decode())
def connection_lost(self, exc: Optional[Exception]) -> None:
logger.debug("Connection closed")
self.on_con_lost.set_result(True)
async def create_udp_tls_socket(address: Tuple[str, int]) -> SSLSocket:
sock = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)
ssock = ssl.wrap_socket(sock)
ssock.bind(address)
return ssock
async def main() -> None:
logging.basicConfig(level=logging.DEBUG)
address = ("", 11111)
sock = await create_udp_tls_socket(address)
loop = asyncio.get_running_loop()
transport, protocol = await loop.create_datagram_endpoint(
lambda: MyProtocol(on_con_lost=loop.create_future()),
sock=sock,
)
# Simulate the reception of data from the network.
loop.call_soon_threadsafe(transport.sendto, b"test-msg", address)
try:
await protocol.on_con_lost
finally:
transport.close()
sock.close()
if __name__ == "__main__":
asyncio.run(main())
【讨论】:
以上是关于如何创建异步 ssl 数据报端点?的主要内容,如果未能解决你的问题,请参考以下文章
(40)C#里使用WebRequest和出错:请求被中止: 未能创建 SSL/TLS 安全通道
(40)C#里使用WebRequest和出错:请求被中止: 未能创建 SSL/TLS 安全通道