mTLS 双向认证工程实战：从证书签发到客户端验证的完整落地

实践教程 · 2026-07-13 · 17 阅读

前言

在等保三级、关基保护和金融行业合规场景中，"双向认证"（Mutual TLS, mTLS）是高频要求。与标准 TLS（只有服务端出示证书）不同，mTLS 要求客户端也必须出示证书用于身份验证。

mTLS 的核心复杂性不在 TLS 协议本身，而在证书体系的正确构建和国密算法的工程集成：

服务端和客户端各需一张由国密 CA 签发的证书
证书链必须使用国密根 CA 或国际根 CA
国密改造时需将国际算法替换为 SM2/SM3/SM4 组合

⚠️ 阅读前声明：本文所有代码使用 Python 的 cryptography 库（基于 SECP256R1 曲线 + SHA256）演示 mTLS 的通用工程框架。真正的国密合规场景需要将这些 SECP256R1 密钥替换为 SM2 密钥（OID: 1.2.156.10197.1.301），SHA256 替换为 SM3。文中每一处都明确标注了替换点，避免誤导。

环境准备

所需工具

Python 3.10+
cryptography 44.x：X.509 证书处理和 TLS 协议栈
FastAPI 0.110+：HTTP 框架
uvicorn：ASGI 服务器
httpx：带证书支持的 HTTP 客户端
OpenSSL 3.x：证书工具链验证

BASH

pip install cryptography==44.0.0 fastapi==0.115.0 uvicorn[standard]==0.30.0 httpx[brotli]==0.27.0

国密改造依赖（可选）

若需完整的国密 mTLS，还需：

gmssl 3.2.x：SM2/SM3 纯软件实现（用于密钥生成和签名）
Tongsuo（铜豌豆 OpenSSL）：支持国密码套件的 OpenSSL 分支
国密密码机/HSM：合规生产环境要求通过密评的硬件模块

一、证书体系构建（通用框架 + 国密改造方案）

1.1 证书层级设计

CODE

┌─────────────────────────────────────────────┐
│           根 CA 证书（自签发）                │
│         有效期: 10年                         │
│         密钥: 硬件保护（HSM）或离线存储        │
└────────────────┬────────────────────────────┘
                 │ 签发
        ┌────────┴────────┐
        │                 │
┌───────┴──────┐  ┌───────┴──────┐
│  中间 CA 证书  │  │  中间 CA 证书  │
│  (TLS 服务端) │  │  (客户端认证) │
└──────┬───────┘  └──────┬───────┘
       │                 │
  ┌────┴────┐        ┌───┴────┐
  │ 服务端   │        │ 客户端  │
  │ 证书     │        │ 证书    │
  │ serverAuth│        │ clientAuth│
  └─────────┘        └────────┘

1.2 根 CA 证书生成

⚠️ 国密改造点 1/6：代码使用 ec.SECP256R1 + SHA256 作为演示。合规国密场景应使用 SM2 密钥 + gmssl/Tongsuo 完成签名。

PYTHON

#!/usr/bin/env python3
"""根 CA 证书生成脚本"""
from cryptography import x509
from cryptography.x509.oid import NameOID, ExtendedKeyUsageOID
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec, utils
import datetime
from pathlib import Path


