关键信息基础设施密码合规实战：第5号令下企业的国密改造路线图

阶段一：差距分析（1-2个月）
    ├── 识别 CII 边界
    ├── 梳理密码应用现状
    ├── 对标 GM/T 0133-2024 评估差距
    └── 输出《密码应用差距分析报告》

阶段二：方案设计（2-3个月）
    ├── 设计密码应用方案（含物理、网络、主机、应用、数据层）
    ├── 选型国密产品（密码机、SSL 网关、Key management 等）
    ├── 编写《密码应用方案》并提交密评机构预审
    └── 制定实施计划

阶段三：工程实施（3-6个月）
    ├── 部署密码基础设施
    ├── 改造应用系统（接口替换、数据加密）
    ├── 集成密钥管理系统
    └── 内部测试验证

阶段四：密评验收（1-2个月）
    ├── 邀请密评机构现场评估
    ├── 整改不符合项
    ├── 获取密评报告
    └── 建立持续合规机制

┌─────────────────┐
                     │   根 CA / 子 CA  │
                     │  (国密 SM2 证书) │
                     └────────┬────────┘
                              │ 签发
                ┌─────────────┼─────────────┐
                │             │             │
        ┌───────▼──────┐ ┌───▼────┐ ┌──────▼───────┐
        │   SSL 网关   │ │  KMS   │ │  时间戳服务器  │
        │ SM2/RSA 双证 │ │ 密钥管理│ │  RFC3161+SM2 │
        └───────┬──────┘ └───┬────┘ └──────┬───────┘
                │            │             │
    ┌───────────┼────────────┼─────────────┼───────────┐
    │           │            │             │           │
┌───▼───┐  ┌───▼───┐  ┌────▼────┐  ┌────▼────┐  ┌───▼───┐
│ Web   │  │ API   │  │ 数据库   │  │ 文件    │  │ 应用   │
│ 前端  │  │ 网关  │  │ (SM4)   │  │ 存储    │  │ 服务   │
└───────┘  └───────┘  └─────────┘  └─────────┘  └───────┘

# Tongsuo 国密 SSL 配置
# ⚠️ 环境要求：Tongsuo 铜锁密码库 + Tengine/ nginx 国密模块
server {
    listen 443 ssl;
    server_name example.com;
    
    # 国密证书（SM2）
    ssl_certificate /etc/nginx/certs/sm2_server.pem;
    ssl_certificate_key /etc/nginx/certs/sm2_server.key;
    
    # 国际证书（RSA/ECDSA）
    ssl_certificate /etc/nginx/certs/rsa_server.pem;
    ssl_certificate_key /etc/nginx/certs/rsa_server.key;
    
    # 国密优先密码套件
    ssl_ciphers ECDHE-SM2-SM4-GCM-SM3:ECDHE-SM2-SM4-CBC-SM3:ECDHE-RSA-AES128-GCM-SHA256;
    ssl_protocols TLSv1.2 TLSv1.3;
    ssl_prefer_server_ciphers on;
    
    location / {
        proxy_pass http://backend;
    }
}

from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.primitives import hashes
import os
import json
import time
from typing import Dict, Optional, Tuple
from dataclasses import dataclass


@dataclass
class KeyMaterial:
    """密钥材料"""
    key_id: str
    version: int
    key_value: bytes
    created_at: float
    expires_at: float
    status: str  # "active", "deprecated", "destroyed"
    algorithm: str  # "SM4", "SM2", "SM3"


