030_密码学实战：ECB模式检测技术深度解析——从分组密码原理到重复块识别的完整指南

分组密码模式对比:
┌───────────┐ ┌───────────┐ ┌───────────┐
│   ECB     │ │   CBC     │ │   CTR     │
├───────────┤ ├───────────┤ ├───────────┤
│ 独立加密  │ │ 链式结构  │ │ 计数器模式 │
│ 相同明文  │ │ 密文链接  │ │ 并行处理  │
│ 相同密文  │ │ 安全性高  │ │ 随机性好  │
└───────────┘ └───────────┘ └───────────┘

C_i = E_K(P_i)

# 以16字节（AES块大小）为一组显示
hexdump -C -v ciphertext.bin | grep -A 1 -B 1 "重复的16字节模式"

# 另一种方法：每16字节显示一行
dd if=ciphertext.bin bs=16 | hexdump -C

def detect_ecb_mode(ciphertext, block_size=16):
    """检测密文是否使用ECB模式加密
    
    Args:
        ciphertext: 字节类型的密文数据
        block_size: 分组密码的块大小，默认为16字节（AES）
    
    Returns:
        tuple: (是否为ECB模式的可能性评分, 重复块信息)
    """
    # 将密文分割成块
    blocks = [ciphertext[i:i+block_size] for i in range(0, len(ciphertext), block_size)]
    
    # 去除最后一个可能不完整的块
    if len(blocks[-1]) != block_size:
        blocks = blocks[:-1]
    
    # 计算重复块
    block_count = {}
    for block in blocks:
        if block in block_count:
            block_count[block] += 1
        else:
            block_count[block] = 1
    
    # 计算重复块数量和比例
    duplicate_blocks = {block: count for block, count in block_count.items() if count > 1}
    total_blocks = len(blocks)
    unique_blocks = len(block_count)
    
    # 计算评分（0-1，越高表示越可能是ECB模式）
    # 基于重复块的数量和比例
    if total_blocks > 0:
        # 有重复块的概率评分
        duplicate_score = len(duplicate_blocks) / total_blocks if total_blocks > 0 else 0
        
        # 考虑重复的程度（每个重复块的重复次数）
        repetition_degree = sum(count-1 for count in duplicate_blocks.values()) / total_blocks
        
        # 综合评分
        score = (duplicate_score + repetition_degree) / 2
    else:
        score = 0
    
    # 格式化重复块信息以便于阅读
    readable_duplicates = {block.hex(): count for block, count in duplicate_blocks.items()}
    
    return {
        'is_ecb_likely': score > 0.1,  # 阈值可根据需要调整
        'score': score,
        'total_blocks': total_blocks,
        'unique_blocks': unique_blocks,
        'duplicate_blocks': readable_duplicates,
        'duplicate_count': len(duplicate_blocks),
        'duplicate_ratio': len(duplicate_blocks) / total_blocks if total_blocks > 0 else 0
    }

import argparse
import binascii
import collections
import matplotlib.pyplot as plt

def load_ciphertext(file_path=None, hex_string=None):
    """加载密文数据"""
    if file_path:
        with open(file_path, 'rb') as f:
            return f.read()
    elif hex_string:
        return binascii.unhexlify(hex_string)
    else:
        raise ValueError("必须提供文件路径或十六进制字符串")

def detect_ecb_with_multiple_block_sizes(ciphertext, min_block_size=8, max_block_size=32):
    """使用多个块大小进行ECB模式检测"""
    results = {}
    
    for block_size in range(min_block_size, max_block_size + 1):
        if len(ciphertext) >= block_size:
            results[block_size] = detect_ecb_mode(ciphertext, block_size)
    
    # 找出最可能的块大小（评分最高的）
    if results:
        best_block_size = max(results, key=lambda k: results[k]['score'])
    else:
        best_block_size = None
    
    return results, best_block_size

