SM2 API 签名认证实战：防重放、时间戳校验与密钥管理的完整方案

#!/usr/bin/env python3
"""
SM2 API 签名生成器
依赖：pip install gmssl==3.2.2
"""
import hashlib
import hmac
import os
import time
import base64
import json
from typing import Optional
from gmssl import sm2, sm3, func
from gmssl.sm2 import default_ecc_table


class SM2ApiSigner:
    """SM2 API 签名生成器"""
    
    def __init__(self, private_key_hex: str, public_key_hex: str):
        """
        初始化签名器
        
        Args:
            private_key_hex: 私钥（64字节十六进制）
            public_key_hex: 公钥（65字节，含04前缀）
        """
        self.private_key = private_key_hex
        self.public_key = public_key_hex
        self.crypt_sm2 = sm2.CryptSM2(
            private_key=private_key_hex,
            public_key=public_key_hex
        )
    
    def _build_sign_message(self, method: str, path: str, query: str, 
                            timestamp: int, nonce: str, body: str = "") -> str:
        """
        构建签名消息
        
        签名消息格式：METHOD + PATH + QUERY + TIMESTAMP + NONCE + BODY
        使用固定分隔符 '+' 避免内容中包含 '+' 时的歧义
        
        Args:
            method: HTTP方法（GET/POST等）
            path: 请求路径
            query: 查询参数（不含'?'）
            timestamp: 毫秒级时间戳
            nonce: 随机字符串
            body: 请求体（JSON字符串）
        
        Returns:
            待签名字符串
        """
        # 标准化：去掉路径末尾斜杠，query排序
        path = path.rstrip('/')
        if query:
            # 对查询参数排序，确保签名一致性
            sorted_query = '&'.join(sorted(query.split('&')))
        else:
            sorted_query = ""
        
        # 构建签名消息
        sign_msg = f"{method}+{path}+{sorted_query}+{timestamp}+{nonce}+{body}"
        return sign_msg
    
    def sign(self, method: str, path: str, query: str = "", 
             timestamp: Optional[int] = None, nonce: Optional[str] = None,
             body: str = "") -> dict:
        """
        生成签名
        
        Args:
            method: HTTP方法
            path: 请求路径
            query: 查询参数
            timestamp: 时间戳（毫秒），默认当前时间
            nonce: 随机字符串（16字节），默认自动生成
            body: 请求体
        
        Returns:
            签名请求字典
        """
        if timestamp is None:
            timestamp = int(time.time() * 1000)
        if nonce is None:
            nonce = os.urandom(16).hex()
        
        # 构建签名消息
        sign_msg = self._build_sign_message(method, path, query, timestamp, nonce, body)
        
        # SM2签名（使用sign_with_sm3，自动处理ZA预处理）
        sign_result = self.crypt_sm2.sign_with_sm3(sign_msg.encode('utf-8'))
        
        # 返回签名结果（Base64编码）
        return {
            "method": method,
            "path": path,
            "query": query,
            "timestamp": timestamp,
            "nonce": nonce,
            "body": body,
            "sign": base64.b64encode(sign_result).decode('utf-8')
        }
    
    def verify(self, sign_request: dict, public_key_hex: str, 
               max_timestamp_diff_ms: int = 300000) -> bool:
        """
        验证签名
        
        Args:
            sign_request: 签名请求字典
            public_key_hex: 对方公钥（65字节）
            max_timestamp_diff_ms: 最大时间差（毫秒），默认5分钟
        
        Returns:
            验证结果
        """
        # 1. 时间戳校验（防重放）
        current_timestamp = int(time.time() * 1000)
        timestamp_diff = abs(current_timestamp - sign_request['timestamp'])
        if timestamp_diff > max_timestamp_diff_ms:
            return False
        
        # 2. 重新构建签名消息
        sign_msg = self._build_sign_message(
            sign_request['method'],
            sign_request['path'],
            sign_request.get('query', ''),
            sign_request['timestamp'],
            sign_request['nonce'],
            sign_request.get('body', '')
        )
        
        # 3. SM2验签
        try:
            sign_bytes = base64.b64decode(sign_request['sign'])
            crypt_verify = sm2.CryptSM2(
                private_key='',
                public_key=public_key_hex
            )
            return crypt_verify.verify_with_sm3(sign_bytes, sign_msg.encode('utf-8'))
        except Exception:
            return False

