国密算法密钥生命周期管理：从生成到销毁的实战指南

import os
import secrets
from gmssl import sm2, func

def generate_sm2_keypair_secure():
    """生成 SM2 密钥对，使用密码学安全随机数"""
    ecc_table = sm2.default_ecc_table
    n = int(ecc_table['n'], 16)

    # 使用 secrets 模块（Python 3.6+），比 random 更安全
    while True:
        rand_bytes = secrets.token_bytes(32)
        d = int.from_bytes(rand_bytes, 'big') % n
        if d != 0:  # 私钥不能为 0
            break

    # 计算公钥 Q = d·G
    crypt = sm2.CryptSM2(public_key='', private_key='')
    pub_point = crypt._kg(d, ecc_table['g'])

    return format(d, '064x'), pub_point

┌────────────────────────────────────────────────────────────┐
│  Level 4: HSM（硬件安全模块）                                 │
│  私钥永不离开硬件，所有签名/解密在 HSM 内部完成                 │
│  适用：根 CA 密钥、主密钥、高价值签名密钥                       │
├────────────────────────────────────────────────────────────┤
│  Level 3: 密钥管理服务（KMS / Vault）                          │
│  密钥加密存储，通过 API 调用，支持审计日志和访问控制             │
│  适用：业务签名密钥、数据加密密钥                               │
├────────────────────────────────────────────────────────────┤
│  Level 2: 加密存储（Envelope Encryption）                     │
│  用主密钥加密业务密钥，加密后的业务密钥可安全存数据库            │
│  适用：应用配置中的密钥、数据库连接凭证                          │
├────────────────────────────────────────────────────────────┤
│  Level 1: 配置文件（仅限开发/测试）                            │
│  密钥以明文或 Base64 存储在配置文件中                           │
│  ⚠️ 生产环境绝对禁止                                          │
└────────────────────────────────────────────────────────────┘

import secrets
from gmssl.sm4 import CryptSM4, SM4_ENCRYPT, SM4_DECRYPT
from gmssl.func import bytes_to_list, list_to_bytes

class EnvelopeEncryption:
    """信封加密：用 SM4 主密钥加密业务密钥"""

    def __init__(self, master_key: bytes):
        if len(master_key) != 16:
            raise ValueError("SM4 密钥必须为 16 字节")
        self.master_key = master_key

    def wrap_key(self, plaintext_key: bytes) -> bytes:
        """用主密钥加密业务密钥，返回 IV + 密文"""
        iv = secrets.token_bytes(16)
        crypt = CryptSM4()
        crypt.set_key(self.master_key, SM4_ENCRYPT)

        # PKCS7 填充
        pad_len = 16 - (len(plaintext_key) % 16)
        padded = plaintext_key + bytes([pad_len] * pad_len)

        encrypted = list_to_bytes(crypt.crypt_ecb(bytes_to_list(padded)))
        return iv + encrypted

    def unwrap_key(self, wrapped_key: bytes) -> bytes:
        """用主密钥解密业务密钥"""
        iv = wrapped_key[:16]
        ciphertext = wrapped_key[16:]

        crypt = CryptSM4()
        crypt.set_key(self.master_key, SM4_DECRYPT)

        decrypted = list_to_bytes(crypt.crypt_ecb(bytes_to_list(ciphertext)))

        # 去除 PKCS7 填充
        pad_len = decrypted[-1]
        if not (1 <= pad_len <= 16):
            raise ValueError("无效的填充")
        for i in range(pad_len):
            if decrypted[-(i + 1)] != pad_len:
                raise ValueError("填充字节不一致")

        return decrypted[:-pad_len]


# 使用示例
def demo_envelope():
    # 1. 从 KMS 获取主密钥（演示用随机生成）
    master_key = secrets.token_bytes(16)

    # 2. 创建信封加密实例
    envelope = EnvelopeEncryption(master_key)

    # 3. 生成业务密钥（DEK）
    dek = secrets.token_bytes(16)

    # 4. 加密业务密钥
    wrapped_dek = envelope.wrap_key(dek)
    # wrapped_dek 可以安全存入数据库

    # 5. 使用时解密
    unwrapped_dek = envelope.unwrap_key(wrapped_dek)
    assert dek == unwrapped_dek

