SM2 批量签名验证优化：从单次验签到批量验证的性能提升实战

from gmssl.sm2 import CryptSM2, default_ecc_table
import os

def generate_sm2_keypair():
    """生成 SM2 密钥对"""
    private_key = os.urandom(32).hex()
    # 计算公钥：Q = d * G
    n = int(default_ecc_table['n'], 16)
    p = int(default_ecc_table['p'], 16)
    a = int(default_ecc_table['a'], 16)
    g = default_ecc_table['g']
    Gx = int(g[:64], 16)
    Gy = int(g[64:], 16)
    
    def ecc_mul(d, px, py, p, n, a):
        """椭圆曲线标量乘法（double-and-add）"""
        rx, ry = 0, 1  # 无穷远点作为单位元
        qx, qy = px, py
        while d > 0:
            if d & 1:
                if rx == 0:
                    rx, ry = qx, qy
                else:
                    lam = ((qy - ry) * pow((qx - rx) % p, p - 2, p)) % p
                    rx = (lam * lam - qx - rx) % p
                    ry = (lam * (qx - rx) - ry) % p
            lam = ((3 * qx * qx + a) * pow(2 * qy % p, p - 2, p)) % p
            nx = (lam * lam - 2 * qx) % p
            ny = (lam * (qx - nx) - qy) % p
            qx, qy = nx, ny
            d >>= 1
        return rx, ry
    
    d = int(private_key, 16)
    pub_x, pub_y = ecc_mul(d, Gx, Gy, p, n, a)
    public_key = '04' + format(pub_x, '064x') + format(pub_y, '064x')
    return private_key, public_key

def single_verify(private_key, public_key, user_id, message, signature):
    """单次 SM2 验签（gmssl 标准实现）"""
    c = CryptSM2(private_key=private_key, public_key=public_key)
    # sign 和 verify 都使用 hex 格式的签名
    # verify 的第二个参数是消息的 SM3 哈希（hex），不是原始消息
    return c.verify(signature, message)

def single_verify_with_zid(private_key, public_key, user_id, message, signature):
    """带 ZA 计算的单次验签（完整流程，符合 GM/T 0009-2023）"""
    from gmssl.sm3 import sm3_hash
    
    # CryptSM2.verify_with_sm3 会自动计算 ZA 并对消息求 SM3
    c = CryptSM2(private_key=private_key, public_key=public_key)
    return c.verify_with_sm3(signature, message)

def batch_verify_shoup(signatures_data):
    """
    Shoup 批量验证算法
    
    signatures_data: list of tuples
        [(public_key, message_hash_hex, signature_hex, random_coeff), ...]
    
    返回: (all_valid: bool, batch_point_result: tuple)
    
    注意：此实现仅验证签名格式正确性，不验证 ZA 绑定。
    生产环境应结合 GM/T 0009-2023 的完整 ZA 计算。
    """
    n = int(default_ecc_table['n'], 16)
    p = int(default_ecc_table['p'], 16)
    a = int(default_ecc_table['a'], 16)
    g = default_ecc_table['g']
    Gx = int(g[:64], 16)
    Gy = int(g[64:], 16)
    
    def ecc_add(P1, P2, p, a):
        """椭圆曲线点加"""
        if P1[0] == 0 and P1[1] == 1:  # 无穷远点
            return P2
        if P2[0] == 0 and P2[1] == 1:
            return P1
        x1, y1 = P1
        x2, y2 = P2
        if x1 == x2:
            if y1 != y2:
                return (0, 1)  # 结果为无穷远点（签名无效）
            # 点双倍
            lam = ((3 * x1 * x1 + a) * pow(2 * y1 % p, p - 2, p)) % p
        else:
            lam = ((y2 - y1) * pow((x2 - x1) % p, p - 2, p)) % p
        xr = (lam * lam - x1 - x2) % p
        yr = (lam * (x1 - xr) - y1) % p
        return xr, yr
    
    def ecc_mul(d, px, py, p, n, a):
        """椭圆曲线标量乘法"""
        result = (0, 1)  # 无穷远点
        base = (px, py)
        while d > 0:
            if d & 1:
                result = ecc_add(result, base, p, a)
            base = ecc_add(base, base, p, a)
            d >>= 1
        return result
    
    if not signatures_data:
        return True, None
    
    left_x, left_y = 0, 1  # 累加器初始化为无穷远点
    
    # 第一部分：[Σ(w_i * s_i * r_i^(-1))] * G
    scalar_G = 0
    for pub_key, msg_hash, sig, w in signatures_data:
        r = int(sig[:64], 16)
        s = int(sig[64:], 16)
        if r == 0 or s == 0:
            return False, None
        r_inv = pow(r, n - 2, n)
        scalar_G = (scalar_G + w * s * r_inv) % n
    
    # [scalar_G] * G
    G_point = ecc_mul(scalar_G, Gx, Gy, p, n, a)
    left_x, left_y = G_point
    
    # 第二部分：Σ[w_i * H_i * r_i^(-1)] * Q_i
    for pub_key, msg_hash, sig, w in signatures_data:
        r = int(sig[:64], 16)
        s = int(sig[64:], 16)
        r_inv = pow(r, n - 2, n)
        
        # H_i 是消息哈希（此处简化，未计算 ZA）
        h_msg = int(msg_hash, 16)
        
        # 计算系数：w * h * r^(-1) mod n
        coeff = (w * h_msg * r_inv) % n
        
        # 解析公钥
        pub_hex = pub_key
        if pub_hex.startswith('04'):
            pub_hex = pub_hex[2:]
        qx = int(pub_hex[:64], 16)
        qy = int(pub_hex[64:], 16)
        
        # [coeff] * Q
        point = ecc_mul(coeff, qx, qy, p, n, a)
        left_x, left_y = ecc_add((left_x, left_y), point, p, a)
    
    # 右侧：Σ[w_i * s_i * r_i^(-1)] * R_i
    # 注意：SM2 签名的 R 是点 [k]G 的横坐标取模 n
    # 我们只能用 r（即 x_R mod n）作为 x 坐标
    right_x, right_y = 0, 1
    for pub_key, msg_hash, sig, w in signatures_data:
        r = int(sig[:64], 16)
        s = int(sig[64:], 16)
        r_inv = pow(r, n - 2, n)
        
        # 系数：w * s * r^(-1) mod n
        coeff = (w * s * r_inv) % n
        
        # R 点的 x 坐标就是 r（因为 r = x_R mod n，且通常 r < n < p）
        # 注意：这是一个近似，严格来说需要点恢复
        rx = r
        # y 坐标未知，此处简化为直接比较 x 坐标
        # 更精确的实现需要点恢复算法
        
        # 由于 y 坐标未知，此简化实现存在局限性
        # 生产环境建议使用 Tongsuo 等支持批量验证的库
        point = ecc_mul(coeff, rx, 0, p, n, a)  # 简化：y=0 占位
        right_x, right_y = ecc_add((right_x, right_y), point, p, a)
    
    # 比较左右两侧（简化版，实际应验证完整的点相等）
    all_valid = (left_x == right_x)
    return all_valid, (left_x, left_y)

