后量子密码迁移第一步：企业密码资产清单（CBOM）构建实战

密码学 · 2026-06-15 · 9 阅读

前言

2026 年 4 月，Meta 安全工程团队发布了业界首个公开的后量子密码（PQC）迁移手册，其中将密码资产清单（Cryptographic Bill of Materials, CBOM）定义为迁移工程的第一步，也是最重要的一步。

这不是孤立的行业实践。美国国家标准与技术研究院（NIST）在 IR 8547《过渡到后量子密码标准》中明确将"密码资产发现与清单化"列为迁移六阶段的第一阶段。NSA 的 CNSA 2.0 政策要求 2027 年 1 月 1 日起新采购的国家安全系统必须合规，而 CBOM 是证明合规的基础证据。

然而，大多数企业面临一个共同的困境：你不知道自己有什么，就不知道怎么保护它。 密码算法和密钥分散在应用程序代码、TLS 证书、VPN 配置、数据库连接字符串、IoT 固件、CI/CD 流水线中——没有完整的清单，PQC 迁移无从谈起。

本文给出 CBOM 的完整构建方法论，并提供一个可直接运行的 Python 扫描工具。

一、为什么 CBOM 是 PQC 迁移的前提

1.1 密码资产的"冰山效应"

PYTHON

# 典型企业的密码资产分布（基于 NIST NCCoE 案例）
crypto_assets = {
    "可见部分（~20%）": {
        "TLS 证书": "公网域名证书、负载均衡器证书",
        "VPN 配置": "IPSec/IKE 预共享密钥、证书",
        "SSH 密钥": "authorized_keys、known_hosts",
    },
    "隐藏部分（~80%": {
        "应用程序代码": "硬编码的 RSA/ECC 密钥对、JWT 签名密钥",
        "数据库连接": "TLS 客户端证书、连接字符串中的密码",
        "CI/CD 流水线": "代码签名密钥、容器镜像签名",
        "第三方 SDK": "嵌入式 TLS 库、证书钉扎（Pinning）",
        "IoT/边缘设备": "固件中的硬编码证书、预置密钥",
        "备份与归档": "加密备份数据的密钥管理",
    },
}

NIST 的实证研究表明，大型企业在进行系统化的密码资产发现后，平均发现密码资产数量是预期的 3-5 倍。一家拥有 200 个应用系统的金融机构，最终识别出超过 1,200 个独立的密码学密钥和证书。

1.2 CBOM 驱动迁移决策

没有 CBOM，企业的典型决策路径是：

CODE

听说 PQC 很紧急 → 升级 OpenSSL → 替换 TLS 证书 → 完成？

有了 CBOM，决策路径变为：

CODE

完整资产清单 → 按敏感度和暴露面评分 → 按优先级分批迁移 → 可验证的进度

Meta 的迁移手册用了一个精炼的表述："你无法保护你不知道自己拥有的东西"（You can't protect what you don't know you have）。

1.3 监管要求的硬性时间表

监管框架	时间节点	CBOM 相关要求
CNSA 2.0（NSA）	2027-01-01	新采购 NSS 必须 CNSA 2.0 合规，需证明密码资产可见性
NIST IR 8547	2035 年	所有联邦系统完成 PQC 迁移，清单化是前提
NIST SP 800-227	当前	建议使用混合 KEM，需识别所有使用密钥交换的系统
Google 内部目标	2029 年	所有 Google 系统 PQC 合规，驱动供应链要求
CA/B 论坛	2026-03-15	证书有效期缩短至 200 天，自动化管理依赖准确的证书清单

二、CBOM 的核心结构

2.1 清单数据模型

Meta 的迁移手册定义了 CBOM 的最小必要字段集。以下是经过实践验证的数据模型：

PYTHON

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Optional

class AlgorithmFamily(Enum):
    RSA = "RSA"
    ECDSA = "ECDSA"
    EdDSA = "EdDSA"
    SM2 = "SM2"
    DH = "DH"
    ECDH = "ECDH"
    AES = "AES"
    SM4 = "SM4"
    SHA256 = "SHA-256"
    SHA3 = "SHA-3"
    SM3 = "SM3"
    OTHER = "Other"

class AssetType(Enum):
    TLS_CERT = "TLS 证书"
    SSH_KEY = "SSH 密钥"
    CODE_SIGNING = "代码签名密钥"
    VPN_KEY = "VPN 密钥"
    DATABASE_ENC = "数据库加密密钥"
    JWT_SIGNING = "JWT 签名密钥"
    API_KEY = "API 密钥"
    IOT_FIRMWARE = "IoT 固件密钥"
    OTHER = "其他"

