X.509 证书密码应用标识合规检查实战：基于 GM/T 0006-2023 与 GB/T 33560-2026 构建国密标识验证工具

from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Optional
from cryptography import x509
from cryptography.x509.oid import NameOID


@dataclass
class AlgorithmIdentifier:
    """算法标识信息"""
    oid: str
    name: str
    is_gm: bool          # 是否国密算法
    category: str        # 签名/加密/哈希/密钥交换


@dataclass
class ComplianceCheck:
    """单项合规检查"""
    item: str            # 检查项名称
    value: str           # 实际值
    expected: str        # 期望值
    passed: bool         # 是否通过
    severity: str        # CRITICAL/WARN/INFO


@dataclass
class CertificateAuditReport:
    """证书审计报告"""
    subject: str
    issuer: str
    serial_number: str
    not_before: str
    not_after: str
    signature_algo: AlgorithmIdentifier
    public_key_algo: AlgorithmIdentifier
    key_size: int
    san_list: List[str]
    compliance_checks: List[ComplianceCheck]
    overall_compliant: bool
    audit_time: str

#!/usr/bin/env python3
"""
X.509 证书密码应用标识合规检查工具
基于 GM/T 0006-2023 和 GB/T 33560-2017 构建

运行方式:
    python cert_identifier_checker.py certificate.pem
    python cert_identifier_checker.py certificate.der --format der
"""

import argparse
import json
import sys
from datetime import datetime, timezone
from typing import Dict, List, Optional, Tuple

from cryptography import x509
from cryptography.x509.oid import ExtensionOID
from cryptography.hazmat.primitives import serialization


# ============================================================
# 1. 国密算法标识字典（GM/T 0006-2023 / GB/T 33560-2017）
GM_ALGORITHM_OIDS: Dict[str, Tuple[str, str]] = {
    # OID → (算法名称, 类别)
    "1.2.156.10197.1.301": ("SM2 椭圆曲线公钥", "公钥加密/密钥交换"),
    "1.2.156.10197.1.501": ("SM2-with-SM3 数字签名", "数字签名"),
    "1.2.156.10197.1.502": ("SM4 分组密码", "对称加密"),
    "1.2.156.10197.1.503": ("SM2 密钥交换协议", "密钥交换"),
    "1.2.156.10197.1.504": ("SM3 密码杂凑算法", "哈希"),
    "1.2.156.10197.1.601": ("SM9 标识签名", "数字签名"),
    "1.2.156.10197.1.602": ("SM9 标识加密", "公钥加密"),
    "1.2.156.10197.1.401": ("ZUC 128 序列密码", "序列密码"),
    "1.2.156.10197.1.402": ("ZUC 完整性算法", "消息认证码"),
}

# 国际算法 OID 参考
INTL_ALGORITHM_OIDS: Dict[str, Tuple[str, str]] = {
    "1.2.840.113549.1.1.1": ("RSA 公钥", "公钥加密"),
    "1.2.840.113549.1.1.11": ("sha256WithRSAEncryption", "数字签名"),
    "1.2.840.113549.1.1.12": ("sha384WithRSAEncryption", "数字签名"),
    "1.2.840.113549.1.1.13": ("sha512WithRSAEncryption", "数字签名"),
    "1.2.840.10045.2.1": ("id-ecPublicKey (ECDSA)", "公钥加密"),
    "1.2.840.10045.4.3.2": ("ecdsa-with-SHA256", "数字签名"),
    "1.2.840.10045.4.3.3": ("ecdsa-with-SHA384", "数字签名"),
    "1.2.840.10045.4.3.4": ("ecdsa-with-SHA512", "数字签名"),
    "2.16.840.1.101.3.4.2.1": ("SHA-256", "哈希"),
    "2.16.840.1.101.3.4.2.2": ("SHA-384", "哈希"),
    "2.16.840.1.101.3.4.2.3": ("SHA-512", "哈希"),
    "2.16.840.1.101.3.4.1.2": ("AES-128-CBC", "对称加密"),
    "2.16.840.1.101.3.4.1.42": ("AES-256-CBC", "对称加密"),
}

