如何将Python同步SSL/TLS握手代码改写为异步以并行检测多站点？

问题描述 #

我有一段用于连接网站并获取SSL/TLS版本及服务器证书详情的传统同步代码：

import ssl, socket

SERVER_HOSTNAME = "www.example.com"

try:
    sock = socket.create_connection((SERVER_HOSTNAME, 443), timeout=3)
    ssl_context = ssl.create_default_context()
    ssock = ssl_context.wrap_socket(sock, server_hostname=SERVER_HOSTNAME)
    tls_info = {
        'version': ssock.version(),
        'peercert': ssock.getpeercert(),
    }
except Exception as e:
    quit(e)

如何将其改写为异步代码，以便并行获取多个站点的上述信息？我发现asyncore已被弃用，aioopenssl找不到合适的使用示例，aiohttp虽支持但大材小用，因为我仅需完成SSL/TLS握手，无需发起实际HTTP请求。

异步实现方案 #

可以直接用Python标准库的asyncio结合ssl模块实现，无需依赖第三方库，完全匹配仅完成SSL/TLS握手的需求，不会像aiohttp那样冗余。

核心思路 #

• 用asyncio.open_connection建立异步TCP连接，该方法支持传入SSL上下文，自动完成SSL握手
• 将单个站点的信息获取逻辑封装为异步函数
• 通过asyncio.gather并行执行多个异步任务，实现多站点同时查询

完整代码示例 #

import asyncio
import ssl

async def fetch_tls_info(hostname: str, port: int = 443, timeout: int = 3) -> dict:
    try:
        ssl_context = ssl.create_default_context()
        # 异步建立连接并完成SSL握手
        reader, writer = await asyncio.wait_for(
            asyncio.open_connection(hostname, port, ssl=ssl_context, server_hostname=hostname),
            timeout=timeout
        )
        # 获取SSL套接对象提取信息
        ssl_socket = writer.get_extra_info('ssl_object')
        tls_info = {
            'hostname': hostname,
            'version': ssl_socket.version(),
            'peercert': ssl_socket.getpeercert()
        }
        # 关闭连接
        writer.close()
        await writer.wait_closed()
        return tls_info
    except Exception as e:
        return {
            'hostname': hostname,
            'error': str(e)
        }

async def main():
    # 待查询的站点列表
    sites = ["www.example.com", "www.google.com", "github.com"]
    # 并行执行所有查询任务
    results = await asyncio.gather(*[fetch_tls_info(site) for site in sites])
    # 输出结果
    for res in results:
        print(f"站点: {res['hostname']}")
        if 'error' in res:
            print(f"错误: {res['error']}\n")
        else:
            print(f"TLS版本: {res['version']}")
            print(f"证书信息: {res['peercert']}\n")

if __name__ == "__main__":
    asyncio.run(main())

代码说明 #

• asyncio.open_connection：替代同步的socket.create_connection，异步建立TCP连接并自动应用SSL上下文完成握手，无需手动调用wrap_socket
• asyncio.wait_for：为单个连接设置超时，避免单个慢站点阻塞整个任务队列
• writer.get_extra_info('ssl_object')：获取底层SSL套接字对象，用于提取TLS版本和证书详情
• asyncio.gather：打包多个异步任务并行执行，大幅提升多站点查询效率

内容的提问来源于stack exchange，提问作者John Heyer