SM3 国密哈希算法实战：HMAC-SM3、PBKDF2-SM3、性能对比与安全防护

"""
SM3 哈希算法完整实现
标准：GM/T 0004-2012 密码杂凑算法
"""

import struct
from typing import Union

# SM3 初始向量（标准第 5.3 节）
SM3_IV = [
    0x7380166f, 0x4914b2b9, 0x172442d7, 0xda8a0600,
    0xa96f30bc, 0x163138aa, 0xe38dee4d, 0xb0fb0e4e,
]

# 常量 T_j（标准第 5.4 节）
T = [0x79cc4519] * 16 + [0x7a879d8a] * 48


def _rotate_left(x: int, n: int) -> int:
    """32位循环左移"""
    n &= 31
    return ((x << n) | (x >> (32 - n))) & 0xFFFFFFFF


def _ff_j(x: int, y: int, z: int, j: int) -> int:
    """布尔函数 FF"""
    if 0 <= j <= 15:
        return x ^ y ^ z
    else:
        return (x & y) | (x & z) | (y & z)


def _gg_j(x: int, y: int, z: int, j: int) -> int:
    """布尔函数 GG"""
    if 0 <= j <= 15:
        return x ^ y ^ z
    else:
        return (x & y) | ((~x) & z)


def _p_0(x: int) -> int:
    """置换函数 P0"""
    return x ^ _rotate_left(x, 9) ^ _rotate_left(x, 17)


def _p_1(x: int) -> int:
    """置换函数 P1"""
    return x ^ _rotate_left(x, 15) ^ _rotate_left(x, 23)


def _message_expansion(b: list) -> tuple:
    """
    消息扩展：将 16 个 32-bit 字扩展为 68 个字(W) + 64 个字(W')
    标准第 5.4.2 节
    """
    w = list(b) + [0] * 52
    for j in range(16, 68):
        w[j] = _p_1(w[j - 16] ^ w[j - 9] ^ _rotate_left(w[j - 3], 15)) ^ \
               _rotate_left(w[j - 13], 7) ^ w[j - 6]
        w[j] &= 0xFFFFFFFF

    w_prime = [0] * 64
    for j in range(64):
        w_prime[j] = w[j] ^ w[j + 4]
        w_prime[j] &= 0xFFFFFFFF

    return w, w_prime


def _compress(cv: list, block: list) -> list:
    """
    压缩函数 CF: (CV_i, B_i) -> CV_{i+1}
    标准第 5.4 节
    """
    w, w_prime = _message_expansion(block)

    a, b, c, d, e, f, g, h = cv

    for j in range(64):
        ss1 = _rotate_left(
            (_rotate_left(a, 12) + e + _rotate_left(T[j], j)) & 0xFFFFFFFF, 7
        )
        ss2 = ss1 ^ _rotate_left(a, 12)
        tt1 = (_ff_j(a, b, c, j) + d + ss2 + w_prime[j]) & 0xFFFFFFFF
        tt2 = (_gg_j(e, f, g, j) + h + ss1 + w[j]) & 0xFFFFFFFF
        d = c
        c = _rotate_left(b, 9)
        b = a
        a = tt1
        h = g
        g = _rotate_left(f, 19)
        f = e
        e = _p_0(tt2)

    return [a ^ cv[0], b ^ cv[1], c ^ cv[2], d ^ cv[3],
            e ^ cv[4], f ^ cv[5], g ^ cv[6], h ^ cv[7]]


def sm3_hash(message: Union[bytes, str]) -> bytes:
    """
    计算 SM3 哈希值

    Args:
        message: 输入消息（bytes 或 str，str 按 UTF-8 编码）

    Returns:
        32 字节（256 比特）哈希值
    """
    if isinstance(message, str):
        message = message.encode("utf-8")

    # 填充（标准第 5.2 节）
    msg_len_bits = len(message) * 8
    padded = bytearray(message)
    padded.append(0x80)  # 追加 '1' 比特

    while (len(padded) % 64) != 56:
        padded.append(0x00)

    # 追加 64 比特原始消息长度（大端序）
    padded.extend(struct.pack(">Q", msg_len_bits))

    # 迭代压缩
    cv = list(SM3_IV)
    for i in range(0, len(padded), 64):
        block = list(struct.unpack(">16I", padded[i:i + 64]))
        cv = _compress(cv, block)

    # 输出
    return struct.pack(">8I", *cv)