def create_root_ca(
    common_name: str = "My Demo Root CA",
    valid_years: int = 10,
    output_dir: str = "./certs"
):
    """
    创建根 CA 证书（SECP256R1 + SHA256 演示版）
    
    ⚠️ 国密合规改造路径：
      1. 将 ec.SECP256R1() 替换为国密 SM2 密钥（via gmssl 或密码机）
      2. 将 hashes.SHA256() 替换为 hashes.SM3()（需 Tongsuo）
      3. 证书 OID 设置为 1.2.156.10197.1.301（SM2 算法标识）
    """
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)
    
    # === 国密改造点：此处用 SECP256R1 演示密钥生成 ===
    # 合规场景：使用 GM/T 0003.1-2012 定义的 SM2 曲线参数
    private_key = ec.SECP256R1().generate()
    public_key = private_key.public_key()
    
    # 构建证书
    subject = issuer = x509.Name([
        x509.NameAttribute(NameOID.COUNTRY_NAME, "CN"),
        x509.NameAttribute(NameOID.ORGANIZATION_NAME, "My Organization"),
        x509.NameAttribute(NameOID.COMMON_NAME, common_name),
    ])
    
    builder = x509.CertificateBuilder()
    builder = builder.subject_name(subject)
    builder = builder.issuer_name(issuer)
    builder = builder.public_key(public_key)
    builder = builder.serial_number(x509.random_serial_number())
    builder = builder.not_valid_before(
        datetime.datetime.utcnow() - datetime.timedelta(days=1)
    )
    builder = builder.not_valid_after(
        datetime.datetime.utcnow() + datetime.timedelta(days=365 * valid_years)
    )
    
    # 约束扩展
    builder = builder.add_extension(
        x509.SubjectKeyIdentifier.from_public_key(public_key),
        critical=False,
    )
    builder = builder.add_extension(
        x509.AuthorityKeyIdentifier.from_issuer_public_key(public_key),
        critical=False,
    )
    builder = builder.add_extension(
        x509.BasicConstraints(ca=True, path_length=1),
        critical=True,
    )
    builder = builder.add_extension(
        x509.KeyUsage(
            key_cert_sign=True,
            crl_sign=True,
            digital_signature=False,
            content_commitment=False,
            key_encipherment=False,
            data_encipherment=False,
            key_agreement=False,
            encipher_only=False,
            decipher_only=False,
        ),
        critical=True,
    )
    
    # === 国密改造点：签名算法 SHA256 → SM3 ===
    cert = builder.sign(private_key, hashes.SHA256())
    
    # 存储
    key_path = output_path / "root-ca.key"
    cert_path = output_path / "root-ca.crt"
    
    key_path.write_bytes(
        private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.BestAvailableEncryption(
                b"root-ca-password"
            )
        )
    )
    cert_path.write_bytes(
        cert.public_bytes(serialization.Encoding.PEM)
    )
    
    print(f"[+] Root CA 生成完成")
    print(f"    密钥: {key_path}")
    print(f"    证书: {cert_path}")
    print(f"    算法: SECP256R1 + SHA256 (演示版，国密合规需替换为 SM2 + SM3)")
    print(f"    指纹(SHA-256): {cert.fingerprint(hashes.SHA256()).hex()}")
    return key_path, cert_path


if __name__ == "__main__":
    create_root_ca()

1.3 服务端证书签发

⚠️ 国密改造点 2/6：服务端密钥使用 SECP256R1 + SHA256 签名。国密合规需替换为 SM2 密钥 + SM3 签名（gmssl + Tongsuo）。

PYTHON

#!/usr/bin/env python3
"""服务端证书签发脚本"""
from cryptography import x509
from cryptography.x509.oid import NameOID, ExtendedKeyUsageOID
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.backends import default_backend
import datetime
from pathlib import Path


def load_ca(key_path: str, password: bytes) -> tuple:
    """加载 CA 密钥和证书"""
    priv = serialization.load_pem_private_key(
        Path(key_path).read_bytes(), password=password,
        backend=default_backend()
    )
    cert = x509.load_pem_x509_certificate(
        Path(key_path.with_suffix(".crt")).read_bytes(),
        backend=default_backend()
    )
    return priv, cert