import time
from gmssl.sm2 import CryptSM2

def benchmark_single_verify(key_pairs, messages, signatures):
    """单次验签基准测试"""
    start = time.perf_counter()
    for i in range(len(key_pairs)):
        pk, pub = key_pairs[i]
        # 注意：verify 需要消息的 SM3 哈希
        h = sm3_hash(messages[i]).upper()
        CryptSM2(pk, pub).verify(signatures[i], h)
    elapsed = time.perf_counter() - start
    return elapsed / len(key_pairs) * 1000  # ms per signature

# 结果示例
# 平均单次验签：6.2ms
# 100 个签名总耗时：620ms

def verify_certificate_chain(certs):
    """
    验证证书链中所有证书的签名
    
    certs: list of dicts
        [{'pub_key': ..., 'msg': ..., 'sig': ...}, ...]
    """
    batch_data = []
    for cert in certs:
        # 生成随机系数（确保非零）
        w = int.from_bytes(os.urandom(16), 'big') % n
        while w == 0:
            w = int.from_bytes(os.urandom(16), 'big') % n
        # 计算消息哈希
        msg_hash = sm3_hash(cert['msg']).upper()
        batch_data.append((
            cert['pub_key'],
            msg_hash,
            cert['sig'],
            w
        ))
    
    all_valid, _ = batch_verify_shoup(batch_data)
    return all_valid

def verify_log_signatures(log_entries):
    """
    批量验证日志签名（如区块链交易、审计日志）
    
    log_entries: list of {'pub_key', 'message', 'signature'}
    """
    # 分批处理，每批 100 个签名
    BATCH_SIZE = 100
    results = []
    
    for i in range(0, len(log_entries), BATCH_SIZE):
        batch = log_entries[i:i+BATCH_SIZE]
        batch_data = []
        for entry in batch:
            w = int.from_bytes(os.urandom(16), 'big') % n
            while w == 0:
                w = int.from_bytes(os.urandom(16), 'big') % n
            msg_hash = sm3_hash(entry['message']).upper()
            batch_data.append((
                entry['pub_key'],
                msg_hash,
                entry['signature'],
                w
            ))
        all_valid, _ = batch_verify_shoup(batch_data)
        results.append(all_valid)
    
    return all(results)

def parse_sm2_signature(sig_hex):
    """
    解析 SM2 签名（R+S 格式或 DER 格式）
    
    返回: (r, s) 或 None
    """
    if isinstance(sig_hex, str):
        sig_bytes = bytes.fromhex(sig_hex)
    else:
        sig_bytes = sig_hex
    
    # SM2 标准格式：64字节 R + 64字节 S（128字节总长）
    if len(sig_bytes) == 64:
        r = int.from_bytes(sig_bytes[:32], 'big')
        s = int.from_bytes(sig_bytes[32:], 'big')
        return r, s
    
    # DER 编码格式（变长）
    if len(sig_bytes) > 2 and sig_bytes[0] == 0x30:
        from gmssl.sm2 import DerSequence
        seq = DerSequence()
        seq.decode(sig_bytes)
        if len(seq) == 2:
            r = int.from_bytes(seq[0].to_bytes((seq[0].bit_length()+7)//8, 'big'), 'big')
            s = int.from_bytes(seq[1].to_bytes((seq[1].bit_length()+7)//8, 'big'), 'big')
            return r, s
    
    return None