# ❌ 错误：明文存储
DATABASE_ENCRYPTION_KEY = "0123456789abcdef"

# ❌ 错误：Base64 不是加密
import base64
DATABASE_ENCRYPTION_KEY = base64.b64decode("ASNFZ4mrze8=")

# ✅ 最小方案：从环境变量读取
import os
DATABASE_ENCRYPTION_KEY = os.environ.get('DB_ENC_KEY')
if not DATABASE_ENCRYPTION_KEY:
    raise RuntimeError("DB_ENC_KEY 环境变量未设置")

# ✅ 更好：从密钥管理服务读取（以 HashiCorp Vault 为例）
from hvac import Client

def get_key_from_vault(path: str) -> bytes:
    client = Client(url='https://vault.internal:8200')
    secret = client.secrets.kv.v2.read_secret_version(path=path)
    return bytes.fromhex(secret['data']['data']['key'])

DATABASE_ENCRYPTION_KEY = get_key_from_vault('secret/data/db/encryption')

import hmac

def constant_time_compare(a: bytes, b: bytes) -> bool:
    """恒定时间比较，防止计时攻击泄露比较位置"""
    return hmac.compare_digest(a, b)

# ❌ 错误：普通比较会在第一个不同字节处提前返回
def unsafe_verify(expected: bytes, actual: bytes) -> bool:
    return expected == actual  # 计时攻击可逐字节猜测

# ✅ 正确：恒定时间比较
def safe_verify(expected: bytes, actual: bytes) -> bool:
    return hmac.compare_digest(expected, actual)

import logging
import json
import time
from dataclasses import dataclass, asdict
from typing import Optional

@dataclass
class KeyAuditEvent:
    timestamp: float
    key_id: str
    operation: str       # 'sign', 'verify', 'encrypt', 'decrypt', 'rotate', 'revoke'
    operator: str        # 操作者标识（服务名或用户 ID）
    success: bool
    client_ip: Optional[str] = None
    request_id: Optional[str] = None  # 用于关联请求链路

class KeyAuditLogger:
    """密钥操作审计日志"""

    def __init__(self, logger_name: str = 'key_audit'):
        self.logger = logging.getLogger(logger_name)

    def log(self, event: KeyAuditEvent):
        """记录密钥操作"""
        self.logger.info(json.dumps(asdict(event), ensure_ascii=False))

    def log_sign(self, key_id: str, operator: str, success: bool, **kwargs):
        self.log(KeyAuditEvent(
            timestamp=time.time(),
            key_id=key_id,
            operation='sign',
            operator=operator,
            success=success,
            **kwargs
        ))

    def log_rotate(self, old_key_id: str, new_key_id: str, operator: str):
        self.log(KeyAuditEvent(
            timestamp=time.time(),
            key_id=f"{old_key_id}->{new_key_id}",
            operation='rotate',
            operator=operator,
            success=True
        ))

    def log_anomaly(self, message: str, **context):
        """记录异常事件"""
        self.logger.warning("ANOMALY: " + message + " | " + json.dumps(context))

import time
from collections import defaultdict

class KeyRateLimiter:
    """密钥操作频率限制器"""

    def __init__(self, max_ops_per_second: int = 100, max_ops_per_key: int = 10):
        self.max_ops_per_second = max_ops_per_second
        self.max_ops_per_key = max_ops_per_key
        self.window = 1.0  # 1 秒窗口
        self.counters: dict = defaultdict(list)  # key_id -> [timestamps]

    def check(self, key_id: str) -> bool:
        """检查是否允许操作，返回 True 表示允许"""
        now = time.time()
        timestamps = self.counters[key_id]

        # 清理窗口外的时间戳
        self.counters[key_id] = [t for t in timestamps if now - t < self.window]
        timestamps = self.counters[key_id]

        # 检查全局频率
        if len(timestamps) >= self.max_ops_per_second:
            return False

        # 检查单密钥频率
        key_count = sum(1 for t in timestamps if now - t < self.window)
        if key_count >= self.max_ops_per_key:
            return False

        # 记录本次操作
        self.counters[key_id].append(now)
        return True