# 公钥类型与 OID 映射（辅助判断）
PUBLIC_KEY_POLICY_OIDS = {
    "ecPublicKey": {
        "1.2.156.10197.1.301": ("SM2", True, "SM2 椭圆曲线公钥（国密合规）"),
    },
    "rsaEncryption": {
        "1.2.840.113549.1.1.1": ("RSA", False, "RSA 公钥（非国密）"),
    },
}


# ============================================================
# 2. 证书加载与解析
# ============================================================

def load_certificate(cert_path: str, cert_format: str = "pem") -> x509.Certificate:
    """加载 X.509 证书（PEM 或 DER）"""
    with open(cert_path, "rb") as f:
        data = f.read()

    if cert_format == "pem":
        return x509.load_pem_x509_certificate(data)
    elif cert_format == "der":
        return x509.load_der_x509_certificate(data)
    else:
        raise ValueError(f"不支持的格式: {cert_format}，仅支持 pem 或 der")


def get_algorithm_identifier(oid: str, algo_dict: Dict) -> Tuple[str, str, bool, str]:
    """
    根据 OID 返回算法标识信息
    
    返回: (名称, 类别, 是否国密, 来源)
    """
    # 优先匹配国密
    if oid in GM_ALGORITHM_OIDS:
        name, category = GM_ALGORITHM_OIDS[oid]
        return name, category, True, "GM/T 0006-2023 / GB/T 33560-2017"

    # 匹配国际算法
    if oid in INTL_ALGORITHM_OIDS:
        name, category = INTL_ALGORITHM_OIDS[oid]
        return name, category, False, "国际算法"

    # 未知 OID
    return f"未知算法 ({oid})", "未知", False, "未知来源"


def get_public_key_info(cert: x509.Certificate) -> Tuple[str, int, bool, str]:
    """
    提取公钥信息
    
    返回: (算法名称, 密钥长度/位数, 是否国密, 详细信息)
    """
    pub_key = cert.public_key()

    # 判断公钥类型
    from cryptography.hazmat.primitives.asymmetric import ec, rsa, dsa

    if isinstance(pub_key, rsa.RSAPublicKey):
        key_size = pub_key.key_size
        return f"RSA", key_size, False, f"RSA {key_size} 位公钥"

    elif isinstance(pub_key, ec.EllipticCurvePublicKey):
        curve_name = pub_key.curve.name
        key_size = pub_key.key_size
        # 判断是否 SM2 曲线（通过曲线名称或 OID）
        curve_oid = ""
        try:
            # SM2 使用 kid 2A811CC9B5A2ED 等特征识别
            # 在 cryptography 中 SM2 曲线可能作为自定义曲线加载
            if "sm2" in curve_name.lower() or "Sm2" in curve_name:
                return f"SM2", key_size, True, f"SM2 {curve_name} 公钥（国密合规）"
        except AttributeError:
            pass

        # NIST 曲线
        if curve_name in ("secp256r1", "secp384r1", "secp521r1"):
            return f"ECDSA ({curve_name})", key_size, False, f"ECDSA {curve_name} {key_size} 位公钥（非国密）"

        # 其他曲线（可能是 Edwards 或自定义）
        return f"ECC ({curve_name})", key_size, False, f"ECC {curve_name} {key_size} 位公钥"

    elif isinstance(pub_key, dsa.DSAPublicKey):
        return f"DSA", pub_key.key_size, False, f"DSA {pub_key.key_size} 位公钥（非国密）"

    else:
        return f"未知 ({type(pub_key).__name__})", 0, False, "未知公钥类型"


# ============================================================
# 3. 合规检查引擎
# ============================================================