class KeyManagementService:
    """
    密钥管理系统核心
    
    功能：
    - 密钥生成、存储、分发、轮换、销毁
    - 密钥分层（主密钥 -> KEK -> DEK）
    - 密钥版本管理
    - 密钥使用审计日志
    
    ⚠️ 生产环境需要使用硬件密码机（HSM）保护主密钥
    """
    
    def __init__(self, hsm_enabled: bool = False):
        self._hsm_enabled = hsm_enabled
        self._keys: Dict[str, KeyMaterial] = {}
        self._audit_log: list = []
    
    def _log(self, action: str, key_id: str, details: str = ""):
        """审计日志"""
        self._audit_log.append({
            "timestamp": time.time(),
            "action": action,
            "key_id": key_id,
            "details": details
        })
    
    def generate_key(
        self,
        key_id: str,
        algorithm: str = "SM4",
        key_length: int = 16,
        validity_days: int = 365,
        key_version: int = 1
    ) -> KeyMaterial:
        """
        生成新密钥
        
        Args:
            key_id: 密钥标识符
            algorithm: 算法类型 (SM4/SM2)
            key_length: 密钥长度（SM4 固定 16 字节）
            validity_days: 有效期（天）
            key_version: 版本号
        """
        if algorithm == "SM4":
            key_value = os.urandom(key_length)
        elif algorithm == "SM2":
            # SM2 密钥需要 GM/T 0003 标准参数
            # 这里简化为生成随机种子，实际应使用 gmssl 或 HSM
            key_value = os.urandom(32)
        else:
            raise ValueError(f"不支持的算法: {algorithm}")
        
        now = time.time()
        key = KeyMaterial(
            key_id=key_id,
            version=key_version,
            key_value=key_value,
            created_at=now,
            expires_at=now + validity_days * 86400,
            status="active",
            algorithm=algorithm
        )
        
        storage_key = f"{key_id}:v{key_version}"
        self._keys[storage_key] = key
        self._log("generate", key_id, f"version={key_version}")
        return key
    
    def get_key(self, key_id: str, version: Optional[int] = None) -> KeyMaterial:
        """获取密钥（默认返回最新版本）"""
        if version:
            storage_key = f"{key_id}:v{version}"
            if storage_key not in self._keys:
                raise KeyError(f"密钥不存在: {storage_key}")
            return self._keys[storage_key]
        
        # 查找最新版本
        latest = None
        for storage_key, key in self._keys.items():
            if key.key_id == key_id and key.status == "active":
                if latest is None or key.version > latest.version:
                    latest = key
        
        if not latest:
            raise KeyError(f"密钥不存在或已过期: {key_id}")
        
        self._log("access", key_id, f"version={latest.version}")
        return latest
    
    def rotate_key(self, key_id: str, validity_days: int = 365) -> KeyMaterial:
        """
        密钥轮换
        
        1. 生成新版本密钥
        2. 旧版本标记为 deprecated（仍可用于解密）
        3. 新数据使用新版本加密
        """
        latest = self.get_key(key_id)
        new_version = latest.version + 1
        
        new_key = self.generate_key(
            key_id=key_id,
            algorithm=latest.algorithm,
            key_length=len(latest.key_value),
            validity_days=validity_days,
            key_version=new_version
        )
        
        # 旧版本标记为 deprecated
        latest.status = "deprecated"
        self._log("rotate", key_id, f"{latest.version} -> {new_version}")
        return new_key
    
    def destroy_key(self, key_id: str, version: int):
        """密钥销毁"""
        storage_key = f"{key_id}:v{version}"
        if storage_key in self._keys:
            # 安全擦除
            key = self._keys[storage_key]
            # 注意：Python 中无法真正擦除内存，需在 HSM 中实现
            key.status = "destroyed"
            key.key_value = b'\x00' * len(key.key_value)
            self._log("destroy", key_id, f"version={version}")
    
    def get_audit_log(self) -> list:
        """获取审计日志"""
        return self._audit_log.copy()


# 使用示例
kms = KeyManagementService()

# 生成数据库加密密钥
db_key = kms.generate_key("database-encryption", algorithm="SM4", validity_days=180)
print(f"生成密钥: {db_key.key_id} v{db_key.version}")

# 轮换
new_key = kms.rotate_key("database-encryption")
print(f"轮换后: v{new_key.version}")

# 获取审计日志
for entry in kms.get_audit_log():
    print(f"  [{entry['action']}] {entry['key_id']}")

import os
import json
import base64
from typing import Optional
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives import padding