import time
import secrets
import threading
from collections import OrderedDict
from dataclasses import dataclass
from typing import Optional, List

@dataclass
class KeyVersion:
    key_id: str
    private_key: str   # 实际中应加密存储或存入 HSM
    public_key: str
    created_at: float
    expires_at: float
    status: str  # 'active', 'deprecated', 'destroyed'

class SM2KeyRotationManager:
    """SM2 密钥轮换管理器"""

    def __init__(self, rotation_interval: int = 86400):
        self.rotation_interval = rotation_interval
        self.active_key: Optional[KeyVersion] = None
        self.key_history: OrderedDict[str, KeyVersion] = OrderedDict()
        self.max_history = 5
        self._lock = threading.Lock()

    def initialize(self):
        """初始化：加载现有密钥或生成新密钥"""
        self._generate_new_key()

    def _generate_new_key(self) -> KeyVersion:
        from gmssl import sm2
        ecc_table = sm2.default_ecc_table
        n = int(ecc_table['n'], 16)

        while True:
            d = int.from_bytes(secrets.token_bytes(32), 'big') % n
            if d != 0:
                break

        crypt = sm2.CryptSM2(public_key='', private_key='')
        pub_point = crypt._kg(d, ecc_table['g'])

        now = time.time()
        return KeyVersion(
            key_id=format(int.from_bytes(secrets.token_bytes(4), 'big'), '08x'),
            private_key=format(d, '064x'),
            public_key=pub_point,
            created_at=now,
            expires_at=now + self.rotation_interval,
            status='active'
        )

    def rotate(self) -> KeyVersion:
        """执行密钥轮换"""
        with self._lock:
            # 1. 将当前密钥移入历史
            if self.active_key:
                self.active_key.status = 'deprecated'
                self.key_history[self.active_key.key_id] = self.active_key

                # 2. 清理过多历史密钥
                while len(self.key_history) > self.max_history:
                    _, old_key = self.key_history.popitem(last=False)
                    old_key.status = 'destroyed'

            # 3. 生成新密钥
            new_key = self._generate_new_key()
            self.active_key = new_key
            return new_key

    def get_active_key(self) -> KeyVersion:
        """获取当前活跃密钥（用于签名）

        ⚠️ 注意：自动轮换发生在获取密钥的瞬间。如果调用方正在执行签名操作，
        轮换可能导致签名失败（旧密钥已标记为 deprecated）。生产环境中建议：
        - 在签名开始前显式获取一次密钥并复用
        - 或在 rotate() 中保留旧密钥的短暂宽限期（grace period）
        """
        with self._lock:
            if not self.active_key:
                raise RuntimeError("未初始化密钥")
            if time.time() > self.active_key.expires_at:
                return self.rotate()
            return self.active_key

    def get_verification_keys(self) -> List[KeyVersion]:
        """获取所有可用于验签的密钥（活跃 + 历史）"""
        with self._lock:
            keys = list(self.key_history.values())
            if self.active_key:
                keys.append(self.active_key)
            return keys

    def find_key_by_id(self, key_id: str) -> Optional[KeyVersion]:
        """根据 key_id 查找密钥（用于验证历史签名）"""
        if self.active_key and self.active_key.key_id == key_id:
            return self.active_key
        return self.key_history.get(key_id)