def visualize_block_distribution(ciphertext, block_size=16):
    """可视化密文块的分布情况"""
    blocks = [ciphertext[i:i+block_size] for i in range(0, len(ciphertext), block_size)]
    if len(blocks[-1]) != block_size:
        blocks = blocks[:-1]
    
    # 计算块的频率
    block_counter = collections.Counter(blocks)
    frequencies = list(block_counter.values())
    
    # 绘制频率分布图
    plt.figure(figsize=(10, 6))
    plt.hist(frequencies, bins=max(frequencies), edgecolor='black')
    plt.title(f'Block Frequency Distribution (Block Size: {block_size} bytes)')
    plt.xlabel('Frequency')
    plt.ylabel('Number of Blocks')
    plt.grid(True, alpha=0.3)
    plt.savefig('block_distribution.png')
    plt.close()
    print("块分布直方图已保存为 'block_distribution.png'")

def main():
    parser = argparse.ArgumentParser(description='ECB模式自动检测工具')
    parser.add_argument('--file', '-f', help='密文文件路径')
    parser.add_argument('--hex', '-x', help='十六进制格式的密文字符串')
    parser.add_argument('--block-size', '-b', type=int, default=16, help='指定块大小（默认：16字节）')
    parser.add_argument('--auto-detect', '-a', action='store_true', help='自动检测块大小')
    parser.add_argument('--visualize', '-v', action='store_true', help='生成块分布可视化图表')
    
    args = parser.parse_args()
    
    # 加载密文
    ciphertext = load_ciphertext(args.file, args.hex)
    print(f"加载的密文长度: {len(ciphertext)} 字节")
    
    # 执行检测
    if args.auto_detect:
        print("正在使用多种块大小进行检测...")
        results, best_block_size = detect_ecb_with_multiple_block_sizes(ciphertext)
        
        print("\n检测结果:")
        for block_size, result in results.items():
            print(f"块大小 {block_size} 字节:")
            print(f"  ECB模式可能性: {'高' if result['is_ecb_likely'] else '低'}")
            print(f"  评分: {result['score']:.4f}")
            print(f"  总块数: {result['total_blocks']}")
            print(f"  唯一块数: {result['unique_blocks']}")
            print(f"  重复块数: {result['duplicate_count']}")
            print(f"  重复比例: {result['duplicate_ratio']:.4f}")
        
        if best_block_size:
            print(f"\n最可能的块大小: {best_block_size} 字节 (评分: {results[best_block_size]['score']:.4f})")
            if args.visualize:
                visualize_block_distribution(ciphertext, best_block_size)
    else:
        print(f"使用指定块大小 {args.block_size} 字节进行检测...")
        result = detect_ecb_mode(ciphertext, args.block_size)
        
        print("\n检测结果:")
        print(f"ECB模式可能性: {'高' if result['is_ecb_likely'] else '低'}")
        print(f"评分: {result['score']:.4f}")
        print(f"总块数: {result['total_blocks']}")
        print(f"唯一块数: {result['unique_blocks']}")
        print(f"重复块数: {result['duplicate_count']}")
        print(f"重复比例: {result['duplicate_ratio']:.4f}")
        
        if result['duplicate_blocks']:
            print("\n重复的块:")
            for block_hex, count in result['duplicate_blocks'].items():
                print(f"  {block_hex}: 出现 {count} 次")
        
        if args.visualize:
            visualize_block_distribution(ciphertext, args.block_size)

if __name__ == "__main__":
    main()

import binascii

def detect_ecb_simple(ciphertext, block_size=16):
    """简单的ECB模式检测"""
    # 分块
    blocks = [ciphertext[i:i+block_size] for i in range(0, len(ciphertext), block_size)]
    
    # 只考虑完整的块
    if len(blocks) > 1 and len(blocks[-1]) != block_size:
        blocks = blocks[:-1]
    
    # 检查是否有重复块
    unique_blocks = set(blocks)
    return len(blocks) != len(unique_blocks)

# 示例使用
# 假设我们有一个十六进制格式的密文
ciphertext_hex = "0123456789abcdef0123456789abcdef"  # 这里是示例，实际应替换为题目密文
ciphertext = binascii.unhexlify(ciphertext_hex)

# 尝试常见的块大小
for block_size in [8, 16, 24, 32]:
    if len(ciphertext) >= block_size:
        is_ecb = detect_ecb_simple(ciphertext, block_size)
        print(f"块大小 {block_size} 字节: {'可能是ECB模式' if is_ecb else '可能不是ECB模式'}")