class ExposureLevel(Enum):
    PUBLIC = "公网暴露"
    DMZ = "DMZ 区域"
    INTERNAL = "内网"
    AIR_GAPPED = "物理隔离"

class SensitivityLevel(Enum):
    CRITICAL = "关键"
    HIGH = "高"
    MEDIUM = "中"
    LOW = "低"

@dataclass
class CryptoAsset:
    """单个密码资产的 CBOM 条目"""
    # 基础标识
    asset_id: str                          # 唯一标识符
    asset_type: AssetType
    description: str
    
    # 算法信息
    algorithm_family: AlgorithmFamily
    key_size: int                          # 位
    protocol: Optional[str] = None         # TLS 1.2/1.3, SSH, IPSec 等
    
    # 位置信息
    host: str = ""
    port: int = 0
    application: str = ""
    environment: str = "production"        # production/staging/dev
    
    # 分类信息
    exposure: ExposureLevel = ExposureLevel.INTERNAL
    sensitivity: SensitivityLevel = SensitivityLevel.MEDIUM
    data_classification: str = ""          # 保护的数据分类
    
    # 生命周期
    created_at: Optional[datetime] = None
    expires_at: Optional[datetime] = None
    rotation_policy: str = ""
    
    # 迁移状态
    migration_priority: int = 0            # 1-5, 5 最高
    migration_status: str = "pending"      # pending/in_progress/migrated/verified
    migration_target: str = ""             # 目标算法
    notes: str = ""

@dataclass
class CBOM:
    """密码资产清单"""
    org_name: str
    created_at: datetime = field(default_factory=datetime.utcnow)
    assets: list = field(default_factory=list)
    
    @property
    def total_assets(self) -> int:
        return len(self.assets)
    
    def by_algorithm(self) -> dict:
        """按算法分组统计"""
        result = {}
        for a in self.assets:
            key = a.algorithm_family.value
            result[key] = result.get(key, 0) + 1
        return dict(sorted(result.items(), key=lambda x: -x[1]))
    
    def by_exposure(self) -> dict:
        """按暴露面分组统计"""
        result = {}
        for a in self.assets:
            key = a.exposure.value
            result[key] = result.get(key, 0) + 1
        return dict(sorted(result.items(), key=lambda x: -x[1]))
    
    def priority_queue(self) -> list:
        """按迁移优先级排序"""
        return sorted(self.assets, key=lambda a: -a.migration_priority)

2.2 优先级评分模型

Meta 的迁移手册和 NIST NCCoE 都推荐基于两个维度进行优先级排序：

PYTHON

def calculate_priority(asset: CryptoAsset) -> int:
    """
    计算迁移优先级分数 (1-5)
    
    维度一：暴露度（Exposure）— 攻击者能否触及
    维度二：敏感度（Sensitivity）— 数据泄露的影响
    """
    exposure_score = {
        ExposureLevel.PUBLIC: 5,
        ExposureLevel.DMZ: 4,
        ExposureLevel.INTERNAL: 2,
        ExposureLevel.AIR_GAPPED: 1,
    }
    
    sensitivity_score = {
        SensitivityLevel.CRITICAL: 5,
        SensitivityLevel.HIGH: 4,
        SensitivityLevel.MEDIUM: 2,
        SensitivityLevel.LOW: 1,
    }
    
    # 加权计算：暴露度 60% + 敏感度 40%
    raw_score = exposure_score[asset.exposure] * 0.6 + sensitivity_score[asset.sensitivity] * 0.4
    
    if raw_score >= 4.0:
        return 5  # 最高优先级：立即处理
    elif raw_score >= 3.0:
        return 4  # 高优先级：本季度处理
    elif raw_score >= 2.0:
        return 3  # 中优先级：半年内处理
    elif raw_score >= 1.5:
        return 2  # 低优先级：年度处理
    else:
        return 1  # 最低优先级：监控即可

三、自动化发现工具链

3.1 工具链架构

CODE

┌─────────────────────────────────────────────────────┐
│                 CBOM 自动化发现平台                     │
│                                                       │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐           │
│  │ 网络扫描  │  │ 配置审计  │  │ 代码分析  │           │
│  │          │  │          │  │          │           │
│  │ OpenSSL  │  │ Ansible  │  │ semgrep  │           │
│  │ s_client │  │ 配置解析 │  │ 静态分析 │           │
│  │ Nmap     │  │ Terraform│  │ 依赖扫描 │           │
│  └─────┬────┘  └─────┬────┘  └─────┬────┘           │
│        │              │              │                │
│        └──────────────┼──────────────┘                │
│                       ▼                               │
│              ┌────────────────┐                       │
│              │  统一 CBOM 数据 │                       │
│              │  模型 + 数据库  │                       │
│              └────────┬───────┘                       │
│                       ▼                               │
│              ┌────────────────┐                       │
│              │ 优先级排序引擎  │                       │
│              │ + 可视化报告    │                       │
│              └────────────────┘                       │
└─────────────────────────────────────────────────────┘