def sm3_hex(message: Union[bytes, str]) -> str:
    """计算 SM3 哈希值，返回十六进制字符串"""
    return sm3_hash(message).hex()


# === 验证向量（GM/T 0004-2012 附录 A）===
if __name__ == "__main__":
    # 测试向量 1：空消息
    result = sm3_hex(b"")
    expected = "1ab21d8355cfa17f8e61194831e81a8f22bec8c728fefb747ed035eb5082aa2b"
    assert result == expected, f"空消息测试失败: {result}"
    print(f"[PASS] 空消息 SM3: {result}")

    # 测试向量 2：短消息 "abc"
    result = sm3_hex("abc")
    expected = "66c7f0f462eeedd9d1f2d46bdc10e4e24167c4875cf2f7a2297da02b8f4ba8e0"
    assert result == expected, f"abc 测试失败: {result}"
    print(f"[PASS] 'abc' SM3: {result}")

    # 测试向量 3：长消息（6个 'abcd' 重复）
    result = sm3_hex("abcd" * 6)
    expected = "debe9ff92275b8a138604889c18e5a4d6fdb70e5387e5765293dcba39c0c5732"
    assert result == expected, f"长消息测试失败: {result}"
    print(f"[PASS] 长消息 SM3: {result}")

    print("\n所有测试向量通过！")

package sm3

import (
	"encoding/binary"
	"fmt"
)

// SM3 初始向量
var IV = [8]uint32{
	0x7380166f, 0x4914b2b9, 0x172442d7, 0xda8a0600,
	0xa96f30bc, 0x163138aa, 0xe38dee4d, 0xb0fb0e4e,
}

func rotateLeft(x uint32, n int) uint32 {
	n &= 31
	return (x << n) | (x >> (32 - n))
}

func ffJ(x, y, z uint32, j int) uint32 {
	if j <= 15 {
		return x ^ y ^ z
	}
	return (x & y) | (x & z) | (y & z)
}

func ggJ(x, y, z uint32, j int) uint32 {
	if j <= 15 {
		return x ^ y ^ z
	}
	return (x & y) | ((^x) & z)
}

func p0(x uint32) uint32 {
	return x ^ rotateLeft(x, 9) ^ rotateLeft(x, 17)
}

func p1(x uint32) uint32 {
	return x ^ rotateLeft(x, 15) ^ rotateLeft(x, 23)
}

func getT(j int) uint32 {
	if j <= 15 {
		return 0x79cc4519
	}
	return 0x7a879d8a
}

// SM3Hash 计算 SM3 哈希值，返回 32 字节
func SM3Hash(msg []byte) []byte {
	// 填充
	bitLen := uint64(len(msg) * 8)
	padded := make([]byte, 0, len(msg)+128)
	padded = append(padded, msg...)
	padded = append(padded, 0x80)

	for len(padded)%64 != 56 {
		padded = append(padded, 0x00)
	}

	// 追加长度（大端序 64-bit）
	var lenBytes [8]byte
	binary.BigEndian.PutUint64(lenBytes[:], bitLen)
	padded = append(padded, lenBytes[:]...)

	// 迭代压缩
	cv := IV
	for i := 0; i < len(padded); i += 64 {
		block := make([]uint32, 16)
		for j := 0; j < 16; j++ {
			block[j] = binary.BigEndian.Uint32(padded[i+j*4:])
		}
		cv = compress(cv, block)
	}

	// 序列化输出
	result := make([]byte, 32)
	for i := 0; i < 8; i++ {
		binary.BigEndian.PutUint32(result[i*4:], cv[i])
	}
	return result
}