def detect_ecb_with_unknown_blocksize(ciphertext):
    """未知块大小时的ECB模式检测"""
    results = []
    
    # 尝试常见的块大小（8字节到32字节）
    for block_size in range(8, 33):
        if len(ciphertext) < block_size * 2:  # 至少需要两个块才能检测重复
            continue
        
        # 分块
        blocks = [ciphertext[i:i+block_size] for i in range(0, len(ciphertext), block_size)]
        if len(blocks[-1]) != block_size:
            blocks = blocks[:-1]
        
        # 计算重复情况
        unique_blocks = set(blocks)
        duplicate_count = len(blocks) - len(unique_blocks)
        duplicate_ratio = duplicate_count / len(blocks) if blocks else 0
        
        results.append({
            'block_size': block_size,
            'total_blocks': len(blocks),
            'unique_blocks': len(unique_blocks),
            'duplicate_count': duplicate_count,
            'duplicate_ratio': duplicate_ratio,
            'is_ecb_likely': duplicate_count > 0  # 有重复块就认为可能是ECB
        })
    
    # 按重复比例排序
    results.sort(key=lambda x: x['duplicate_ratio'], reverse=True)
    
    return results

# 示例使用
with open('ciphertext.bin', 'rb') as f:
    ciphertext = f.read()

results = detect_ecb_with_unknown_blocksize(ciphertext)

print("ECB模式检测结果（按可能性排序）:")
for i, result in enumerate(results[:5]):  # 显示前5个最可能的结果
    print(f"\n排名 {i+1}:")
    print(f"  块大小: {result['block_size']} 字节")
    print(f"  总块数: {result['total_blocks']}")
    print(f"  唯一块数: {result['unique_blocks']}")
    print(f"  重复块数: {result['duplicate_count']}")
    print(f"  重复比例: {result['duplicate_ratio']:.4f}")
    print(f"  可能是ECB模式: {result['is_ecb_likely']}")

import requests
import base64

def detect_block_size(encryption_function):
    """检测加密使用的块大小"""
    # 从一个字节开始，逐渐增加，观察密文长度变化
    initial_length = len(encryption_function(b''))
    
    for i in range(1, 64):  # 尝试最多64个字节
        length = len(encryption_function(b'A' * i))
        if length != initial_length:
            # 块大小 = 新长度 - 初始长度
            return length - initial_length
    
    return None  # 没有找到块大小

def detect_ecb_in_oracle(encryption_function, block_size):
    """通过加密预言机检测ECB模式"""
    # 构造包含足够重复内容的明文，确保产生重复的密文块
    plaintext = b'A' * (block_size * 3)
    ciphertext = encryption_function(plaintext)
    
    # 检查是否有重复的块
    blocks = [ciphertext[i:i+block_size] for i in range(0, len(ciphertext), block_size)]
    return len(blocks) != len(set(blocks))

def find_prefix_length(encryption_function, block_size):
    """找到前缀的长度（对齐到块边界）"""
    # 记录空输入的密文
    ciphertext_empty = encryption_function(b'')
    
    # 添加相同的字符，直到密文块发生变化
    for prefix_padding in range(block_size):
        test_input = b'A' * prefix_padding
        current_ciphertext = encryption_function(test_input)
        
        # 检查哪个块开始变化
        for i in range(0, len(current_ciphertext), block_size):
            if current_ciphertext[i:i+block_size] != ciphertext_empty[i:i+block_size]:
                return i - prefix_padding + block_size
    
    return 0

def byte_at_a_time_ecb_decryption(encryption_function, block_size, prefix_length, suffix_length):
    """逐字节ECB解密攻击"""
    # 计算需要填充多少字节才能让前缀+填充刚好对齐到块边界
    prefix_padding = (block_size - (prefix_length % block_size)) % block_size
    
    # 要恢复的明文
    recovered_plaintext = b''
    
    # 计算后缀的总长度（包括填充）
    total_length = len(encryption_function(b'')) - prefix_length
    
    # 逐字节恢复
    for byte_position in range(total_length):
        # 计算当前要攻击的块
        target_block = (prefix_length + prefix_padding + byte_position) // block_size
        
        # 构造足够的填充，使得目标字节是块中的最后一个字节
        padding_length = block_size - 1 - ((prefix_length + prefix_padding + byte_position) % block_size)
        padding = b'A' * (prefix_padding + padding_length)
        
        # 记录此时的密文
        ciphertext = encryption_function(padding)
        target_block_data = ciphertext[target_block*block_size:(target_block+1)*block_size]
        
        # 尝试所有可能的字节值
        for i in range(256):
            # 构造测试输入：padding + known_text + candidate_byte
            test_input = padding + recovered_plaintext + bytes([i])
            test_ciphertext = encryption_function(test_input)
            
            # 比较相应块
            if test_ciphertext[target_block*block_size:(target_block+1)*block_size] == target_block_data:
                recovered_plaintext += bytes([i])
                print(f"恢复的字节: {bytes([i])}，当前明文: {recovered_plaintext}")
                break
    
    return recovered_plaintext

