国密改造 Kubernetes Secret 存储：SM4 加密的完整实践指南

# /etc/kubernetes/encryption-config.yaml
apiVersion: apiserver.config.k8s.io/v1
kind: EncryptionConfig
resources:
  - resources:
      - secrets
    providers:
      - kms:
          name: sm4-kms-plugin
          endpoint: unix:///var/run/kms-plugin/plugin.sock
          timeout: 3s
          cacheSize: 1000
      - identity: {}  # 兜底：不加密

"""
SM4 KMS Plugin - Kubernetes KMS gRPC 服务实现
依赖: pip install grpcio grpcio-tools gmssl
"""
import os
import base64

from gmssl import sm4, func


def sm4_pkcs7_pad(data: bytes, block_size: int = 16) -> bytes:
    """SM4 PKCS#7 填充"""
    pad_len = block_size - (len(data) % block_size)
    return data + bytes([pad_len] * pad_len)


def sm4_pkcs7_unpad(data: bytes) -> bytes:
    """SM4 PKCS#7 去填充"""
    pad_len = data[-1]
    if pad_len < 1 or pad_len > 16:
        raise ValueError("Invalid PKCS#7 padding")
    return data[:-pad_len]


def sm4_ecb_encrypt_block(key: bytes, block: bytes) -> bytes:
    """SM4 ECB 单块加密（使用 gmssl 库的 one_round 接口）"""
    crypt = sm4.CryptSM4()
    crypt.set_key(key, sm4.SM4_ENCRYPT)
    return bytes(crypt.one_round(crypt.sk, list(block)))


def sm4_ecb_decrypt_block(key: bytes, block: bytes) -> bytes:
    """SM4 ECB 单块解密"""
    crypt = sm4.CryptSM4()
    crypt.set_key(key, sm4.SM4_DECRYPT)
    return bytes(crypt.one_round(crypt.sk, list(block)))


def sm4_cbc_encrypt(key: bytes, iv: bytes, plaintext: bytes) -> bytes:
    """SM4 CBC 模式加密"""
    padded = sm4_pkcs7_pad(plaintext)
    prev = iv
    ciphertext = b""
    for i in range(0, len(padded), 16):
        block = padded[i:i + 16]
        xored = bytes(a ^ b for a, b in zip(block, prev))
        encrypted = sm4_ecb_encrypt_block(key, xored)
        ciphertext += encrypted
        prev = encrypted
    return ciphertext


def sm4_cbc_decrypt(key: bytes, iv: bytes, ciphertext: bytes) -> bytes:
    """SM4 CBC 模式解密"""
    prev = iv
    plaintext = b""
    for i in range(0, len(ciphertext), 16):
        block = ciphertext[i:i + 16]
        decrypted = sm4_ecb_decrypt_block(key, block)
        plain = bytes(a ^ b for a, b in zip(decrypted, prev))
        plaintext += plain
        prev = block
    return sm4_pkcs7_unpad(plaintext)


def generate_sm4_key() -> str:
    """生成随机 SM4 密钥（16 字节 = 128 位），base64 编码"""
    key = os.urandom(16)
    return base64.b64encode(key).decode("utf-8")


def generate_key_identity() -> str:
    """生成密钥标识符（用于密钥轮换）"""
    return base64.b64encode(os.urandom(16)).decode("utf-8")


if __name__ == "__main__":
    # 生成密钥
    key_b64 = generate_sm4_key()
    key = base64.b64decode(key_b64)
    identity = generate_key_identity()
    print(f"SM4 Key (base64): {key_b64}")
    print(f"Key Identity: {identity}")
    print(f"Key length: {len(key)} bytes = {len(key) * 8} bits")

    # 加密测试
    iv = os.urandom(16)
    secret_data = b'{"password": "secret123", "token": "abc"}'
    encrypted = sm4_cbc_encrypt(key, iv, secret_data)
    decrypted = sm4_cbc_decrypt(key, iv, encrypted)

    print(f"\n加密测试:")
    print(f"  原始: {secret_data}")
    print(f"  密文 (base64): {base64.b64encode(encrypted).decode()}")
    print(f"  解密: {decrypted}")
    print(f"  验证: {secret_data == decrypted}")

    print(f"\n请妥善保存以上密钥，丢失后将无法解密已加密数据")