def issue_server_cert(
    ca_key_path: str,
    ca_password: bytes,
    common_name: str = "server.example.com",
    san_dns: list = None,
    valid_days: int = 365,
    output_dir: str = "./certs"
) -> tuple:
    """
    签发 TLS 服务端证书（SECP256R1 + SHA256 演示版）
    
    ⚠️ 国密合规改造路径：
      1. server_key = gmssl_sm2_keygen() 替换 ec.SECP256R1()
      2. 证书签名哈希从 SHA256 改为 SM3
      3. OID 指定 1.2.156.10197.1.301（国密 SM2）
    """
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)
    
    ca_key, ca_cert = load_ca(ca_key_path, ca_password)
    
    # === 国密改造点：密钥生成改为 SM2 ===
    server_key = ec.SECP256R1().generate()
    server_public = server_key.public_key()
    
    # 构建主题
    subject = x509.Name([
        x509.NameAttribute(NameOID.COUNTRY_NAME, "CN"),
        x509.NameAttribute(NameOID.ORGANIZATION_NAME, "My Organization"),
        x509.NameAttribute(NameOID.COMMON_NAME, common_name),
    ])
    
    builder = x509.CertificateBuilder()
    builder = builder.subject_name(subject)
    builder = builder.issuer_name(ca_cert.subject)
    builder = builder.public_key(server_public)
    builder = builder.serial_number(x509.random_serial_number())
    builder = builder.not_valid_before(
        datetime.datetime.utcnow() - datetime.timedelta(hours=1)
    )
    builder = builder.not_valid_after(
        datetime.datetime.utcnow() + datetime.timedelta(days=valid_days)
    )
    
    # Subject Alternative Name
    if san_dns is None:
        san_dns = [common_name, "localhost"]
    builder = builder.add_extension(
        x509.SubjectAlternativeName([x509.DNSName(d) for d in san_dns]),
        critical=False,
    )
    
    # 扩展密钥用法: serverAuth
    builder = builder.add_extension(
        x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]),
        critical=False,
    )
    
    # 基础约束: 非 CA
    builder = builder.add_extension(
        x509.BasicConstraints(ca=False, path_length=None),
        critical=True,
    )
    
    builder = builder.add_extension(
        x509.KeyUsage(
            key_cert_sign=False,
            crl_sign=False,
            digital_signature=True,
            content_commitment=False,
            key_encipherment=True,
            data_encipherment=False,
            key_agreement=True,
            encipher_only=False,
            decipher_only=False,
        ),
        critical=True,
    )
    
    # === 国密改造点：SHA256 → SM3 ===
    cert = builder.sign(ca_key, hashes.SHA256())
    
    # 存储
    key_path = output_path / "server.key"
    cert_path = output_path / "server.crt"
    chain_path = output_path / "server-chain.crt"
    
    key_path.write_bytes(
        server_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption()
        )
    )
    cert_pem = cert.public_bytes(serialization.Encoding.PEM)
    cert_path.write_bytes(cert_pem)
    
    # 证书链文件 (server cert + CA cert)
    ca_pem = Path(str(ca_key_path).replace(".key", ".crt")).read_bytes()
    chain_pem = cert_pem + ca_pem
    chain_path.write_bytes(chain_pem)
    
    print(f"[+] Server 证书签发完成")
    print(f"    CN: {common_name}")
    print(f"    SAN: {san_dns}")
    print(f"    有效期: {valid_days} 天")
    print(f"    算法: SECP256R1 + SHA256 (演示版)")
    return key_path, cert_path, chain_path


if __name__ == "__main__":
    issue_server_cert(
        ca_key_path=Path("./certs/root-ca.key"),
        ca_password=b"root-ca-password",
        common_name="server.example.com",
        san_dns=["server.example.com", "localhost", "127.0.0.1"]
    )

1.4 客户端证书签发

⚠️ 国密改造点 3/6：与 1.3 相同的改造方案，区别仅在 ExtendedKeyUsage 改为 clientAuth。

PYTHON

#!/usr/bin/env python3
"""客户端证书签发脚本（含客户端认证 EKU）"""
# [完整 import 同上文，此处省略]
from cryptography import x509
from cryptography.x509.oid import NameOID, ExtendedKeyUsageOID
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.backends import default_backend
import datetime
from pathlib import Path


