云HSM密码服务实战：SM2密钥托管与SM4数据加密API集成

import hashlib
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import hashes, padding
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from gmssl import sm2, sm4
import os

class HSMClient:
    """云HSM客户端封装"""
    
    def __init__(self, host: str, port: int, slot_id: int = 0):
        self.host = host
        self.port = port
        self.slot_id = slot_id
        # 实际环境中通过PKCS#11 API连接
        # 这里使用gmssl进行模拟演示
        self._sm2_ctx = None
        self._sm4_key = None
        
    def generate_sm2_keypair(self) -> dict:
        """
        在HSM内生成SM2密钥对
        私钥永不离开HSM，仅返回公钥
        """
        # 实际HSM调用：
        # C_GenerateKey(key_type=SM2_PRIVATE, attributes={CKA_EXTRACTABLE: False})
        
        # 模拟演示（生产环境请替换为真实HSM API）
        private_key = os.urandom(32)
        # 使用gmssl的default_ecc_table进行标量乘法计算公钥
        from gmssl.func import bytes_to_list
        from gmssl.sm2 import default_ecc_table
        
        # 简化处理：实际应调用HSM返回密钥句柄
        return {
            'key_handle': f'k_{hashlib.sha256(private_key).hexdigest()[:16]}',
            'public_key': '04' + hashlib.sha256(private_key).hexdigest(),
            'warning': '演示用，生产环境请使用真实HSM'
        }
    
    def sm2_sign(self, key_handle: str, data: bytes) -> bytes:
        """
        使用HSM内SM2私钥对数据进行签名
        私钥不出HSM，仅传入数据哈希
        """
        # 实际HSM调用：
        # C_SignInit(h SM2WithSM3)
        # C_Sign(data_hash)
        # 返回签名结果R||S
        
        # 模拟演示
        raise NotImplementedError('请使用真实HSM SDK')
    
    def sm2_verify(self, public_key: str, data: bytes, signature: bytes) -> bool:
        """
        验签：使用公钥验证签名
        注意：此处为演示代码，生产环境应调用真实HSM SDK
        """
        # 实际HSM调用：
        # C_VerifyInit(h SM2WithSM3)
        # result = C_Verify(data, signature)
        # return result == CKR_OK
        
        # 演示用：调用gmssl验证
        from gmssl.sm2 import CryptSM2
        try:
            sm2 = CryptSM2(public_key=public_key, private_key=None)
            return sm2.verify_with_sm3(signature, data)
        except Exception:
            return False
    
    def sm4_encrypt(self, key_handle: str, plaintext: bytes, iv: bytes = None) -> bytes:
        """
        SM4-CBC加密
        密钥由HSM托管，仅传入明文
        """
        if iv is None:
            iv = os.urandom(16)
        
        # 实际HSM调用：
        # C_EncryptInit(mode=SM4_CBC, key_handle)
        # C_Encrypt(plaintext)
        
        # 模拟演示
        from gmssl import sm4
        key_bytes = bytes.fromhex(key_handle.replace('k_', ''))[:16]
        cipher = sm4.CryptSM4()
        ciphertext = cipher.encrypt_cbc(key_bytes, iv, plaintext)
        return iv + ciphertext  # 返回IV+Ciphertext
    
    def sm4_decrypt(self, key_handle: str, ciphertext: bytes) -> bytes:
        """
        SM4-CBC解密
        """
        iv = ciphertext[:16]
        ciphertext = ciphertext[16:]
        
        # 实际HSM调用：
        # C_DecryptInit(mode=SM4_CBC, key_handle)
        # C_Decrypt(ciphertext)
        
        from gmssl import sm4
        key_bytes = bytes.fromhex(key_handle.replace('k_', ''))[:16]
        cipher = sm4.CryptSM4()
        plaintext = cipher.decrypt_cbc(key_bytes, iv, ciphertext)
        return plaintext

# 需要先安装：pip install ctKCS11
# 并配置厂商提供的 .so 库路径
from ctKCS11 import C_KCS11

hsm = C_KCS11("/path/to/hsm_pkcs11.so")
hsm.C_Initialize()
session = hsm.C_OpenSession(slot_id=0, flags=hsm.CKF_SERIAL_SESSION)
hsm.C_Login(session, hsm.CKF_USER, pin="12345678")

# SM2签名（密钥句柄由HSM分配）
signature = hsm.C_SignInit(session, mechanism="CKM_SM2_WITH_SM3", key_handle="key_001")
signature = hsm.C_Sign(session, data_hash)

hsm.C_CloseSession(session)
hsm.C_Finalize()

from gmssl.sm2 import CryptSM2
from gmssl.sm4 import CryptSM4
import os

# SM2密钥生成（生产环境应在HSM内完成）
priv_key = os.urandom(32).hex()
pub_key = '04' + os.urandom(64).hex()  # 未压缩格式

sm2 = CryptSM2(priv_key, pub_key)
data = b"test message"
signature = sm2.sign_with_sm3(data)  # 内部自动计算 ZA + SM3