# 默认时间窗口：5分钟（300000毫秒）
MAX_TIMESTAMP_DIFF_MS = 300000

def check_timestamp(timestamp: int, max_diff_ms: int = MAX_TIMESTAMP_DIFF_MS) -> bool:
    """
    检查时间戳是否在允许窗口内
    
    生产环境建议：
    - 内网服务：30秒~5分钟
    - 公网API：15秒~1分钟（防止大规模重放）
    """
    current = int(time.time() * 1000)
    return abs(current - timestamp) <= max_diff_ms

import redis
from functools import wraps

# Redis 存储 Nonce（实际生产用分布式缓存）
_nonce_store = redis.Redis(host='localhost', port=6379, db=0)
NONCE_TTL = 600  # 10分钟过期


def check_nonce(nonce: str) -> bool:
    """
    检查Nonce是否已使用
    
    Returns:
        True: 首次使用（合法）
        False: 已使用（重放攻击）
    """
    key = f"sm2_nonce:{nonce}"
    # SETNX: 如果key不存在则设置，返回1；否则返回0
    result = _nonce_store.setnx(key, "1")
    if result:
        # 设置过期时间
        _nonce_store.expire(key, NONCE_TTL)
    return bool(result)

# 客户端示例
from sm2_api_signer import SM2ApiSigner

# 私钥（生产环境应从HSM或密钥管理系统获取）
PRIVATE_KEY = "0e23456789abcdef0e23456789abcdef0e23456789abcdef0e23456789abcdef"
PUBLIC_KEY = "04" + "a" * 128  # 65字节公钥

signer = SM2ApiSigner(PRIVATE_KEY, PUBLIC_KEY)

# 生成签名请求
request = signer.sign(
    method="POST",
    path="/api/v1/transfer",
    query="amount=1000&to=account_123",
    body=json.dumps({"source": "account_456", "memo": "test"})
)

print(f"签名请求: {json.dumps(request, indent=2)}")

from flask import Flask, request, jsonify
from functools import wraps
import base64

app = Flask(__name__)

# 预置的客户端公钥映射（生产环境应从密钥管理系统动态获取）
CLIENT_PUBLIC_KEYS = {
    "client_001": "04" + "a" * 128,
    "client_002": "04" + "b" * 128,
}

# Nonce 去重存储（生产环境用Redis）
used_nonces = set()
NONCE_EXPIRE = 600  # 秒


def require_sm2_signature(f):
    """SM2 签名验证装饰器"""
    @wraps(f)
    def decorated_function(*args, **kwargs):
        # 1. 提取签名信息
        sign_header = request.headers.get('X-SM2-Signature')
        if not sign_header:
            return jsonify({"code": 401, "msg": "Missing signature header"}), 401
        
        try:
            sign_data = json.loads(base64.b64decode(sign_header))
        except Exception:
            return jsonify({"code": 401, "msg": "Invalid signature format"}), 401
        
        # 2. 时间戳校验
        timestamp = sign_data.get('timestamp', 0)
        current_time = int(time.time() * 1000)
        if abs(current_time - timestamp) > 300000:  # 5分钟窗口
            return jsonify({"code": 401, "msg": "Timestamp expired"}), 401
        
        # 3. Nonce 去重
        nonce = sign_data.get('nonce', '')
        if nonce in used_nonces:
            return jsonify({"code": 403, "msg": "Nonce reused (replay attack)"}), 403
        used_nonces.add(nonce)
        
        # 4. 构建签名消息
        sign_msg = f"{sign_data['method']}+{sign_data['path']}+{sign_data.get('query', '')}+{timestamp}+{nonce}+{sign_data.get('body', '')}"
        
        # 5. SM2验签
        client_id = sign_data.get('client_id')
        public_key = CLIENT_PUBLIC_KEYS.get(client_id)
        if not public_key:
            return jsonify({"code": 401, "msg": "Unknown client"}), 401
        
        try:
            from gmssl import sm2
            crypt = sm2.CryptSM2(private_key='', public_key=public_key)
            sign_bytes = base64.b64decode(sign_data['sign'])
            if not crypt.verify_with_sm3(sign_bytes, sign_msg.encode('utf-8')):
                return jsonify({"code": 401, "msg": "Signature verification failed"}), 401
        except Exception as e:
            return jsonify({"code": 500, "msg": f"Verification error: {str(e)}"}), 500
        
        # 验签通过，执行原函数
        return f(*args, **kwargs)
    
    return decorated_function