# 示例：与服务器交互的函数

def oracle_function(user_input):
    """模拟加密预言机，实际应用中替换为与服务器的交互"""
    # 假设这是服务器端的加密函数
    # prefix = 未知前缀
    # suffix = 未知后缀，是我们要恢复的内容
    prefix = b'unknown_prefix_'
    suffix = b'this_is_the_secret_message_we_want_to_recover'
    
    # 将前缀、用户输入和后缀连接起来
    data = prefix + user_input + suffix
    
    # 使用ECB模式加密（这里只是模拟，实际应用中需要使用真实的加密方法）
    # 注意：这只是示例，真正的函数应该调用服务器API
    from Crypto.Cipher import AES
    from Crypto.Util.Padding import pad
    
    key = b'YELLOW SUBMARINE'  # 示例密钥
    cipher = AES.new(key, AES.MODE_ECB)
    padded_data = pad(data, AES.block_size)
    ciphertext = cipher.encrypt(padded_data)
    
    return ciphertext

# 攻击流程

# 1. 检测块大小
block_size = detect_block_size(oracle_function)
print(f"检测到的块大小: {block_size} 字节")

# 2. 确认是ECB模式
is_ecb = detect_ecb_in_oracle(oracle_function, block_size)
print(f"是否使用ECB模式: {is_ecb}")

# 3. 找出前缀长度
prefix_length = find_prefix_length(oracle_function, block_size)
print(f"前缀长度: {prefix_length} 字节")

# 4. 计算后缀长度
empty_ciphertext_length = len(oracle_function(b''))
suffix_length = empty_ciphertext_length - prefix_length
print(f"估计的后缀长度: {suffix_length} 字节")

# 5. 逐字节解密
if is_ecb and block_size:
    print("开始逐字节解密...")
    recovered_suffix = byte_at_a_time_ecb_decryption(oracle_function, block_size, prefix_length, suffix_length)
    print(f"\n恢复的后缀: {recovered_suffix}")
else:
    print("不是ECB模式，无法使用此攻击方法")

from PIL import Image
import numpy as np
import matplotlib.pyplot as plt

def load_image(image_path):
    """加载图像"""
    with Image.open(image_path) as img:
        return np.array(img)

def visualize_ecb_patterns(ciphertext_image_path):
    """可视化ECB模式加密的图像中的块模式"""
    # 加载加密图像
    cipher_img = load_image(ciphertext_image_path)
    
    # 假设使用16x16像素的块（对应AES的16字节块，每像素3字节RGB）
    block_size = 16
    height, width, channels = cipher_img.shape
    
    # 创建一个副本用于绘制块边界
    marked_img = np.copy(cipher_img)
    
    # 绘制水平块边界
    for i in range(0, height, block_size):
        marked_img[i:i+1, :, :] = [255, 0, 0]  # 红色水平线
    
    # 绘制垂直块边界
    for j in range(0, width, block_size):
        marked_img[:, j:j+1, :] = [255, 0, 0]  # 红色垂直线
    
    # 显示原始加密图像和标记块边界的图像
    plt.figure(figsize=(15, 10))
    
    plt.subplot(121)
    plt.imshow(cipher_img)
    plt.title('ECB加密图像')
    plt.axis('off')
    
    plt.subplot(122)
    plt.imshow(marked_img)
    plt.title(f'ECB加密图像（块大小: {block_size}x{block_size}）')
    plt.axis('off')
    
    plt.tight_layout()
    plt.savefig('ecb_pattern_visualization.png')
    plt.show()
    
    print("ECB模式可视化结果已保存为 'ecb_pattern_visualization.png'")