3.2 实战：TLS 证书扫描工具

以下脚本扫描指定网段的所有 TLS 服务，提取证书信息并识别使用的算法：

PYTHON

#!/usr/bin/env python3
"""
CBOM TLS 扫描器 — 发现公网/内网 TLS 证书并识别密码算法
依赖: pip install cryptography
"""
import socket
import ssl
import json
import csv
import sys
from datetime import datetime, timezone
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa, ec, ed25519, ed448

# 国密 OID 映射
SM2_OID = "1.2.156.10197.1.301"
SM3_OID = "1.2.156.10197.1.401"
SM4_OID = "1.2.156.10197.1.104"

ALGORITHM_MAP = {
    "RSA": "RSA",
    "EC": "ECDSA/SM2",
    "Ed25519": "EdDSA",
    "Ed448": "EdDSA",
}

def get_cert_signature_algorithm(cert):
    """识别证书签名算法，包括国密算法"""
    sig_alg = cert.signature_algorithm_oid._name
    oid_str = str(cert.signature_algorithm_oid.dotted_string)
    
    if SM2_OID in oid_str or "SM2" in sig_alg:
        return "SM2-with-SM3"
    elif SM3_OID in oid_str or "SM3" in sig_alg:
        if "SM2" in sig_alg:
            return "SM2-with-SM3"
        return "SM3"
    elif "sha256" in sig_alg.lower():
        return "SHA-256"
    elif "sha384" in sig_alg.lower():
        return "SHA-384"
    elif "sha512" in sig_alg.lower():
        return "SHA-512"
    elif "sha1" in sig_alg.lower():
        return "SHA-1"  # 已弃用，标记为高风险
    return sig_alg

def get_public_key_info(cert):
    """提取公钥类型和长度"""
    pub_key = cert.public_key()
    info = {"type": "Unknown", "size": 0}
    
    if isinstance(pub_key, rsa.RSAPublicKey):
        info = {"type": "RSA", "size": pub_key.key_size}
    elif isinstance(pub_key, ec.EllipticCurvePublicKey):
        info = {"type": "EC", "size": pub_key.key_size, "curve": pub_key.curve.name}
        # 检查是否为 SM2 曲线
        if "sm2" in pub_key.curve.name.lower():
            info["type"] = "SM2"
    elif isinstance(pub_key, ed25519.Ed25519PublicKey):
        info = {"type": "Ed25519", "size": 256}
    elif isinstance(pub_key, ed448.Ed448PublicKey):
        info = {"type": "Ed448", "size": 456}
    
    return info

def scan_tls_service(host, port=443, timeout=5):
    """扫描单个主机的 TLS 证书"""
    result = {
        "host": host,
        "port": port,
        "reachable": False,
        "certificate": None,
        "tls_version": None,
        "cipher_suite": None,
        "error": None,
    }
    
    try:
        context = ssl.create_default_context()
        context.check_hostname = False
        context.verify_mode = ssl.CERT_NONE  # 扫描模式，不验证证书链
        
        with socket.create_connection((host, port), timeout=timeout) as sock:
            with context.wrap_socket(sock, server_hostname=host) as ssock:
                result["reachable"] = True
                result["tls_version"] = ssock.version()
                result["cipher_suite"] = ssock.cipher()
                
                # 获取对端证书（DER 格式）
                der_cert = ssock.getpeercert(binary_form=True)
                if der_cert:
                    cert = x509.load_der_x509_certificate(der_cert, default_backend())
                    
                    pub_key_info = get_public_key_info(cert)
                    sig_alg = get_cert_signature_algorithm(cert)
                    
                    # 计算证书剩余有效期
                    now = datetime.now(timezone.utc)
                    expires = cert.not_valid_after_utc if hasattr(cert, 'not_valid_after_utc') else cert.not_valid_after.replace(tzinfo=timezone.utc)
                    days_remaining = (expires - now).days
                    
                    result["certificate"] = {
                        "subject": str(cert.subject),
                        "issuer": str(cert.issuer),
                        "serial_number": str(cert.serial_number),
                        "not_before": str(cert.not_valid_before_utc if hasattr(cert, 'not_valid_before_utc') else cert.not_valid_before),
                        "not_after": str(expires),
                        "days_remaining": days_remaining,
                        "public_key": pub_key_info,
                        "signature_algorithm": sig_alg,
                        "is_self_signed": cert.issuer == cert.subject,
                        "san_names": [
                            name.value for name in cert.extensions.get_extension_for_oid(
                                x509.oid.ExtensionOID.SUBJECT_ALTERNATIVE_NAME
                            ).value if isinstance(name, x509.DNSName)
                        ] if cert.extensions.get_extension_for_oid(
                            x509.oid.ExtensionOID.SUBJECT_ALTERNATIVE_NAME
                        , None) else [],
                    }
    except socket.timeout:
        result["error"] = "connection_timeout"
    except ConnectionRefusedError:
        result["error"] = "connection_refused"
    except ssl.SSLError as e:
        result["error"] = f"ssl_error: {e}"
    except Exception as e:
        result["error"] = f"error: {e}"
    
    return result

