国密通用密码服务接口GM/T 0019-2023实战：从标准解读到Python工程实现

┌──────────────────────────────────────────────────────────────┐
│                   应用层（业务系统）                           │
│   "帮我加密这段数据" / "帮我签名这段消息"                      │
├──────────────────────────────────────────────────────────────┤
│              GM/T 0019-2023 通用密码服务接口                  │
│   Encrypt() / Decrypt() / Sign() / Verify() / Hash() / ...  │
├──────────────────────────────────────────────────────────────┤
│                   密码服务实现层                              │
│   ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐   │
│   │ 软加密库  │  │  HSM 设备  │  │  云密码服务 │  │  硬件加速  │   │
│   │ gmssl    │  │  Thales  │  │  阿里云KMS │  │  Intel SGX│   │
│   └──────────┘  └──────────┘  └──────────┘  └──────────┘   │
├──────────────────────────────────────────────────────────────┤
│                   密码算法标准层                              │
│   GM/T 0002-2012 (SM2) / GM/T 0004-2012 (SM3)               │
│   GM/T 0001-2012 (SM4)                                       │
└──────────────────────────────────────────────────────────────┘

# 输入/输出数据封装
class CryptoRequest:
    algorithm: str           # "SM2" / "SM4" / "SM3"
    operation: str           # "encrypt" / "decrypt" / "sign" / "verify" / "hash"
    key_id: str              # 密钥标识符
    data: bytes              # 输入数据
    params: dict             # 附加参数（IV、AAD 等）

class CryptoResponse:
    success: bool
    result: bytes            # 输出数据
    error_code: int          # 错误码
    error_msg: str           # 错误信息

#!/usr/bin/env python3
"""
GM/T 0019-2023 通用密码服务接口 Python 实现

依赖：pip install gmssl==3.2.2 cryptography>=42.0
"""

from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
from typing import Optional
import os
import hashlib

class Algorithm(str, Enum):
    SM2 = "SM2"
    SM3 = "SM3"
    SM4 = "SM4"


class Operation(str, Enum):
    ENCRYPT = "encrypt"
    DECRYPT = "decrypt"
    SIGN = "sign"
    VERIFY = "verify"
    HASH = "hash"
    KEY_WRAP = "key_wrap"
    KEY_UNWRAP = "key_unwrap"
    GENERATE_KEY = "generate_key"
    GENERATE_RANDOM = "generate_random"


class SM4Mode(str, Enum):
    ECB = "ecb"
    CBC = "cbc"
    CTR = "ctr"
    GCM = "gcm"


# === 数据结构 ===

@dataclass
class CryptoRequest:
    algorithm: Algorithm
    operation: Operation
    key_id: str
    data: bytes
    params: dict  # iv, aad, tag_length 等

    def __post_init__(self):
        if not self.key_id:
            raise ValueError("key_id 不能为空")


@dataclass
class CryptoResponse:
    success: bool
    result: Optional[bytes] = None
    error_code: int = 0
    error_msg: str = ""


# === 错误码常量 ===

ERROR_SUCCESS = 0
ERROR_INVALID_PARAM = -1
ERROR_KEY_NOT_FOUND = -2
ERROR_KEY_EXPIRED = -3
ERROR_KEY_DISABLED = -4
ERROR_ALGORITHM_NOT_SUPPORTED = -100
ERROR_DATA_LENGTH_INVALID = -101
ERROR_HARDWARE_COMM_FAIL = -200

from gmssl import sm2, sm3, func
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend

