GM/T 0132-2023 信息系统密码应用实施指南：从标准到工程落地的实战手册

┌─────────────────────────────────────────────┐
│  第一层：基础算法标准                          │
│  GM/T 0003-2012 (SM2) / GM/T 0004-2012 (SM3)    │
│  GM/T 0002-2012 (SM4)                            │
├─────────────────────────────────────────────┤
│  第二层：应用要求标准                          │
│  GM/T 0054-2018 信息系统密码应用基本要求       │
│  GM/T 0028-2014 密码模块安全技术要求           │
│  GM/T 0034-2014 CA 证书认证系统规范            │
├─────────────────────────────────────────────┤
│  第三层：实施指南标准                          │
│  GM/T 0132-2023 信息系统密码应用实施指南  ◄── 本文 │
└─────────────────────────────────────────────┘

┌──────────────────────────────────────┐
│  根密钥（Key Encryption Key, KEK）    │
│  用途：保护下层密钥                    │
│  存储：HSM 内部，永不导出              │
├──────────────────────────────────────┤
│  密钥加密密钥（Key Wrapping Key）      │
│  用途：加密工作密钥                    │
│  存储：加密后存于数据库/配置文件        │
├──────────────────────────────────────┤
│  工作密钥（Data Encryption Key, DEK）  │
│  用途：直接加密业务数据                │
│  存储：信封加密，随密文一起存储         │
└──────────────────────────────────────┘

from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives import hashes, padding
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
import os

# 验证 SM4-CBC 支持
key = os.urandom(16)  # SM4-128
iv = os.urandom(16)
cipher = Cipher(algorithms.SM4(key), modes.CBC(iv))
encryptor = cipher.encryptor()
decryptor = cipher.decryptor()
print("SM4-CBC 支持: OK")

# 验证 SM3 哈希支持
digest = hashes.Hash(hashes.SM3())
digest.update(b"test")
print("SM3 支持:", digest.finalize().hex()[:16] + "...")

"""
GM/T 0132-2023 三层密钥体系实施示例
符合 GM/T 0002-2012 / GM/T 0003-2012 / GM/T 0004-2012

加密方案：SM4-CBC + HMAC-SM3 (Encrypt-then-MAC)
此构造在密码学强度上等价于 AEAD，且不依赖 Tongsuo/BabaSSL。

⚠️ 安全注意：本示例为教学演示。生产环境应：
  1. 使用独立的加密密钥和 MAC 密钥（通过 HKDF 从根密钥派生）
  2. 使用物理 HSM 存储根密钥
  3. 所有密钥操作在 HSM 边界内完成
"""
import os
import time
from dataclasses import dataclass
from typing import Optional
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives import hashes, padding
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.primitives.hmac import HMAC


@dataclass
class KeyMetadata:
    """密钥元数据 — 用于审计和生命周期管理"""
    key_id: str
    key_type: str          # root / wrapping / working
    purpose: str           # 用途说明
    created_at: float      # 创建时间戳
    expires_at: float      # 过期时间戳
    is_active: bool = True
    rotation_count: int = 0


def _derive_subkeys(root_key: bytes, context: bytes) -> tuple:
    """
    从根密钥派生加密密钥和 MAC 密钥。
    生产环境应使用 HKDF 派生独立密钥，而非复用根密钥。
    """
    enc_key = HKDF(
        algorithm=hashes.SM3(),
        length=16,
        salt=None,
        info=b"enc:" + context,
    ).derive(root_key)
    mac_key = HKDF(
        algorithm=hashes.SM3(),
        length=16,
        salt=None,
        info=b"mac:" + context,
    ).derive(root_key)
    return enc_key, mac_key