def assess_pqc_risk(cert_info):
    """评估证书的 PQC 迁移风险"""
    if not cert_info:
        return {"risk_level": "unknown", "pqc_ready": False, "issues": ["no_certificate"]}
    
    issues = []
    risk_score = 0
    
    # 检查签名算法
    sig_alg = cert_info.get("signature_algorithm", "")
    if "SHA-1" in sig_alg:
        issues.append("使用已弃用的 SHA-1 签名算法")
        risk_score += 3
    elif "SM2" in sig_alg or "SM3" in sig_alg:
        issues.append("使用国密 SM2/SM3 算法，需关注后量子替代方案")
        risk_score += 1
    
    # 检查公钥类型
    pub_key = cert_info.get("public_key", {})
    key_type = pub_key.get("type", "")
    key_size = pub_key.get("size", 0)
    
    if key_type == "RSA" and key_size < 2048:
        issues.append(f"RSA 密钥长度 {key_size} 位不足 2048 位")
        risk_score += 3
    elif key_type == "RSA" and key_size >= 2048:
        issues.append("RSA 密钥需迁移至后量子算法")
        risk_score += 2
    elif key_type in ("EC", "SM2"):
        if key_size < 256:
            issues.append(f"EC 密钥长度 {key_size} 位不足 256 位")
            risk_score += 2
        issues.append("EC/SM2 密钥需迁移至后量子签名算法")
        risk_score += 2
    
    # 检查证书有效期
    days = cert_info.get("days_remaining", 0)
    if days < 0:
        issues.append("证书已过期")
        risk_score += 5
    elif days < 30:
        issues.append(f"证书将在 {days} 天内过期")
        risk_score += 2
    
    # 检查是否自签名
    if cert_info.get("is_self_signed"):
        issues.append("自签名证书，不在 Web PKI 信任链中")
        risk_score += 1
    
    risk_level = "critical" if risk_score >= 5 else "high" if risk_score >= 3 else "medium" if risk_score >= 1 else "low"
    
    return {
        "risk_level": risk_level,
        "risk_score": risk_score,
        "pqc_ready": False,  # 当前 TLS 证书都不是 PQC 的
        "issues": issues,
        "recommended_action": "plan_migration" if risk_score >= 3 else "monitor",
    }

def generate_cbom_report(scan_results, output_path="cbom_report.json"):
    """生成 CBOM 报告"""
    cbom = {
        "metadata": {
            "generated_at": datetime.now(timezone.utc).isoformat(),
            "total_hosts_scanned": len(scan_results),
            "reachable_hosts": sum(1 for r in scan_results if r["reachable"]),
            "hosts_with_certificates": sum(1 for r in scan_results if r.get("certificate")),
        },
        "algorithm_summary": {},
        "risk_summary": {},
        "assets": [],
    }
    
    for result in scan_results:
        cert = result.get("certificate")
        if not cert:
            continue
        
        # 算法统计
        key_type = cert["public_key"]["type"]
        sig_alg = cert["signature_algorithm"]
        cbom["algorithm_summary"][key_type] = cbom["algorithm_summary"].get(key_type, 0) + 1
        cbom["algorithm_summary"][f"sig:{sig_alg}"] = cbom["algorithm_summary"].get(f"sig:{sig_alg}", 0) + 1
        
        # 风险评估
        risk = assess_pqc_risk(cert)
        risk_level = risk["risk_level"]
        cbom["risk_summary"][risk_level] = cbom["risk_summary"].get(risk_level, 0) + 1
        
        cbom["assets"].append({
            "host": result["host"],
            "port": result["port"],
            "tls_version": result["tls_version"],
            "certificate": cert,
            "risk_assessment": risk,
        })
    
    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(cbom, f, ensure_ascii=False, indent=2, default=str)
    
    return cbom