func compress(cv [8]uint32, block []uint32) [8]uint32 {
	// 消息扩展
	w := make([]uint32, 68)
	for j := 0; j < 16; j++ {
		w[j] = block[j]
	}
	for j := 16; j < 68; j++ {
		w[j] = p1(w[j-16]^w[j-9]^rotateLeft(w[j-3], 15)) ^
			rotateLeft(w[j-13], 7) ^ w[j-6]
	}

	wPrime := make([]uint32, 64)
	for j := 0; j < 64; j++ {
		wPrime[j] = w[j] ^ w[j+4]
	}

	a, b, c, d, e, f, g, h := cv[0], cv[1], cv[2], cv[3], cv[4], cv[5], cv[6], cv[7]

	for j := 0; j < 64; j++ {
		ss1 := rotateLeft((rotateLeft(a, 12)+e+rotateLeft(getT(j), j)), 7)
		ss2 := ss1 ^ rotateLeft(a, 12)
		tt1 := ffJ(a, b, c, j) + d + ss2 + wPrime[j]
		tt2 := ggJ(e, f, g, j) + h + ss1 + w[j]
		d = c
		c = rotateLeft(b, 9)
		b = a
		a = tt1
		h = g
		g = rotateLeft(f, 19)
		f = e
		e = p0(tt2)
	}

	return [8]uint32{
		a ^ cv[0], b ^ cv[1], c ^ cv[2], d ^ cv[3],
		e ^ cv[4], f ^ cv[5], g ^ cv[6], h ^ cv[7],
	}
}

// SM3Hex 返回十六进制哈希字符串
func SM3Hex(msg []byte) string {
	return fmt.Sprintf("%x", SM3Hash(msg))
}

"""
HMAC-SM3 实现
基于 RFC 2104 HMAC 框架 + GM/T 0004-2012 SM3
"""

from sm3 import sm3_hash  # 使用上文实现的 sm3_hash

BLOCK_SIZE = 64  # SM3 分组大小：512 bit = 64 byte
DIGEST_SIZE = 32  # SM3 输出：256 bit = 32 byte


def hmac_sm3(key: bytes, message: bytes) -> bytes:
    """
    计算 HMAC-SM3

    Args:
        key: 密钥（任意长度）
        message: 消息

    Returns:
        32 字节 HMAC 值
    """
    # 密钥处理
    if len(key) > BLOCK_SIZE:
        key = sm3_hash(key)
    key = key.ljust(BLOCK_SIZE, b'\x00')

    # 构造 ipad 和 opad
    ipad = bytes(b ^ 0x36 for b in key)
    opad = bytes(b ^ 0x5c for b in key)

    # HMAC = SM3(opad || SM3(ipad || message))
    inner_hash = sm3_hash(ipad + message)
    outer_hash = sm3_hash(opad + inner_hash)

    return outer_hash


def hmac_sm3_hex(key: bytes, message: bytes) -> str:
    """返回十六进制 HMAC-SM3 字符串"""
    return hmac_sm3(key, message).hex()


# === 使用示例 ===
if __name__ == "__main__":
    key = b"my-secret-key-2024"
    msg = b"transfer: amount=10000, to=account_12345"

    mac = hmac_sm3_hex(key, msg)
    print(f"HMAC-SM3: {mac}")

    # 验证：相同输入产生相同输出
    assert hmac_sm3_hex(key, msg) == mac
    # 验证：不同密钥产生不同输出
    assert hmac_sm3_hex(b"wrong-key", msg) != mac
    # 验证：不同消息产生不同输出
    assert hmac_sm3_hex(key, b"tampered message") != mac

    print("HMAC-SM3 验证通过")

package hmac_sm3

import (
	"bytes"
	"crypto/subtle"
	"github.com/example/sm3" // 使用上文 SM3 实现
)

const BlockSize = 64

func HmacSm3(key, message []byte) []byte {
	if len(key) > BlockSize {
		key = sm3.SM3Hash(key)
	}
	key = append(key, make([]byte, BlockSize-len(key))...)

	ipad := make([]byte, BlockSize)
	opad := make([]byte, BlockSize)
	for i := 0; i < BlockSize; i++ {
		ipad[i] = key[i] ^ 0x36
		opad[i] = key[i] ^ 0x5c
	}

	// 内层哈希
	inner := append(ipad, message...)
	innerHash := sm3.SM3Hash(inner)

	// 外层哈希
	outer := append(opad, innerHash...)
	return sm3.SM3Hash(outer)
}

// VerifyHmacSm3 使用恒定时间比较验证 HMAC（防时序攻击）
func VerifyHmacSm3(key, message, expectedMac []byte) bool {
	computed := HmacSm3(key, message)
	return subtle.ConstantTimeCompare(computed, expectedMac) == 1
}

"""
PBKDF2-SM3 密钥派生实现
基于 RFC 8018 PBKDF2 框架 + HMAC-SM3
"""

import struct
from hmac_sm3 import hmac_sm3  # 使用上文 HMAC-SM3 实现