class SoftwareCryptoProvider:
    """
    基于 gmssl + cryptography 的软加密实现
    
    对应 GM/T 0019-2023 的软件实现层
    """

    def __init__(self):
        # 密钥存储（生产环境应使用 HSM 或密钥管理服务）
        self._keys: dict[str, dict] = {}

    def _get_key(self, key_id: str) -> dict:
        """获取密钥，检查有效性"""
        if key_id not in self._keys:
            return None
        key_info = self._keys[key_id]
        if key_info.get('disabled'):
            raise CryptoError(ERROR_KEY_DISABLED, f"密钥 {key_id} 已被禁用")
        return key_info

    def generate_key(self, request: CryptoRequest) -> CryptoResponse:
        """生成密钥"""
        algo = request.algorithm
        key_type = request.params.get('key_type', 'symmetric')

        try:
            if algo == Algorithm.SM2:
                # SM2 密钥对生成
                priv_key = func.random_hex(64)
                # 注意：gmssl 没有 generate_keypair，需要手动计算公钥
                # 这里简化处理，实际生产应使用 ECC 标量乘法
                pub_key = self._compute_sm2_public_key(priv_key)
                key_data = {
                    'private_key': priv_key,
                    'public_key': pub_key,
                    'key_type': 'asymmetric',
                    'algorithm': 'SM2',
                    'created_at': os.times().elapsed,
                }
            elif algo == Algorithm.SM4:
                # SM4 密钥生成（128 位）
                key_bytes = os.urandom(16)
                key_data = {
                    'key': key_bytes.hex(),
                    'key_type': 'symmetric',
                    'algorithm': 'SM4',
                    'created_at': os.times().elapsed,
                }
            else:
                return CryptoResponse(
                    success=False,
                    error_code=ERROR_ALGORITHM_NOT_SUPPORTED,
                    error_msg=f"不支持的算法: {algo}"
                )

            self._keys[request.key_id] = key_data
            return CryptoResponse(
                success=True,
                result=request.key_id.encode(),  # 返回密钥 ID
            )

        except Exception as e:
            return CryptoResponse(
                success=False,
                error_code=ERROR_INVALID_PARAM,
                error_msg=str(e)
            )

    def encrypt(self, request: CryptoRequest) -> CryptoResponse:
        """SM4 加密"""
        key_info = self._get_key(request.key_id)
        if not key_info:
            return CryptoResponse(
                success=False,
                error_code=ERROR_KEY_NOT_FOUND,
                error_msg=f"密钥 {request.key_id} 不存在"
            )

        mode = request.params.get('mode', 'cbc').lower()
        data = request.data
        key_bytes = bytes.fromhex(key_info['key'])

        try:
            if mode == 'ecb':
                cipher = Cipher(algorithms.SM4(key_bytes), modes.ECB(), backend=default_backend())
                encryptor = cipher.encryptor()
                # PKCS7 填充
                padded = self._pkcs7_pad(data, 16)
                result = encryptor.update(padded) + encryptor.finalize()
            elif mode == 'cbc':
                iv = request.params.get('iv', os.urandom(16))
                cipher = Cipher(algorithms.SM4(key_bytes), modes.CBC(iv), backend=default_backend())
                encryptor = cipher.encryptor()
                padded = self._pkcs7_pad(data, 16)
                result = encryptor.update(padded) + encryptor.finalize()
                # 将 IV 附加到结果前面
                result = iv + result
            elif mode == 'gcm':
                iv = request.params.get('iv', os.urandom(12))
                cipher = Cipher(algorithms.SM4(key_bytes), modes.GCM(iv), backend=default_backend())
                encryptor = cipher.encryptor()
                aad = request.params.get('aad', b'')
                padded = self._pkcs7_pad(data, 16)
                result = encryptor.update(padded) + encryptor.finalize()
                tag = encryptor.tag
                result = iv + tag + result
            else:
                return CryptoResponse(
                    success=False,
                    error_code=ERROR_INVALID_PARAM,
                    error_msg=f"不支持的 SM4 模式: {mode}"
                )

            return CryptoResponse(success=True, result=result)

        except Exception as e:
            return CryptoResponse(
                success=False,
                error_code=ERROR_INVALID_PARAM,
                error_msg=str(e)
            )

    def decrypt(self, request: CryptoRequest) -> CryptoResponse:
        """SM4 解密"""
        key_info = self._get_key(request.key_id)
        if not key_info:
            return CryptoResponse(
                success=False,
                error_code=ERROR_KEY_NOT_FOUND,
                error_msg=f"密钥 {request.key_id} 不存在"
            )

        mode = request.params.get('mode', 'cbc').lower()
        data = request.data
        key_bytes = bytes.fromhex(key_info['key'])

        try:
            if mode == 'ecb':
                cipher = Cipher(algorithms.SM4(key_bytes), modes.ECB(), backend=default_backend())
                decryptor = cipher.decryptor()
                result = decryptor.update(data) + decryptor.finalize()
                result = self._pkcs7_unpad(result)
            elif mode == 'cbc':
                iv = data[:16]
                ciphertext = data[16:]
                cipher = Cipher(algorithms.SM4(key_bytes), modes.CBC(iv), backend=default_backend())
                decryptor = cipher.decryptor()
                padded = decryptor.update(ciphertext) + decryptor.finalize()
                result = self._pkcs7_unpad(padded)
            elif mode == 'gcm':
                iv = data[:12]
                tag = data[12:28]
                ciphertext = data[28:]
                cipher = Cipher(algorithms.SM4(key_bytes), modes.GCM(iv, tag), backend=default_backend())
                decryptor = cipher.decryptor()
                result = decryptor.update(ciphertext) + decryptor.finalize()
                result = self._pkcs7_unpad(result)
            else:
                return CryptoResponse(
                    success=False,
                    error_code=ERROR_INVALID_PARAM,
                    error_msg=f"不支持的 SM4 模式: {mode}"
                )

            return CryptoResponse(success=True, result=result)

        except Exception as e:
            return CryptoResponse(
                success=False,
                error_code=ERROR_INVALID_PARAM,
                error_msg=str(e)
            )

    def sign(self, request: CryptoRequest) -> CryptoResponse:
        """SM2 签名"""
        key_info = self._get_key(request.key_id)
        if not key_info:
            return CryptoResponse(
                success=False,
                error_code=ERROR_KEY_NOT_FOUND,
                error_msg=f"密钥 {request.key_id} 不存在"
            )

        priv_key = key_info['private_key']
        pub_key = key_info['public_key']

        try:
            crypt = sm2.CryptSM2(
                public_key=pub_key,
                private_key=priv_key
            )
            # sign_with_sm3 自动计算 ZA + SM3 哈希
            signature = crypt.sign_with_sm3(request.data)
            return CryptoResponse(success=True, result=bytes.fromhex(signature))
        except Exception as e:
            return CryptoResponse(
                success=False,
                error_code=ERROR_INVALID_PARAM,
                error_msg=str(e)
            )

    def verify(self, request: CryptoRequest) -> CryptoResponse:
        """SM2 验签"""
        key_info = self._get_key(request.key_id)
        if not key_info:
            return CryptoResponse(
                success=False,
                error_code=ERROR_KEY_NOT_FOUND,
                error_msg=f"密钥 {request.key_id} 不存在"
            )

        pub_key = key_info['public_key']
        signature = request.data

        try:
            crypt = sm2.CryptSM2(
                public_key=pub_key,
                private_key=""
            )
            sig_hex = signature.hex()
            result = crypt.verify_with_sm3(sig_hex, request.params.get('message', b''))
            return CryptoResponse(success=result)
        except Exception as e:
            return CryptoResponse(
                success=False,
                error_code=ERROR_INVALID_PARAM,
                error_msg=str(e)
            )

    def hash(self, request: CryptoRequest) -> CryptoResponse:
        """SM3 哈希"""
        try:
            # SM3 实现
            digest = sm3.sm3_hash(func.bytes_to_list(request.data))
            return CryptoResponse(success=True, result=bytes.fromhex(digest))
        except Exception as e:
            return CryptoResponse(
                success=False,
                error_code=ERROR_INVALID_PARAM,
                error_msg=str(e)
            )

    def _compute_sm2_public_key(self, priv_key: str) -> str:
        """
        计算 SM2 公钥（从私钥推导）
        
        注意：gmssl 库没有提供公钥生成 API，需要自行实现椭圆曲线标量乘法
        """
        from gmssl import sm2
        
        ecc = sm2.default_ecc_table
        p = int(ecc['p'], 16)
        n = int(ecc['n'], 16)
        Gx = int(ecc['g'][:64], 16)
        Gy = int(ecc['g'][64:], 16)
        a = int(ecc['a'], 16)

        d = int(priv_key, 16)
        
        # 标量乘法 d*G
        Q = self._scalar_mul(d, Gx, Gy, p, n, a)
        if Q is None:
            raise ValueError("公钥计算失败")
        
        return format(Q[0], '064x') + format(Q[1], '064x')

    def _scalar_mul(self, d, Gx, Gy, p, n, a):
        """椭圆曲线标量乘法（double-and-add）"""
        def point_add(P1, P2):
            if P1 is None: return P2
            if P2 is None: return P1
            x1, y1 = P1
            x2, y2 = P2
            if x1 == x2:
                if y1 != y2: return None
                if y1 == 0: return None
                lam = (3 * x1 * x1 + a) * pow(2 * y1, -1, p) % p
            else:
                lam = ((y2 - y1) * pow(x2 - x1, -1, p)) % p
            x3 = (lam * lam - x1 - x2) % p
            y3 = (lam * (x1 - x3) - y1) % p
            return (x3, y3)

        def point_double(P):
            if P is None or P[1] == 0: return None
            x, y = P
            lam = (3 * x * x + a) * pow(2 * y, -1, p) % p
            x3 = (lam * lam - 2 * x) % p
            y3 = (lam * (x - x3) - y) % p
            return (x3, y3)

        result = None
        addend = (Gx, Gy)
        while d:
            if d & 1:
                result = point_add(result, addend)
            addend = point_double(addend)
            d >>= 1
        return result

    def _pkcs7_pad(self, data: bytes, block_size: int) -> bytes:
        """PKCS7 填充"""
        pad_len = block_size - (len(data) % block_size)
        return data + bytes([pad_len] * pad_len)

    def _pkcs7_unpad(self, data: bytes) -> bytes:
        """PKCS7 去填充"""
        if not data:
            raise ValueError("空数据")
        pad_len = data[-1]
        if pad_len > len(data) or pad_len > 16:
            raise ValueError("无效的 PKCS7 填充")
        if data[-pad_len:] != bytes([pad_len] * pad_len):
            raise ValueError("PKCS7 填充验证失败")
        return data[:-pad_len]

    def key_wrap(self, request: CryptoRequest) -> CryptoResponse:
        """SM4 密钥封装（简化版）"""
        # 生产环境应使用 ESKM 或专用密钥封装算法
        return CryptoResponse(
            success=False,
            error_code=ERROR_ALGORITHM_NOT_SUPPORTED,
            error_msg="密钥封装需要专用算法支持"
        )