@app.route('/api/v1/transfer', methods=['POST'])
@require_sm2_signature
def transfer():
    """转账接口（需要SM2签名验证）"""
    data = request.json
    # 业务逻辑...
    return jsonify({"code": 0, "msg": "success"})

# 错误码定义
ERROR_CODES = {
    401: "UNAUTHORIZED",      # 签名验证失败
    403: "FORBIDDEN",         # Nonce重放
    410: "TIMESTAMP_EXPIRED", # 时间戳过期
    500: "VERIFICATION_ERROR" # 验签过程异常
}

def build_error_response(code: int, message: str) -> tuple:
    """构建标准化错误响应"""
    return jsonify({
        "code": code,
        "message": message,
        "timestamp": int(time.time() * 1000)
    }), code

import hashlib
from datetime import datetime, timedelta
from typing import Dict, Optional


class SM2KeyManager:
    """SM2 密钥生命周期管理器"""
    
    def __init__(self, key_ttl_days: int = 90):
        """
        Args:
            key_ttl_days: 密钥有效期（天），默认90天
        """
        self.key_ttl_days = key_ttl_days
        self.keys: Dict[str, dict] = {}  # client_id -> {private, public, created_at, expires_at}
    
    def generate_keypair(self, client_id: str) -> dict:
        """
        生成新的SM2密钥对
        
        Returns:
            {
                "client_id": str,
                "private_key": str,  # 64字节hex
                "public_key": str,   # 65字节（含04前缀）
                "created_at": int,   # 毫秒时间戳
                "expires_at": int    # 毫秒时间戳
            }
        """
        # 使用gmssl生成密钥（生产环境应从HSM获取）
        # 注意：gmssl的CryptSM2需要同时传入私钥和公钥
        # 这里使用固定私钥生成公钥（演示用，生产环境需改用HSM）
        import os
        private_key_hex = os.urandom(32).hex()
        
        # 计算公钥 Q = [d]G
        d = int(private_key_hex, 16)
        Gx = int(default_ecc_table["g"][:64], 16)
        Gy = int(default_ecc_table["g"][64:], 16)
        p = int(default_ecc_table["p"], 16)
        n = int(default_ecc_table["n"], 16)
        a = int(default_ecc_table["a"], 16)
        
        # 标量乘法 Q = [d]G
        Qx, Qy = ecc_mul(d, Gx, Gy, p, n, a)
        public_key_hex = "04" + format(Qx, '064x') + format(Qy, '064x')
        
        created_at = int(time.time() * 1000)
        expires_at = created_at + self.key_ttl_days * 24 * 3600 * 1000
        
        self.keys[client_id] = {
            "private_key": private_key_hex,
            "public_key": public_key_hex,
            "created_at": created_at,
            "expires_at": expires_at
        }
        
        return {
            "client_id": client_id,
            "private_key": private_key_hex,
            "public_key": public_key_hex,
            "created_at": created_at,
            "expires_at": expires_at
        }
    
    def get_public_key(self, client_id: str) -> Optional[str]:
        """获取客户端公钥（用于验签）"""
        key_info = self.keys.get(client_id)
        if not key_info:
            return None
        
        # 检查是否过期
        current_time = int(time.time() * 1000)
        if current_time > key_info['expires_at']:
            return None  # 密钥已过期
        
        return key_info['public_key']
    
    def rotate_key(self, client_id: str) -> dict:
        """
        密钥轮换：生成新密钥，保留旧密钥一段时间
        
        生产环境建议：
        1. 新旧密钥并行验证期：7天
        2. 轮换通知：提前30天通知客户端
        3. 旧密钥归档：保留90天用于历史请求验证
        """
        old_key = self.keys.get(client_id)
        if not old_key:
            raise ValueError(f"Client {client_id} not found")
        
        # 生成新密钥
        new_key_info = self.generate_keypair(client_id)
        
        # 保留旧密钥用于并行验证期
        new_key_info['old_public_key'] = old_key['public_key']
        new_key_info['parallel_verify_until'] = int(time.time() * 1000) + 7 * 24 * 3600 * 1000
        
        return new_key_info