def issue_client_cert(
    ca_key_path: str,
    ca_password: bytes,
    common_name: str = "client-user-001",
    org_unit: str = "security-team",
    valid_days: int = 365,
    output_dir: str = "./certs"
) -> tuple:
    """
    签发 TLS 客户端证书（演示版）
    
    关键差异: ExtendedKeyUsage 包含 clientAuth
    
    ⚠️ 国密改造同 1.3
    """
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)
    
    ca_key, ca_cert = load_ca(ca_key_path, ca_password)
    
    # === 国密改造点：密钥生成改为 SM2 ===
    client_key = ec.SECP256R1().generate()
    client_public = client_key.public_key()
    
    subject = x509.Name([
        x509.NameAttribute(NameOID.COUNTRY_NAME, "CN"),
        x509.NameAttribute(NameOID.ORGANIZATION_NAME, "My Organization"),
        x509.NameAttribute(NameOID.ORGANIZATIONAL_UNIT_NAME, org_unit),
        x509.NameAttribute(NameOID.COMMON_NAME, common_name),
    ])
    
    builder = x509.CertificateBuilder()
    builder = builder.subject_name(subject)
    builder = builder.issuer_name(ca_cert.subject)
    builder = builder.public_key(client_public)
    builder = builder.serial_number(x509.random_serial_number())
    builder = builder.not_valid_before(
        datetime.datetime.utcnow() - datetime.timedelta(hours=1)
    )
    builder = builder.not_valid_after(
        datetime.datetime.utcnow() + datetime.timedelta(days=valid_days)
    )
    
    # 扩展密钥用法: clientAuth
    builder = builder.add_extension(
        x509.ExtendedKeyUsage([ExtendedKeyUsageOID.CLIENT_AUTH]),
        critical=False,
    )
    
    builder = builder.add_extension(
        x509.BasicConstraints(ca=False, path_length=None),
        critical=True,
    )
    
    builder = builder.add_extension(
        x509.KeyUsage(
            key_cert_sign=False,
            crl_sign=False,
            digital_signature=True,
            content_commitment=False,
            key_encipherment=False,
            data_encipherment=False,
            key_agreement=False,
            encipher_only=False,
            decipher_only=False,
        ),
        critical=True,
    )
    
    # === 国密改造点：SHA256 → SM3 ===
    cert = builder.sign(ca_key, hashes.SHA256())
    
    key_path = output_path / f"client-{common_name}.key"
    cert_path = output_path / f"client-{common_name}.crt"
    
    key_path.write_bytes(
        client_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption()
        )
    )
    cert_path.write_bytes(cert.public_bytes(serialization.Encoding.PEM))
    print(f"[+] Client 证书签发完成")
    print(f"    CN: {common_name}")
    print(f"    OU: {org_unit}")
    print(f"    有效期: {valid_days} 天")
    print(f"    算法: SECP256R1 + SHA256 (演示版)")
    return key_path, cert_path


if __name__ == "__main__":
    issue_client_cert(
        ca_key_path=Path("./certs/root-ca.key"),
        ca_password=b"root-ca-password",
        common_name="client-user-001",
        org_unit="security-team"
    )

二、Python 服务端实现（FastAPI + uvicorn）

2.1 证书加载与 SSL 上下文构建

PYTHON

#!/usr/bin/env python3
"""mTLS FastAPI 服务端"""
import ssl
import tempfile
from pathlib import Path
from typing import Optional

from fastapi import FastAPI, Request, HTTPException, Depends
from fastapi.responses import JSONResponse
import uvicorn


def create_ssl_context(
    server_key: str,
    server_cert: str,
    ca_cert: str,
    verify_mode: str = "required"  # "required" | "optional"
) -> ssl.SSLContext:
    """
    构建 mTLS 所需的 SSL 上下文
    
    Args:
        server_key: 服务端密钥路径
        server_cert: 服务端证书路径
        ca_cert: 可信 CA 证书路径
        verify_mode: "required" 强制客户端证书，"optional" 可选
    """
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    
    # 加载证书链
    ctx.load_cert_chain(
        certfile=server_cert,
        keyfile=server_key,
    )
    
    # 设置最低 TLS 版本 1.2
    ctx.minimum_version = ssl.TLSVersion.TLSv1_2
    
    # 加载 CA 证书（用于验证客户端证书）
    ctx.load_verify_locations(cafile=ca_cert)
    
    if verify_mode == "required":
        ctx.verify_mode = ssl.CERT_REQUIRED
    elif verify_mode == "optional":
        ctx.verify_mode = ssl.CERT_OPTIONAL
    else:
        ctx.verify_mode = ssl.CERT_NONE
    
    # 国密改造点 4/6：
    # 标准 Python ssl 模块不原生支持国密密码套件（如 TLS_SM4_GCM_SM3）
    # 需 Tongsuo/国密 OpenSSL 做 SSL offload
    # ctx.set_ciphers("TLS_SM4_GCM_SM3:TLS_SM4_CCM_SM3:ECDHE-SM2-WITH-SM4-SM3")
    
    return ctx


app = FastAPI(title="mTLS Demo Server")


def verify_client_cert(request: Request) -> dict:
    """从请求中提取并验证客户端证书"""
    cert = request.scope.get("client_cert")
    if cert is None:
        raise HTTPException(status_code=401, detail="缺少客户端证书")
    
    # 解析证书信息
    peer_cert = request.scope.get("transport", {}).get("getpeercert", lambda: {})()
    subject = peer_cert.get("subject", ())
    
    cn = ""
    for rdn in subject:
        for attr in rdn:
            if attr[0] == "commonName":
                cn = attr[2]
                break
    
    issuer = peer_cert.get("issuer", ())
    not_after = peer_cert.get("notAfter", "unknown")
    serial = peer_cert.get("serialNumber", "unknown")
    
    return {
        "cn": cn,
        "issuer": issuer,
        "not_after": not_after,
        "serial": serial,
    }