class CryptoGateway:
    """
    通用密码服务网关
    
    实现 GM/T 0019-2023 定义的统一接口，支持多 Provider 切换
    """

    def __init__(self):
        self._providers: dict[str, "CryptoProvider"] = {}
        self._default_provider: Optional["CryptoProvider"] = None

    def register_provider(self, name: str, provider: "CryptoProvider"):
        """注册密码服务提供者"""
        self._providers[name] = provider
        if self._default_provider is None:
            self._default_provider = provider

    def set_default_provider(self, name: str):
        """设置默认提供者"""
        if name not in self._providers:
            raise KeyError(f"未找到提供者: {name}")
        self._default_provider = self._providers[name]

    def encrypt(self, key_id: str, data: bytes, **params) -> bytes:
        """加密接口"""
        request = CryptoRequest(
            algorithm=Algorithm.SM4,
            operation=Operation.ENCRYPT,
            key_id=key_id,
            data=data,
            params=params
        )
        response = self._default_provider.process(request)
        if not response.success:
            raise CryptoError(response.error_code, response.error_msg)
        return response.result

    def decrypt(self, key_id: str, data: bytes, **params) -> bytes:
        """解密接口"""
        request = CryptoRequest(
            algorithm=Algorithm.SM4,
            operation=Operation.DECRYPT,
            key_id=key_id,
            data=data,
            params=params
        )
        response = self._default_provider.process(request)
        if not response.success:
            raise CryptoError(response.error_code, response.error_msg)
        return response.result

    def sign(self, key_id: str, message: bytes) -> bytes:
        """签名接口"""
        request = CryptoRequest(
            algorithm=Algorithm.SM2,
            operation=Operation.SIGN,
            key_id=key_id,
            data=message,
            params={}
        )
        response = self._default_provider.process(request)
        if not response.success:
            raise CryptoError(response.error_code, response.error_msg)
        return response.result

    def verify(self, key_id: str, message: bytes, signature: bytes) -> bool:
        """验签接口"""
        request = CryptoRequest(
            algorithm=Algorithm.SM2,
            operation=Operation.VERIFY,
            key_id=key_id,
            data=signature,
            params={'message': message}
        )
        response = self._default_provider.process(request)
        return response.success

    def hash(self, data: bytes) -> bytes:
        """哈希接口"""
        request = CryptoRequest(
            algorithm=Algorithm.SM3,
            operation=Operation.HASH,
            key_id="",
            data=data,
            params={}
        )
        response = self._default_provider.process(request)
        if not response.success:
            raise CryptoError(response.error_code, response.error_msg)
        return response.result

    def generate_key(self, key_id: str, algorithm: Algorithm, **params) -> None:
        """密钥生成接口"""
        request = CryptoRequest(
            algorithm=algorithm,
            operation=Operation.GENERATE_KEY,
            key_id=key_id,
            data=b'',
            params=params
        )
        response = self._default_provider.process(request)
        if not response.success:
            raise CryptoError(response.error_code, response.error_msg)

