038_密码学实战：高级哈希碰撞技术深度解析——从生日攻击原理到MD5碰撞构造的完整指南

哈希函数安全特性:
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ 单向性      │ │ 抗弱碰撞    │ │ 抗强碰撞    │
├─────────────┤ ├─────────────┤ ├─────────────┤
│ 无法逆向    │ │ 难找到原像  │ │ 难找到碰撞  │
│ 计算容易    │ │ 给定h找M    │ │ 找M≠M'同h   │
└─────────────┘ └─────────────┘ └─────────────┘

MD5差分路径示例：
状态A → ΔA → 状态A'
状态B → ΔB → 状态B'
状态C → ΔC → 状态C'
状态D → ΔD → 状态D'

def modify_message(message, target_condition, position):
    """修改消息的特定位以满足目标条件"""
    modified = bytearray(message)
    # 计算需要翻转的位
    bit_to_flip = calculate_bit_to_flip(target_condition, position)
    # 执行位翻转
    modified[position] ^= (1 << bit_to_flip)
    return bytes(modified)

def check_conditions(message, conditions):
    """检查消息是否满足所有差分条件"""
    state = md5_initial_state()
    for i, block in enumerate(split_into_blocks(message)):
        state = md5_compress(state, block)
        # 检查当前块的条件
        if i < len(conditions) and not verify_conditions(state, conditions[i]):
            return False
    return True

import hashlib
import struct

# 这是一个简化的示例，实际的MD5碰撞构造需要更复杂的算法
# 这里我们使用已知的碰撞对来演示

def generate_md5_collision():
    """生成MD5碰撞对"""
    # 这是两个已知的MD5碰撞块（128字节）
    # 这些块来自于Marc Stevens等人的研究
    block1 = bytes.fromhex(
        'd131dd02c5e6eec4693d9a0698aff95c2fcab58712467eab4004583eb8fb7f8955ad340609f4b30283e488832571415a085125e8f7cdc99fd91dbdf280373c5bd8823e3156348f5bae6dacd436c919c6dd53e2b487da03fd02396306d248cda0e99f33420f577ee8ce54b67080a80d1ec69821bcb6a8839396f9652b6ff72a70'
    )
    block2 = bytes.fromhex(
        'd131dd02c5e6eec4693d9a0698aff95c2fcab50712467eab4004583eb8fb7f8955ad340609f4b30283e4888325f1415a085125e8f7cdc99fd91dbd7280373c5bd8823e3156348f5bae6dacd436c919c6dd53e23487da03fd02396306d248cda0e99f33420f577ee8ce54b67080280d1ec69821bcb6a8839396f965ab6ff72a70'
    )
    
    # 计算两个块的MD5哈希值，应该相同
    hash1 = hashlib.md5(block1).hexdigest()
    hash2 = hashlib.md5(block2).hexdigest()
    
    return block1, block2, hash1, hash2

# 生成并验证碰撞
def verify_collision():
    block1, block2, hash1, hash2 = generate_md5_collision()
    
    print(f"块1的MD5哈希: {hash1}")
    print(f"块2的MD5哈希: {hash2}")
    print(f"两个块是否不同: {block1 != block2}")
    print(f"哈希值是否相同: {hash1 == hash2}")
    
    # 打印两个块的差异位置
    diff_positions = []
    for i in range(len(block1)):
        if block1[i] != block2[i]:
            diff_positions.append(i)
    
    print(f"差异位置数量: {len(diff_positions)}")
    print(f"差异位置: {diff_positions}")

if __name__ == "__main__":
    verify_collision()

def sha1_collision_search():
    # 1. 初始化两个消息块，它们具有特定的差分
    msg1 = generate_initial_message_with_pattern()
    msg2 = msg1 ^ create_difference_pattern()
    
    # 2. 应用消息修改技术满足差分路径条件
    iterations = 0
    while iterations < MAX_ITERATIONS:
        # 计算前几轮的状态差异
        state_diff1 = calculate_state_differences(msg1)
        state_diff2 = calculate_state_differences(msg2)
        
        # 检查是否满足差分条件
        if check_differential_conditions(state_diff1, state_diff2):
            # 3. 寻找完全碰撞
            if find_full_collision(msg1, msg2):
                return msg1, msg2
        
        # 修改消息并继续
        msg1 = modify_message_for_collision(msg1)
        msg2 = modify_message_for_collision(msg2)
        iterations += 1
    
    return None, None

