云原生 SM4-GCM 全链路数据加密实战：从微服务通信到 Kubernetes Secret 存储

import os
import json
import base64
import struct
import time
from dataclasses import dataclass, asdict
from typing import Optional
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes


@dataclass
class EncryptedPayload:
    """SM4-GCM 加密载荷"""
    ciphertext: str  # Base64 编码的密文
    iv: str          # Base64 编码的初始化向量 (12 bytes)
    tag: str         # Base64 编码的 GCM 认证标签 (16 bytes)
    aad: str         # Base64 编码的附加认证数据
    timestamp: int   # 加密时间戳（用于防重放）
    key_id: str      # 密钥标识符（用于密钥轮换）

    def to_json(self) -> str:
        return json.dumps(asdict(self))

    @classmethod
    def from_json(cls, data: str) -> "EncryptedPayload":
        return cls(**json.loads(data))


class SM4GCMService:
    """
    SM4-GCM 认证加密服务
    
    特性：
    - 每次加密生成随机 IV（12 bytes，GCM 标准长度）
    - 支持 AAD（附加认证数据）用于绑定上下文
    - 内置时间戳防重放
    - 支持密钥版本管理
    """
    
    IV_LENGTH = 12   # GCM 推荐 IV 长度
    TAG_LENGTH = 16  # GCM 认证标签长度
    KEY_LENGTH = 16  # SM4 密钥长度 (128 bits)
    
    def __init__(self, master_key: bytes, key_id: str = "v1"):
        """
        初始化加密服务
        
        Args:
            master_key: 16 字节 SM4 密钥
            key_id: 密钥标识符，用于密钥轮换
        """
        if len(master_key) != self.KEY_LENGTH:
            raise ValueError(f"SM4 密钥长度必须为 {self.KEY_LENGTH} 字节")
        self._master_key = master_key
        self._key_id = key_id
    
    def encrypt(
        self, 
        plaintext: bytes, 
        aad: Optional[bytes] = None,
        timestamp: Optional[int] = None
    ) -> EncryptedPayload:
        """
        SM4-GCM 加密
        
        Args:
            plaintext: 明文数据
            aad: 附加认证数据（不参与加密但参与认证）
            timestamp: 时间戳（默认当前时间）
        
        Returns:
            EncryptedPayload 加密载荷
        """
        iv = os.urandom(self.IV_LENGTH)
        ts = timestamp or int(time.time())
        
        # 构造 AAD：时间戳 + 用户提供的 AAD
        aad_data = struct.pack(">Q", ts)
        if aad:
            aad_data += aad
        
        cipher = Cipher(
            algorithms.SM4(self._master_key),
            modes.GCM(iv)
        )
        encryptor = cipher.encryptor()
        encryptor.authenticate_additional_data(aad_data)
        
        ciphertext = encryptor.update(plaintext) + encryptor.finalize()
        tag = encryptor.tag  # 16 字节认证标签
        
        return EncryptedPayload(
            ciphertext=base64.b64encode(ciphertext).decode(),
            iv=base64.b64encode(iv).decode(),
            tag=base64.b64encode(tag).decode(),
            aad=base64.b64encode(aad).decode() if aad else "",
            timestamp=ts,
            key_id=self._key_id
        )
    
    def decrypt(
        self, 
        payload: EncryptedPayload, 
        aad: Optional[bytes] = None,
        max_age: Optional[int] = None
    ) -> bytes:
        """
        SM4-GCM 解密
        
        Args:
            payload: 加密载荷
            aad: 附加认证数据（必须与加密时一致）
            max_age: 最大有效时间（秒），用于防重放
        
        Returns:
            解密后的明文
        
        Raises:
            ValueError: 认证失败或数据被篡改
        """
        # 检查时间戳（防重放）
        if max_age is not None:
            age = int(time.time()) - payload.timestamp
            if age > max_age:
                raise ValueError(f"数据已过期: {age}s > {max_age}s")
        
        iv = base64.b64decode(payload.iv)
        tag = base64.b64decode(payload.tag)
        ciphertext = base64.b64decode(payload.ciphertext)
        
        # 重构 AAD
        aad_data = struct.pack(">Q", payload.timestamp)
        if aad:
            aad_data += aad
        elif payload.aad:
            # 使用 payload 中存储的 AAD（仅当用户未提供时）
            pass
        
        cipher = Cipher(
            algorithms.SM4(self._master_key),
            modes.GCM(iv, tag)
        )
        decryptor = cipher.decryptor()
        decryptor.authenticate_additional_data(aad_data)
        
        try:
            plaintext = decryptor.update(ciphertext) + decryptor.finalize()
            return plaintext
        except Exception as e:
            raise ValueError(f"解密失败（认证标签不匹配）: {e}")