def check_compliance(cert: x509.Certificate) -> List[dict]:
    """
    对证书执行密码应用标识合规检查
    
    检查维度：
    1. 签名算法是否为国密
    2. 公钥算法是否为国密
    3. 密钥长度是否满足要求
    4. 证书有效期合理性
    5. 是否包含必要的扩展
    """
    checks = []

    # ---- 检查 1: 签名算法 ----
    sig_algo_oid = cert.signature_algorithm_oid.dotted_string
    sig_name, sig_category, sig_is_gm, sig_source = get_algorithm_identifier(sig_algo_oid)

    checks.append({
        "item": "签名算法标识",
        "oid": sig_algo_oid,
        "value": sig_name,
        "expected": "国密算法 (SM2-with-SM3 / SM3)",
        "passed": sig_is_gm,
        "severity": "CRITICAL",
        "detail": f"签名算法 OID {sig_algo_oid} → {sig_name}（{'国密' if sig_is_gm else '国际算法'}）"
    })

    # ---- 检查 2: 公钥算法 ----
    pub_key_algo, key_size, pub_is_gm, pub_detail = get_public_key_info(cert)

    checks.append({
        "item": "公钥算法标识",
        "oid": "—",
        "value": f"{pub_key_algo} ({key_size} bits)",
        "expected": "国密公钥 (SM2 / SM9)",
        "passed": pub_is_gm,
        "severity": "CRITICAL",
        "detail": pub_detail
    })

    # ---- 检查 3: 密钥长度 ----
    min_key_size = 256 if pub_is_gm else 2048  # SM2 256 位 vs RSA 2048 位
    key_size_ok = key_size >= min_key_size

    checks.append({
        "item": "密钥长度",
        "oid": "—",
        "value": f"{key_size} bits",
        "expected": f"≥ {min_key_size} bits",
        "passed": key_size_ok,
        "severity": "WARN",
        "detail": f"{'满足' if key_size_ok else '不满足'} 国密/国际最小密钥长度要求（{min_key_size} bits）"
    })

    # ---- 检查 4: 证书有效期 ----
    now = datetime.now(timezone.utc)
    not_before = cert.not_valid_before_utc if hasattr(cert, 'not_valid_before_utc') else cert.not_valid_before
    not_after = cert.not_valid_after_utc if hasattr(cert, 'not_valid_after_utc') else cert.not_valid_after

    validity_days = (not_after - not_before).days
    validity_reasonable = 0 < validity_days < 3650  # 10 年以内

    # 国密改造场景中建议不超过 395 天（13 个月）
    gm_recommended = validity_days <= 395

    checks.append({
        "item": "证书有效期",
        "oid": "—",
        "value": f"{validity_days} 天",
        "expected": "≤ 395 天（国密改造推荐）",
        "passed": gm_recommended,
        "severity": "INFO",
        "detail": f"有效期 {validity_days} 天（{not_before.strftime('%Y-%m-%d')} 至 {not_after.strftime('%Y-%m-%d')}），"
                  f"{'符合国密改造推荐' if gm_recommended else '超过 395 天建议值'}"
    })

    # ---- 检查 5: SAN 扩展 ----
    try:
        san_ext = cert.extensions.get_extension_for_oid(ExtensionOID.SUBJECT_ALTERNATIVE_NAME)
        san_values = san_ext.value.get_values_for_type(x509.DNSName)
        san_ok = len(san_values) > 0
        san_detail = f"包含 {len(san_values)} 个 DNS 名称: {', '.join(san_values[:3])}"
    except x509.ExtensionNotFound:
        san_ok = False
        san_values = []
        san_detail = "缺少 SAN 扩展（不符合 TLS 证书最佳实践）"

    checks.append({
        "item": "SAN 扩展 (subjectAltName)",
        "oid": ExtensionOID.SUBJECT_ALTERNATIVE_NAME.dotted_string,
        "value": f"{len(san_values)} 个 DNS",
        "expected": "≥ 1 个 DNS 名称",
        "passed": san_ok,
        "severity": "WARN",
        "detail": san_detail
    })

    return checks


# ============================================================
# 4. 报告生成
# ============================================================

