国密SM2证书链验证实战：从gmssl到生产环境的完整方案

# 第1步：验证中间证书由根CA签发
gmssl verify -CAfile root_ca.crt intermediate.crt

# 第2步：验证叶子证书由中间CA签发
gmssl verify -CAfile root_ca.crt -untrusted intermediate.crt leaf.crt

# 第3步：检查叶子证书有效期
gmssl x509 -in leaf.crt -checkend 0 -noout
# 返回 0 表示未过期，返回 1 表示已过期

# 第4步：检查密钥用法
gmssl x509 -in leaf.crt -text -noout | grep -A2 "Key Usage"
gmssl x509 -in leaf.crt -text -noout | grep -A3 "Extended Key Usage"

from gmssl import sm2, func
from gmssl.sm2 import CryptSM2
from cryptography import x509
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.x509.oid import ExtensionOID
import datetime

def verify_sm2_chain(
    leaf_cert_path: str,
    intermediate_certs: list[str],
    root_ca_path: str,
    expected_dns_names: list[str] | None = None,
    purpose: str = "sslserver"
) -> dict:
    """
    验证 SM2 证书链的完整函数。
    
    Args:
        leaf_cert_path: 叶子证书路径（PEM格式）
        intermediate_certs: 中间证书路径列表（按顺序：离叶子最近的在前）
        root_ca_path: 根CA证书路径
        expected_dns_names: 期望匹配的域名列表（None 则跳过域名检查）
        purpose: 验证目的，"sslserver" 或 "sslclient"
    
    Returns:
        dict: {
            "valid": bool,
            "chain_valid": bool,
            "expiry_valid": bool,
            "eku_valid": bool,
            "name_match": bool,
            "errors": list[str]
        }
    """
    errors = []
    
    # === 1. 加载证书 ===
    with open(leaf_cert_path, "rb") as f:
        leaf_cert = x509.load_pem_x509_certificate(f.read())
    
    with open(root_ca_path, "rb") as f:
        root_cert = x509.load_pem_x509_certificate(f.read())
    
    inter_certs = []
    for cert_path in intermediate_certs:
        with open(cert_path, "rb") as f:
            inter_certs.append(x509.load_pem_x509_certificate(f.read()))
    
    # === 2. 信任链完整性验证 ===
    chain_valid = True
    current_cert = leaf_cert
    
    # 验证叶子 → 中间证书链
    for inter_cert in inter_certs:
        if not _verify_signature(current_cert, inter_cert):
            errors.append(f"叶子证书签名验证失败：Issuer不匹配")
            chain_valid = False
            break
        current_cert = inter_cert
    
    # 验证最后一级中间证书 → 根CA
    if chain_valid and inter_certs:
        if not _verify_signature(current_cert, root_cert):
            errors.append("中间证书链验证失败：最后一级中间证书签名无效")
            chain_valid = False
    
    # === 3. 有效期验证 ===
    now = datetime.datetime.now(datetime.timezone.utc)
    expiry_valid = True
    
    for cert in [leaf_cert] + inter_certs + [root_cert]:
        if cert.not_valid_before_utc > now:
            errors.append(f"证书已未生效：{cert.subject.rfc4514_string()}")
            expiry_valid = False
        if cert.not_valid_after_utc < now:
            errors.append(f"证书已过期：{cert.subject.rfc4514_string()}")
            expiry_valid = False
    
    # === 4. 密钥用途（EKU）验证 ===
    eku_valid = True
    try:
        eku_ext = leaf_cert.extensions.get_extension_for_oid(
            ExtensionOID.EXTENDED_KEY_USAGE
        )
        usage_values = eku_ext.value
        
        if purpose == "sslserver":
            required = x509.oid.ExtendedKeyUsageOID.SERVER_AUTH
        else:
            required = x509.oid.ExtendedKeyUsageOID.CLIENT_AUTH
        
        if required not in usage_values:
            errors.append(
                f"证书EKU不包含{'serverAuth' if purpose=='sslserver' else 'clientAuth'}"
            )
            eku_valid = False
    except x509.ExtensionNotFound:
        # 无EKU扩展时，部分实现允许所有用途（宽松模式）
        # 生产环境建议：严格模式下应报错
        pass
    
    # === 5. 域名匹配验证 ===
    name_match = True
    if expected_dns_names:
        try:
            san_ext = leaf_cert.extensions.get_extension_for_oid(
                ExtensionOID.SUBJECT_ALTERNATIVE_NAME
            )
            dns_names = san_ext.value.get_values_for_type(x509.DNSName)
            if not any(name in dns_names for name in expected_dns_names):
                errors.append(f"域名不匹配：期望{expected_dns_names}，实际{dns_names}")
                name_match = False
        except x509.ExtensionNotFound:
            # 回退到CN检查（不推荐，CN在RFC 6125中已废弃）
            cn = leaf_cert.subject.get_attributes_for_oid(NameOID.COMMON_NAME)
            if cn and cn[0].value not in expected_dns_names:
                errors.append(f"域名不匹配：CN={cn[0].value}")
                name_match = False
    
    return {
        "valid": chain_valid and expiry_valid and eku_valid and name_match,
        "chain_valid": chain_valid,
        "expiry_valid": expiry_valid,
        "eku_valid": eku_valid,
        "name_match": name_match,
        "errors": errors,
    }