import unittest
from sm2_api_signer import SM2ApiSigner


class TestSM2ApiSigner(unittest.TestCase):
    """SM2 API签名测试用例"""
    
    def setUp(self):
        """测试前准备：生成密钥对"""
        import os
        # 生成测试密钥（生产环境应从HSM获取）
        private_key_hex = os.urandom(32).hex()
        d = int(private_key_hex, 16)
        Gx = int(default_ecc_table["g"][:64], 16)
        Gy = int(default_ecc_table["g"][64:], 16)
        p = int(default_ecc_table["p"], 16)
        n = int(default_ecc_table["n"], 16)
        a = int(default_ecc_table["a"], 16)
        Qx, Qy = ecc_mul(d, Gx, Gy, p, n, a)
        public_key_hex = "04" + format(Qx, '064x') + format(Qy, '064x')
        
        self.signer = SM2ApiSigner(private_key_hex, public_key_hex)
        self.public_key = public_key_hex
    
    def test_sign_and_verify(self):
        """测试签名和验证"""
        request = self.signer.sign(
            method="POST",
            path="/api/v1/test",
            body='{"key": "value"}'
        )
        
        self.assertIn('sign', request)
        self.assertTrue(len(request['sign']) > 0)
        
        # 验证签名
        self.assertTrue(self.signer.verify(request, self.public_key))
    
    def test_tampered_request_fails(self):
        """测试篡改请求验证失败"""
        request = self.signer.sign(
            method="POST",
            path="/api/v1/test",
            body='{"key": "value"}'
        )
        
        # 篡改body
        request['body'] = '{"key": "tampered"}'
        self.assertFalse(self.signer.verify(request, self.public_key))
    
    def test_expired_timestamp_rejected(self):
        """测试过期时间戳被拒绝"""
        request = self.signer.sign(
            method="GET",
            path="/api/v1/test",
            timestamp=int(time.time() * 1000) - 600000  # 10分钟前
        )
        
        # 使用5分钟窗口验证
        self.assertFalse(self.signer.verify(request, self.public_key, max_timestamp_diff_ms=300000))
    
    def test_nonce_reuse_detected(self):
        """测试Nonce重放检测"""
        nonce = "fixed_nonce_for_test"
        request1 = self.signer.sign(
            method="POST",
            path="/api/v1/test",
            nonce=nonce
        )
        
        # 第一次验证通过
        self.assertTrue(self.signer.verify(request1, self.public_key))
        
        # 第二次使用相同Nonce（模拟重放）
        request2 = self.signer.sign(
            method="POST",
            path="/api/v1/test",
            nonce=nonce
        )
        
        # 第二次验证也应通过（验签本身没问题），但应用层应检查Nonce重复
        self.assertTrue(self.signer.verify(request2, self.public_key))
    
    def test_signature_format(self):
        """测试签名格式"""
        request = self.signer.sign(method="GET", path="/test")
        
        # 检查必需字段
        self.assertIn('method', request)
        self.assertIn('path', request)
        self.assertIn('timestamp', request)
        self.assertIn('nonce', request)
        self.assertIn('sign', request)
        
        # 检查时间戳是正整数
        self.assertGreater(request['timestamp'], 0)
        
        # 检查Nonce长度
        self.assertEqual(len(request['nonce']), 32)  # 16字节 = 32位hex


if __name__ == '__main__':
    unittest.main()

# 生产环境必做
SECURITY_CHECKLIST = [
    # 1. 密钥管理
    "私钥存储在HSM或密钥管理服务中",
    "禁止在代码中硬编码私钥",
    "密钥轮换周期不超过90天",
    
    # 2. 签名验证
    "验签失败返回统一错误码（不暴露细节）",
    "记录所有验签失败日志（用于审计）",
    "限制单个IP的签名请求频率",
    
    # 3. 时间戳
    "时间戳窗口不超过5分钟",
    "服务端时钟同步（NTP）",
    "拒绝未来时间戳请求",
    
    # 4. Nonce
    "Nonce唯一性检查（Redis setnx）",
    "设置合理的TTL（建议600秒）",
    "定期清理过期Nonce",
    
    # 5. 监控告警
    "验签失败率超过1%触发告警",
    "单IP高频请求触发限流",
    "密钥轮换成功/失败通知"
]