def test_crypto_gateway():
    """完整功能测试"""
    
    # 初始化网关
    gateway = CryptoGateway()
    provider = SoftwareCryptoProvider()
    gateway.register_provider("software", provider)
    
    # 测试 SM4-CBC 加密解密
    print("=== SM4-CBC 加密解密测试 ===")
    key_id = "sm4-key-001"
    gateway.generate_key(key_id, Algorithm.SM4)
    
    plaintext = b"Hello, GM/T 0019-2023!"
    ciphertext = gateway.encrypt(key_id, plaintext, mode='cbc')
    decrypted = gateway.decrypt(key_id, ciphertext, mode='cbc')
    
    assert decrypted == plaintext, "SM4 解密失败"
    print(f"✓ 明文: {plaintext}")
    print(f"✓ 密文: {ciphertext.hex()[:32]}...")
    print(f"✓ 解密: {decrypted}")
    
    # 测试 SM2 签名验签
    print("\n=== SM2 签名验签测试 ===")
    sm2_key_id = "sm2-key-001"
    gateway.generate_key(sm2_key_id, Algorithm.SM2)
    
    message = b"SM2 signature test message"
    signature = gateway.sign(sm2_key_id, message)
    assert gateway.verify(sm2_key_id, message, signature), "验签失败"
    print(f"✓ 消息: {message}")
    print(f"✓ 签名: {signature.hex()[:32]}...")
    print(f"✓ 验签: 通过")
    
    # 测试 SM3 哈希
    print("\n=== SM3 哈希测试 ===")
    data = b"SM3 hash test"
    digest = gateway.hash(data)
    print(f"✓ 输入: {data}")
    print(f"✓ 哈希: {digest.hex()}")
    
    # 验证 SM3 标准测试向量
    expected = sm3.sm3_hash(func.bytes_to_list(b"abc"))
    assert digest.hex() == expected, "SM3 哈希值不匹配"
    print(f"✓ 标准向量验证: 通过")
    
    print("\n=== 所有测试通过 ===")