def pbkdf2_sm3(
    password: bytes,
    salt: bytes,
    iterations: int = 100_000,
    dk_len: int = 32
) -> bytes:
    """
    PBKDF2-SM3 密钥派生

    Args:
        password: 密码
        salt: 盐值（建议至少 16 字节）
        iterations: 迭代次数（生产环境建议 >= 100000）
        dk_len: 派生密钥长度（字节）

    Returns:
        派生密钥
    """
    h_len = 32  # SM3 输出长度

    # 计算需要的块数
    num_blocks = (dk_len + h_len - 1) // h_len

    dk = b""
    for i in range(1, num_blocks + 1):
        # U_1 = HMAC-SM3(password, salt || i)
        u = hmac_sm3(password, salt + struct.pack(">I", i))
        result = u

        # U_2 到 U_c
        for _ in range(iterations - 1):
            u = hmac_sm3(password, u)
            # XOR 累加
            result = bytes(a ^ b for a, b in zip(result, u))

        dk += result

    return dk[:dk_len]


# === 使用示例 ===
if __name__ == "__main__":
    import os
    import time

    password = b"MyStr0ng!Passw0rd"
    salt = os.urandom(16)  # 随机盐值

    # 派生 32 字节密钥
    start = time.time()
    key = pbkdf2_sm3(password, salt, iterations=100_000, dk_len=32)
    elapsed = time.time() - start

    print(f"派生密钥: {key.hex()}")
    print(f"迭代次数: 100,000")
    print(f"耗时: {elapsed:.2f}s")
    print(f"盐值: {salt.hex()}")

    # 验证：相同输入产生相同输出
    key2 = pbkdf2_sm3(password, salt, iterations=100_000, dk_len=32)
    assert key == key2, "确定性验证失败"

    # 验证：不同盐值产生不同输出
    salt2 = os.urandom(16)
    key3 = pbkdf2_sm3(password, salt2, iterations=100_000, dk_len=32)
    assert key != key3, "盐值敏感性验证失败"

    print("PBKDF2-SM3 验证通过")

"""
SM3 vs SHA-256 性能对比测试
"""

import hashlib
import time
from sm3 import sm3_hash  # 纯 Python 实现


def benchmark(func, data, iterations=100):
    """基准测试"""
    start = time.perf_counter()
    for _ in range(iterations):
        func(data)
    elapsed = time.perf_counter() - start
    throughput = (len(data) * iterations) / elapsed / 1024 / 1024
    return elapsed, throughput


if __name__ == "__main__":
    data_1kb = b"x" * 1024
    data_1mb = b"x" * (1024 * 1024)

    print("=" * 60)
    print(f"{'算法':<15} {'数据大小':<10} {'耗时(s)':<12} {'吞吐(MB/s)':<12}")
    print("=" * 60)

    for label, data in [("1KB", data_1kb), ("1MB", data_1mb)]:
        # SHA-256（hashlib，C 实现）
        t, tp = benchmark(lambda d: hashlib.sha256(d).digest(), data)
        print(f"{'SHA-256 (C)':<15} {label:<10} {t:<12.4f} {tp:<12.1f}")

        # SM3（纯 Python 实现）
        t, tp = benchmark(sm3_hash, data)
        print(f"{'SM3 (Python)':<15} {label:<10} {t:<12.4f} {tp:<12.1f}")

        print("-" * 60)

"""
SM3 长度扩展攻击演示
仅用于安全研究，请勿用于非法用途
"""

from sm3 import sm3_hash, SM3_IV, _compress
import struct


def sm3_length_extension(original_hash: bytes, original_len: int, 
                         extension: bytes) -> bytes:
    """
    长度扩展攻击：已知 SM3(M) 和 len(M)，计算 SM3(M || pad || extension)
    """
    # 从哈希值恢复链变量
    cv = list(struct.unpack(">8I", original_hash))

    # 构造扩展消息的填充
    # 注意：总长度 = original_len + padding + len(extension)
    total_bits = original_len * 8

    # 先构造原始消息的填充
    pad = b'\x80'
    while (original_len + len(pad)) % 64 != 56:
        pad += b'\x00'
    pad += struct.pack(">Q", total_bits)

    # 新的总长度（比特）
    new_total_bits = (original_len + len(pad) + len(extension)) * 8

    # 对扩展数据进行填充
    ext_padded = extension + b'\x80'
    while len(ext_padded) % 64 != 56:
        ext_padded += b'\x00'
    ext_padded += struct.pack(">Q", new_total_bits)

    # 用恢复的 CV 继续压缩
    for i in range(0, len(ext_padded), 64):
        block = list(struct.unpack(">16I", ext_padded[i:i + 64]))
        cv = _compress(cv, block)

    return struct.pack(">8I", *cv)