def print_summary(cbom):
    """打印 CBOM 摘要"""
    print("\n" + "=" * 60)
    print("  CBOM 扫描报告摘要")
    print("=" * 60)
    print(f"  扫描主机数: {cbom['metadata']['total_hosts_scanned']}")
    print(f"  可达主机数: {cbom['metadata']['reachable_hosts']}")
    print(f"  有效证书数: {cbom['metadata']['hosts_with_certificates']}")
    
    print("\n  算法分布:")
    for algo, count in cbom["algorithm_summary"].items():
        if not algo.startswith("sig:"):
            print(f"    {algo}: {count}")
    
    print("\n  签名算法分布:")
    for algo, count in cbom["algorithm_summary"].items():
        if algo.startswith("sig:"):
            print(f"    {algo.replace('sig:', '')}: {count}")
    
    print("\n  风险等级分布:")
    for level, count in cbom["risk_summary"].items():
        marker = "🔴" if level == "critical" else "🟠" if level == "high" else "🟡" if level == "medium" else "🟢"
        print(f"    {marker} {level}: {count}")
    
    # PQC 迁移就绪度
    total = len(cbom["assets"])
    non_pqc = sum(1 for a in cbom["assets"] if not a["risk_assessment"]["pqc_ready"])
    pqc_ready_pct = ((total - non_pqc) / total * 100) if total > 0 else 0
    print(f"\n  PQC 迁移就绪度: {pqc_ready_pct:.1f}% ({total - non_pqc}/{total})")
    print("=" * 60)

if __name__ == "__main__":
    # 示例：扫描一组目标
    targets = [
        ("example.com", 443),
        ("www.google.com", 443),
        ("www.baidu.com", 443),
    ]
    
    if len(sys.argv) > 1:
        # 支持命令行参数: python3 cbom_scanner.py host1:port host2:port
        targets = []
        for arg in sys.argv[1:]:
            parts = arg.split(":")
            host = parts[0]
            port = int(parts[1]) if len(parts) > 1 else 443
            targets.append((host, port))
    
    print(f"开始 CBOM 扫描，共 {len(targets)} 个目标...")
    results = []
    for host, port in targets:
        print(f"  扫描 {host}:{port}...", end=" ", flush=True)
        r = scan_tls_service(host, port)
        if r["reachable"]:
            print(f"✓ ({r['tls_version']})")
        else:
            print(f"✗ ({r['error']})")
        results.append(r)
    
    cbom = generate_cbom_report(results)
    print_summary(cbom)

3.3 配置层发现：SSH 密钥审计

PYTHON

#!/usr/bin/env python3
"""
CBOM SSH 密钥扫描器 — 发现企业环境中的 SSH 密钥对
"""
import os
import subprocess
import json
from pathlib import Path
from cryptography.hazmat.primitives.serialization import load_ssh_public_key
from cryptography.hazmat.backends import default_backend

def scan_ssh_keys(search_paths=None):
    """扫描指定路径下的 SSH 密钥"""
    if search_paths is None:
        search_paths = [
            os.path.expanduser("~/.ssh"),
            "/etc/ssh",
        ]
    
    keys = []
    for search_path in search_paths:
        path = Path(search_path)
        if not path.exists():
            continue
        
        for key_file in path.rglob("*"):
            if not key_file.is_file():
                continue
            
            # 跳过私钥文件（仅扫描公钥）
            if key_file.name in ("known_hosts", "authorized_keys", "config"):
                if key_file.name == "authorized_keys":
                    # 解析 authorized_keys，提取每个公钥
                    keys.extend(_parse_authorized_keys(key_file))
                continue
            
            if key_file.suffix == ".pub" or _is_public_key(key_file):
                key_info = _analyze_ssh_key(key_file)
                if key_info:
                    keys.append(key_info)
    
    return keys

def _parse_authorized_keys(file_path):
    """解析 authorized_keys 文件"""
    keys = []
    try:
        with open(file_path, "r") as f:
            for line_num, line in enumerate(f, 1):
                line = line.strip()
                if not line or line.startswith("#"):
                    continue
                parts = line.split()
                if len(parts) >= 2:
                    key_type = parts[0]
                    keys.append({
                        "file": str(file_path),
                        "line": line_num,
                        "key_type": key_type,
                        "source": "authorized_keys",
                    })
    except Exception:
        pass
    return keys

def _is_public_key(file_path):
    """判断文件是否为 SSH 公钥"""
    try:
        with open(file_path, "r") as f:
            first_line = f.readline().strip()
            return first_line.startswith(("ssh-rsa", "ssh-ed25519", "ecdsa-sha2-", "ssh-sm2"))
    except Exception:
        return False

