You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

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格式的证书文件和私钥文件:

  1. 导出客户端证书:
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证书:

  1. 先安装依赖库:
pip install pyopenssl cryptography
  1. 修改测试脚本,新增证书解析逻辑:
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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.09.29 19:06:03