@app.get("/")
async def root():
    return {"message": "mTLS 服务端运行中", "protocol": "TLS 1.2+"}


@app.get("/api/users/me")
async def get_current_user(cert: dict = Depends(verify_client_cert)):
    """受 mTLS 保护的 API 端点"""
    return {
        "status": "authenticated",
        "cn": cert["cn"],
        "serial": cert["serial"],
        "expires": cert["not_after"],
        "note": "客户端证书已验证"
    }


@app.get("/api/secure/data")
async def secure_data(cert: dict = Depends(verify_client_cert)):
    """模拟敏感数据端点"""
    return {
        "status": "ok",
        "data": "敏感业务数据（已使用 mTLS 传输通道保护）",
        "authenticated_user": cert["cn"],
    }


def run_server(
    host: str = "0.0.0.0",
    port: int = 8443,
    cert_dir: str = "./certs"
):
    ssl_ctx = create_ssl_context(
        server_key=f"{cert_dir}/server.key",
        server_cert=f"{cert_dir}/server-chain.crt",
        ca_cert=f"{cert_dir}/root-ca.crt",
        verify_mode="required",
    )
    
    config = uvicorn.Config(
        app,
        host=host,
        port=port,
        ssl_keyfile=f"{cert_dir}/server.key",
        ssl_certfile=f"{cert_dir}/server-chain.crt",
        ssl_ca_certs=f"{cert_dir}/root-ca.crt",
        ssl_verify_mode=ssl.CERT_REQUIRED,
        log_level="info",
    )
    server = uvicorn.Server(config)
    server.run()


if __name__ == "__main__":
    run_server()

三、Python 客户端实现

3.1 带证书的 HTTPS 客户端

PYTHON

#!/usr/bin/env python3
"""mTLS Python 客户端"""
import ssl
import httpx
from pathlib import Path


def create_mtls_client(
    client_key: str,
    client_cert: str,
    ca_cert: str,
    verify: bool = True,
) -> httpx.Client:
    """
    创建 mTLS httpx 客户端
    
    Args:
        client_key: 客户端密钥路径
        client_cert: 客户端证书路径
        ca_cert: 可信 CA 证书路径（验证服务端）
        verify: 是否验证服务端证书
    """
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    ctx.minimum_version = ssl.TLSVersion.TLSv1_2
    
    # 加载客户端证书（用于 mTLS 认证）
    ctx.load_cert_chain(
        certfile=client_cert,
        keyfile=client_key,
    )
    
    # 验证服务端证书
    ctx.load_verify_locations(cafile=ca_cert)
    ctx.check_hostname = True
    
    client = httpx.Client(
        verify=ctx,
        cert=(client_cert, client_key),
    )
    return client


def call_secure_api(base_url: str, client: httpx.Client):
    """调用受保护的 API"""
    # 测试公共端点
    r = client.get(f"{base_url}/")
    print(f"[GET /] {r.status_code}: {r.json()}")
    
    # 测试受保护端点
    r = client.get(f"{base_url}/api/users/me")
    print(f"[GET /api/users/me] {r.status_code}: {r.json()}")
    
    # 测试数据端点
    r = client.get(f"{base_url}/api/secure/data")
    print(f"[GET /api/secure/data] {r.status_code}: {r.json()}")


if __name__ == "__main__":
    base_url = "https://localhost:8443"
    cert_dir = "./certs"
    
    client = create_mtls_client(
        client_key=f"{cert_dir}/client-client-user-001.key",
        client_cert=f"{cert_dir}/client-client-user-001.crt",
        ca_cert=f"{cert_dir}/root-ca.crt",
    )
    
    try:
        call_secure_api(base_url, client)
    except Exception as e:
        print(f"[!] 错误: {e}")
    finally:
        client.close()

四、国密 TLS 接入层配置（Nginx）

标准 Python ssl 模块不支持国密密码套件。企业实践通常由 Nginx 做国密 TLS offload：

⚠️ 环境要求：以下配置中的 TLS_SM4_GCM_SM3、ECDHE-SM2-WITH-SM4-SM3 等密码套件需要 Tongsuo（铜豌豆 OpenSSL） 或 BabaSSL 支持的国密 OpenSSL 版本。标准 OpenSSL 不支持国密密码套件。Python 后端通过 HTTP 接收 Nginx 转发的请求，Nginx 注入客户端证书信息到 HTTP Header。