def _analyze_ssh_key(file_path):
    """分析 SSH 密钥类型"""
    try:
        with open(file_path, "r") as f:
            content = f.read().strip()
        
        parts = content.split()
        if len(parts) < 2:
            return None
        
        key_type = parts[0]
        
        # 映射到算法族
        algo_map = {
            "ssh-rsa": "RSA",
            "ssh-ed25519": "Ed25519",
            "ssh-ed448": "Ed448",
            "ecdsa-sha2-nistp256": "ECDSA (P-256)",
            "ecdsa-sha2-nistp384": "ECDSA (P-384)",
            "ecdsa-sha2-nistp521": "ECDSA (P-521)",
            "ssh-sm2": "SM2",
        }
        
        return {
            "file": str(file_path),
            "key_type": key_type,
            "algorithm": algo_map.get(key_type, f"Unknown ({key_type})"),
            "key_size": _estimate_key_size(key_type),
            "pqc_vulnerable": key_type in ("ssh-rsa", "ecdsa-sha2-nistp256", "ecdsa-sha2-nistp384",
                                            "ecdsa-sha2-nistp521"),
            "recommendation": _pqc_migration_recommendation(key_type),
        }
    except Exception:
        return None

def _estimate_key_size(key_type):
    """估算密钥长度"""
    sizes = {
        "ssh-rsa": 2048,  # 需进一步解析实际长度
        "ssh-ed25519": 256,
        "ecdsa-sha2-nistp256": 256,
        "ecdsa-sha2-nistp384": 384,
        "ecdsa-sha2-nistp521": 521,
        "ssh-sm2": 256,
    }
    return sizes.get(key_type, 0)

def _pqc_migration_recommendation(key_type):
    """给出 PQC 迁移建议"""
    recommendations = {
        "ssh-rsa": "迁移至 FIDO2/WebAuthn 或 ML-KEM 密钥交换",
        "ecdsa-sha2-nistp256": "迁移至 ML-KEM + 混合密钥交换",
        "ecdsa-sha2-nistp384": "迁移至 ML-KEM + 混合密钥交换",
        "ecdsa-sha2-nistp521": "迁移至 ML-KEM + 混合密钥交换",
        "ssh-ed25519": "短期安全，长期需迁移至 PQC 方案",
        "ssh-sm2": "国密算法，需关注后量子替代方案",
    }
    return recommendations.get(key_type, "评估后确定迁移路径")

3.4 代码层发现：依赖库中的硬编码密码学

PYTHON

#!/usr/bin/env python3
"""
CBOM 代码扫描器 — 发现代码中硬编码的密码学算法和密钥
依赖: pip install PyYAML toml
"""
import re
import json
from pathlib import Path

# 检测模式
PATTERNS = {
    "RSA 密钥": [
        r'-----BEGIN RSA PRIVATE KEY-----',
        r'-----BEGIN RSA PUBLIC KEY-----',
        r'RSAPrivateKey|RSAPublicKey',
        r'generate_rsa\(|RSA\.generate\(',
    ],
    "ECC 密钥": [
        r'-----BEGIN EC PRIVATE KEY-----',
        r'-----BEGIN PUBLIC KEY-----.*EC',
        r'EC\.generate\(|generate_ecdsa\(',
        r'SECP256R1|SECP384R1|prime256v1',
    ],
    "AES 密钥/IV": [
        r'AES\.new\(|AES\.Cipher\(',
        r'aes_key\s*=|aes_iv\s*=',
        r'AES_KEY\s*=|AES_IV\s*=',
    ],
    "硬编码密码": [
        r'password\s*=\s*["\'][^"\']{8,}["\']',
        r'secret\s*=\s*["\'][^"\']{8,}["\']',
        r'api_key\s*=\s*["\'][^"\']{16,}["\']',
    ],
    "弱哈希算法": [
        r'md5\s*\(|hashlib\.md5',
        r'sha1\s*\(|hashlib\.sha1',
        r'MD5\.new\(|SHA1\.new\(',
    ],
    "不安全的随机数": [
        r'random\.random\(\)|random\.randint\(',
        r'Random\.getInstance\(\)',
        r'Math\.random\(\)',
    ],
    "国密算法": [
        r'SM2|SM3|SM4',
        r'sm2_encrypt|sm2_sign|sm3_hash|sm4_encrypt',
        r'SM4_GCM|SM4_CBC|SM4_ECB',
    ],
    "TLS/SSL 配置": [
        r'SSLContext\(|ssl\.create_default_context',
        r'TLSv1\.0|TLSv1\.1|PROTOCOL_SSLv23',
        r'CERT_NONE|CERT_OPTIONAL',
    ],
    "JWT 签名": [
        r'jwt\.encode\(|\.sign\(['"](HS|RS|ES)',
        r'algorithm\s*=\s*["\'](HS256|RS256|ES256)["\']',
    ],
}

