wiki文档

前言

在加密学和计算机科学中,哈希树或默克尔树是一种树形数据结构,其中每个"叶子"节点都被标记为数据块的密码学哈希值,而不是叶子的节点(称为分支、内部节点或inode)则被标记为其子节点标签的密码学哈希值。哈希树允许有效和安全地验证大型数据结构的内容。哈希树是哈希列表和哈希链的一种推广。

证明叶子节点是给定二进制哈希树的一部分需要计算与叶子节点数量的对数成正比的哈希值。相比之下,在哈希列表中,所需的哈希值数量与叶子节点数量本身成正比。因此,默克尔树是加密承诺方案的一个高效示例,其中树的根被视为一个承诺,而叶子节点可以被揭示并证明是原始承诺的一部分。

定义

• 默克尔树是一种基于哈希算法的树形数据结构。它的每个非叶子节点包含其子节点哈希值的组合,根节点的哈希值能够唯一标识整个树中包含的所有数据。

默克尔树的有序性

默克尔树是一种有序的数据结构时,并不是指它的节点数据本身存在大小关系,而是指它的结构存在明确的关联性,从而能够高效地查找和验证数据。

具体来说:

• 默克尔树中的节点并不需要按照任何大小顺序排列,它们可以是无序的原始数据。
• 但是,默克尔树的结构本身是有序的,因为每个非叶子节点的哈希值都是由其两个子节点的哈希值计算得出的。
• 这种自底向上的哈希计算过程,建立了节点之间的有序关联关系,使得整棵树可以被高效地遍历和验证。
• 根节点的哈希值作为整棵树的数字签名,可以快速确认树中任意节点的完整性。

产生背景

• 默克尔树由拉尔夫·默克尔(Ralph Merkle)在1979年提出,旨在解决数字签名的问题。
• 在分布式系统中,如何有效地验证数据的完整性和一致性是一个重要问题。
• 传统的方法是对整个数据集进行哈希计算,但这需要下载整个数据集,效率低下。

工作原理

• 将需要存储的数据分割成固定大小的数据块。
• 计算每个数据块的哈希值,作为默克尔树的叶子节点。
• 将相邻的两个叶子节点的哈希值组合起来,计算出父节点的哈希值。
• 依此递归计算,直到得到整个树的根节点哈希值。
• 根节点的哈希值就是整个默克尔树的标识符,也称为根哈希值。

优势

• 默克尔树可以高效地验证数据的完整性和一致性。
• 在分布式系统中,只需要下载根哈希值,就可以验证某个数据块是否包含在整个数据集中,大大减少了带宽开销。
• 默克尔树的存储开销也很低,只需要存储根哈希值就可以。

数据的构建

这种基于固定大小数据块的默克尔树结构非常灵活和通用。它可以适用于各种类型的二进制数据,只要能够将数据划分成固定大小的块。这样做的好处包括:

• 简单易实现
• 可以并行处理数据块
• 支持增量更新和部分验证
• 适用于各种分布式系统和存储应用

数据块格式

每个数据块都可以是任意格式的二进制数据,比如字符串、图片、视频等。关键是这些数据块的大小需要是固定的。

数据块列表

这些固定大小的数据块会被组织成一个列表或数组,作为输入传递给默克尔树的构建函数。

构建过程

默克尔树的构建函数会遍历这个数据块列表,计算每个数据块的哈希值,然后递归地构建出默克尔树的层级结构,最终得到根节点的哈希值。

应用场景

区块链技术

• 在比特币和以太坊等区块链系统中,默克尔树被广泛应用于存储和验证交易数据。
• 每个区块中都包含一个默克尔树的根哈希值,用于验证该区块中所有交易的完整性。
• 这样即使只下载区块头部信息(包含根哈希值),也可以验证某笔交易是否包含在区块中。

分布式存储

• 在分布式云存储系统中,默克尔树用于存储和验证存储在不同节点上的文件块。
• 只需下载根哈希值,就可以验证某个文件块是否存在于整个文件系统中。这大大减少了下载开销。

数字版权管理

• 在数字内容版权管理中,默克尔树可用于有效地管理和验证数字资产的所有权。
• 每个数字内容可以对应一棵默克尔树,根哈希值作为该内容的唯一标识。

数据完整性验证

• 在任何需要验证数据完整性的场景中,默克尔树都可以发挥作用。
• 例如在软件更新包、数字证书、日志文件等场景中,使用默克尔树可以有效地验证数据的完整性。

点对点网络

• 在点对点(P2P)网络中,默克尔树可用于有效地传输和验证文件。
• 只需下载文件的根哈希值,就可以验证文件的完整性,而不需要下载整个文件。

python构建默克尔树

import hashlib
from typing import List, Optional, Tuple

class MerkleNode:
    def __init__(self, data: bytes = None, left: 'MerkleNode' = None, right: 'MerkleNode' = None):
        self.data = data
        self.left = left
        self.right = right
        self.hash = None

    def calculate_hash(self, hash_function=hashlib.sha256) -> None:
        if self.data:
            self.hash = hash_function(self.data).digest()
        elif self.left and self.right:
            self.hash = hash_function(self.left.hash + self.right.hash).digest()
        else:
            self.hash = None