if __name__ == "__main__":
    test_crypto_gateway()

# 开发环境：纯软加密
dev_gateway = CryptoGateway()
dev_gateway.register_provider("soft", SoftwareCryptoProvider())
dev_gateway.set_default_provider("soft")

# 生产环境：HSM 设备
prod_gateway = CryptoGateway()
prod_gateway.register_provider("hsm", HSMCryptoProvider())  # 厂商 SDK
prod_gateway.register_provider("soft", SoftwareCryptoProvider())
prod_gateway.set_default_provider("hsm")

# 灾难恢复：降级到软加密
def failover_to_soft():
    prod_gateway.set_default_provider("soft")

class KeyManager:
    """密钥生命周期管理（符合 GM/T 0019 要求）"""
    
    def __init__(self, gateway: CryptoGateway):
        self.gateway = gateway
        self._key_metadata: dict = {}
    
    def create_key(self, key_id: str, algorithm: Algorithm, 
                   params: dict = None) -> str:
        """创建密钥"""
        self.gateway.generate_key(key_id, algorithm, **(params or {}))
        self._key_metadata[key_id] = {
            'created_at': datetime.now(),
            'algorithm': algorithm.value,
            'status': 'active'
        }
        return key_id
    
    def disable_key(self, key_id: str):
        """禁用密钥（不删除，保留历史记录）"""
        if key_id in self._key_metadata:
            self._key_metadata[key_id]['status'] = 'disabled'
            # 更新底层 provider
            # ... 具体实现取决于 provider
    
    def destroy_key(self, key_id: str):
        """销毁密钥（永久删除）"""
        if key_id in self._key_metadata:
            del self._key_metadata[key_id]
            # 通知 provider 删除密钥
            # ...
    
    def rotate_key(self, old_key_id: str, new_key_id: str, 
                   algorithm: Algorithm) -> None:
        """密钥轮换"""
        # 1. 生成新密钥
        self.create_key(new_key_id, algorithm)
        # 2. 迁移数据（使用新密钥重新加密）
        # 3. 禁用旧密钥
        self.disable_key(old_key_id)