def scan_codebase(root_path, extensions=None):
    """扫描代码库中的密码学使用"""
    if extensions is None:
        extensions = {".py", ".js", ".ts", ".java", ".go", ".c", ".cpp", ".h", ".rs"}
    
    findings = {}
    root = Path(root_path)
    
    for file_path in root.rglob("*"):
        if file_path.suffix.lower() not in extensions:
            continue
        
        try:
            content = file_path.read_text(encoding="utf-8", errors="ignore")
        except Exception:
            continue
        
        for category, patterns in PATTERNS.items():
            for pattern in patterns:
                matches = re.finditer(pattern, content)
                for match in matches:
                    line_num = content[:match.start()].count("\n") + 1
                    key = str(file_path)
                    if key not in findings:
                        findings[key] = []
                    findings[key].append({
                        "category": category,
                        "pattern": pattern,
                        "line": line_num,
                        "match": match.group()[:50],  # 截断过长的匹配
                    })
    
    return findings

def assess_codebase_pqc_readiness(findings):
    """评估代码库的 PQC 就绪度"""
    total_files = len(findings)
    categories = {}
    
    for file_path, file_findings in findings.items():
        for finding in file_findings:
            cat = finding["category"]
            categories[cat] = categories.get(cat, 0) + 1
    
    # 计算风险
    risk_items = []
    if "弱哈希算法" in categories:
        risk_items.append(f"发现 {categories['弱哈希算法']} 处弱哈希算法（MD5/SHA-1）")
    if "不安全的随机数" in categories:
        risk_items.append(f"发现 {categories['不安全的随机数']} 处不安全随机数生成")
    if "硬编码密码" in categories:
        risk_items.append(f"发现 {categories['硬编码密码']} 处硬编码密码")
    if "RSA 密钥" in categories:
        risk_items.append(f"发现 {categories['RSA 密钥']} 处 RSA 密钥使用")
    
    return {
        "total_files_with_crypto": total_files,
        "category_counts": categories,
        "risk_items": risk_items,
        "pqc_readiness_score": max(0, 100 - len(risk_items) * 15),
    }

四、从 CBOM 到迁移路线图

4.1 Meta 的六步迁移框架

Meta 2026 年 4 月发布的 PQC 迁移手册定义了六个步骤：

CODE

Step 1: 库存（Inventory）
  └── CBOM 构建 — 你有什么密码资产？

Step 2: 分类（Classification）
  └── 按敏感度和暴露面评分 — 哪些最关键？

Step 3: 评估（Assessment）
  └── 量子威胁分析 — 哪些数据需要长期保密？

Step 4: 规划（Roadmap）
  └── 分批迁移计划 — 先做什么，后做什么？

Step 5: 执行（Execution）
  └── 混合部署 — 先上混合方案，逐步替换

Step 6: 验证（Verification）
  └── 持续监控 — 确保没有遗留量子脆弱资产

4.2 双轨迁移模型

PQC Migration Framework v2.0（2026 年 6 月更新）提出了双轨迁移模型，将密钥交换和签名/认证分为两个独立轨道：

CODE

轨道 A：密钥交换（Key Exchange）
  驱动因素：先存储后解密（Harvest Now, Decrypt Later）威胁
  当前状态：ML-KEM 已标准化（FIPS 203），混合方案可立即部署
  部署方式：ECDHE + ML-KEM 混合密钥交换（IETF draft-ietf-tls-hybrid-design）
  优先级：高 — 保护当前通信免受未来量子攻击

轨道 B：签名/认证（Signature/Authentication）
  驱动因素：信任伪造（Trust Now, Forge Later）威胁
  当前状态：MTC 标准化中，FIPS 140-3 PQC 验证刚起步
  部署方式：取决于场景（Web 用 MTC，企业 PKI 用 ML-DSA）
  优先级：中高 — 依赖 PKI 架构决策

4.3 环境分级

不同部署环境对 PQC 的接受程度不同：

环境级别	PQC 部署状态	典型系统	建议动作
无限制	立即部署混合 ML-KEM	公网 Web 应用	本季度完成
FIPS 感知	验证后部署	企业应用	等待 FIPS 140-3 PQC 模块
FIPS 必需	需等验证模块	金融支付系统	SafeLogic 2026 年 5 月已提交首个模块
CNSA 2.0	强制合规	国家安全系统	2027 年 1 月前完成