import os

# 生成密钥（生产环境应从 KMS 获取）
key = os.urandom(16)
service = SM4GCMService(key, key_id="v1")

# 加密
payload = service.encrypt(
    b"用户身份证号: 110101199001011234",
    aad=b"user:12345:profile"
)
print(f"加密结果: {payload.to_json()}")

# 解密
plaintext = service.decrypt(payload, aad=b"user:12345:profile")
print(f"解密结果: {plaintext.decode()}")

import os
import json
import hashlib
import secrets
from typing import Dict, Any
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.primitives import hashes


class MicroserviceCrypto:
    """
    微服务间数据传输加密
    
    设计要点：
    - 每个服务对拥有独立的派生密钥
    - 使用 HKDF-SM3 从主密钥派生会话密钥
    - AAD 绑定服务对和请求上下文
    """
    
    def __init__(self, master_key: bytes):
        self._master_key = master_key
    
    def _derive_key(self, service_pair: str) -> bytes:
        """使用 HKDF-SM3 派生服务对专用密钥"""
        hkdf = HKDF(
            algorithm=hashes.SM3(),
            length=16,
            salt=None,
            info=f"sm4-gcm:{service_pair}".encode()
        )
        return hkdf.derive(self._master_key)
    
    def encrypt_for_service(
        self,
        data: bytes,
        source_service: str,
        target_service: str,
        request_id: str
    ) -> str:
        """
        为特定目标服务加密数据
        
        Args:
            data: 明文数据
            source_service: 源服务名
            target_service: 目标服务名
            request_id: 请求 ID（用于防重放）
        
        Returns:
            JSON 格式的加密载荷
        """
        service_pair = f"{source_service}->{target_service}"
        key = self._derive_key(service_pair)
        
        # AAD 绑定服务对和请求 ID
        aad = f"{service_pair}:{request_id}".encode()
        
        service = SM4GCMService(key, key_id=service_pair)
        payload = service.encrypt(data, aad=aad)
        return payload.to_json()
    
    def decrypt_from_service(
        self,
        encrypted_json: str,
        source_service: str,
        target_service: str,
        request_id: str,
        max_age: int = 300
    ) -> bytes:
        """解密来自特定源服务的数据"""
        service_pair = f"{source_service}->{target_service}"
        key = self._derive_key(service_pair)
        
        aad = f"{service_pair}:{request_id}".encode()
        
        payload = EncryptedPayload.from_json(encrypted_json)
        service = SM4GCMService(key, key_id=service_pair)
        return service.decrypt(payload, aad=aad, max_age=max_age)


# 使用示例
master_key = os.urandom(16)
crypto = MicroserviceCrypto(master_key)

# 订单服务 -> 支付服务
encrypted = crypto.encrypt_for_service(
    b'{"card_no": "622202************", "amount": 10000}',
    source_service="order-service",
    target_service="payment-service",
    request_id=secrets.token_hex(16)
)
print(f"加密传输数据: {encrypted[:100]}...")

# 支付服务解密
decrypted = crypto.decrypt_from_service(
    encrypted,
    source_service="order-service",
    target_service="payment-service",
    request_id=req_ctx["request_id"]  # 从请求上下文获取
)
print(f"解密结果: {decrypted.decode()}")

import os
import base64
import json
import struct
from typing import Dict