消息1: PREFIX || SUFFIX1
消息2: PREFIX || SUFFIX2
H(消息1) = H(消息2)

def perform_length_extension_attack(original_message, original_hash, known_prefix_length, new_suffix):
    # 从哈希值恢复内部状态
    internal_state = extract_internal_state(original_hash)
    
    # 构造扩展消息
    padding = create_padding(original_message)
    extended_message = original_message + padding + new_suffix
    
    # 继续哈希计算
    new_hash = continue_hash_computation(internal_state, new_suffix)
    
    return extended_message, new_hash

def second_preimage_attack(original_message, hash_function):
    # 获取原始哈希值
    target_hash = hash_function(original_message)
    
    # 尝试修改原消息的特定部分
    for i in range(MAX_ATTEMPTS):
        modified = modify_controlled_portion(original_message)
        if hash_function(modified) == target_hash and modified != original_message:
            return modified
    
    return None

from concurrent.futures import ProcessPoolExecutor

def parallel_collision_search(initial_seed, num_processes, hash_function):
    # 划分子空间
    workspaces = divide_search_space(initial_seed, num_processes)
    
    # 并行搜索
    with ProcessPoolExecutor(max_workers=num_processes) as executor:
        futures = [executor.submit(search_subspace, ws, hash_function) for ws in workspaces]
        
        # 检查结果
        for future in futures:
            result = future.result()
            if result:
                return result
    
    return None

import hashlib
import itertools

def solve_hash_collision_challenge(prefix, target_hash_length, max_attempts=1000000):
    # 挑战要求：构造消息 prefix + x，使得其哈希值前 target_hash_length 位为特定值
    target_prefix = "0" * target_hash_length
    
    # 尝试不同的附加数据
    for i in itertools.count():
        if i >= max_attempts:
            return None
        
        # 生成附加数据
        suffix = f"{i:x}"
        message = prefix + suffix
        
        # 计算哈希值
        hash_value = hashlib.sha256(message.encode()).hexdigest()
        
        # 检查是否满足条件
        if hash_value.startswith(target_prefix):
            return message, hash_value

def solve_prefix_collision_challenge(required_prefix, target_hash):
    # 挑战要求：构造以 required_prefix 开头且哈希值为 target_hash 的消息
    padding_length = len(required_prefix) + 8  # 假设8字节的填充区域
    
    # 使用生日攻击的变种
    seen = {}
    for i in range(1000000):
        # 生成部分可控的消息
        control_bytes = f"{i:x}".zfill(8)
        message = required_prefix + control_bytes
        
        # 计算哈希值
        current_hash = hashlib.md5(message.encode()).hexdigest()
        
        # 检查是否匹配
        if current_hash == target_hash:
            return message
        
        # 记录用于可能的碰撞
        if current_hash in seen:
            print(f"找到碰撞：{seen[current_hash]} 和 {message}")
        seen[current_hash] = message
    
    return None

def evaluate_hash_security(hash_function):
    results = {
        "collision_resistance": {
            "theoretical_complexity": get_theoretical_collision_complexity(hash_function),
            "best_known_attack": get_best_known_collision_attack(hash_function),
            "safety_margin": calculate_safety_margin(hash_function, "collision")
        },
        "preimage_resistance": {
            "theoretical_complexity": get_theoretical_preimage_complexity(hash_function),
            "best_known_attack": get_best_known_preimage_attack(hash_function),
            "safety_margin": calculate_safety_margin(hash_function, "preimage")
        },
        "implementation_security": check_implementation_vulnerabilities(hash_function),
        "performance": benchmark_hash_performance(hash_function)
    }
    
    # 综合评估
    security_level = calculate_overall_security_level(results)
    results["overall_security_level"] = security_level
    
    return results

import hashlib
import os

def secure_hash(message, salt=None):
    # 如果没有提供盐值，生成随机盐
    if salt is None:
        salt = os.urandom(16)
    
    # 第一步：使用SHA-256计算加盐哈希
    h1 = hashlib.sha256(salt + message).digest()
    
    # 第二步：使用BLAKE2b进一步处理
    h2 = hashlib.blake2b(h1 + salt).hexdigest()
    
    return salt, h2

def verify_secure_hash(message, salt, expected_hash):
    # 重新计算哈希并验证
    _, computed_hash = secure_hash(message, salt)
    return computed_hash == expected_hash