NGINX

# nginx.conf - 国密 mTLS 接入层
# ⚠️ 需编译 Tongsuo / GmSSL / BabaSSL 国密模块
server {
    listen 8443 ssl;
    
    # 服务端证书（国密 SM2）
    ssl_certificate         /etc/nginx/certs/server-chain.crt;
    ssl_certificate_key     /etc/nginx/certs/server.key;
    
    # 国密密码套件（需 Tongsuo/BabaSSL）
    ssl_ciphers TLS_SM4_GCM_SM3:TLS_SM4_CCM_SM3:ECDHE-SM2-WITH-SM4-SM3;
    ssl_protocols TLSv1.2 TLSv1.3;
    
    # ---- mTLS 配置 ----
    ssl_client_certificate  /etc/nginx/certs/root-ca.crt;
    ssl_verify_client       on;
    ssl_verify_depth        2;
    
    # 可选：OCSP Stapling（国密环境支持有限）
    ssl_stapling on;
    ssl_stapling_verify on;
    
    location / {
        # 转发到后端 Python 应用（HTTP）
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-SSL-Client-CN $ssl_client_s_dn_legacy;
        proxy_set_header X-SSL-Client-Serial $ssl_client_serial;
        proxy_set_header X-SSL-Client-Verify $ssl_client_verify;
    }
}

五、手动证书链验证

PYTHON

#!/usr/bin/env python3
"""手动验证证书链"""
from cryptography import x509
from cryptography.x509.oid import ExtensionOID, AuthorityInformationAccessOID
from cryptography.hazmat.primitives.asymmetric import ec, utils
from cryptography.hazmat.primitives import hashes
from cryptography.exceptions import InvalidSignature
import datetime
from typing import List


class CertificateChainVerifier:
    """证书链验证器（SECP256R1 + SHA256 演示版）
    
    ⚠️ 国密改造点 5/6：
      验证国密 SM2 证书时需要：
      1. 使用 SM2 公钥验证签名（而非 ECDSA）
      2. 使用 SM3 哈希代替 SHA256
      3. 检查公钥算法 OID 为 1.2.156.10197.1.301
    
    生产环境建议使用 Tongsuo 的 verify 命令或国密专用验证库。
    """
    
    def __init__(self, root_ca_cert: x509.Certificate):
        self.root_ca = root_ca_cert
        self.trusted = {root_ca_cert.subject: root_ca_cert}
    
    def add_intermediate(self, cert: x509.Certificate):
        """添加中间 CA"""
        self.trusted[cert.subject] = cert
    
    def _verify_signature(self, cert: x509.Certificate,
                         issuer_cert: x509.Certificate) -> bool:
        """验证证书签名（SECP256R1 + SHA256 演示版）"""
        try:
            issuer_public_key = issuer_cert.public_key()
            if isinstance(issuer_public_key, ec.EllipticCurvePublicKey):
                issuer_public_key.verify(
                    cert.signature,
                    cert.tbs_certificate_bytes,
                    ec.ECDSA(cert.signature_hash_algorithm),
                )
            else:
                issuer_public_key.verify(
                    cert.signature,
                    cert.tbs_certificate_bytes,
                    cert.signature_hash_algorithm,
                )
            return True
        except InvalidSignature:
            return False
    
    def _check_time(self, cert: x509.Certificate) -> bool:
        """检查证书有效期"""
        now = datetime.datetime.utcnow()
        return cert.not_valid_before <= now <= cert.not_valid_after
    
    def _check_constraints(self, cert: x509.Certificate, is_ca: bool) -> bool:
        """检查基本约束"""
        try:
            bc = cert.extensions.get_extension_for_oid(
                ExtensionOID.BASIC_CONSTRAINTS
            ).value
            if is_ca and not bc.ca:
                return False
            return True
        except x509.ExtensionNotFound:
            return False
    
    def _check_key_usage(self, cert: x509.Certificate) -> bool:
        """检查用途"""
        try:
            ku = cert.extensions.get_extension_for_oid(
                ExtensionOID.KEY_USAGE
            ).value
            return ku.digital_signature or ku.key_cert_sign
        except x509.ExtensionNotFound:
            return False
    
    def verify_chain(self, cert_chain: List[x509.Certificate]) -> dict:
        """验证证书链"""
        if not cert_chain:
            return {"valid": False, "reason": "空证书链", "depth": 0}
        
        leaf = cert_chain[0]
        
        if not self._check_time(leaf):
            return {"valid": False, "reason": "叶子证书已过期或未生效", "depth": 0}
        
        if not self._check_constraints(leaf, is_ca=False):
            return {"valid": False, "reason": "叶子证书 BasicConstraints 标记为 CA", "depth": 0}
        
        current = leaf
        depth = 0
        
        for issuer in cert_chain[1:]:
            if not self._check_time(issuer):
                return {"valid": False, "reason": f"CA 证书不在有效期内 (深度 {depth})", "depth": depth}
            
            if not self._verify_signature(current, issuer):
                return {"valid": False, "reason": f"签名验证失败 (深度 {depth})", "depth": depth}
            
            current = issuer
            depth += 1
        
        if current.subject not in [s for s in self.trusted]:
            return {"valid": False, "reason": "证书链未到达可信根", "depth": depth}
        
        return {"valid": True, "reason": "验证通过", "depth": depth}