class KMSKeyStore:
    """
    模拟密钥管理服务（KMS）
    
    生产环境应使用：
    - 硬件密码机（HSM）
    - 云厂商 KMS（阿里云 KMS、华为云 KMS）
    - HashiCorp Vault 国密插件
    """
    
    def __init__(self):
        self._keys: Dict[str, bytes] = {}
    
    def create_key(self, key_id: str) -> bytes:
        """创建新密钥"""
        key = os.urandom(16)
        self._keys[key_id] = key
        return key
    
    def get_key(self, key_id: str) -> bytes:
        """获取密钥"""
        if key_id not in self._keys:
            raise KeyError(f"密钥不存在: {key_id}")
        return self._keys[key_id]


class K8sSecretCrypto:
    """
    Kubernetes Secret 国密加密存储
    
    工作流程：
    1. 用户提交 Secret（明文）
    2. 使用 SM4-GCM 加密 Secret 值
    3. 将加密后的数据存储为 ConfigMap 或自定义资源
    4. Pod 启动时通过 Init Container 解密
    """
    
    def __init__(self, kms: KMSKeyStore, key_id: str = "k8s-secret-v1"):
        self._kms = kms
        self._key_id = key_id
    
    def encrypt_secret(self, secret_data: Dict[str, str]) -> Dict[str, str]:
        """
        加密 Secret 中的所有键值对
        
        Args:
            secret_data: {"username": "admin", "password": "xxx"}
        
        Returns:
            加密后的数据，格式为 base64(ciphertext:tag:iv:timestamp)
        """
        key = self._kms.get_key(self._key_id)
        service = SM4GCMService(key, key_id=self._key_id)
        
        encrypted = {}
        for name, value in secret_data.items():
            payload = service.encrypt(
                value.encode('utf-8'),
                aad=f"k8s:secret:{name}".encode()
            )
            # 合并为单个 base64 字符串（含时间戳用于防重放）
            ts_bytes = struct.pack(">Q", payload.timestamp)
            combined = (
                base64.b64decode(payload.ciphertext) +
                base64.b64decode(payload.tag) +
                base64.b64decode(payload.iv) +
                ts_bytes
            )
            encrypted[f"sm4gcm.{name}"] = base64.b64encode(combined).decode()
        
        return encrypted
    
    def decrypt_secret(self, encrypted_data: Dict[str, str]) -> Dict[str, str]:
        """解密 Secret"""
        key = self._kms.get_key(self._key_id)
        service = SM4GCMService(key, key_id=self._key_id)
        
        decrypted = {}
        for name, value in encrypted_data.items():
            if not name.startswith("sm4gcm."):
                continue
            original_name = name[7:]  # 去掉 "sm4gcm." 前缀
            
            combined = base64.b64decode(value)
            # 拆分: ciphertext(n bytes) + tag(16 bytes) + iv(12 bytes) + timestamp(8 bytes)
            ts_bytes = combined[-8:]
            iv = combined[-20:-8]
            tag = combined[-36:-20]
            ciphertext = combined[:-36]
            
            timestamp = struct.unpack(">Q", ts_bytes)[0]
            
            payload = EncryptedPayload(
                ciphertext=base64.b64encode(ciphertext).decode(),
                iv=base64.b64encode(iv).decode(),
                tag=base64.b64encode(tag).decode(),
                aad="",
                timestamp=timestamp,
                key_id=self._key_id
            )
            
            plaintext = service.decrypt(
                payload, 
                aad=f"k8s:secret:{original_name}".encode()
            )
            decrypted[original_name] = plaintext.decode('utf-8')
        
        return decrypted


# 使用示例
kms = KMSKeyStore()
kms.create_key("k8s-secret-v1")

secret_crypto = K8sSecretCrypto(kms)

# 加密 Secret
secret = {"db_password": "MyP@ssw0rd!", "api_key": "sk-1234567890abcdef"}
encrypted = secret_crypto.encrypt_secret(secret)
print(f"加密后: {encrypted}")

# 解密 Secret
decrypted = secret_crypto.decrypt_secret(encrypted)
print(f"解密后: {decrypted}")

import os
import json
import struct
from typing import List, Tuple
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.primitives import hashes