class CryptoMetrics:
    """密码服务性能监控"""
    
    def __init__(self):
        self._stats: dict = {}
    
    def record_operation(self, operation: str, duration_ms: float, 
                         success: bool, provider: str):
        """记录操作统计"""
        key = f"{operation}_{provider}"
        if key not in self._stats:
            self._stats[key] = {
                'count': 0,
                'total_time': 0,
                'success': 0,
                'error': 0
            }
        
        self._stats[key]['count'] += 1
        self._stats[key]['total_time'] += duration_ms
        if success:
            self._stats[key]['success'] += 1
        else:
            self._stats[key]['error'] += 1
    
    def get_stats(self) -> dict:
        """获取统计信息"""
        result = {}
        for key, stats in self._stats.items():
            result[key] = {
                'ops': stats['count'],
                'avg_latency_ms': stats['total_time'] / max(stats['count'], 1),
                'success_rate': stats['success'] / max(stats['count'], 1) * 100,
            }
        return result

# ✅ 方案一：使用环境变量 + 密钥管理服务
import os
key_id = os.environ.get('CRYPTO_KEY_ID')
key_encrypted = self.kms.decrypt(key_id)  # 从 KMS 获取

# ✅ 方案二：使用 HSM
from hsm_sdk import HSMDriver
hsm = HSMDriver()
key_handle = hsm.import_key(key_data)  # 密钥进入 HSM，不可导出

# ✅ 方案三：使用加密配置文件
from cryptography.fernet import Fernet
key = Fernet.generate_key()
f = Fernet(key)
encrypted_config = f.encrypt(config_bytes)