Python Requests SSL 证书验证详解

SSL/TLS证书概述

SSL证书验证流程

当客户端(你的Python程序)连接到HTTPS服务器时，会进行以下验证流程：

Requests默认的SSL验证行为

Requests库默认启用SSL证书验证，这意味着：

import requests

# 默认情况下，verify=True
response = requests.get('https://api.github.com')

# 上面的代码等价于：
response = requests.get('https://api.github.com', verify=True)

import requests

try:
    # 访问一个证书无效的网站
    response = requests.get('https://expired.badssl.com')
except requests.exceptions.SSLError as e:
    print(f"SSL证书验证失败: {e}")

信任的证书颁发机构(CA)

Requests使用certifi库作为其默认的CA证书包。certifi是一个精心维护的Mozilla CA证书包副本。

import certifi
import requests

# 查看certifi的CA证书路径
print("CA证书路径:", certifi.where())

# 使用certifi的证书路径进行验证
response = requests.get('https://api.github.com', verify=certifi.where())

# 也可以为Session设置默认的CA证书
session = requests.Session()
session.verify = certifi.where()

禁用SSL证书验证

如何禁用SSL验证

import requests
import warnings

# 禁用SSL警告（可选）
warnings.filterwarnings('ignore', message='Unverified HTTPS request')

# 为单次请求禁用SSL验证
response = requests.get('https://self-signed.badssl.com', verify=False)

# 检查响应
print(f"状态码: {response.status_code}")
print("警告：连接不安全，证书未验证！")

import requests

# 创建Session并禁用SSL验证
session = requests.Session()
session.verify = False

# 或者使用上下文管理器
with requests.Session() as session:
    session.verify = False
    response = session.get('https://self-signed.badssl.com')
    print(f"状态码: {response.status_code}")

# 注意：这会影响该Session的所有请求

import os
import requests

# 通过环境变量全局禁用SSL验证（不推荐）
# 这会影响到当前进程的所有requests请求
os.environ['REQUESTS_CA_BUNDLE'] = ''
os.environ['CURL_CA_BUNDLE'] = ''

# 或者使用Python的urllib3环境变量
os.environ['PYTHONHTTPSVERIFY'] = '0'

# 现在所有requests请求都不会验证SSL证书
response = requests.get('https://self-signed.badssl.com')
print(f"状态码: {response.status_code}")

# 强烈不推荐这种方法，因为它会影响所有请求

何时可以禁用SSL验证？

虽然不推荐，但在以下特定情况下可能需要临时禁用SSL验证：

使用自定义CA证书

如果你需要连接到使用自签名证书或内部CA颁发证书的服务器，最佳实践是使用自定义CA证书。

1. 指定自定义CA证书文件

import requests

# 指定自定义CA证书文件路径
# 证书文件可以是PEM或DER格式
response = requests.get(
    'https://internal-api.company.com',
    verify='/path/to/custom/ca-bundle.crt'
)

# 对于Session对象
session = requests.Session()
session.verify = '/path/to/custom/ca-bundle.crt'

# 或者同时使用系统证书和自定义证书
import certifi
import os

# 创建一个合并的证书文件
with open('merged-ca-bundle.crt', 'w') as outfile:
    # 写入系统CA证书
    with open(certifi.where(), 'r') as infile:
        outfile.write(infile.read())

    # 写入自定义CA证书
    with open('/path/to/custom/ca.crt', 'r') as infile:
        outfile.write(infile.read())

# 使用合并的证书文件
response = requests.get(
    'https://internal-api.company.com',
    verify='merged-ca-bundle.crt'
)

2. 创建自签名证书并信任它

对于开发和测试环境，可以创建自签名证书：

# 生成私钥
openssl genrsa -out server.key 2048

# 生成证书签名请求
openssl req -new -key server.key -out server.csr \
  -subj "/C=CN/ST=Beijing/L=Beijing/O=MyCompany/CN=localhost"

# 生成自签名证书
openssl x509 -req -days 365 -in server.csr \
  -signkey server.key -out server.crt

# 创建PEM格式的证书包
cat server.crt > ca-bundle.crt

import requests

# 使用自签名证书
response = requests.get(
    'https://localhost:8443/api/data',
    verify='./ca-bundle.crt'  # 包含自签名证书的CA包
)

# 如果需要，也可以同时禁用主机名验证
response = requests.get(
    'https://localhost:8443/api/data',
    verify='./ca-bundle.crt',
    # 注意：Requests没有直接提供禁用主机名验证的参数
    # 如果需要，可以创建自定义的SSL上下文
)

