Locust调用PFX证书报错ssl.SSLError: [SSL] PEM lib (_ssl.c:4065)求助
报错原因分析
该SSL错误是因为Locust底层依赖的requests库的cert参数仅支持PEM格式的客户端证书/密钥文件,你传入的是PFX(PKCS#12)格式的证书,无法被PEM解析库识别,因此触发报错。
代码中设置的self.client.verify = False仅用于跳过服务端SSL证书校验,和客户端身份验证的证书逻辑无关,不会解决当前报错。
解决方案
方案1:将PFX证书转换为PEM格式(推荐,复杂度最低)
使用openssl工具执行以下两条命令,把PFX证书拆分为PEM格式的证书文件和私钥文件:
- 导出客户端证书:
openssl pkcs12 -in 你的证书文件.pfx -clcerts -nokeys -out client.crt
执行时会要求输入原PFX证书的密码,按提示输入即可。
2. 导出客户端私钥:
# 导出无密码保护的私钥 openssl pkcs12 -in 你的证书文件.pfx -nocerts -nodes -out client.key # 如果需要私钥加密码,去掉命令中的 -nodes 参数即可
转换完成后修改请求代码即可,修改后的示例:
@task def get_status(self): self.client.headers = self.headers self.client.verify = False self.client.post("/uam/Link/Get/v1", data=self.payload, # 替换为转换后的PEM证书、私钥路径 cert=("path/to/client.crt", "path/to/client.key"), # 如果私钥设置了密码,新增以下参数即可 # cert_password="你的私钥密码" )
方案2:代码中直接解析PFX证书(无需手动转换格式)
如果不想手动转换证书格式,可以借助加密库在代码运行时动态解析PFX证书:
- 先安装依赖库:
pip install pyopenssl cryptography
- 修改测试脚本,新增证书解析逻辑:
from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.serialization import pkcs12 import tempfile import os class MyHttpTest(HttpUser): wait_time = constant(5) host = "https://abc.myorg.net" headers = { 'Content-Type': "application/json", 'Ocp-Apim-Trace': "true", 'Ocp-Apim-Subscription-Key': "<APIM subscription Key>", 'Cache-Control': "no-cache" } payload = "{\n\t\"name\": \"abc\"\n}" def on_start(self): # 加载PFX证书 with open("<path to pfx certificate>", "rb") as f: pfx_data = f.read() private_key, cert, _ = pkcs12.load_key_and_certificates( pfx_data, password=b"<password of pfx certificate>" ) # 生成临时PEM证书文件 self.cert_file = tempfile.NamedTemporaryFile(delete=False, suffix=".crt") self.cert_file.write(cert.public_bytes(encoding=serialization.Encoding.PEM)) self.cert_file.close() # 生成临时PEM私钥文件 self.key_file = tempfile.NamedTemporaryFile(delete=False, suffix=".key") self.key_file.write(private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.TraditionalOpenSSL, encryption_algorithm=serialization.NoEncryption() )) self.key_file.close() # 给客户端预设置证书 self.client.cert = (self.cert_file.name, self.key_file.name) def on_stop(self): # 测试结束清理临时文件 os.unlink(self.cert_file.name) os.unlink(self.key_file.name) @task def get_status(self): self.client.headers = self.headers self.client.verify = False # 无需再传cert参数,已在初始化时设置 self.client.post("/uam/Link/Get/v1", data=self.payload)
内容的提问来源于stack exchange,提问作者Ajay
相关产品推荐
相关产品推荐