def generate_report(cert: x509.Certificate) -> dict:
    """生成完整的合规审计报告"""

    # 解析证书基本信息
    subject_cn = cert.subject.get_attributes_for_oid(NameOID.COMMON_NAME)
    issuer_cn = cert.issuer.get_attributes_for_oid(NameOID.COMMON_NAME)

    # 签名算法 OID
    sig_algo_oid = cert.signature_algorithm_oid.dotted_string
    sig_name, _, sig_is_gm, _ = get_algorithm_identifier(sig_algo_oid)

    # 公钥信息
    pub_key_algo, key_size, pub_is_gm, _ = get_public_key_info(cert)

    # 执行检查
    checks = check_compliance(cert)

    # 汇总
    overall_compliant = all(c["passed"] for c in checks if c["severity"] == "CRITICAL")
    critical_pass = sum(1 for c in checks if c["severity"] == "CRITICAL" and c["passed"])
    critical_total = sum(1 for c in checks if c["severity"] == "CRITICAL")
    warn_pass = sum(1 for c in checks if c["severity"] == "WARN" and c["passed"])
    warn_total = sum(1 for c in checks if c["severity"] == "WARN")

    report = {
        "certificate": {
            "subject": subject_cn[0].value if subject_cn else str(cert.subject),
            "issuer": issuer_cn[0].value if issuer_cn else str(cert.issuer),
            "serial_number": format(cert.serial_number, 'X'),
            "not_before": str(cert.not_valid_before),
            "not_after": str(cert.not_valid_after),
            "signature_algorithm": {
                "oid": sig_algo_oid,
                "name": sig_name,
                "is_gm": sig_is_gm,
            },
            "public_key": {
                "algorithm": pub_key_algo,
                "key_size": key_size,
                "is_gm": pub_is_gm,
            },
        },
        "compliance": {
            "overall": "COMPLIANT" if overall_compliant else "NON_COMPLIANT",
            "checks": checks,
            "summary": {
                "critical": f"{critical_pass}/{critical_total} 通过",
                "warning": f"{warn_pass}/{warn_total} 通过",
                "total_checks": len(checks),
            },
        },
        "audit_metadata": {
            "tool": "cert-identifier-checker",
            "version": "1.0.0",
            "standard_reference": "GM/T 0006-2023, GB/T 33560-2017",
            "audit_time": datetime.now(timezone.utc).isoformat(),
        },
    }

    return report


def print_report(report: dict):
    """控制台彩色输出报告"""
    cert = report["certificate"]
    comp = report["compliance"]

    print("\n" + "=" * 70)
    print("  X.509 证书密码应用标识合规审计报告")
    print("  依据: GM/T 0006-2023 / GB/T 33560-2017")
    print("=" * 70)

    # 证书信息
    print(f"\n📋 证书基本信息:")
    print(f"  主体 (Subject)     : {cert['subject']}")
    print(f"  颁发者 (Issuer)    : {cert['issuer']}")
    print(f"  序列号             : {cert['serial_number']}")
    print(f"  有效期             : {cert['not_before'][:10]} → {cert['not_after'][:10]}")
    print(f"  签名算法           : {cert['signature_algorithm']['name']}")
    print(f"  └─ OID            : {cert['signature_algorithm']['oid']}")
    print(f"  └─ 是否国密        : {'✅ 是' if cert['signature_algorithm']['is_gm'] else '❌ 否'}")
    print(f"  公钥算法           : {cert['public_key']['algorithm']} ({cert['public_key']['key_size']} bits)")
    print(f"  └─ 是否国密        : {'✅ 是' if cert['public_key']['is_gm'] else '❌ 否'}")

    # 合规结论
    overall = comp["overall"]
    status_icon = "✅ 合规" if overall == "COMPLIANT" else "❌ 不合规"
    print(f"\n🔍 合规结论: {status_icon}")
    print(f"  关键项 (CRITICAL)  : {comp['summary']['critical']}")
    print(f"  警告项 (WARNING)   : {comp['summary']['warning']}")

    # 检查详情
    print(f"\n📊 检查项明细:")
    for i, check in enumerate(comp["checks"], 1):
        icon = "✅" if check["passed"] else "⚠️"
        severity = f"[{check['severity']}]"
        print(f"  {icon} [{i}] {severity} {check['item']}")
        print(f"       实际值: {check['value']}")
        print(f"       期望值: {check['expected']}")
        if not check["passed"]:
            print(f"       建议  : {check['detail']}")

    print("\n" + "=" * 70)
    print(f"  审计时间: {report['audit_metadata']['audit_time']}")
    print("=" * 70)


# ============================================================
# 5. 主程序
# ============================================================

