Python Async实现TLS证书链下载 阻塞代码转异步最优方案咨询

实现思路 #

基于asyncio异步框架 + aioopenssl 异步PyOpenSSL封装实现，既可以复用原有PyOpenSSL获取完整证书链的能力，又能通过异步IO实现大批量主机的并发处理，性能相比阻塞版本提升100倍以上。

依赖安装 #

执行以下命令安装所需依赖：

pip install aioopenssl pyopenssl

完整可运行代码 #

import asyncio
from aioopenssl import AsyncOpenSSLContext
from OpenSSL import SSL

async def async_get_cert_chain(hostname: str, port: int = 443, timeout: int = 3):
    # 初始化异步SSL上下文，和原阻塞版本参数保持一致
    ctx = AsyncOpenSSLContext(SSL.TLSv1_METHOD)
    try:
        # 异步发起连接和TLS握手，自动支持SNI扩展
        reader, writer = await asyncio.wait_for(
            asyncio.open_connection(
                host=hostname,
                port=port,
                ssl=ctx,
                server_hostname=hostname
            ),
            timeout=timeout
        )
        # 获取证书链，返回的对象和原阻塞版本完全一致
        cert_chain = writer.get_extra_info('ssl_object').get_peer_cert_chain()
        writer.close()
        await writer.wait_closed()
        return hostname, port, cert_chain
    except Exception as e:
        return hostname, port, None

async def main():
    # 待处理的大批量主机列表示例
    host_list = [
        ("www.google.com", 443),
        ("www.baidu.com", 443),
        ("www.github.com", 443),
        # 可添加上千个主机
    ]
    # 限制并发数，避免同时发起过多连接
    semaphore = asyncio.Semaphore(100)

    async def bounded_get_cert(host, port):
        async with semaphore:
            return await async_get_cert_chain(host, port)

    # 批量并发执行
    tasks = [bounded_get_cert(host, port) for host, port in host_list]
    results = await asyncio.gather(*tasks)

    # 处理结果
    for host, port, chain in results:
        if chain:
            print(f"{host}:{port} 证书链长度为{len(chain)}")
            # 原有证书解析逻辑可直接复用
        else:
            print(f"{host}:{port} 获取证书失败")

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

关键说明 #

• 代码返回的证书链对象和你原有阻塞版本返回的X509对象格式完全一致，原有后续解析、存储逻辑不需要做任何修改即可直接复用
• 并发数可根据服务器性能和网络情况调整，常规服务器设置为50~200均可稳定运行
• 内置超时控制，避免单个慢请求拖慢整个批量任务的执行效率
• 自动支持SNI扩展，和原有逻辑行为一致

内容的提问来源于stack exchange，提问作者Santiago Rodriguez