# ===== 使用示例 =====
if __name__ == "__main__":
    root_ca = x509.load_pem_x509_certificate(
        Path("./certs/root-ca.crt").read_bytes()
    )
    server_cert = x509.load_pem_x509_certificate(
        Path("./certs/server.crt").read_bytes()
    )
    
    verifier = CertificateChainVerifier(root_ca)
    result = verifier.verify_chain([server_cert, root_ca])
    
    print(f"证书链验证结果: {result}")

六、吊销检查（CRL）

PYTHON

#!/usr/bin/env python3
"""证书吊销检查：CRL 验证"""
from cryptography import x509
from cryptography.x509.oid import ExtensionOID
import datetime


def check_revocation_crl(
    cert: x509.Certificate,
    issuer_ca: x509.Certificate,
    crl: x509.CertificateRevocationList
) -> dict:
    """通过 CRL 检查证书是否被吊销"""
    # 1. 检查 CRL 签发者是否匹配
    if crl.issuer != issuer_ca.subject:
        return {"ok": False, "status": "CRL签发者与CA不匹配", "revoked_date": None}
    
    # 2. 检查 CRL 有效期
    now = datetime.datetime.utcnow()
    if not (crl.last_update <= now <= crl.next_update):
        return {"ok": False, "status": "CRL不在有效期内", "revoked_date": None}
    
    # 3. 检查是否被吊销
    revoked = crl.get_revoked_certificate_by_serial_number(cert.serial_number)
    if revoked:
        return {
            "ok": False,
            "status": "证书已被吊销",
            "revoked_date": revoked.revocation_date
        }
    
    return {"ok": True, "status": "证书有效（未吊销）", "revoked_date": None}


def extract_crl_distribution_point(cert: x509.Certificate) -> str:
    """提取 CRL 分发点 URL"""
    try:
        crl_dp = cert.extensions.get_extension_for_oid(
            ExtensionOID.CRL_DISTRIBUTION_POINTS
        ).value
        for dp in crl_dp:
            if dp.full_name:
                for name in dp.full_name:
                    if isinstance(name, x509.UniformResourceIdentifier):
                        return name.value
    except x509.ExtensionNotFound:
        return ""
    return ""

七、6 个真实踩坑记录

⛔ 坑 1：密钥格式兼容

现象：不同密码库生成的密钥互相导入失败。

原因：gmssl 的公钥是 04 + x(64hex) + y(64hex) 格式（130 字符 hex），而 cryptography 需要 SubjectPublicKeyInfo (SPKI) 编码格式。

解决：转换时注意编码格式差异。

⚠️ 国密合规说明：SM2 公钥有独立 OID（1.2.156.10197.1.301），完整的 SPKI 编码需使用 RFC 5480 的国密曲线标识。以下示例用 SECP256R1 仅演示格式转换逻辑，生产环境请使用 Tongsuo 的 openssl 命令生成符合 GM/T 0015-2023 的 CSR。

PYTHON