def main():
    parser = argparse.ArgumentParser(
        description="X.509 证书密码应用标识合规检查工具",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
使用示例:
  %(prog)s cert.pem                            # PEM 格式
  %(prog)s cert.der --format der               # DER 格式
  %(prog)s cert.pem --json report.json         # 输出 JSON 报告
        """
    )
    parser.add_argument("cert_file", help="证书文件路径")
    parser.add_argument("--format", choices=["pem", "der"], default="pem", help="证书格式 (默认: pem)")
    parser.add_argument("--json", help="输出 JSON 报告到文件")

    args = parser.parse_args()

    # 加载证书
    try:
        cert = load_certificate(args.cert_file, args.format)
    except Exception as e:
        print(f"❌ 证书加载失败: {e}", file=sys.stderr)
        sys.exit(1)

    # 生成报告
    report = generate_report(cert)

    # 控制台输出
    print_report(report)

    # JSON 输出
    if args.json:
        with open(args.json, "w", encoding="utf-8") as f:
            json.dump(report, f, ensure_ascii=False, indent=2)
        print(f"\n📄 JSON 报告已保存到: {args.json}")

    # 退出码：不合规返回 1（便于 CI/CD 集成）
    if report["compliance"]["overall"] != "COMPLIANT":
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""生成测试用自签 RSA 证书"""
from datetime import datetime, timedelta
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.backends import default_backend

private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048, backend=default_backend())

subject = issuer = x509.Name([
    x509.NameAttribute(NameOID.COMMON_NAME, "test-international.example.com"),
    x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Test International"),
])

cert = (
    x509.CertificateBuilder()
    .subject_name(subject)
    .issuer_name(issuer)
    .public_key(private_key.public_key())
    .serial_number(x509.random_serial_number())
    .not_valid_before(datetime.utcnow())
    .not_valid_after(datetime.utcnow() + timedelta(days=365))
    .add_extension(
        x509.SubjectAlternativeName([x509.DNSName("test-international.example.com")]),
        critical=False,
    )
    .sign(private_key, hashes.SHA256(), default_backend())
)

with open("/tmp/test_international.pem", "wb") as f:
    f.write(cert.public_bytes(serialization.Encoding.PEM))

print("已生成: /tmp/test_international.pem (RSA + SHA256)")

{
  "certificate": {
    "subject": "test-international.example.com",
    "issuer": "test-international.example.com",
    "serial_number": "A1B2C3D4...",
    "not_before": "2024-01-15",
    "not_after": "2025-01-15",
    "signature_algorithm": {
      "oid": "1.2.840.113549.1.1.11",
      "name": "sha256WithRSAEncryption",
      "is_gm": false
    },
    "public_key": {
      "algorithm": "RSA",
      "key_size": 2048,
      "is_gm": false
    }
  },
  "compliance": {
    "overall": "NON_COMPLIANT",
    "checks": [
      {
        "item": "签名算法标识",
        "oid": "1.2.840.113549.1.1.11",
        "value": "sha256WithRSAEncryption",
        "expected": "国密算法 (SM2-with-SM3 / SM3)",
        "passed": false,
        "severity": "CRITICAL",
        "detail": "..."
      }
    ],
    "summary": {
      "critical": "0/2 通过",
      "warning": "2/2 通过",
      "total_checks": 5
    }
  },
  "audit_metadata": {
    "tool": "cert-identifier-checker",
    "version": "1.0.0",
    "standard_reference": "GM/T 0006-2023, GB/T 33560-2017",
    "audit_time": "2026-07-23T10:30:00+00:00"
  }
}

import glob

def scan_pki_inventory(cert_dir: str) -> dict:
    """批量扫描目录中的证书并汇总合规率"""
    results = {"total": 0, "gm_compliant": 0, "non_compliant": 0}
    for cert_file in glob.glob(f"{cert_dir}/**/*.pem", recursive=True):
        cert = x509.load_pem_x509_certificate(open(cert_file, "rb").read())
        report = generate_report(cert)
        results["total"] += 1
        if report["compliance"]["overall"] == "COMPLIANT":
            results["gm_compliant"] += 1
        else:
            results["non_compliant"] += 1
    return results