class DataKeyRotator:
    """数据密钥轮换器：用新 DEK 重新加密数据"""

    def __init__(self, kek: bytes, old_dek_wrapped: bytes, new_dek_wrapped: bytes):
        """
        kek: 密钥加密密钥（用于解包 DEK）
        old_dek_wrapped: 旧 DEK 的加密包裹
        new_dek_wrapped: 新 DEK 的加密包裹
        """
        envelope = EnvelopeEncryption(kek)
        self.old_dek = envelope.unwrap_key(old_dek_wrapped)
        self.new_dek = envelope.unwrap_key(new_dek_wrapped)

    def reencrypt_record(self, iv: bytes, ciphertext: bytes) -> tuple:
        """重新加密单条记录，返回 (new_iv, new_ciphertext)"""
        from gmssl.sm4 import CryptSM4, SM4_DECRYPT, SM4_ENCRYPT
        from gmssl.func import bytes_to_list, list_to_bytes

        # 1. 用旧 DEK 解密
        crypt_old = CryptSM4()
        crypt_old.set_key(self.old_dek, SM4_DECRYPT)
        plaintext_padded = list_to_bytes(
            crypt_old.crypt_ecb(bytes_to_list(ciphertext))
        )

        # 去除 PKCS7 填充得到明文
        pad_len = plaintext_padded[-1]
        plaintext = plaintext_padded[:-pad_len]

        # 2. 用新 DEK 加密（生成新 IV）
        new_iv = secrets.token_bytes(16)
        crypt_new = CryptSM4()
        crypt_new.set_key(self.new_dek, SM4_ENCRYPT)

        # PKCS7 填充
        pad_len = 16 - (len(plaintext) % 16)
        padded = plaintext + bytes([pad_len] * pad_len)

        new_ciphertext = list_to_bytes(crypt_new.crypt_ecb(bytes_to_list(padded)))
        return new_iv, new_ciphertext

# 吊销 SM2 证书
/usr/local/gmssl/bin/openssl ca -revoke /path/to/compromised-cert.pem \
    -keyfile /path/to/ca-key.pem \
    -cert /path/to/ca-cert.pem \
    -config /path/to/openssl.cnf

# 生成 CRL
/usr/local/gmssl/bin/openssl ca -gencrl \
    -keyfile /path/to/ca-key.pem \
    -cert /path/to/ca-cert.pem \
    -out /path/to/crl.pem

# 验证 CRL
/usr/local/gmssl/bin/openssl crl -in /path/to/crl.pem -text -noout

import time
import threading
from typing import Set

class KeyRevocationList:
    """应用层密钥吊销列表"""

    def __init__(self, refresh_interval: int = 3600):
        self.revoked_keys: Set[str] = set()
        self.refresh_interval = refresh_interval
        self.last_refresh = 0
        self._lock = threading.Lock()

    def revoke(self, key_id: str):
        """吊销指定密钥"""
        with self._lock:
            self.revoked_keys.add(key_id)

    def is_revoked(self, key_id: str) -> bool:
        """检查密钥是否已被吊销"""
        self._refresh_if_needed()
        return key_id in self.revoked_keys

    def _refresh_if_needed(self):
        """定期从服务端拉取最新吊销列表"""
        now = time.time()
        if now - self.last_refresh > self.refresh_interval:
            self._fetch_revocation_list()

    def _fetch_revocation_list(self):
        """从 Vault/CA 拉取吊销列表（示例）"""
        # 实际中从 CRL 端点或 Vault API 拉取
        # response = requests.get('https://ca.internal/crl')
        # self.revoked_keys = parse_crl(response.content)
        self.last_refresh = time.time()

import ctypes
import mmap

def secure_erase_bytes(data: bytearray):
    """安全擦除内存中的密钥数据"""
    # 覆写 3 次（DoD 5220.22-M 标准简化版）
    for _ in range(3):
        for i in range(len(data)):
            data[i] = 0x00
        for i in range(len(data)):
            data[i] = 0xFF

    # 释放
    data.clear()

def secure_erase_file(filepath: str):
    """安全擦写文件（DoD 5220.22-M 简化版）"""
    import os

    if not os.path.exists(filepath):
        return

    file_size = os.path.getsize(filepath)

    with open(filepath, 'r+b') as f:
        # 第 1 轮：覆写 \x00
        f.seek(0)
        f.write(b'\x00' * file_size)
        f.flush()
        os.fsync(f.fileno())

        # 第 2 轮：覆写 \xFF（必须 seek 回开头，否则是追加）
        f.seek(0)
        f.truncate()
        f.write(b'\xFF' * file_size)
        f.flush()
        os.fsync(f.fileno())

        # 第 3 轮：覆写随机数据
        f.seek(0)
        f.truncate()
        f.write(os.urandom(file_size))
        f.flush()
        os.fsync(f.fileno())

    # 删除文件
    os.unlink(filepath)