# 概念代码：仅展示格式转换逻辑，未使用 SM2 OID
def convert_pub_to_spki_demo(gmssl_pub_hex: str) -> bytes:
    from cryptography.hazmat.primitives.asymmetric import ec
    from cryptography.hazmat.backends import default_backend
    
    pub_hex = gmssl_pub_hex[2:] if gmssl_pub_hex.startswith("04") else gmssl_pub_hex
    x = int(pub_hex[:64], 16)
    y = int(pub_hex[64:], 16)
    
    public_numbers = ec.EllipticCurvePublicNumbers(
        x, y, ec.SECP256R1()  # 国密场景需使用 SM2 曲线参数
    )
    public_key = public_numbers.public_key(default_backend())
    return public_key.public_bytes(
        encoding=serialization.Encoding.DER,
        format=serialization.PublicFormat.SubjectPublicKeyInfo,
    )

⛔ 坑 2：证书链顺序错误

现象：客户端报 unable to get local issuer certificate。

原因：证书链文件中顺序错误。

解决：必须按「叶子 → 中间 CA → 根 CA」顺序拼接：

BASH

cat server.crt intermediate-ca.crt root-ca.crt > server-chain.crt
openssl verify -CAfile root-ca.crt -untrusted intermediate-ca.crt server.crt

⛔ 坑 3：客户端证书缺少 clientAuth EKU

现象：服务端报 TLS alert。

原因：客户端证书的 ExtendedKeyUsage 不包含 clientAuth。

解决：签发客户端证书时必须显式设置。

⛔ 坑 4：Python ssl 模块对 SM2 证书不原生兼容

现象：Python 3.12+ 的 ssl 模块无法正确验证链中含国密算法证书的客户端。

原因：cryptography 库对 SM2 签名算法 OID 的处理存在边界情况。

解决：将证书验证逻辑移到应用层手动执行，或使用 Nginx 做 TLS 终结。

⛔ 坑 5：国密 OCSP 响应器对 SM2 证书支持有限

现象：OCSP 查询返回 unknown 或签名验签失败。

优先方案：使用 CRL 作为吊销检查机制。

⛔ 坑 6：SAN 缺失导致主机名验证失败

现象：httpx 客户端报 hostname mismatch。

解决：签发证书时必须在 SAN 中包含所有使用的主机名和 IP。

八、完整运行流程

BASH

mkdir -p certs
python3 create_root_ca.py
python3 issue_server_cert.py
python3 issue_client_cert.py
python3 server.py &
sleep 2
python3 client.py

九、生产环境部署建议

密钥保护：CA 私钥必须离线存储（HSM 或物理隔离）
证书层级：使用根 CA → 中间 CA → 终端实体三级结构，根 CA 离线
吊销机制：必须部署 CRL 分发点并设置合理的有效期（建议 7 天以内）
监控告警：监控证书有效期，设置 30 天/14 天/7 天分级告警
国密改造：TLS 层优先使用 Nginx + Tongsuo，Python 应用走 HTTP 后端
密评准备：记录所有密码模块的型号、版本、密评证书号

总结

本文构建了一套完整的 mTLS 双向认证工程框架：

✅ 根 CA / 服务端证书 / 客户端证书三级结构
✅ Python FastAPI 服务端 mTLS 框架
✅ Python httpx 客户端证书加载
✅ 手动证书链验证器
✅ CRL 吊销检查
✅ Nginx 国密 TLS offload 配置
✅ 6 个真实踩坑及解法
✅ 6 处国密改造点明确标注

mTLS 的核心不在于手写密码算法，而在于证书体系的正确构建与 TLS 接入层的国密集成。 Python 应用负责业务逻辑，国密密码层由专业 TLS 服务器或密码硬件完成。

参考来源

GM/T 0003.1-2012《SM2 密码算法第 1 部分：总则》
GM/T 0003.2-2012《SM2 密码算法第 2 部分：数字签名算法》
GM/T 0003.3-2012《SM2 密码算法第 3 部分：密钥交换协议》
GM/T 0003.4-2012《SM2 密码算法第 4 部分：公钥加密算法》
GM/T 0003.5-2012《SM2 密码算法第 5 部分：参数定义》
GM/T 0009-2023《SM2 密码算法使用规范》
GM/T 0015-2023《SM2 密码算法加密签名消息语法规范》
RFC 5280《Internet X.509 Public Key Infrastructure Certificate and CRL Profile》
Tongsuo 项目：https://github.com/Tongsuo-Project/Tongsuo
gmssl 库：https://github.com/guanzhi/GmSSL

mTLS TLS 双向认证 FastAPI 证书链 SECP256R1 国密改造

← 返回文章列表