"""
SM4 加密工具 - 用于应用层加密 Secret 数据
依赖: pip install gmssl
"""
import os
import base64
from gmssl import sm4


def sm4_pkcs7_pad(data: bytes, block_size: int = 16) -> bytes:
    pad_len = block_size - (len(data) % block_size)
    return data + bytes([pad_len] * pad_len)


def sm4_pkcs7_unpad(data: bytes) -> bytes:
    pad_len = data[-1]
    if pad_len < 1 or pad_len > 16:
        raise ValueError("Invalid PKCS#7 padding")
    return data[:-pad_len]


def sm4_encrypt(key_b64: str, plaintext: str, iv_b64: str = None) -> dict:
    """
    SM4-CBC 加密
    
    Args:
        key_b64: base64 编码的 SM4 密钥（16 字节）
        plaintext: 待加密的明文字符串
        iv_b64: 可选的 base64 编码 IV（16 字节），不提供则随机生成
    
    Returns:
        {
            "iv": base64 编码的 IV,
            "ciphertext": base64 编码的密文
        }
    """
    key = base64.b64decode(key_b64)
    if len(key) != 16:
        raise ValueError(f"SM4 密钥长度必须为 16 字节，实际为 {len(key)} 字节")
    
    iv = base64.b64decode(iv_b64) if iv_b64 else os.urandom(16)
    plaintext_bytes = plaintext.encode("utf-8")
    
    # CBC 模式加密
    prev = iv
    ciphertext = b""
    padded = sm4_pkcs7_pad(plaintext_bytes)
    for i in range(0, len(padded), 16):
        block = padded[i:i + 16]
        xored = bytes(a ^ b for a, b in zip(block, prev))
        crypt = sm4.CryptSM4()
        crypt.set_key(key, sm4.SM4_ENCRYPT)
        enc_block = bytes(crypt.one_round(crypt.sk, list(xored)))
        ciphertext += enc_block
        prev = enc_block
    
    return {
        "iv": base64.b64encode(iv).decode("utf-8"),
        "ciphertext": base64.b64encode(ciphertext).decode("utf-8")
    }


def sm4_decrypt(key_b64: str, iv_b64: str, ciphertext_b64: str) -> str:
    """SM4-CBC 解密"""
    key = base64.b64decode(key_b64)
    iv = base64.b64decode(iv_b64)
    ciphertext = base64.b64decode(ciphertext_b64)
    
    prev = iv
    plaintext = b""
    for i in range(0, len(ciphertext), 16):
        block = ciphertext[i:i + 16]
        crypt = sm4.CryptSM4()
        crypt.set_key(key, sm4.SM4_DECRYPT)
        dec_block = bytes(crypt.one_round(crypt.sk, list(block)))
        plain = bytes(a ^ b for a, b in zip(dec_block, prev))
        plaintext += plain
        prev = block
    
    return sm4_pkcs7_unpad(plaintext).decode("utf-8")


# 使用示例
if __name__ == "__main__":
    # 生成密钥（生产环境应从 KMS 获取）
    key = base64.b64encode(os.urandom(16)).decode("utf-8")
    print(f"SM4 Key: {key}")
    
    secret = '{"username": "admin", "password": "s3cret!"}'
    result = sm4_encrypt(key, secret)
    decrypted = sm4_decrypt(key, result["iv"], result["ciphertext"])
    
    print(f"加密: {result}")
    print(f"解密: {decrypted}")
    assert secret == decrypted, "加解密验证失败"
    print("加解密验证通过")

apiVersion: apiserver.config.k8s.io/v1
kind: EncryptionConfig
resources:
  - resources:
      - secrets
    providers:
      - kms:
          name: sm4-kms-v2
          keyId: sm4-key-2026-07
          endpoint: unix:///var/run/kms-plugin/plugin.sock
      - kms:
          name: sm4-kms-v1
          keyId: sm4-key-2026-01
          endpoint: unix:///var/run/kms-plugin/plugin.sock
      - identity: {}