from PyKCS11 import PyKCS11Lib, Mechanism, KeyType

class HSMKeyManager:
    """通过 PKCS#11 接口管理 HSM 中的国密密钥"""

    def __init__(self, lib_path: str, slot: int, pin: str):
        self.pkcs11 = PyKCS11Lib()
        self.pkcs11.load(lib_path)
        self.slot = self.pkcs11.getSlotList(tokenPresent=True)[slot]
        self.session = self.pkcs11.openSession(self.slot)
        self.session.login(pin)

    def generate_sm2_keypair(self, label: str) -> tuple:
        """在 HSM 内部生成 SM2 密钥对，私钥永不离开 HSM"""
        # SM2 密钥属性
        public_key_template = {
            'mechanism': 'SM2_KEY_PAIR_GEN',
            'label': label + '_pub',
            'token': True,        # 持久化存储
            'verify': True,       # 可用于验签
            'encrypt': True,      # 可用于加密
        }

        private_key_template = {
            'label': label + '_priv',
            'token': True,
            'sensitive': True,    # 敏感密钥
            'extractable': False,  # 不可导出
            'sign': True,
            'decrypt': True,
        }

        # 生成密钥对（实际属性需根据具体 HSM 调整）
        pub_key, priv_key = self.session.generateKeyPair(
            public_key_template, private_key_template
        )
        return pub_key, priv_key

    def sign_with_sm2(self, data: bytes, key_label: str) -> bytes:
        """使用 HSM 中的 SM2 私钥签名"""
        # 先做 SM3 哈希（依据 GM/T 0004-2012《SM3 密码杂凑算法》）
        from gmssl import sm3, func
        hash_value = sm3.sm3_hash(func.bytes_to_list(data))

        # 在 HSM 中签名
        priv_key = self.session.findObjects({
            'label': key_label + '_priv',
        })[0]

        signature = self.session.sign(
            priv_key,
            hash_value,
            Mechanism('SM2_SIGN')
        )
        return bytes(signature)

┌──────────────┐     ┌──────────────┐
│   HSM 主节点   │◄───►│   HSM 备节点   │
│  (Active)     │ 同步  │  (Standby)    │
└──────┬───────┘     └──────┬───────┘
       │                    │
       └────────┬───────────┘
                │
        ┌───────▼───────┐
        │  负载均衡层     │
        │  (PKCS#11 代理) │
        └───────┬───────┘
                │
        ┌───────▼───────┐
        │   应用服务      │
        └───────────────┘

### 密钥生成
- [ ] 使用密码学安全随机数生成器（secrets / /dev/urandom / HSM）
- [ ] 私钥生成后立即加密或存入 HSM
- [ ] 密钥生成过程有审计日志

### 密钥存储
- [ ] 生产环境不存在明文密钥
- [ ] 信封加密的 KEK 与 DEK 分开存储
- [ ] 密钥访问有权限控制（最小权限原则）

### 密钥轮换
- [ ] 有自动轮换机制（定时或按事件触发）
- [ ] 轮换过程不需要停机
- [ ] 历史密钥保留用于验签（SM2）或解密旧数据（SM4）

### 密钥吊销
- [ ] 有明确的吊销流程和响应时间
- [ ] CRL 端点可访问且定期更新
- [ ] 应用层有吊销检查机制

### 密钥销毁
- [ ] 密钥销毁有安全擦除流程
- [ ] 销毁记录可追溯
- [ ] 内存中的密钥使用后立即清除

### 监控与审计
- [ ] 所有密钥操作有审计日志
- [ ] 异常访问有告警（如单用户短时间大量签名）
- [ ] 密钥使用量有监控（防止资源耗尽攻击）