# === 演示 ===
if __name__ == "__main__":
    # 原始消息（攻击者不知道内容，但知道长度）
    secret = b"secret-key:abc123"
    original_hash = sm3_hash(secret)
    print(f"原始哈希: {original_hash.hex()}")

    # 攻击者附加数据
    extension = b"&admin=true"

    # 通过长度扩展计算新哈希
    forged_hash = sm3_length_extension(
        original_hash, len(secret), extension
    )

    # 验证：直接计算 SM3(secret || padding || extension)
    # 构造完整消息
    total_bits = len(secret) * 8
    full_msg = secret + b'\x80'
    while (len(full_msg)) % 64 != 56:
        full_msg += b'\x00'
    full_msg += struct.pack(">Q", total_bits)
    full_msg += extension

    real_hash = sm3_hash(full_msg)

    print(f"伪造哈希: {forged_hash.hex()}")
    print(f"真实哈希: {real_hash.hex()}")
    print(f"攻击成功: {forged_hash == real_hash}")

"""
file_integrity.py - 基于 SM3 的文件完整性校验工具
功能：生成校验文件、验证文件完整性
"""

import os
import sys
import json
import hmac
import argparse
from pathlib import Path
from sm3 import sm3_hex
from hmac_sm3 import hmac_sm3_hex


def hash_file(filepath: str, chunk_size: int = 65536) -> str:
    """增量计算大文件的 SM3 哈希"""
    # 增量实现需要修改 sm3_hash 支持 update 接口
    # 这里简化为直接读取
    with open(filepath, "rb") as f:
        data = f.read()
    return sm3_hex(data)


def generate_manifest(directory: str, key: bytes = None) -> dict:
    """
    为目录下所有文件生成完整性清单

    Args:
        directory: 目标目录
        key: 如果提供，使用 HMAC-SM3；否则使用普通 SM3

    Returns:
        清单字典 {filename: hash_value}
    """
    manifest = {}
    base = Path(directory)

    for filepath in sorted(base.rglob("*")):
        if filepath.is_file():
            rel_path = str(filepath.relative_to(base))
            with filepath.open("rb") as f:
                data = f.read()

            if key:
                value = hmac_sm3_hex(key, data)
                mode = "HMAC-SM3"
            else:
                value = sm3_hex(data)
                mode = "SM3"

            manifest[rel_path] = value

    return {"mode": mode, "files": manifest}


def verify_manifest(directory: str, manifest: dict, key: bytes = None) -> list:
    """验证文件完整性，返回被修改的文件列表"""
    modified = []
    base = Path(directory)

    for rel_path, expected_hash in manifest["files"].items():
        filepath = base / rel_path
        if not filepath.exists():
            modified.append((rel_path, "文件缺失"))
            continue

        with filepath.open("rb") as f:
            data = f.read()

        if key:
            actual_hash = hmac_sm3_hex(key, data)
        else:
            actual_hash = sm3_hex(data)

        if actual_hash != expected_hash:
            modified.append((rel_path, "哈希不匹配"))

    return modified


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="SM3 文件完整性校验工具")
    parser.add_argument("action", choices=["sign", "verify"])
    parser.add_argument("directory", help="目标目录")
    parser.add_argument("--key", help="HMAC 密钥（可选）")
    parser.add_argument("--output", default="manifest.json", help="清单文件路径")

    args = parser.parse_args()
    key = args.key.encode() if args.key else None

    if args.action == "sign":
        manifest = generate_manifest(args.directory, key)
        with open(args.output, "w") as f:
            json.dump(manifest, f, indent=2, ensure_ascii=False)
        print(f"已生成清单：{args.output}")
        print(f"模式：{manifest['mode']}")
        print(f"文件数：{len(manifest['files'])}")

    elif args.action == "verify":
        with open(args.output) as f:
            manifest = json.load(f)
        modified = verify_manifest(args.directory, manifest, key)
        if modified:
            print("⚠️  发现篡改：")
            for path, reason in modified:
                print(f"  {path}: {reason}")
            sys.exit(1)
        else:
            print("✅ 所有文件完整性验证通过")