import ssl
import requests
from requests.adapters import HTTPAdapter
from urllib3.poolmanager import PoolManager

class CustomSSLAdapter(HTTPAdapter):
    """自定义SSL适配器，用于更精细的SSL控制"""

    def init_poolmanager(self, *args, **kwargs):
        # 创建自定义SSL上下文
        context = ssl.create_default_context()

        # 加载自定义CA证书
        context.load_verify_locations(cafile='./ca-bundle.crt')

        # 禁用主机名验证（如果需要）
        # context.check_hostname = False

        # 设置其他SSL选项
        context.options |= ssl.OP_NO_SSLv2
        context.options |= ssl.OP_NO_SSLv3

        kwargs['ssl_context'] = context
        return super().init_poolmanager(*args, **kwargs)

# 使用自定义适配器
session = requests.Session()
adapter = CustomSSLAdapter()
session.mount('https://', adapter)

response = session.get('https://localhost:8443/api/data')

使用客户端证书

某些服务器要求客户端提供证书进行双向认证（mutual TLS）。这种情况下，除了验证服务器证书外，还需要提供客户端证书。

使用客户端证书进行请求

import requests

# 提供客户端证书文件（包含证书和私钥）
response = requests.get(
    'https://api.secure-bank.com/user',
    cert='/path/to/client-cert.pem'
)

# 证书文件可以是以下格式：
# 1. PEM格式：包含证书和私钥
# 2. 证书和私钥分别的文件（见下一个选项卡）

import requests

# 分别提供证书和私钥文件
response = requests.get(
    'https://api.secure-bank.com/user',
    cert=('/path/to/client.crt', '/path/to/client.key')
)

# 对于Session对象
session = requests.Session()
session.cert = ('/path/to/client.crt', '/path/to/client.key')

# 如果私钥有密码
from requests.auth import HTTPBasicAuth

# 注意：Requests不直接支持加密的私钥文件
# 你需要先解密私钥文件，或者使用无密码的私钥
response = requests.get(
    'https://api.secure-bank.com/user',
    cert=('/path/to/client.crt', '/path/to/decrypted.key')
)

import requests

# 创建包含证书和私钥的PEM文件
# client.pem 内容格式：
"""
-----BEGIN CERTIFICATE-----
（客户端证书内容）
-----END CERTIFICATE-----
-----BEGIN PRIVATE KEY-----
（客户端私钥内容）
-----END PRIVATE KEY-----
"""

# 使用PEM文件
response = requests.get(
    'https://api.secure-bank.com/user',
    cert='/path/to/client.pem'
)

# 或者使用字符串形式的证书（不推荐，仅适用于测试）
cert_data = """-----BEGIN CERTIFICATE-----
MIIE...（证书内容）
-----END CERTIFICATE-----"""

key_data = """-----BEGIN PRIVATE KEY-----
MIIE...（私钥内容）
-----END PRIVATE KEY-----"""

# 将证书和私钥写入临时文件
import tempfile
import os

with tempfile.NamedTemporaryFile(mode='w', suffix='.pem', delete=False) as cert_file:
    cert_file.write(cert_data + '\n' + key_data)
    cert_path = cert_file.name

try:
    response = requests.get(
        'https://api.secure-bank.com/user',
        cert=cert_path
    )
finally:
    # 删除临时文件
    os.unlink(cert_path)

客户端证书最佳实践

• 保护私钥：客户端私钥应该严格保密，文件权限应设为600 (仅所有者可读写)
• 使用密码保护：为私钥设置强密码，但注意Requests需要解密后的私钥
• 定期轮换：定期更新客户端证书，增强安全性
• 环境变量：不要将证书路径硬编码在代码中，使用环境变量或配置文件

常见SSL错误与解决方法

调试SSL连接

import requests
import ssl
import socket
import certifi

