国密算法在区块链中的实战应用：从SM2签名到智能合约

#!/usr/bin/env python3
"""
SM2在区块链交易签名中的实战应用
基于gmssl库实现
"""
from gmssl import sm2, func
from gmssl.sm3 import sm3_hash
import hashlib
import json
import secrets


class Sm2BlockchainSigner:
    """国密SM2区块链签名器"""
    
    def __init__(self, private_key_hex: str, user_id: str = 'default'):
        """
        初始化签名器
        
        Args:
            private_key_hex: 16进制私钥（64位十六进制字符串）
            user_id: 用户标识，用于计算ZA值
        """
        self.private_key = private_key_hex
        self.user_id = user_id.encode('utf-8')
        self.crypt_sm2 = sm2.CryptSM2(
            public_key='04' + '00' * 64,  # 占位，实际应计算
            private_key=private_key_hex
        )
    
    def sign_transaction(self, tx_data: dict) -> dict:
        """
        对交易数据进行SM2签名
        
        Args:
            tx_data: 交易数据字典
        
        Returns:
            包含签名的完整交易对象
        """
        # 1. 序列化交易数据
        tx_copy = {k: v for k, v in tx_data.items() if k != 'signature'}
        tx_json = json.dumps(tx_copy, sort_keys=True, separators=(',', ':'))
        
        # 2. 计算交易哈希
        tx_hash_bytes = hashlib.sha256(hashlib.sha256(tx_json.encode()).digest()).digest()
        tx_hash_list = func.bytes_to_list(tx_hash_bytes)
        
        # 3. SM2签名（使用sign_with_sm3方法）
        signature_hex = self.crypt_sm2.sign_with_sm3(tx_hash_bytes)
        
        # 4. 组装完整交易
        tx_data['signature'] = signature_hex
        tx_data['tx_hash'] = tx_hash_bytes.hex()
        
        return tx_data
    
    def verify_transaction(self, tx_data: dict) -> bool:
        """验签交易"""
        if 'signature' not in tx_data:
            return False
        
        # 重新计算交易哈希
        tx_copy = {k: v for k, v in tx_data.items() if k not in ['signature', 'tx_hash']}
        tx_json = json.dumps(tx_copy, sort_keys=True, separators=(',', ':'))
        tx_hash_bytes = hashlib.sha256(hashlib.sha256(tx_json.encode()).digest()).digest()
        
        # SM2验签
        try:
            result = self.crypt_sm2.verify_with_sm3(tx_hash_bytes, tx_data['signature'])
            return result
        except Exception:
            return False


class BlockChainNode:
    """简化的区块链节点（演示用途）"""
    
    def __init__(self, signer: Sm2BlockchainSigner):
        self.signer = signer
        self.chain = []
        self.pending_transactions = []
    
    def create_genesis_block(self):
        """创建创世块"""
        genesis_tx = {
            'from': 'SYSTEM',
            'to': 'GENESIS',
            'amount': 1000000,
            'nonce': 0,
            'user_id': self.signer.user_id.decode()
        }
        signed_tx = self.signer.sign_transaction(genesis_tx)
        genesis_block = {
            'index': 0,
            'timestamp': '2026-10-01T00:00:00Z',
            'transactions': [signed_tx],
            'previous_hash': '0' * 64,
            'nonce': 0
        }
        self.chain.append(genesis_block)
        return genesis_block
    
    def mine_pending_transactions(self):
        """挖矿产出新区块"""
        if not self.pending_transactions:
            return None
        
        previous_block = self.chain[-1]
        previous_hash = self._calculate_hash(previous_block)
        
        block = {
            'index': len(self.chain),
            'timestamp': '2026-10-01T00:00:00Z',
            'transactions': self.pending_transactions.copy(),
            'previous_hash': previous_hash,
            'nonce': secrets.randbelow(1000000)
        }
        
        self.chain.append(block)
        self.pending_transactions.clear()
        return block
    
    def _calculate_hash(self, block: dict) -> str:
        """计算区块哈希（使用SM3替代SHA-256）"""
        block_copy = {k: v for k, v in block.items() if k not in ['nonce', 'signature']}
        block_json = json.dumps(block_copy, sort_keys=True, separators=(',', ':'))
        return sm3_hash(func.bytes_to_list(block_json.encode()))
    
    def add_transaction(self, tx_data: dict):
        """添加交易到待确认池"""
        signed_tx = self.signer.sign_transaction(tx_data)
        if self.signer.verify_transaction(signed_tx):
            self.pending_transactions.append(signed_tx)
            return True
        return False
    
    def validate_chain(self) -> bool:
        """验证区块链完整性"""
        for i in range(1, len(self.chain)):
            current_block = self.chain[i]
            previous_block = self.chain[i-1]
            
            # 验证哈希链
            computed_hash = self._calculate_hash(previous_block)
            if current_block['previous_hash'] != computed_hash:
                return False
            
            # 验证所有交易签名
            for tx in current_block['transactions']:
                if not self.signer.verify_transaction(tx):
                    return False
        
        return True