class DatabaseFieldEncryptor:
    """
    数据库字段级加密器
    
    加密策略：
    - SM4-CBC 加密
    - 每列独立 IV
    - 支持确定性加密（用于等值查询）和随机加密（高安全性）
    """
    
    def __init__(self, master_key: bytes):
        self._master_key = master_key
    
    def encrypt_field(
        self,
        plaintext: str,
        field_name: str,
        deterministic: bool = False
    ) -> str:
        """
        加密单个字段
        
        Args:
            plaintext: 明文值
            field_name: 字段名（用于派生列密钥）
            deterministic: 是否确定性加密（相同输入产生相同输出，支持查询）
        
        Returns:
            Base64 编码的密文（格式: version:iv:ciphertext）
        """
        # 派生列密钥
        col_key = self._derive_column_key(field_name)
        
        # 确定性加密：使用固定 IV；随机加密：使用随机 IV
        if deterministic:
            iv = self._derive_iv(plaintext, col_key)
        else:
            iv = os.urandom(16)
        
        # PKCS#7 填充
        padder = padding.PKCS7(128).padder()
        padded = padder.update(plaintext.encode('utf-8')) + padder.finalize()
        
        cipher = Cipher(algorithms.SM4(col_key), modes.CBC(iv))
        encryptor = cipher.encryptor()
        ciphertext = encryptor.update(padded) + encryptor.finalize()
        
        # 格式: v1:base64(iv):base64(ciphertext)
        return f"v1:{base64.b64encode(iv).decode()}:{base64.b64encode(ciphertext).decode()}"
    
    def decrypt_field(self, encrypted: str, field_name: str) -> str:
        """解密字段"""
        parts = encrypted.split(":", 2)
        if len(parts) != 3 or parts[0] != "v1":
            raise ValueError(f"不支持的加密格式: {encrypted[:20]}...")
        
        iv = base64.b64decode(parts[1])
        ciphertext = base64.b64decode(parts[2])
        
        col_key = self._derive_column_key(field_name)
        cipher = Cipher(algorithms.SM4(col_key), modes.CBC(iv))
        decryptor = cipher.decryptor()
        padded = decryptor.update(ciphertext) + decryptor.finalize()
        
        unpadder = padding.PKCS7(128).unpadder()
        plaintext = unpadder.update(padded) + unpadder.finalize()
        return plaintext.decode('utf-8')
    
    def _derive_column_key(self, field_name: str) -> bytes:
        """派生列密钥"""
        from cryptography.hazmat.primitives.kdf.hkdf import HKDF
        from cryptography.hazmat.primitives import hashes
        
        hkdf = HKDF(
            algorithm=hashes.SM3(),
            length=16,
            salt=None,
            info=f"db:column:{field_name}".encode()
        )
        return hkdf.derive(self._master_key)
    
    def _derive_iv(self, plaintext: str, key: bytes) -> bytes:
        """确定性 IV：使用 HKDF-SM3 从明文派生"""
        from cryptography.hazmat.primitives import hashes
        from cryptography.hazmat.primitives.kdf.hkdf import HKDF

        hkdf = HKDF(algorithm=hashes.SM3(), length=16, salt=None, info=b"deterministic-iv")
        return hkdf.derive(plaintext.encode() + key)


# 使用示例
key = os.urandom(16)
encryptor = DatabaseFieldEncryptor(key)

# 随机加密（高安全性，不支持查询）
encrypted_id = encryptor.encrypt_field("110101199001011234", "id_number")
print(f"身份证号密文: {encrypted_id[:40]}...")

# 确定性加密（支持等值查询）
encrypted_phone = encryptor.encrypt_field("13800138000", "phone", deterministic=True)
print(f"手机号密文: {encrypted_phone[:40]}...")

# 解密
decrypted = encryptor.decrypt_field(encrypted_id, "id_number")
print(f"解密后: {decrypted}")

# 国密客户端证书验证示例（Python Flask）
from flask import Flask, request, jsonify
from cryptography import x509
from cryptography.hazmat.primitives import hashes
import datetime

