Zeep同步请求正常,异步结合httpx出现SSLV3握手失败求助
Zeep异步请求遭遇SSL握手失败
问题概述
Zeep同步调用正常运行:
from zeep import Client from zeep.transports import Transport import requests session = requests.Session() session.cert = cert_path transport = Transport(session=session, operation_timeout=60) client = Client( "url.wsdl", transport=transport, ) client.service.GetSomething(request) # 执行成功
异步实现尝试
改用Zeep异步客户端时出现错误:
from zeep import AsyncClient as AsyncClientZeep from zeep.transports import AsyncTransport import httpx async def async_request(): async with httpx.AsyncClient(cert=cert_path) as client_httpx: wsdl_response = await client_httpx.get('url.wsdl') print(f'WSDL Status Code: {wsdl_response.status_code}') # 此部分正常 # 创建带httpx客户端的Zeep AsyncTransport transport = AsyncTransport(client=client_httpx, operation_timeout=60) # 初始化Zeep客户端 client = AsyncClientZeep("url.wsdl", transport=transport) response = await client.service.GetSomething(request) # 此处报错 # 调用示例 import asyncio asyncio.run(async_request())
报错信息
ConnectError: [SSL: SSLV3_ALERT_HANDSHAKE_FAILURE] sslv3 alert handshake failure (_ssl.c:1000)
已尝试的排查步骤
试过多种SSL配置,例如:
import ssl ssl_context = ssl.create_default_context() ssl_context.set_ciphers('DEFAULT') ssl_context.options |= ssl.OP_NO_SSLv3 ssl_context.options |= ssl.OP_NO_TLSv1 ssl_context.options |= ssl.OP_NO_TLSv1_1 async with httpx.AsyncClient(verify=ssl_context): # 仍出现相同错误
但所有配置均未解决问题。
备注
Zeep当前支持使用httpx进行异步操作。
解决方案
- 让AsyncTransport直接处理证书配置
避免手动管理httpx客户端,直接将证书参数传给AsyncTransport,由Zeep内部处理客户端初始化:
from zeep import AsyncClient as AsyncClientZeep from zeep.transports import AsyncTransport import asyncio async def async_request(): # 直接在AsyncTransport中传入证书 transport = AsyncTransport(cert=cert_path, operation_timeout=60) # 使用AsyncClient的上下文管理器确保资源正确释放 async with AsyncClientZeep("url.wsdl", transport=transport) as client: response = await client.service.GetSomething(request) print(response) asyncio.run(async_request())
- 确保SSL上下文和证书正确关联
如果需要自定义SSL上下文,必须加载证书链,同时确保httpx客户端和AsyncTransport都使用该上下文:
import ssl from zeep import AsyncClient as AsyncClientZeep from zeep.transports import AsyncTransport import httpx import asyncio async def async_request(): ssl_context = ssl.create_default_context() # 加载证书文件(若cert_path是证书+密钥的组合,可分别指定certfile和keyfile) ssl_context.load_cert_chain(cert_path) # 指定与服务端兼容的密码套件(可通过openssl命令查询服务端支持的套件) ssl_context.set_ciphers('ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256') # 禁用旧版本协议 ssl_context.options |= ssl.OP_NO_SSLv3 ssl_context.options |= ssl.OP_NO_TLSv1 ssl_context.options |= ssl.OP_NO_TLSv1_1 async with httpx.AsyncClient(cert=cert_path, verify=ssl_context) as client_httpx: transport = AsyncTransport(client=client_httpx, operation_timeout=60) async with AsyncClientZeep("url.wsdl", transport=transport) as client: response = await client.service.GetSomething(request) print(response) asyncio.run(async_request())
- 匹配服务端的TLS配置
同步请求成功说明requests的默认TLS配置和服务端兼容,可通过以下方式让httpx对齐配置:
- 查看requests使用的SSL上下文参数,在httpx中复制相同设置
- 用
openssl s_client -connect your-service-domain:port命令获取服务端支持的TLS版本和密码套件,然后在ssl_context中明确指定
内容的提问来源于stack exchange,提问作者Simon
相关产品推荐
相关产品推荐