def demo_blockchain_with_sm2():
    """演示国密区块链的基本流程"""
    print("=" * 60)
    print("国密SM2区块链交易签名演示")
    print("=" * 60)
    
    # 生成测试密钥对
    private_key_hex = secrets.token_hex(32)
    print(f"\n私钥（演示用）: {private_key_hex[:16]}...")
    
    # 初始化签名器
    signer = Sm2BlockchainSigner(private_key_hex, user_id='alice')
    
    # 创建区块链节点
    node = BlockChainNode(signer)
    
    # 创建创世块
    print("\n[1] 创建创世块...")
    genesis = node.create_genesis_block()
    print(f"    创世块高度: {genesis['index']}")
    
    # 创建并验证交易
    print("\n[2] 创建交易A->B...")
    tx1 = {
        'from': 'Alice',
        'to': 'Bob',
        'amount': 100,
        'nonce': 1,
        'user_id': 'alice'
    }
    signed_tx1 = signer.sign_transaction(tx1)
    print(f"    交易哈希: {signed_tx1['tx_hash'][:16]}...")
    print(f"    签名长度: {len(signed_tx1['signature'])} 字符")
    
    # 添加交易到节点
    print("\n[3] 交易上链...")
    if node.add_transaction(tx1):
        print("    ✓ 交易签名验证通过，已加入待确认池")
    
    # 挖矿出块
    print("\n[4] 挖矿打包...")
    block = node.mine_pending_transactions()
    if block:
        print(f"    ✓ 新区块高度: {block['index']}")
        print(f"    包含交易数: {len(block['transactions'])}")
    
    # 验证链完整性
    print("\n[5] 验证区块链完整性...")
    if node.validate_chain():
        print("    ✓ 区块链验证通过，所有签名有效")
    else:
        print("    ✗ 区块链验证失败")
    
    print("\n" + "=" * 60)
    print("演示完成")
    print("=" * 60)


if __name__ == '__main__':
    demo_blockchain_with_sm2()

from gmssl.sm3 import sm3_hash
from gmssl import func
import json

def calculate_block_hash(block: dict) -> str:
    """
    使用SM3计算区块哈希
    
    Args:
        block: 区块数据结构
    
    Returns:
        64位十六进制哈希值
    """
    # 1. 确定序列化方式
    block_copy = {k: v for k, v in block.items() 
                  if k not in ['nonce', 'miner_signature']}
    block_json = json.dumps(block_copy, sort_keys=True, separators=(',', ':'))
    
    # 2. SM3哈希计算
    hash_bytes = sm3_hash(func.bytes_to_list(block_json.encode()))
    
    return hash_bytes

from gmssl import sm4, func
import os

class Sm4ContractEncryptor:
    """SM4智能合约数据加密器"""
    
    def __init__(self, key_hex: str):
        """
        初始化加密器
        
        Args:
            key_hex: 128位密钥（32位十六进制）
        """
        if len(key_hex) != 32:
            raise ValueError("SM4密钥必须是128位（32位十六进制）")
        self.key = bytes.fromhex(key_hex)
        self.crypt_sm4 = sm4.CryptSM4()
        self.crypt_sm4.set_key(self.key, sm4.SM4_ENCRYPT)
    
    def encrypt_field(self, plaintext: str, field_name: str) -> dict:
        """加密合约字段"""
        # 生成随机IV
        iv = os.urandom(16)
        
        # SM4-CBC加密
        plaintext_bytes = plaintext.encode('utf-8')
        padded_data = func.pkcs7_padding(plaintext_bytes, 16)
        ciphertext = self.crypt_sm4.crypt_cbc(iv, padded_data)
        
        return {
            'field': field_name,
            'iv': iv.hex(),
            'ciphertext': ciphertext.hex(),
            'algorithm': 'SM4-CBC',
            'padding': 'PKCS7'
        }
    
    def decrypt_field(self, encrypted_data: dict) -> str:
        """解密合约字段"""
        iv = bytes.fromhex(encrypted_data['iv'])
        ciphertext = bytes.fromhex(encrypted_data['ciphertext'])
        
        crypt_sm4_dec = sm4.CryptSM4()
        crypt_sm4_dec.set_key(self.key, sm4.SM4_DECRYPT)
        decrypted_padded = crypt_sm4_dec.crypt_cbc(iv, ciphertext)
        decrypted = func.pkcs7_unpadding(decrypted_padded)
        
        return decrypted.decode('utf-8')


def demo_contract_encryption():
    """演示智能合约数据加密"""
    print("\n" + "=" * 60)
    print("SM4智能合约数据加密演示")
    print("=" * 60)
    
    # 生成加密密钥
    key_hex = os.urandom(16).hex()
    print(f"\n加密密钥: {key_hex[:8]}...")
    
    encryptor = Sm4ContractEncryptor(key_hex)
    
    # 模拟合约数据
    contract_data = {
        'user_name': '张三',
        'id_card': '110101199001011234',
        'phone': '13800138000',
        'balance': '10000.50'
    }
    
    print("\n原始合约数据:")
    for k, v in contract_data.items():
        print(f"  {k}: {v}")
    
    # 加密敏感字段
    encrypted_fields = {}
    for field_name, value in contract_data.items():
        if field_name in ['id_card', 'phone']:
            enc_result = encryptor.encrypt_field(value, field_name)
            encrypted_fields[field_name] = enc_result
            print(f"\n加密 {field_name}:")
            print(f"  IV: {enc_result['iv']}")
            print(f"  密文: {enc_result['ciphertext'][:16]}...")
        else:
            encrypted_fields[field_name] = value
    
    # 解密验证
    print("\n解密验证:")
    for field_name, enc_data in encrypted_fields.items():
        if isinstance(enc_data, dict) and 'ciphertext' in enc_data:
            decrypted = encryptor.decrypt_field(enc_data)
            original = contract_data[field_name]
            match = "✓" if decrypted == original else "✗"
            print(f"  {field_name}: {match}")
    
    print("\n" + "=" * 60)