# SM4-CBC加密
key = os.urandom(16)
iv = os.urandom(16)
crypt_sm4 = CryptSM4()
ciphertext = crypt_sm4.encrypt_cbc(key, iv, b"plaintext")

# key_lifecycle.py
import hashlib
from datetime import datetime, timedelta
from typing import Optional

class KeyLifecycleManager:
    """密钥生命周期管理器"""
    
    def __init__(self, hsm_client: HSMClient):
        self.hsm = hsm_client
        self.keys = {}  # 密钥存储（实际应存入HSM或安全数据库）
    
    def create_sm2_key(self, alias: str, valid_days: int = 365) -> dict:
        """创建SM2密钥对"""
        result = self.hsm.generate_sm2_keypair()
        
        key_info = {
            'alias': alias,
            'key_handle': result['key_handle'],
            'public_key': result['public_key'],
            'created_at': datetime.now(),
            'expires_at': datetime.now() + timedelta(days=valid_days),
            'status': 'active',
            'usage': ['sign', 'verify', 'key_exchange']
        }
        
        self.keys[alias] = key_info
        return key_info
    
    def rotate_key(self, old_alias: str, new_alias: str) -> dict:
        """密钥轮换"""
        old_key = self.keys.get(old_alias)
        if not old_key:
            raise ValueError(f'密钥 {old_alias} 不存在')
        
        # 标记旧密钥为过期
        old_key['status'] = 'retired'
        old_key['retired_at'] = datetime.now()
        
        # 创建新密钥
        new_key = self.create_sm2_key(new_alias)
        return new_key
    
    def get_active_key(self, alias: str) -> Optional[dict]:
        """获取活跃密钥"""
        key = self.keys.get(alias)
        if not key:
            return None
        if key['status'] != 'active':
            return None
        if datetime.now() > key['expires_at']:
            key['status'] = 'expired'
            return None
        return key
    
    def cleanup_expired_keys(self):
        """清理过期密钥"""
        now = datetime.now()
        expired = []
        for alias, key in self.keys.items():
            if key['status'] == 'expired' or now > key['expires_at']:
                expired.append(alias)
        
        for alias in expired:
            del self.keys[alias]
        
        return expired

# async_hsm_service.py
import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import Callable

class AsyncHSMService:
    """异步HSM服务"""
    
    def __init__(self, max_workers: int = 10):
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
        self._pool = []  # HSM连接池
    
    async def _run_in_executor(self, func: Callable, *args):
        """在线程池中执行HSM操作"""
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(self.executor, func, *args)
    
    async def sign_batch(self, key_handle: str, data_list: list) -> list:
        """批量签名"""
        tasks = [self._sign(key_handle, data) for data in data_list]
        return await asyncio.gather(*tasks)
    
    async def _sign(self, key_handle: str, data: bytes) -> bytes:
        """单次签名"""
        return await self._run_in_executor(self.hsm.sm2_sign, key_handle, data)
    
    async def encrypt_stream(self, key_handle: str, data_chunks: list) -> list:
        """流式加密"""
        tasks = [self._encrypt(key_handle, chunk) for chunk in data_chunks]
        return await asyncio.gather(*tasks)
    
    async def _encrypt(self, key_handle: str, data: bytes) -> bytes:
        """单次加密"""
        return await self._run_in_executor(self.hsm.sm4_encrypt, key_handle, data)

# benchmark.py（参考模板，需替换为实际HSM连接）
import time
from gmssl.sm2 import CryptSM2
from gmssl.sm4 import CryptSM4
import os

def benchmark_local_performance():
    """本地gmssl性能基准（仅供参考，非HSM性能）"""
    priv_key = os.urandom(32).hex()
    pub_key = '04' + os.urandom(64).hex()
    sm2 = CryptSM2(priv_key, pub_key)
    data = b'test data for SM2 signature'
    
    key = os.urandom(16)
    iv = os.urandom(16)
    crypt_sm4 = CryptSM4()
    plaintext = os.urandom(1024)  # 1KB
    
    # SM2签名测试
    times_sign = []
    for _ in range(100):
        start = time.perf_counter()
        sig = sm2.sign_with_sm3(data)
        elapsed = time.perf_counter() - start
        times_sign.append(elapsed * 1000)
    
    # SM4加密测试
    times_encrypt = []
    for _ in range(100):
        start = time.perf_counter()
        ciphertext = crypt_sm4.encrypt_cbc(key, iv, plaintext)
        elapsed = time.perf_counter() - start
        times_encrypt.append(elapsed * 1000)
    
    print('=' * 50)
    print('Local SM2/SM4 Performance Benchmark')
    print('=' * 50)
    print(f"SM2 Sign:     {sum(times_sign)/len(times_sign):.3f} ms/op")
    print(f"SM4 Encrypt:  {sum(times_encrypt)/len(times_encrypt):.3f} ms/op")
    print('=' * 50)
    print("注意：以上为纯软件实现性能，HSM硬件加速通常快5-10倍")

if __name__ == '__main__':
    benchmark_local_performance()