def _verify_signature(cert: x509.Certificate, issuer: x509.Certificate) -> bool:
    """验证 cert 是否由 issuer 签发（通用X.509实现，需适配国密）"""
    try:
        issuer_pubkey = issuer.public_key()
        # 注意：标准 cryptography 库不支持 SM2 曲线
        # 如需验证 SM2 证书，请使用 gmssl 命令行：
        #   gmssl verify -CAfile issuer.pem -untrusted chain.pem leaf.pem
        # 或 Tongsuo 库（国密 OpenSSL 分支）
        if hasattr(issuer_pubkey, 'algorithm_identifier'):
            oid = issuer_pubkey.algorithm_identifier
            # 此处仅为演示框架，实际 SM2 验签需使用专用库
            return False
        return False
    except Exception:
        return False

package main

import (
	"fmt"
	"os"

	"github.com/tongsuo-project/tongsuo-go"
)

func verifySM2Chain(leafPEM, intermediatePEM, rootPEM []byte) error {
	// 使用 Tongsuo 提供的 SM2 支持
	// Tongsuo Go 绑定的 API 与标准库不同，请参考官方文档
	_ = leafPEM
	_ = intermediatePEM
	_ = rootPEM

	return nil
}

import "os/exec"

func verifyWithGMSSL(leaf, intermediate, root string) (bool, error) {
    cmd := exec.Command(
        "gmssl", "verify",
        "-CAfile", root,
        "-untrusted", intermediate,
        leaf,
    )
    output, err := cmd.CombinedOutput()
    if err != nil {
        return false, fmt.Errorf("验证失败: %s\n%s", err, output)
    }
    return true, nil
}

#!/usr/bin/env python3
"""
生产环境证书链验证脚本
用法: python3 verify_sm2_chain.py leaf.crt intermediate.crt root.crt example.com
"""

import sys
import subprocess
import argparse
import re


def run_gmssl(args: list[str]) -> tuple[int, str, str]:
    """运行 gmssl 命令，返回 (returncode, stdout, stderr)"""
    result = subprocess.run(
        ["gmssl"] + args,
        capture_output=True,
        text=True,
        timeout=30,
    )
    return result.returncode, result.stdout, result.stderr


def verify_chain(leaf: str, intermediate: str, root: str) -> dict:
    """执行完整的证书链验证"""
    checks = {}
    
    # 1. 基础链验证
    rc, stdout, stderr = run_gmssl([
        "verify",
        "-CAfile", root,
        "-untrusted", intermediate,
        leaf,
    ])
    checks["chain"] = rc == 0 and ": OK" in stdout
    
    # 2. 有效期检查
    rc, _, _ = run_gmssl(["x509", "-in", leaf, "-checkend", "0", "-noout"])
    checks["expiry"] = rc == 0
    
    # 3. 用途验证（sslserver）
    rc, stdout, stderr = run_gmssl([
        "verify",
        "-CAfile", root,
        "-untrusted", intermediate,
        "-purpose", "sslserver",
        leaf,
    ])
    checks["purpose"] = rc == 0 and "verify ok" in stdout.lower()
    
    # 4. 主题提取（用于后续域名匹配）
    rc, stdout, _ = run_gmssl(["x509", "-in", leaf, "-text", "-noout"])
    san_match = re.search(r"DNS:[\w.\-]+", stdout)
    checks["san"] = san_match is not None
    checks["san_value"] = san_match.group(0).replace("DNS:", "") if san_match else None
    
    return checks


def main():
    parser = argparse.ArgumentParser(description="验证 SM2 证书链")
    parser.add_argument("leaf", help="叶子证书路径")
    parser.add_argument("intermediate", help="中间证书路径")
    parser.add_argument("root", help="根CA证书路径")
    parser.add_argument("--hostname", help="期望匹配的域名")
    args = parser.parse_args()
    
    results = verify_chain(args.leaf, args.intermediate, args.root)
    
    all_passed = all(results.values())
    
    print("=" * 50)
    print("SM2 证书链验证结果")
    print("=" * 50)
    for check, passed in results.items():
        status = "✓" if passed else "✗"
        print(f"  {status} {check}: {'通过' if passed else '失败'}")
    
    if args.hostname and results.get("san_value"):
        domain_ok = args.hostname in results["san_value"] or results["san_value"].endswith(f".{args.hostname}")
        status = "✓" if domain_ok else "✗"
        print(f"  {status} 域名匹配: {results['san_value']} vs {args.hostname}")
        all_passed = all_passed and domain_ok
    
    print("=" * 50)
    print(f"总体结果: {'验证通过' if all_passed else '验证失败'}")
    sys.exit(0 if all_passed else 1)


if __name__ == "__main__":
    main()