app = Flask(__name__)

# 国密根 CA 证书（用于验证客户端证书链）
# 生产环境应从可信来源加载
GM_CA_CERT = None  # 加载 SM2 根证书


@app.route('/api/sensitive', methods=['POST'])
def sensitive_operation():
    """敏感操作接口 - 要求客户端国密证书"""
    
    # 从请求中获取客户端证书（由 SSL 网关注入到 Header）
    cert_pem = request.headers.get('X-SSL-Client-Cert')
    if not cert_pem:
        return jsonify({"error": "缺少客户端证书"}), 401
    
    try:
        # 解析证书
        cert = x509.load_pem_x509_certificate(cert_pem.encode())
        
        # 验证证书有效期
        now = datetime.datetime.utcnow()
        if now < cert.not_valid_before or now > cert.not_valid_after:
            return jsonify({"error": "证书已过期"}), 401
        
        # 验证是否为 SM2 算法（通过 OID 判断）
        # SM2 签名算法 OID: 1.2.156.10197.1.301
        sig_oid = cert.signature_algorithm_oid.dotted_string
        if sig_oid != "1.2.156.10197.1.301":
            return jsonify({"error": "证书非国密 SM2 算法"}), 401
        
        # 提取用户 ID（从证书 Subject CN 字段）
        cn = cert.subject.get_attributes_for_oid(x509.oid.NameOID.COMMON_NAME)
        user_id = cn[0].value if cn else "unknown"
        
        # 业务逻辑...
        return jsonify({
            "status": "success",
            "user": user_id,
            "operation": "sensitive_access"
        })
    
    except Exception as e:
        return jsonify({"error": f"证书验证失败: {e}"}), 401


if __name__ == '__main__':
    # 注意：实际部署时 HTTPS 由 SSL 网关终止，此处为 HTTP
    app.run(host='0.0.0.0', port=8080)

# CII 密码合规自检清单（代码化展示）

CII_COMPLIANCE_CHECKLIST = {
    "制度建设": {
        "密码管理制度": {"required": True, "status": "待确认"},
        "密码人员管理": {"required": True, "status": "待确认"},
        "密码应急处置": {"required": True, "status": "待确认"},
        "密码培训": {"required": True, "status": "待确认"},
    },
    "算法技术": {
        "对称加密": {"algorithm": "SM4", "status": "待核查"},
        "非对称加密": {"algorithm": "SM2", "status": "待核查"},
        "哈希算法": {"algorithm": "SM3", "status": "待核查"},
        "随机数生成": {"standard": "GM/T 0005", "status": "待核查"},
    },
    "产品服务": {
        "服务器密码机": {"cert": "需认证", "status": "待核查"},
        "SSL 安全网关": {"cert": "需认证", "status": "待核查"},
        "数字证书认证系统": {"cert": "需认证", "status": "待核查"},
        "密钥管理系统": {"cert": "需认证", "status": "待核查"},
    },
    "三同步": {
        "同步规划": {"required": True, "status": "待确认"},
        "同步建设": {"required": True, "status": "待确认"},
        "同步使用": {"required": True, "status": "待确认"},
    },
    "密码应用方案": {
        "方案编写": {"required": True, "status": "待确认"},
        "密评预审": {"required": True, "status": "待确认"},
        "方案审批": {"required": True, "status": "待确认"},
    },
}


def print_checklist(checklist: dict, indent: int = 0):
    """打印自检清单"""
    for category, items in checklist.items():
        prefix = "  " * indent + f"□ {category}"
        print(prefix)
        if isinstance(items, dict):
            for item, details in items.items():
                if isinstance(details, dict):
                    required = details.get('required', details.get('algorithm', ''))
                    status = details.get('status', '')
                    print(f"{'  ' * (indent + 1)}  - {item}: {required} [{status}]")
                else:
                    print(f"{'  ' * (indent + 1)}  - {item}: {details}")


# 执行自检清单输出
print("=== CII 密码合规自检清单 ===")
print_checklist(CII_COMPLIANCE_CHECKLIST)