def debug_ssl_connection(hostname, port=443):
    """调试SSL连接问题"""

    print(f"调试 {hostname}:{port}")
    print("=" * 50)

    # 1. 测试基本连接
    try:
        response = requests.get(f'https://{hostname}', timeout=10)
        print(f"✓ 基本HTTPS连接成功: {response.status_code}")
    except requests.exceptions.SSLError as e:
        print(f"✗ SSL错误: {e}")
    except Exception as e:
        print(f"✗ 连接错误: {e}")

    # 2. 检查证书
    print("\n检查证书信息:")
    context = ssl.create_default_context()

    try:
        with socket.create_connection((hostname, port), timeout=10) as sock:
            with context.wrap_socket(sock, server_hostname=hostname) as ssock:
                cert = ssock.getpeercert()

                print(f"  证书主题: {cert.get('subject', 'N/A')}")
                print(f"  颁发者: {cert.get('issuer', 'N/A')}")

                # 获取证书的文本表示
                der_cert = ssock.getpeercert(binary_form=True)
                if der_cert:
                    import pem
                    cert_pem = ssl.DER_cert_to_PEM_cert(der_cert)
                    cert_obj = pem.parse(cert_pem.encode())[0]
                    print(f"  有效期: {cert_obj.not_before} 到 {cert_obj.not_after}")
    except Exception as e:
        print(f"  无法获取证书信息: {e}")

    # 3. 测试不同验证方式
    print("\n测试不同验证方式:")

    # 使用系统证书
    try:
        response = requests.get(f'https://{hostname}', verify=certifi.where())
        print("  ✓ 使用系统证书验证: 成功")
    except:
        print("  ✗ 使用系统证书验证: 失败")

    # 禁用验证
    try:
        response = requests.get(f'https://{hostname}', verify=False)
        print("  ✓ 禁用验证: 成功")
    except:
        print("  ✗ 禁用验证: 失败")

# 使用示例
debug_ssl_connection('api.github.com')

SSL证书验证最佳实践

配置管理建议

import os
import requests
from functools import lru_cache

class SecureRequestClient:
    """安全的HTTP客户端，处理SSL证书验证"""

    def __init__(self, base_url, env='production'):
        self.base_url = base_url
        self.env = env
        self.session = requests.Session()

        # 根据环境配置SSL验证
        self._configure_ssl()

        # 设置默认请求头
        self.session.headers.update({
            'User-Agent': 'SecureClient/1.0',
            'Accept': 'application/json'
        })

    def _configure_ssl(self):
        """根据环境配置SSL验证"""

        if self.env == 'production':
            # 生产环境：严格验证
            self.session.verify = True

        elif self.env == 'staging':
            # 预发布环境：使用自定义CA包
            custom_ca_path = os.getenv('STAGING_CA_BUNDLE', '')
            if os.path.exists(custom_ca_path):
                self.session.verify = custom_ca_path
            else:
                self.session.verify = True

        elif self.env == 'development':
            # 开发环境：根据配置决定
            ssl_verify = os.getenv('DEV_SSL_VERIFY', 'true').lower()
            if ssl_verify == 'false':
                import warnings
                warnings.filterwarnings('ignore',
                                      message='Unverified HTTPS request')
                self.session.verify = False
            else:
                # 使用开发环境的CA证书
                dev_ca_path = os.getenv('DEV_CA_BUNDLE', '')
                if dev_ca_path and os.path.exists(dev_ca_path):
                    self.session.verify = dev_ca_path
                else:
                    self.session.verify = True

        elif self.env == 'test':
            # 测试环境：可配置验证
            self.session.verify = os.getenv('TEST_SSL_VERIFY', True)

    @lru_cache(maxsize=128)
    def get_certificate_info(self, hostname):
        """缓存获取证书信息"""
        import ssl
        import socket

        context = ssl.create_default_context()

        try:
            with socket.create_connection((hostname, 443), timeout=5) as sock:
                with context.wrap_socket(sock,
                                       server_hostname=hostname) as ssock:
                    cert = ssock.getpeercert()
                    return {
                        'subject': cert.get('subject'),
                        'issuer': cert.get('issuer'),
                        'version': cert.get('version'),
                        'notBefore': cert.get('notBefore'),
                        'notAfter': cert.get('notAfter')
                    }
        except Exception as e:
            return {'error': str(e)}

    def get(self, endpoint, **kwargs):
        """发送GET请求"""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        return self.session.get(url, **kwargs)

    def close(self):
        """关闭Session"""
        self.session.close()

# 使用示例
def main():
    # 从环境变量读取配置
    env = os.getenv('APP_ENV', 'development')

    client = SecureRequestClient(
        base_url='https://api.example.com',
        env=env
    )

    try:
        # 获取证书信息
        cert_info = client.get_certificate_info('api.example.com')
        print(f"证书信息: {cert_info}")

        # 发送请求
        response = client.get('/users')
        print(f"响应: {response.status_code}")

    finally:
        client.close()

if __name__ == '__main__':
    main()