def analyze_block_repetition(ciphertext_image_path, block_size=16):
    """分析图像中的块重复情况"""
    cipher_img = load_image(ciphertext_image_path)
    height, width, channels = cipher_img.shape
    
    # 提取所有块
    blocks = {}
    block_positions = {}
    
    for i in range(0, height - block_size + 1, block_size):
        for j in range(0, width - block_size + 1, block_size):
            # 提取块
            block = cipher_img[i:i+block_size, j:j+block_size].tobytes()
            
            # 记录块及其位置
            if block not in blocks:
                blocks[block] = 1
                block_positions[block] = [(i, j)]
            else:
                blocks[block] += 1
                block_positions[block].append((i, j))
    
    # 分析重复块
    total_blocks = len(blocks)
    duplicate_blocks = {b: count for b, count in blocks.items() if count > 1}
    duplicate_count = sum(count-1 for count in duplicate_blocks.values())
    
    print(f"图像分析结果:")
    print(f"图像尺寸: {width}x{height}")
    print(f"分析块大小: {block_size}x{block_size}")
    print(f"总块数: {total_blocks}")
    print(f"重复块数: {len(duplicate_blocks)}")
    print(f"重复次数: {duplicate_count}")
    print(f"重复率: {duplicate_count / total_blocks:.4f}")
    
    # 找出重复次数最多的块
    if duplicate_blocks:
        most_common_block = max(duplicate_blocks.items(), key=lambda x: x[1])
        print(f"\n重复次数最多的块: 出现 {most_common_block[1]} 次")
        print(f"位置示例: {block_positions[most_common_block[0]][:3]}")  # 显示前三个位置

# 示例使用
if __name__ == "__main__":
    # 替换为实际的加密图像路径
    encrypted_image_path = "encrypted_image.png"
    
    # 可视化ECB模式特征
    visualize_ecb_patterns(encrypted_image_path)
    
    # 分析块重复情况
    analyze_block_repetition(encrypted_image_path)

加密模式迁移流程图:
评估现有系统 → 选择替代模式 → 开发新实现 → 并行测试 → 逐步切换 → 安全审计

def cpa_attack_ecb(encryption_oracle, known_prefix, target_block_size=16):
    """
    模拟选择明文攻击ECB模式
    
    Args:
        encryption_oracle: 加密预言机函数，接受明文字节并返回密文
        known_prefix: 已知的前缀内容
        target_block_size: 目标块大小
    
    Returns:
        找到的可能匹配项
    """
    # 生成不同长度的已知前缀，观察密文变化
    results = []
    
    for i in range(1, target_block_size + 1):
        # 构造测试明文：已知前缀 + 填充
        test_plaintext = known_prefix + b'A' * i
        test_ciphertext = encryption_oracle(test_plaintext)
        
        # 分析密文块
        blocks = [test_ciphertext[j:j+target_block_size] for j in range(0, len(test_ciphertext), target_block_size)]
        
        # 记录结果
        results.append({
            'padding_length': i,
            'ciphertext_length': len(test_ciphertext),
            'blocks': blocks
        })
    
    # 分析结果，寻找模式
    print("选择明文攻击结果分析:")
    for result in results:
        print(f"填充长度: {result['padding_length']}, 密文长度: {result['ciphertext_length']}, 块数: {len(result['blocks'])}")
    
    return results

from Crypto.Cipher import AES
from Crypto.Util.Padding import pad

# 生成随机密钥
key = b'0123456789abcdef'  # 16字节密钥

# 创建ECB模式的AES加密器
cipher = AES.new(key, AES.MODE_ECB)

# 加密数据（需要填充到块大小的倍数）
plaintext = b'This is a test message for ECB mode'
padded_data = pad(plaintext, AES.block_size)
ciphertext = cipher.encrypt(padded_data)

# 解密数据
decrypted_padded = cipher.decrypt(ciphertext)

from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend

# 生成随机密钥
key = b'0123456789abcdef'  # 16字节密钥

# 创建ECB模式的AES加密器（注意：cryptography库会警告ECB模式不安全）
cipher = Cipher(algorithms.AES(key), modes.ECB(), backend=default_backend())
encryptor = cipher.encryptor()

# 加密数据（需要自行处理填充）
plaintext = b'This is a test message for ECB mode'  # 确保是块大小的倍数
ciphertext = encryptor.update(plaintext) + encryptor.finalize()