class RootKeyStore:
    """
    根密钥存储 — 模拟 HSM 行为
    
    生产环境应使用物理 HSM（如信安世纪、渔人等国密密码机），
    根密钥永不离开 HSM 边界。本示例用内存模拟，仅用于演示。
    """
    
    def __init__(self):
        # 根密钥：SM4-128 密钥，用于加密工作密钥
        self._root_key = os.urandom(16)
        self._metadata = KeyMetadata(
            key_id="root-key-001",
            key_type="root",
            purpose="HSM根密钥-保护工作密钥",
            created_at=time.time(),
            expires_at=time.time() + 365 * 86400 * 2,  # 2年有效期
        )
    
    def wrap_key(self, plaintext_key: bytes, context: bytes) -> bytes:
        """用根密钥加密工作密钥（信封加密，Encrypt-then-MAC）"""
        enc_key, mac_key = _derive_subkeys(self._root_key, context)
        
        # SM4-CBC 加密
        iv = os.urandom(16)
        cipher = Cipher(algorithms.SM4(enc_key), modes.CBC(iv))
        encryptor = cipher.encryptor()
        padder = padding.PKCS7(128).padder()
        padded_data = padder.update(plaintext_key) + padder.finalize()
        ciphertext = encryptor.update(padded_data) + encryptor.finalize()
        
        # HMAC-SM3 完整性校验（AAD = context）
        h = HMAC(mac_key, hashes.SM3())
        h.update(iv + ciphertext + context)
        mac_tag = h.finalize()
        
        return iv + ciphertext + mac_tag  # iv || ciphertext || tag
    
    def unwrap_key(self, wrapped_key: bytes, context: bytes) -> bytes:
        """解密被信封加密的工作密钥"""
        iv = wrapped_key[:16]
        ciphertext = wrapped_key[16:-32]
        mac_tag = wrapped_key[-32:]
        
        enc_key, mac_key = _derive_subkeys(self._root_key, context)
        
        # 验证 MAC
        h = HMAC(mac_key, hashes.SM3())
        h.update(iv + ciphertext + context)
        try:
            h.verify(mac_tag)
        except Exception:
            raise ValueError("MAC 验证失败：密钥已被篡改或上下文不匹配")
        
        # SM4-CBC 解密
        cipher = Cipher(algorithms.SM4(enc_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
    
    @property
    def metadata(self) -> KeyMetadata:
        return self._metadata


class WorkingKeyManager:
    """
    工作密钥管理器 — 实现密钥生成、信封加密存储、轮换
    
    符合 GM/T 0054-2018 的密钥管理要求：
    - 工作密钥应加密存储
    - 支持密钥轮换
    - 密钥销毁不可恢复
    """
    
    def __init__(self, root_store: RootKeyStore):
        self._root = root_store
        self._keys: dict[str, tuple[KeyMetadata, bytes]] = {}
    
    def generate_working_key(self, key_id: str, purpose: str, ttl_days: int = 180) -> str:
        """生成新的工作密钥"""
        plaintext_key = os.urandom(16)  # SM4-128
        context = f"working-key:{key_id}:{purpose}".encode()
        wrapped_key = self._root.wrap_key(plaintext_key, context)
        
        meta = KeyMetadata(
            key_id=key_id,
            key_type="working",
            purpose=purpose,
            created_at=time.time(),
            expires_at=time.time() + ttl_days * 86400,
        )
        
        self._keys[key_id] = (meta, wrapped_key)
        return key_id
    
    def get_working_key(self, key_id: str) -> Optional[bytes]:
        """获取明文工作密钥（仅在内存中解密）"""
        if key_id not in self._keys:
            return None
        
        meta, wrapped_key = self._keys[key_id]
        if time.time() > meta.expires_at:
            meta.is_active = False
            return None
        
        context = f"working-key:{key_id}:{meta.purpose}".encode()
        return self._root.unwrap_key(wrapped_key, context)
    
    def rotate_key(self, key_id: str) -> str:
        """密钥轮换 — 生成新密钥，旧密钥保留用于解密历史数据"""
        if key_id not in self._keys:
            raise KeyError(f"密钥不存在: {key_id}")
        
        old_meta, _ = self._keys[key_id]
        old_meta.is_active = False
        
        new_key_id = f"{key_id}-v{old_meta.rotation_count + 1}"
        self.generate_working_key(new_key_id, old_meta.purpose)
        return new_key_id
    
    def destroy_key(self, key_id: str) -> None:
        """安全销毁密钥"""
        if key_id in self._keys:
            meta, wrapped_key = self._keys[key_id]
            self._keys[key_id] = (meta, b'\x00' * len(wrapped_key))
            del self._keys[key_id]
    
    def list_keys(self) -> list[dict]:
        result = []
        for key_id, (meta, _) in self._keys.items():
            result.append({
                "key_id": meta.key_id,
                "key_type": meta.key_type,
                "purpose": meta.purpose,
                "created_at": time.strftime("%Y-%m-%d %H:%M", time.localtime(meta.created_at)),
                "expires_at": time.strftime("%Y-%m-%d %H:%M", time.localtime(meta.expires_at)),
                "is_active": meta.is_active,
                "rotation_count": meta.rotation_count,
            })
        return result

class FieldEncryptionService:
    """
    字段级加密服务 — 实现数据加密存储和完整性校验
    
    加密方案：SM4-CBC + HMAC-SM3（Encrypt-then-MAC）
    密钥派生：HKDF-SM3（符合 GM/T 0004-2012）
    """
    
    def __init__(self, key_manager: WorkingKeyManager, key_id: str):
        self._km = key_manager
        self._key_id = key_id
    
    def _derive_field_key(self, field_name: str, record_id: str) -> bytes:
        """
        派生字段专属密钥 — 不同字段使用不同密钥，防止密文挪移攻击
        
        使用 HKDF-SM3，salt 基于字段名和记录 ID 确定性派生
        （salt 必须确定性，否则每次加密产生不同密钥导致解密失败）
        """
        master_key = self._km.get_working_key(self._key_id)
        if not master_key:
            raise RuntimeError("工作密钥不可用或已过期")
        
        # 确定性 salt：基于字段名和记录 ID
        salt = HKDF(
            algorithm=hashes.SM3(),
            length=16,
            salt=None,
            info=f"salt:{field_name}:{record_id}".encode("utf-8"),
        ).derive(b"deterministic-salt-seed")
        
        # HKDF-SM3 派生字段专属密钥
        field_key = HKDF(
            algorithm=hashes.SM3(),
            length=16,
            salt=salt,
            info=f"field:{field_name}:record:{record_id}".encode("utf-8"),
        ).derive(master_key)
        
        return field_key
    
    def encrypt_field(self, plaintext: str, field_name: str, record_id: str) -> str:
        """
        加密字段值
        
        返回格式: base64(iv || ciphertext || hmac_tag)
        """
        field_key = self._derive_field_key(field_name, record_id)
        iv = os.urandom(16)
        
        # SM4-CBC 加密
        cipher = Cipher(algorithms.SM4(field_key), modes.CBC(iv))
        encryptor = cipher.encryptor()
        padder = padding.PKCS7(128).padder()
        padded = padder.update(plaintext.encode("utf-8")) + padder.finalize()
        ciphertext = encryptor.update(padded) + encryptor.finalize()
        
        # HMAC-SM3 完整性校验
        aad = f"{field_name}:{record_id}".encode("utf-8")
        h = HMAC(field_key, hashes.SM3())
        h.update(iv + ciphertext + aad)
        mac_tag = h.finalize()
        
        # 组合: iv (16B) || ciphertext || mac_tag (32B)
        return self._encode(iv + ciphertext + mac_tag)
    
    def decrypt_field(self, encrypted: str, field_name: str, record_id: str) -> str:
        """
        解密字段值
        
        解密失败（密文被篡改、密钥错误、AAD 不匹配）会抛出 InvalidSignature 异常
        """
        field_key = self._derive_field_key(field_name, record_id)
        raw = self._decode(encrypted)
        
        iv = raw[:16]
        ciphertext = raw[16:-32]
        mac_tag = raw[-32:]
        
        # 验证 MAC
        aad = f"{field_name}:{record_id}".encode("utf-8")
        h = HMAC(field_key, hashes.SM3())
        h.update(iv + ciphertext + aad)
        try:
            h.verify(mac_tag)
        except Exception:
            raise ValueError("MAC 验证失败：密文已被篡改或上下文不匹配")
        
        # SM4-CBC 解密
        cipher = Cipher(algorithms.SM4(field_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")
    
    @staticmethod
    def _encode(data: bytes) -> str:
        return __import__('base64').b64encode(data).decode('ascii')
    
    @staticmethod
    def _decode(data: str) -> bytes:
        return __import__('base64').b64decode(data)

def main():
    # 初始化根密钥存储（模拟 HSM）
    root_store = RootKeyStore()
    
    # 初始化工作密钥管理器
    km = WorkingKeyManager(root_store)
    
    # 创建工作密钥
    km.generate_working_key("dek-order-phone", "订单表手机号字段加密", ttl_days=180)
    print("✅ 工作密钥已创建")
    
    # 初始化字段加密服务
    svc = FieldEncryptionService(km, "dek-order-phone")
    
    # 加密测试
    plaintext = "1*********8"
    field_name = "phone"
    record_id = "order-20260801-001"
    
    encrypted = svc.encrypt_field(plaintext, field_name, record_id)
    print(f"明文: {plaintext}")
    print(f"密文: {encrypted}")
    
    # 解密测试
    decrypted = svc.decrypt_field(encrypted, field_name, record_id)
    print(f"解密: {decrypted}")
    assert decrypted == plaintext, "解密结果与原文不一致！"
    print("✅ 加解密验证通过")
    
    # 密钥轮换测试
    new_key_id = km.rotate_key("dek-order-phone")
    print(f"🔄 密钥轮换完成: {new_key_id}")
    
    # 列出密钥
    print("\n密钥清单:")
    for k in km.list_keys():
        status = "活跃" if k["is_active"] else "已轮换"
        print(f"  [{status}] {k['key_id']} | {k['purpose']} | 过期: {k['expires_at']}")


if __name__ == "__main__":
    main()