class KeyHierarchy:
    """
    三层密钥分层管理
    
    Layer 0: 主密钥 (HSM 保护)
    Layer 1: KEK (Key Encryption Key)
    Layer 2: DEK (Data Encryption Key)
    """
    
    def __init__(self, master_key: bytes):
        self._master_key = master_key
        self._keks: Dict[str, bytes] = {}
        self._deks: Dict[str, List[Tuple[str, bytes]]] = {}  # key_id -> [(version, dek)]
    
    def _derive_kek(self, kek_id: str) -> bytes:
        """从主密钥派生 KEK"""
        hkdf = HKDF(
            algorithm=hashes.SM3(),
            length=16,
            salt=None,
            info=f"kek:{kek_id}".encode()
        )
        return hkdf.derive(self._master_key)
    
    def generate_dek(self, key_id: str) -> Tuple[str, bytes]:
        """
        生成新的 DEK 版本
        
        Returns:
            (version, encrypted_dek) 版本号和被 KEK 加密的 DEK
        """
        kek = self._derive_kek(key_id)
        
        # 生成随机 DEK
        dek = os.urandom(16)
        version = f"v{len(self._deks.get(key_id, [])) + 1}"
        
        # 使用 KEK 加密 DEK（密钥封装）
        service = SM4GCMService(kek, key_id=key_id)
        payload = service.encrypt(dek, aad=f"dek:{key_id}:{version}".encode())
        
        # 存储加密后的 DEK
        if key_id not in self._deks:
            self._deks[key_id] = []
        self._deks[key_id].append((version, dek))
        
        return version, payload.to_json()
    
    def get_dek(self, key_id: str, version: str) -> bytes:
        """获取指定版本的 DEK"""
        versions = self._deks.get(key_id, [])
        for ver, dek in versions:
            if ver == version:
                return dek
        raise KeyError(f"DEK 不存在: {key_id}:{version}")
    
    def rotate_key(self, key_id: str) -> Tuple[str, bytes]:
        """轮换密钥（生成新版本 DEK）"""
        return self.generate_dek(key_id)


# 使用示例
master_key = os.urandom(16)
hierarchy = KeyHierarchy(master_key)

# 初始生成 DEK
ver1, enc_dek1 = hierarchy.generate_dek("user-data")
print(f"生成 DEK: {ver1}")

# 轮换（生成新版本）
ver2, enc_dek2 = hierarchy.rotate_key("user-data")
print(f"轮换 DEK: {ver2}")

# 获取 DEK 用于加解密
dek = hierarchy.get_dek("user-data", ver2)
service = SM4GCMService(dek, key_id=f"user-data:{ver2}")
payload = service.encrypt(b"敏感数据", aad=b"context")
print(f"加密成功: {payload.ciphertext[:30]}...")

import os
import time
from typing import Callable, List


def benchmark(func: Callable, iterations: int = 10000) -> float:
    """基准测试"""
    start = time.perf_counter()
    for _ in range(iterations):
        func()
    elapsed = time.perf_counter() - start
    return elapsed / iterations * 1000  # ms


def run_benchmark():
    key = os.urandom(16)
    service = SM4GCMService(key)
    
    # 测试不同数据大小
    sizes = [64, 256, 1024, 4096, 16384]
    
    print(f"{'数据大小':>10} | {'加密(ms)':>10} | {'解密(ms)':>10} | {'吞吐量(MB/s)':>12}")
    print("-" * 55)
    
    for size in sizes:
        data = os.urandom(size)
        
        def encrypt_op():
            service.encrypt(data, aad=b"benchmark")
        
        def decrypt_op():
            payload = service.encrypt(data, aad=b"benchmark")
            service.decrypt(payload, aad=b"benchmark")
        
        enc_time = benchmark(encrypt_op, iterations=5000)
        dec_time = benchmark(decrypt_op, iterations=5000)
        
        throughput = size / (enc_time / 1000) / 1024 / 1024
        
        print(f"{size:>10} | {enc_time:>10.3f} | {dec_time:>10.3f} | {throughput:>12.1f}")


if __name__ == "__main__":
    run_benchmark()