class MerkleTree:
    def __init__(self, data_blocks: List[bytes], hash_function=hashlib.sha256):
        self.hash_function = hash_function
        self.root = self.build_merkle_tree(data_blocks)

    def build_merkle_tree(self, data_blocks: List[bytes]) -> Optional[MerkleNode]:
        """
        递归构建默克尔树
        :param data_blocks: 数据块列表
        :return: 根节点
        """
        if not data_blocks:
            return None

        if len(data_blocks) == 1:
            return MerkleNode(data=data_blocks[0])

        mid = len(data_blocks) // 2
        left_root = self.build_merkle_tree(data_blocks[:mid])
        right_root = self.build_merkle_tree(data_blocks[mid:])
        root = MerkleNode(left=left_root, right=right_root)
        self.update_hashes(root)
        return root

    def update_hashes(self, node: MerkleNode) -> None:
        """
        递归更新节点的哈希值
        :param node: 当前节点
        """
        if node:
            self.update_hashes(node.left)
            self.update_hashes(node.right)
            node.calculate_hash(self.hash_function)

    def update_data_block(self, index: int, new_data: bytes) -> None:
        """
        递归更新指定位置的数据块,并重构默克尔树
        :param index: 数据块索引
        :param new_data: 新的数据块内容
        """
        self.root = self.rebuild_from_index(self.root, index, new_data)

    def add_data_block(self, new_data: bytes) -> None:
        """
        递归添加新的数据块,并重构默克尔树
        :param new_data: 新的数据块内容
        """
        self.root = self.rebuild_from_index(self.root, len(self.get_leaves()), new_data)

    def remove_data_block(self, index: int) -> None:
        """
        递归删除指定位置的数据块,并重构默克尔树
        :param index: 数据块索引
        """
        self.root = self.rebuild_from_index(self.root, index)

    def rebuild_from_index(self, node: MerkleNode, index: int, new_data: bytes = None) -> MerkleNode:
        """
        递归从指定索引开始重构默克尔树
        :param node: 当前节点
        :param index: 数据块索引
        :param new_data: 可选的新数据块内容
        :return: 重构后的根节点
        """
        if not node:
            return None

        if index == 0:
            if new_data:
                return MerkleNode(data=new_data)
            else:
                return self.build_merkle_tree(self.get_leaves()[1:])

        left_count = self.get_leaf_count(node.left)
        if index < left_count:
            node.left = self.rebuild_from_index(node.left, index, new_data)
        else:
            node.right = self.rebuild_from_index(node.right, index - left_count, new_data)

        self.update_hashes(node)
        return node

    def get_leaves(self) -> List[MerkleNode]:
        """
        递归获取所有叶子节点
        :return: 叶子节点列表
        """
        return self.get_leaves_helper(self.root, [])

    def get_leaves_helper(self, node: MerkleNode, leaves: List[MerkleNode]) -> List[MerkleNode]:
        if not node:
            return leaves
        if not node.left and not node.right:
            leaves.append(node)
        else:
            leaves = self.get_leaves_helper(node.left, leaves)
            leaves = self.get_leaves_helper(node.right, leaves)
        return leaves

    def get_leaf_count(self, node: MerkleNode) -> int:
        """
        递归获取叶子节点的数量
        :param node: 当前节点
        :return: 叶子节点数量
        """
        if not node:
            return 0
        if not node.left and not node.right:
            return 1
        return self.get_leaf_count(node.left) + self.get_leaf_count(node.right)

    def verify_proof(self, data: bytes, proof: List[bytes]) -> bool:
        """
        递归验证数据块的默克尔证明
        :param data: 待验证的数据块
        :param proof: 默克尔证明列表
        :return: 验证结果
        """
        return self.verify_proof_helper(data, proof, self.root)

    def verify_proof_helper(self, data: bytes, proof: List[bytes], node: MerkleNode) -> bool:
        if not node:
            return False

        computed_hash = self.hash_function(data).digest()
        for node_hash in proof:
            computed_hash = self.hash_function(computed_hash + node_hash).digest()

        return computed_hash == node.hash

# 示例用法
data_blocks = [b"block1", b"block2", b"block3", b"block4"]
merkle_tree = MerkleTree(data_blocks)
print(f"Merkle tree root hash: {merkle_tree.root.hash.hex()}")

# 更新数据块
merkle_tree.update_data_block(1, b"updated_block2")
print(f"Merkle tree root hash after update: {merkle_tree.root.hash.hex()}")

# 添加数据块
merkle_tree.add_data_block(b"block5")
print(f"Merkle tree root hash after addition: {merkle_tree.root.hash.hex()}")

# 删除数据块
merkle_tree.remove_data_block(0)
print(f"Merkle tree root hash after deletion: {merkle_tree.root.hash.hex()}")

# 验证数据块
block = b"block2"
proof = [node.hash for node in merkle_tree.get_leaves() if node.data != block]
print(f"Verified: {merkle_tree.verify_proof(block, proof)}")

这个实现使用了递归方式来构建、更新、重构和验证默克尔树。主要功能如下:

构建默克尔树: 通过递归地构建二叉树结构,计算每个节点的哈希值。
更新数据块: 可以递归地更新指定位置的数据块,并重构整个默克尔树。