SM3-HMAC 消息认证码实战：API 签名、消息防篡改与完整性验证的完整方案

"""
HMAC-SM3 完整实现
基于 GM/T 0004-2012 (SM3) + RFC 2104 (HMAC)
"""

import hmac
import hashlib
import struct
import time
from gmssl import sm3, func


def sm3_hash(data: bytes) -> bytes:
    """SM3 哈希函数，返回 32 字节摘要"""
    return bytes.fromhex(sm3.sm3_hash(func.bytes_to_list(data)))


def hmac_sm3(key: bytes, message: bytes) -> bytes:
    """
    HMAC-SM3 实现，遵循 RFC 2104
    参数:
        key: 密钥（任意长度，建议 >= 32 字节）
        message: 待认证消息
    返回:
        32 字节 MAC 值
    """
    BLOCK_SIZE = 64  # SM3 块大小

    # Step 1: 密钥处理
    if len(key) > BLOCK_SIZE:
        key = sm3_hash(key)
    key = key.ljust(BLOCK_SIZE, b'\x00')

    # Step 2: 构造 ipad 和 opad
    ipad = bytes(b ^ 0x36 for b in key)
    opad = bytes(b ^ 0x5C for b in key)

    # Step 3: 内层哈希 H((K' ⊕ ipad) || message)
    inner = ipad + message
    inner_hash = sm3_hash(inner)

    # Step 4: 外层哈希 H((K' ⊕ opad) || inner_hash)
    outer = opad + inner_hash
    return sm3_hash(outer)


def hmac_sm3_hex(key: bytes, message: bytes) -> str:
    """返回十六进制字符串格式的 MAC 值"""
    return hmac_sm3(key, message).hex()


# ==================== 验证测试 ====================
if __name__ == "__main__":
    # 测试向量：验证实现正确性
    key = b"this-is-a-test-key-for-hmac-sm3-32bytes!!"
    message = b"Hello, HMAC-SM3 message authentication!"

    mac_result = hmac_sm3_hex(key, message)
    print(f"Message: {message.decode()}")
    print(f"HMAC-SM3: {mac_result}")
    print(f"Length: {len(bytes.fromhex(mac_result))} bytes")

    # 验证：相同输入产生相同输出
    mac_verify = hmac_sm3_hex(key, message)
    assert mac_result == mac_verify, "MAC verification failed!"
    print("✅ 一致性验证通过")

    # 验证：不同消息产生不同 MAC
    different_msg = b"Hello, HMAC-SM3 modified message!"
    mac_different = hmac_sm3_hex(key, different_msg)
    assert mac_result != mac_different, "Different messages should produce different MACs"
    print(f"HMAC-SM3 (modified): {mac_different}")
    print("✅ 消息区分性验证通过")

import hmac
import hashlib

def hmac_sm3_stdlib(key: bytes, message: bytes) -> bytes:
    """
    使用 hmac.new 配合 hashlib 的 SM3 实现
    注意：需要 OpenSSL 3.0+ 且编译时启用了 SM3 支持
    """
    try:
        mac = hmac.new(key, message, digestmod='sm3')
        return mac.digest()
    except ValueError:
        raise RuntimeError(
            "当前 Python/OpenSSL 不支持 SM3 digestmod。"
            "请使用 gmssl 库方案，或升级 OpenSSL 到 3.0+ 并重新编译 Python。"
        )

import os
import json
import time
import base64
from typing import Optional, Tuple


class HMACSM3Auth:
    """HMAC-SM3 API 认证器"""

    def __init__(self, secret_key: bytes, nonce_ttl: int = 300):
        """
        Args:
            secret_key: 共享密钥（建议 32 字节以上）
            nonce_ttl: 随机数有效期（秒），默认 5 分钟
        """
        if len(secret_key) < 16:
            raise ValueError("密钥长度不能小于 16 字节")
        self.secret_key = secret_key
        self.nonce_ttl = nonce_ttl

    def sign_request(self, method: str, path: str,
                     body: bytes = b"",
                     ) -> dict:
        """
        对 API 请求生成签名
        返回包含认证头的字典
        """
        timestamp = str(int(time.time()))
        nonce = os.urandom(16).hex()

        # 构造待签名字符串
        sign_content = f"{method}\n{path}\n{timestamp}\n{nonce}"
        if body:
            body_hash = sm3_hash(body).hex()
            sign_content += f"\n{body_hash}"

        mac = hmac_sm3_hex(self.secret_key, sign_content.encode())

        return {
            "X-HMAC-Signature": mac,
            "X-HMAC-Timestamp": timestamp,
            "X-HMAC-Nonce": nonce,
            "X-HMAC-Algorithm": "SM3",
        }

    def verify_request(self, method: str, path: str,
                       headers: dict,
                       body: bytes = b""
                       ) -> Tuple[bool, Optional[str]]:
        """
        验证 API 请求的签名
        返回: (是否通过, 失败原因)
        """
        mac = headers.get("X-HMAC-Signature")
        timestamp = headers.get("X-HMAC-Timestamp")
        nonce = headers.get("X-HMAC-Nonce")

        if not all([mac, timestamp, nonce]):
            return False, "缺少认证头"

        # 检查时间戳
        try:
            ts = int(timestamp)
            if abs(time.time() - ts) > self.nonce_ttl:
                return False, "请求已过期"
        except ValueError:
            return False, "时间戳格式错误"

        # 重新计算签名
        sign_content = f"{method}\n{path}\n{timestamp}\n{nonce}"
        if body:
            body_hash = sm3_hash(body).hex()
            sign_content += f"\n{body_hash}"

        expected_mac = hmac_sm3_hex(self.secret_key, sign_content.encode())

        # 恒定时间比较，防止时序攻击
        if hmac.compare_digest(mac, expected_mac):
            return True, None
        return False, "签名不匹配"