五、国密体系的特殊考量

5.1 国密资产的双重迁移压力

对于使用国密算法的企业，CBOM 构建需要额外关注：

PYTHON

# 国密特有的 PQC 迁移路径
GM_MIGRATION_PATHS = {
    "SM2 签名": {
        "当前状态": "GM/T 0003-2012 标准，256 位素数域椭圆曲线",
        "量子威胁": "Shor 算法可破解 ECDLP，SM2 与 ECDSA 面临相同威胁",
        "短期方案": "SM2 + ML-DSA 混合签名（需自定义实现）",
        "长期方案": "等待国密后量子标准或全面迁移至 NIST PQC",
        "注意": "国密体系的后量子迁移尚无明确时间表，需持续关注",
    },
    "SM3 哈希": {
        "当前状态": "GM/T 0004-2012 标准，256 位哈希",
        "量子威胁": "Grover 算法将安全强度降至 128 位，仍可接受",
        "短期方案": "维持 SM3，安全强度足够",
        "长期方案": "若需 256 位量子安全，可迁移至 SHA-3-512",
        "注意": "SM3 的量子安全性优于 SHA-256",
    },
    "SM4 分组加密": {
        "当前状态": "GM/T 0002-2012 标准，128 位密钥",
        "量子威胁": "Grover 算法将安全强度降至 64 位",
        "短期方案": "SM4-256（若支持）或 AES-256",
        "长期方案": "迁移至 AES-256 或后量子对称加密方案",
        "注意": "SM4 的 128 位密钥在量子场景下偏弱",
    },
}

5.2 密评合规与 CBOM 的关系

GB/T 39786-2021《信息安全技术信息系统密码应用基本要求》要求：

密码算法：应采用国家密码管理部门认可的算法
密钥管理：密钥的生成、存储、分发、使用、备份、恢复、销毁应有完整记录
密码设备：应采用经过国家密码管理部门核准的密码设备

CBOM 直接支撑密评合规中的"密码算法清单"和"密钥管理记录"两项要求。

六、总结

CBOM 构建是 PQC 迁移的第一步，也是最关键的一步。核心要点：

自动化发现：不要依赖手动盘点。使用 OpenSSL、端口扫描、代码分析等工具链系统化发现。
结构化数据：用 CBOM 数据模型统一管理，避免散落在 Excel 和 Wiki 中。
优先级驱动：按暴露度和敏感度评分，优先迁移公网暴露且保护敏感数据的系统。
双轨并行：密钥交换（Track A）立即部署混合 ML-KEM，签名认证（Track B）根据 PKI 架构决策。
持续更新：CBOM 不是一次性项目，应与 CI/CD 集成，在每次部署时自动更新。

下一步：完成 CBOM 构建后，下一步是按优先级对资产进行混合密钥交换部署。这将在后续文章中详细讨论。

参考来源

NIST IR 8547: Transition to Post-Quantum Cryptography Standards — NIST 后量子迁移框架
Meta Engineering: PQC Migration Playbook — Meta 2026 年 4 月迁移手册
NIST SP 800-227: Recommendations for Key Encapsulation Mechanisms — KEM 实施指南
PQC Migration Framework v2.0 (2026-06-08) — 后量子迁移框架 v2.0
CNSA 2.0 Algorithm Suites — NSA 商用国家安全算法套件 2.0
NIST NCCoE PQC Migration Project — NIST 国家网络安全卓越中心 PQC 迁移项目
Encryption Consulting: PQC Migration Heatmaps — PQC 迁移进度月度热力图
PQCFramework.com — Applied Quantum PQC 迁移框架（CC BY 4.0）
GB/T 39786-2021 — 信息安全技术信息系统密码应用基本要求

后量子密码迁移第一步：企业密码资产清单（CBOM）构建实战

前言

一、为什么 CBOM 是 PQC 迁移的前提

1.1 密码资产的"冰山效应"

1.2 CBOM 驱动迁移决策

1.3 监管要求的硬性时间表

二、CBOM 的核心结构

2.1 清单数据模型

2.2 优先级评分模型

三、自动化发现工具链

3.1 工具链架构

3.2 实战：TLS 证书扫描工具

3.3 配置层发现：SSH 密钥审计

3.4 代码层发现：依赖库中的硬编码密码学

四、从 CBOM 到迁移路线图

4.1 Meta 的六步迁移框架

4.2 双轨迁移模型

4.3 环境分级

五、国密体系的特殊考量

5.1 国密资产的双重迁移压力

5.2 密评合规与 CBOM 的关系

六、总结

参考来源

相关实践