# ==================== 使用示例 ====================
if __name__ == "__main__":
    # 服务端和客户端共享密钥（通过安全通道分发）
    shared_key = os.urandom(32)

    # 客户端：生成签名请求头
    auth = HMACSM3Auth(shared_key)
    headers = auth.sign_request(
        method="POST",
        path="/api/v1/transfer",
        body=b'''{"to": "alice", "amount": 100.00}'''
    )
    print("Request Headers:")
    for k, v in headers.items():
        print(f"  {k}: {v}")

    # 服务端：验证签名
    valid, reason = auth.verify_request(
        method="POST",
        path="/api/v1/transfer",
        headers=headers,
        body=b'''{"to": "alice", "amount": 100.00}'''
    )
    print(f"\nVerification: {'✅ PASSED' if valid else f'❌ FAILED: {reason}'}")

import redis

class APIVerifier:
    """服务端 API 签名验证器（含防重放保护）"""

    def __init__(self, secret_key: bytes, redis_client=None):
        self.auth = HMACSM3Auth(secret_key, nonce_ttl=300)
        self.redis = redis_client  # 用于 nonce 去重

    def verify(self, request) -> Tuple[bool, str]:
        """完整验证流程"""
        # 1. 提取认证头
        headers = {
            "X-HMAC-Signature": request.headers.get("X-HMAC-Signature"),
            "X-HMAC-Timestamp": request.headers.get("X-HMAC-Timestamp"),
            "X-HMAC-Nonce": request.headers.get("X-HMAC-Nonce"),
        }

        # 2. 基本验证（时间戳、签名匹配）
        valid, reason = self.auth.verify_request(
            method=request.method,
            path=request.path,
            headers=headers,
            body=request.body
        )
        if not valid:
            return False, reason

        # 3. Nonce 去重（防重放）
        if self.redis:
            nonce_key = f"nonce:{headers['X-HMAC-Nonce']}"
            if self.redis.exists(nonce_key):
                return False, "重复请求（重放攻击）"
            self.redis.setex(nonce_key, self.auth.nonce_ttl, "1")

        return True, "OK"

import json
import base64
import time


def create_sm3_jwt(payload: dict, secret_key: bytes) -> str:
    """创建 HMAC-SM3 签名的 JWT"""
    header = {"alg": "HS384", "typ": "JWT"}  # HS384 = HMAC-SM3

    # 添加标准声明
    payload["iat"] = int(time.time())
    payload["exp"] = payload["iat"] + 3600  # 1 小时过期

    h = base64.urlsafe_b64encode(json.dumps(header).encode()).rstrip(b"=").decode()
    p = base64.urlsafe_b64encode(json.dumps(payload).encode()).rstrip(b"=").decode()

    signing_input = f"{h}.{p}"
    signature = hmac_sm3(secret_key, signing_input.encode())
    s = base64.urlsafe_b64encode(signature).rstrip(b"=").decode()

    return f"{signing_input}.{s}"


def verify_sm3_jwt(token: str, secret_key: bytes) -> dict:
    """验证 HMAC-SM3 签名的 JWT"""
    parts = token.split(".")
    if len(parts) != 3:
        raise ValueError("Invalid JWT format")

    h, p, s = parts

    # 验证签名
    signing_input = f"{h}.{p}"
    expected_sig = hmac_sm3(secret_key, signing_input.encode())
    actual_sig = base64.urlsafe_b64decode(s + "==")

    if not hmac.compare_digest(expected_sig, actual_sig):
        raise ValueError("Signature verification failed")

    # 解码 payload
    padded = p + "=" * (4 - len(p) % 4)
    payload = json.loads(base64.urlsafe_b64decode(padded))

    # 检查过期
    if payload.get("exp", 0) < time.time():
        raise ValueError("Token expired")

    return payload