理解区块链背后的Merkle Tree

你可以在Github上获取最新的源代码（C#）

目录

• 简介
  • 本文中的术语
• Merkle Tree被应用在哪里?
  • 数字货币
  • 全球供应链
  • 保健行业
  • 资本市场
  • Git 和 Mercurial
• 为什么使用Merkle Tree?
  • 一致性检验
  • 数据校验
  • 数据同步
  • 证明的重要性
• Merkle Tree实践
  • 数据校验（审计证明）是如何实现的?
  • 一致性检验（一致性证明）是如何实现的?
    • 一致性证明演示
    • 重建旧的根哈希
    • 一致性证明的注意事项
    • 一致性证明的常规算法
• 示例
  • 改变叶节点数
  • 审计证明测试
  • 一致性证明测试
• 集成Merkle Tree
  • MerkleHash 类
  • MerkleNode 类
  • MerkleTree 类
  • 属性和字段
    • Contract 方法
    • 构造器
    • 追加叶节点
    • 在现有的树上添加树
    • 根据树的叶节点构造树
    • AuditProof（审计证明）
    • ConsistencyProof（一致性证明）
    • VerifyAudit（审计证明校验）
    • 将审计证明作为哈希对进行校验
    • VerifyConsistency（一致性证明校验）
    • FindLeaf（在叶节点列表中寻找某一个叶节点）
    • 根据自定义的节点要求来创建MerkleNode
    • 其他
  • MerkleProofHash 类
• 单元测试
  • 一致性证明的单元测试
• 单调哈希和区块链
• 结论
• 参考文献

简介

在1979年，Ralph Merkle取得了哈希树即现在广为人知的MerkleTree的专利权（改专利在2002年过期）。其概括的描述为：“该发明包含了一种提供信息验证的数字签名的方法，该方法利用单向的认证树对密码数字进行校验。”

可能你更喜欢维基百科中的定义：“在密码学和计算机科学当中，哈希树或Merkle tree是一种特殊的树结构，其每个非叶节点通过其子节点的标记或者值（子节点为叶节点）的哈希值来进行标注。哈希树为大型的数据结构提供了高效安全的验证手段。哈希树也可以理解为哈希列表和哈希链表的泛化产物”

本文中的术语

除了对外部资料的引用外，我尽量保持了本文中术语的一致性。

记录(Record)——一个用于描述对应Merkle tree中叶节点哈希后对应的数据包。当你阅读Merkle tree相关的内容的时候，根据上下文的不同，它也可能被称为“交易”或“凭证”。

区块(Block)——从比特币中引用的概念，我将用“区块”指代永久存储在Merkle tree叶节点中的记录集。引用来源：“交易数据被永久存储在称为区块的文件当中”。它可以被理解为城市档案或者股票交易账本中独立的几页。换言之：“记录被永久存储在称为区块的文件当中。”

Log——Merkle tree的别称，log是从哈希的记录中构造得到的哈希树。除了用来表示哈希树之外，log还有一个特殊的属性：新的记录总会被作为新的叶节点追加在树最后的叶节点上。除此之外，对于交易系统来说（比如说货币），一旦某个记录被“logged”，它就不能再被更改了——相反地，对交易的更改将在log为表示为新的记录条目，为交易提供完整的审计线索。与之相反的，在分布式存储中（像NoSQL数据库）可以更改记录，同时会触发树中收到影响的记录的哈希值的更新。在这种场景下，Merkle tree可以快速高效地识别已经更改的记录以便同步分布式系统中的节点。

Merkle Tree被应用在哪里？

数字货币

Merkle tree（以及其变体）被应用于比特币，以太坊，Apache Cassandra以及其他用于提供以下服务的系统当中：

• 一致性验证
• 数据校验
• 数据同步（这一部分在本文中不做讲述，因为数据同步本身的内容就可以再写一篇文章）

这些术语都是什么意思呢？我后面会一一讲解。

利用了Merkle tree的区块链技术，现今受欢迎的程度不亚于比特币。需要跟踪数据并且保证数据一致性的企业也开始看到区块链技术对这一过程的帮助。

全球供应链

拿实例来说，IBM和Maersk正在合作使用区块链来管理全球的供应链：

“科技巨头IBM和领先的运输物流公司Maersk宣布了一个潜在的开创性合作——使用区块链技术来数字化全球范围的交易信息和货运管理商、海运承运商、参与供应链的港口和海关部门组成的托运网络。

据IBM和Maersk所说，如果这项技术被广泛应用，可以改变全球的跨境供应链并为行业节省数十亿美元。”

保健行业

“驱动Google健康技术的AI子公司DeepMind Health计划使用基于比特币的新技术来赋予医院、国民保健系统（NHS）以至患者实时监控个人身体情况的能力。它也被称为可验证的数据审计，这项计划将会将患者的行为以加密可验证的方式记录下来据此创建每个人的特殊的数字记录 ‘账本’ ”。这意味着对数据的任何修改、访问行为都将是可见的。

资本市场

虽然比特币技术最初就被当做一个可以被自由获取的，实用的技术来替代传统分布式共享网络中各方资产记录和交易信息的储存和记录手段，但是2015年许多金融科技初创公司都将重点放在了只有通过预先授权的参与者才能访问的私有区块链开发上。GreySpark认为这与金融初创公司关键的商业诉求——为银行和其他买方公司设计、提供一个广泛、通用的区块链解决方案实现交易前到交易后整个生命周期中分布式账本技术的正常运转。

Git 和 Mercurial

显然（尽管我没有找到关于此事的权威论据）Git 和 Mercurial 使用了特殊处理过的Merkle trees来进行版本管理。

为什么使用Merkle Tree？

使用像Merkle tree一样的哈希树

1. 显著减少了要达到证明数据完整性的置信度所需的数据量。
2. 显著减少了维护一致性、数据校验以及数据同步所需的网络I/O数据包大小。
3. 将数据校验和数据本身分离——Merkle tree可以存在本地，也可以存放与受信任的权威机构上，也可以存在分布式系统上（你只需要维护属于你的树即可。）将“我可以证明这个数据是合法的”和数据本身解耦意味着你可以为Merkle Tree和数据存储提供适当的分离（包括冗余）持久性。

所以本节标题问题的答案可以分为以下三点：

1. Merkle tree提供了一种证明数据完整性/有效性的手段。
2. Merkle tree可以利用很少的内存/磁盘空间实现简单高效的计算。
3. Merkle tree的论证和管理只需要很少量的网络传输流量。

一致性验证

也被称为“一致性证明”，你可以用它验证两份日志的版本是否一致：

1. 最新的版本包含了之前所有版本的信息。
2. 日志中的记录顺序是一致的。
3. 所有新的记录都是跟在旧版本记录的后面的。

如果你证明了日志的一致性，那么意味着：

• 日志中没有凭证记录被回滚或者插入。
• 日志中没有凭证被修改。
• 日志并没有经过分支（breached）或者被fork

一致性验证是保证你的日志没有损坏的关键手段。“监督员和审计员经常使用一致性验证来确认日志行为是否正常”

数据校验

也被称为“审计证明”，这是因为它可以让你知道某一条具体的记录是否存在于日志当中。与一致性验证一样，维护日志的服务器需要提供给客户端特定的记录存在于日志当中的证据。任何人都可以用一份日志来请求Merkle审计证明，校验某条凭证记录确实存在于日志当中，审计者会将这些类型的请求发送至日志以便它们检验TLS客户端的证书。如果Merkle审计证明不能生成与Merkle Tree哈希值匹配的根哈希值，则表示证书没有在日志当中。（根节点包含什么、审计证明是如何工作的这些内容会在稍后提到。）

向客户端发送证明还有另外一个原因：它证明了服务器本身并没有创造正确的答案，它只是为你和客户端提供相关的证明，而伪造一个证明在计算上是不可能的。

数据同步

Merkle tree在分布式数据存储中的数据同步中发挥着重要的作用，这是因为它允许分布式系统中的每个节点可以迅速高效地识别已经更改的记录而无需将发送所有的数据来进行比对。一旦树中有特定的叶节点的变更被识别，我们只需要将与该特定叶节点相关的数据上传至网络即可。注意Merkle tree并没有直接提供解决冲突和将多个写入者同步到相同记录的机制。我们后面将演示这是如何实现的。

正如我开头所说，因为数据同步本身的内容就很多，所以你必须要等下一篇文章。基本的数据同步（叶节点的更改）是很简单的，但是动态环境下（存在叶节点的新增和删除）的数据同步就要复杂得多了，这是一个非平凡的问题。从技术上来讲，你可能不希望为此使用Merkle tree，因为这里数据的审计证明和一致性证明通常是无用的，但我认为在分布式数据同步的场景下这仍然是值得的，因为有可能过程中有叶节点在没有被彻底删除时就被标记为被删除。所以，Merkle tree的垃圾回收是数据同步中的一个问题，至少从我看这个问题的角度是这样的。

证明的重要性

一致性证明和审计证明的重要性在于客户端可以自己进行验证。这意味着当客户端请求服务器来验证一致性或者某个订单是否存在时，服务器并不是简单地回复答案“是”或“不是”，即使在“是”的情况下也会向你发送客户端可以验证的相关的证明。这些证明是基于服务器对当前Merkle Tree的已有认知，而这是不能被某些希望客户端相信它们的数据是有效的恶意用户复制重复的。

在分布式系统中，每个节点都维护着它自己数据所在的Merkle tree，在同步的过程中，任何已经修改的节点都隐性地向其他的节点证明了自身的有效性。这也保证了任何节点不可能跳到网络的层级上说“我有了一个新的记录”或者“我有一个记录来替换另一个的记录”，因为每个节点缺乏必要的信息来向其他节点证明自身。

Merkle Tree实践

Merkle一般来说就是一个二叉树，它的每个叶节点的值为与它包含的记录的哈希值。其他的内部节点的值为和它相连的两个子节点中哈希值合并后再次哈希的结果。将子节点的哈希值合并后再次哈希创建节点的过程将不断重复直至抵达顶部的根节点，也称为“根哈希”。

上面的图示模拟了子节点哈希值的级联，我们在下面的文章中也将沿用这个模拟方式。

数据校验（审计证明）是如何实现的？

如下图所示，你是图表中记录“2”的拥有者。你也拥有来自权威机构提供的根哈希值，在我们的图示中就是“01234567”。你询问服务器来证明你的记录“2”确实在树当中。服务器返回给你的将是下面黄色标记的哈希值“3”，“01”，“4567”。

利用这些返回的数据（包含这些数据叠加所需的左右位置信息），可以进行如下证明过程：

• 2 + 3 得到 23
• 01 + 23 得到 0123
• 0123 + 4567 得到 01234567

由于你已经拥有来自权威机构得到的根哈希值“01234567”，计算结果与其一直证明了记录“2”确实存在于树当中。此外，你获取的证明来自的系统也证明了它的“权威性”，因为你可以用你的记录“2”和它提供的哈希值重建出根哈希值“01234567”。任何假冒的验证系统都不能为你提供以上的中间哈希值，因为你只是向服务器索要证明，并没有提供给服务器你的根哈希值——它不知道你的哈希值，只有你自己知道你的哈希值。

为了完成以上校验，需要提供给你的树的信息是很少的。相应的，要完成这个证明所需的数据包也非常小，这使得完成计算所需的信息在网络中的发送更为高效。

一致性检验（一致性证明）是如何实现的？

适用于树的一致性验证只限于新增节点，就像日志记录一样。它通常不会被用在叶节点需要更新的系统当中，因为这需要同步旧的根哈希值。新增的一致性检验是可行的，下面我们将看到这个过程可能并不是我们想象的那样。RFC 6962中的Section2.1.4提供了一些很好的框图来说明一致性证明是如何实现的，但是初看可能会比较费解，所以我在这里尽力解释地更清楚一些。使用他们的例子（但是还是用我上面的框图以及其中的哈希值来演示），我们从包含三个记录的树开始：

最初的三个记录“012”被创建：

第四个记录“3”被添加之后，这个树变成了：

两个新的的记录“45”被添加：

最后添加记录“6”：

每次我们附加的子树（记录集）：

012

3

45

6

我们有：

• 所有的子树附加在原始的树上之后得到的新的根哈希
• 在附加子树之前每个树的原始哈希值

我们现在想要验证尽管子树附加在原始的树上后整个树的根哈希已经改变，但是之前的子树的根哈希仍然可以被重建。这验证了可信的权威服务器上和客户端所在的机器上记录的顺序以及与记录相关的哈希值并没有被改动。

一致性证明演示

每个子树附加在主树（master tree）上的时候，一致性证明为新树重建了哈希值（上面的最后一张图）：

1. 第一个树的根哈希值为“012”。
2. 当第二个子树附加之后，根哈希值变为了“0123”。
3. 当第三个子树附加之后，根哈希值变为了”012345“
4. 当最后一个子树附加之后，根哈希值变为了”0123456“

第一颗树的一致性检验

我们怎样检验第一个子树呢？它的三个叶节点是否还在新的树当中呢？

如上图所示，我们需要节点"01"和"2"来重建第一棵树的根哈希值"012"

第二棵树的一致性检验

当我们在第一棵树上附加只有一个叶节点"3"的第二棵树之后，生成的树具有四个叶节点，其根哈希值变为了"0123"。

第二棵树的一致性证明就是"0123"这个节点。

第三棵树的一致性证明

第三棵树添加了两个叶节点”45“，根哈希值变为了”012345“，我们获得的一致性证明如下图所示：

另一种情况下的一致性证明

假设我们依次单独添加了叶节点”4“，"5"，"6"。当我们添加叶节点"4"的时候，我们会得到树此时的根哈希值"01234"。在"4"，“5”，“6”被添加后我们的树共有7个叶节点，要重建根哈希值“01234”需要的节点如下图黄色标注所示：

以上实例中一致性验证的最后一个情况

这个情况与上述不同的地方在于需要三个节点的哈希值来重建原始的根哈希值。给定的树一共有8个叶节点，当第七个叶节点“6”被添加后，此时的根哈希值为“0123456”。要重建它需要图示中的黄色节点：

重建旧的根哈希值

最后的例子展示了一致性证明如何在已有节点中重建出旧的根哈希值“0123456”。一致性证明给予我们的节点顺序如下：

0123

45

6

要重建出原始的根哈希值，我们需要按次序依次结合这些哈希值

45 + 6 = 456

0123 + 456 = 0123456

一致性验证中的注意事项

一致性验证并不是一个简单的算法，我们在实现的时候需要遵循以下说明和规则：

• 一致性验证是基于计算来验证树中前m个叶节点的。当一个新的节点被添加至树上之后，m为合并后的叶节点数。为了保证数据的所有哈希值的顺序和内容没有被更改，这一步是必不可少的。
• 不过当知道中间根哈希值代表的叶节点数量之后，这个算法可以快速地识别出具有以下特殊的节点：
  • 匹配的哈希值（如果叶节点的数量为2的n次方）
  • 重建中间哈希值所需的子节点组合。
• 这意味着当一棵树被添加至“日志”（已有的记录）当中时，树中节点的数将和根哈希值一起被保存。
• 在一些情况下，一些特定的系统会单独负责树的添加，因为这个系统知道每次添加的树的叶节点数，所以这个过程会变得很简单。
• 在分布式系统当中，其他的参与者也可以在树上新增节点（以及需要避免两个及以上的参与者同时执行添加操作引入的复杂性），参与者在添加的时候需要声明自己添加的叶节点的数量以便后续一致性检验的进行。
• 为了减少所需的额外知识，完整的Merkle tree的维护者可以利用字典在需要被添加的树的根节点与叶节点的数目之间建立映射。这也许是管理树所需信息的更安全的方式。

另一方面，由于这个算法从最左侧的深度为log2(m)的节点开始，它的效率也是很高的，这种方法在避免从第一个节点开始计算很多叶子的哈希值的同时依然保证了生成的根哈希值的有效性和记录、记录次序的合法性。

一致性证明的常规算法

你可以直接通读RFC 6962的2.1.2节或者通过下面展示的图表来理解一致性证明的常规算法。

规则1：

寻找到树最左侧的节点来开始我们的一致性证明。一般来说，它会是获取旧的哈希值所需的叶节点之一。

在给定添加子树之后主树的叶节点数之后，我们需要n=log2(m)步来找到最多代表m个叶节点的内部节点索引。索引值从最树左侧的枝开始计数，如下图所示。

例：

3个叶节点：log2(3) = 1.58(我们从节点“01”开始)

4个叶节点：log2(4) = 2(我们从节点“0123”开始)

6个叶节点：log2(6) = 2.58(我们从节点“0123”开始)

这一规则提供的索引值告诉了我们应当从哪个节点开始计算哈希值。从正确的地方开始的同时我们也知道了子树将从这个节点开始叠加或这个节点是当前节点哈希和子树节点哈希组合的位置（如果log2(m)存在余数的话）。下面的图示可以帮助你更好地理解（注意到计算得到的索引为3不代表着就有2^3=8个节点，如下图所示，可能还有部分节点还没有添加进来）：

除此之外，我们还要设定当前节点包含的子节点数（因为如上图所示最后一片叶节点可能会丢失，所以我们必须计算他们的叶节点数量）

我们还要将初始的兄弟节点设置为规则一获取得到的节点的兄弟节点（如果获取得到的话）。

如果 m-k==0，我们将继续执行规则3。

下面我们用上面的图示和三个实例一起说明：

m=2：index=1（节点“01”），有两个叶节点，m-k=0，所以我们继续执行规则3。

m=4：index=2（节点“0123”），有四个叶节点，m-k=0，我们继续执行规则3。

m=3：index=1（节点“01”），有两个叶节点，m-k=1不等于0，所以我们继续执行规则2。

规则2：

如果m-k == 兄弟节点（SN）的叶节点数，将兄弟节点的哈希值和旧的根哈希值串联之后取代原先的根哈希值。

如果m-k < 兄弟节点的叶节点数，将兄弟节点设定为兄弟节点的左节点并重复规则2。

如果m-k > 兄弟节点的叶节点数，将兄弟节点的哈希值串联并将k加上兄弟节点的叶节点数，设定兄弟节点为它的父节点的右节点。

按照这样的规则我们总是可以从结果的哈希值中重建出旧的根哈希值。

分情况讨论演示：

m=3。k=2（节点“01”下的叶节点数），SN=“23”

SN的叶节点数为2，m-k<2，所以令SN = SN的左子节点“2”

此时SN的节点数为1（SN此时就是叶节点）

m-k= SN的叶节点数，此时可以看到我们可以用哈希值“01”和哈希值“2”重组得到根哈希值“012”。

m=4时由规则1处理。

m=5,k=4（节点“0123”下的叶节点数）。SN=“456”

SN的叶节点数为3

m-k<SN的叶节点数，所以SN=SN的左子节点“45”

SN的叶节点数变为2

m-k<SN的叶节点数，所以SN=SN的左子节点“4”

SN的叶节点数变为1（SN此时为叶节点）

m-k=SN的叶节点数，由此我们得知我们需要使用哈希值“0123”和“4”重建原先的根哈希值“01234”。

这是为从第一个节点开始的任意节点构建一致性树的很好的示例，我们不需要考虑旧的根节点是否作为节点被添加至新的主树上。不过以上情况并不是典型的情况，下面再给出两个示例作为补充：

m=6，k=4（节点“0123”下的叶节点数），SN=“456”

SN的叶节点数为3

m-k<SN的叶节点数，所以SN=SN的左子节点“45”

SN的叶节点数变为2

m-k=SN的叶节点数，由此我们得知我们需要使用哈希值“0123”和“45”重建原先的根哈希值“012345”。

m=7，k=4（节点“0123”下的叶节点数），SN=“456”

m-k=SN的叶节点数，由此我们得知我们需要使用哈希值“0123”和“456”重建原先的根哈希值“0123456”。

规则3：

为一致性证明中的最后一个节点（不一定是叶节点）做审计证明（使用恰当的左兄弟节点或右兄弟节点）。

举例来说，如果我们需要为树的旧哈希值“01234”做审计证明（这是一个比上面更有趣的实例）：

在证明中我们获取根哈希值设计的其他节点哈希值为图示的“5”和“67”

这是因为我们需要“5”来获得“45”，用“67”获得“4567”。我觉得这是一个很酷的东西！

示例

这个demo程序可以让你体验创建Merkle Tree和执行审计声明、一致性证明的过程。其中的图形界面是通过嵌入集成FlowSharp服务实现的。

注意：这个demo程序使用了1100端口上的websocket进行通信以在画布（canvas）上创建连接和图形。你可能在首次运行的时候看到以下授权提醒：

请点击“Allow access”。

改变叶节点数

你最多可以设置数量为16。为了方便起见，叶子的哈希值被模拟为0-F

审计证明测试

你可以指定你想要进行审计证明的叶节点来进行审计证明（叶子的编号为0-15，10-15在图示中被表示为A-F），当你点击“Show Me”的时候，下方会展示审计证明的过程，同时图示中也会高亮参与审计证明的节点）

一致性证明测试

你可以通过数字来设定执行一致性证明的叶节点数。参与计算的旧的根节点将会被黄色高亮显示，完成一致性证明所需的其他节点会显示为紫色：

复选框“only to root node”只是为了方便我讨论一致性证明时创建截图所用——在上面讨论的第二步审计证明中并没有设计紫色节点。

集成Merkle Tree

接下来我们要集成Merkle Tree。它包括三个类：

• MerkleHash
• MerkleNode
• MerkleTree

让我们一个一个来实现。

MerkleHash类

这个类提供了一些静态的创建方法以及判定相等的测试方法，当然还有用字节数组，字符串或者其他的两个MerkleHash实例计算哈希值的方法。

namespace Clifton.Blockchain
{
  public class MerkleHash
  {
    public byte[] Value { get; protected set; }

    protected MerkleHash()
    {
    }

    public static MerkleHash Create(byte[] buffer)
    {
      MerkleHash hash = new MerkleHash();
      hash.ComputeHash(buffer);

      return hash;
    }

    public static MerkleHash Create(string buffer)
    {
      return Create(Encoding.UTF8.GetBytes(buffer));
    }

    public static MerkleHash Create(MerkleHash left, MerkleHash right)
    {
      return Create(left.Value.Concat(right.Value).ToArray());
    }

    public static bool operator ==(MerkleHash h1, MerkleHash h2)
    {
      return h1.Equals(h2);
    }

    public static bool operator !=(MerkleHash h1, MerkleHash h2)
    {
      return !h1.Equals(h2);
    }

    public override int GetHashCode()
    {
      return base.GetHashCode();
    }

    public override bool Equals(object obj)
    {
      MerkleTree.Contract(() => obj is MerkleHash, "rvalue is not a MerkleHash");
      return Equals((MerkleHash)obj);
    }

    public override string ToString()
    {
      return BitConverter.ToString(Value).Replace("-", "");
    }

    public void ComputeHash(byte[] buffer)
    {
      SHA256 sha256 = SHA256.Create();
      SetHash(sha256.ComputeHash(buffer));
    }

    public void SetHash(byte[] hash)
    {
      MerkleTree.Contract(() => hash.Length == Constants.HASH_LENGTH, "Unexpected hash length.");
      Value = hash;
    }

    public bool Equals(byte[] hash)
    {
      return Value.SequenceEqual(hash);
    }

    public bool Equals(MerkleHash hash)
    {
      bool ret = false;

      if (((object)hash) != null)
      {
        ret = Value.SequenceEqual(hash.Value);
      }

      return ret;
    }
  }
}

MerkleNode类

这个类包含了一个节点的所有信息——它的父节点、子节点以及它的哈希值。MerkleNode唯一有趣的特性是它实现了一个自底向上/自左向右的迭代器。

namespace Clifton.Blockchain
{
  public class MerkleHash
  {
    public byte[] Value { get; protected set; }

    protected MerkleHash()
    {
    }

    public static MerkleHash Create(byte[] buffer)
    {
      MerkleHash hash = new MerkleHash();
      hash.ComputeHash(buffer);

      return hash;
    }

    public static MerkleHash Create(string buffer)
    {
      return Create(Encoding.UTF8.GetBytes(buffer));
    }

    public static MerkleHash Create(MerkleHash left, MerkleHash right)
    {
      return Create(left.Value.Concat(right.Value).ToArray());
    }

    public static bool operator ==(MerkleHash h1, MerkleHash h2)
    {
      return h1.Equals(h2);
    }

    public static bool operator !=(MerkleHash h1, MerkleHash h2)
    {
      return !h1.Equals(h2);
    }

    public override int GetHashCode()
    {
      return base.GetHashCode();
    }

    public override bool Equals(object obj)
    {
      MerkleTree.Contract(() => obj is MerkleHash, "rvalue is not a MerkleHash");
      return Equals((MerkleHash)obj);
    }

    public override string ToString()
    {
      return BitConverter.ToString(Value).Replace("-", "");
    }

    public void ComputeHash(byte[] buffer)
    {
      SHA256 sha256 = SHA256.Create();
      SetHash(sha256.ComputeHash(buffer));
    }

    public void SetHash(byte[] hash)
    {
      MerkleTree.Contract(() => hash.Length == Constants.HASH_LENGTH, "Unexpected hash length.");
      Value = hash;
    }

    public bool Equals(byte[] hash)
    {
      return Value.SequenceEqual(hash);
    }

    public bool Equals(MerkleHash hash)
    {
      bool ret = false;

      if (((object)hash) != null)
      {
        ret = Value.SequenceEqual(hash.Value);
      }

      return ret;
    }
  }
}

MerkleTree 类

这个类负责构建树及执行一致性证明和审计证明。我已经写了一组静态方法来实现审计证明和一致性证明，所以你不需要再亲自编写验证算法。我下面已经把类拆分为了组成它的各个组件。

属性和字段

namespace Clifton.Blockchain
{
  public class MerkleTree
  {
    public MerkleNode RootNode { get; protected set; }

    protected List<MerkleNode> nodes;
    protected List<MerkleNode> leaves;

这是很简单的。

Contract方法

public static void Contract(Func<bool> action, string msg)
{
  if (!action())
  {
    throw new MerkleException(msg);
  }
}

我使用这个方法进行参数和状态验证。

构造器

public MerkleTree()
{
  nodes = new List<MerkleNode>();
  leaves = new List<MerkleNode>();
}

在已有的树上添加树

public MerkleNode AppendLeaf(MerkleNode node)
{
  nodes.Add(node);
  leaves.Add(node);

  return node;
}

public void AppendLeaves(MerkleNode[] nodes)
{
  nodes.ForEach(n => AppendLeaf(n));
}

public MerkleNode AppendLeaf(MerkleHash hash)
{
  var node = CreateNode(hash);
  nodes.Add(node);
  leaves.Add(node);

  return node;
}

public List<MerkleNode> AppendLeaves(MerkleHash[] hashes)
{
  List<MerkleNode> nodes = new List<MerkleNode>();
  hashes.ForEach(h => nodes.Add(AppendLeaf(h)));

  return nodes;
}

根据树的叶节点构造树

/// <summary>
/// Builds the tree for leaves and returns the root node.
/// </summary>
public MerkleHash BuildTree()
{
  // We do not call FixOddNumberLeaves because we want the ability to append 
  // leaves and add additional trees without creating unecessary wasted space in the tree.
  Contract(() => leaves.Count > 0, "Cannot build a tree with no leaves.");
  BuildTree(leaves);

  return RootNode.Hash;
}

/// <summary>
/// Recursively reduce the current list of n nodes to n/2 parents.
/// </summary>
/// <param name="nodes"></param>
protected void BuildTree(List<MerkleNode> nodes)
{
  Contract(() => nodes.Count > 0, "node list not expected to be empty.");

  if (nodes.Count == 1)
  {
    RootNode = nodes[0];
  }
  else
  {
    List<MerkleNode> parents = new List<MerkleNode>();

    for (int i = 0; i < nodes.Count; i += 2)
    {
      MerkleNode right = (i + 1 < nodes.Count) ? nodes[i + 1] : null;
      MerkleNode parent = CreateNode(nodes[i], right);
      parents.Add(parent);
    }

    BuildTree(parents);
  }
}

审计证明

/// <summary>
/// Returns the audit proof hashes to reconstruct the root hash.
/// </summary>
/// <param name="leafHash">The leaf hash we want to verify exists in the tree.</param>
/// <returns>The audit trail of hashes needed to create the root, or an empty list if the leaf hash doesn't exist.</returns>
public List<MerkleProofHash> AuditProof(MerkleHash leafHash)
{
  List<MerkleProofHash> auditTrail = new List<MerkleProofHash>();
  var leafNode = FindLeaf(leafHash);

  if (leafNode != null)
  {
    Contract(() => leafNode.Parent != null, "Expected leaf to have a parent.");
    var parent = leafNode.Parent;
    BuildAuditTrail(auditTrail, parent, leafNode);
  }

  return auditTrail;
}

protected void BuildAuditTrail(List<MerkleProofHash> auditTrail, MerkleNode parent, MerkleNode child)
{
  if (parent != null)
  {
    Contract(() => child.Parent == parent, "Parent of child is not expected parent.");
    var nextChild = parent.LeftNode == child ? parent.RightNode : parent.LeftNode;
    var direction = parent.LeftNode == child ? MerkleProofHash.Branch.Left : MerkleProofHash.Branch.Right;

    // For the last leaf, the right node may not exist. In that case, we ignore it because it's
    // the hash we are given to verify.
    if (nextChild != null)
    {
      auditTrail.Add(new MerkleProofHash(nextChild.Hash, direction));
    }

    BuildAuditTrail(auditTrail, child.Parent.Parent, child.Parent);
  }
}

一致性证明

/// <summary>
/// Verifies ordering and consistency of the first n leaves, such that we reach the expected subroot.
/// This verifies that the prior data has not been changed and that leaf order has been preserved.
/// m is the number of leaves for which to do a consistency check.
/// </summary>
public List<MerkleProofHash> ConsistencyProof(int m)
{
  // Rule 1:
  // Find the leftmost node of the tree from which we can start our consistency proof.
  // Set k, the number of leaves for this node.
  List<MerkleProofHash> hashNodes = new List<MerkleProofHash>();
  int idx = (int)Math.Log(m, 2);

  // Get the leftmost node.
  MerkleNode node = leaves[0];

  // Traverse up the tree until we get to the node specified by idx.
  while (idx > 0)
  {
    node = node.Parent;
    --idx;
  }

  int k = node.Leaves().Count();
  hashNodes.Add(new MerkleProofHash(node.Hash, MerkleProofHash.Branch.OldRoot));

  if (m == k)
  {
    // Continue with Rule 3 -- the remainder is the audit proof
  }
  else
  {
    // Rule 2:
    // Set the initial sibling node (SN) to the sibling of the node acquired by Rule 1.
    // if m-k == # of SN's leaves, concatenate the hash of the sibling SN and exit Rule 2, as this represents the hash of the old root.
    // if m - k < # of SN's leaves, set SN to SN's left child node and repeat Rule 2.

    // sibling node:
    MerkleNode sn = node.Parent.RightNode;
    bool traverseTree = true;

    while (traverseTree)
    {
      Contract(() => sn != null, "Sibling node must exist because m != k");
      int sncount = sn.Leaves().Count();

      if (m - k == sncount)
      {
        hashNodes.Add(new MerkleProofHash(sn.Hash, MerkleProofHash.Branch.OldRoot));
        break;
      }

      if (m - k > sncount)
      {
        hashNodes.Add(new MerkleProofHash(sn.Hash, MerkleProofHash.Branch.OldRoot));
        sn = sn.Parent.RightNode;
        k += sncount;
      }
      else // (m - k < sncount)
      {
        sn = sn.LeftNode;
      }
    }
  }

  // Rule 3: Apply ConsistencyAuditProof below.

  return hashNodes;
}

/// <summary>
/// Completes the consistency proof with an audit proof using the last node in the consistency proof.
/// </summary>
public List<MerkleProofHash> ConsistencyAuditProof(MerkleHash nodeHash)
{
  List<MerkleProofHash> auditTrail = new List<MerkleProofHash>();

  var node = RootNode.Single(n => n.Hash == nodeHash);
  var parent = node.Parent;
  BuildAuditTrail(auditTrail, parent, node);

  return auditTrail;
}

####审计证明校验

/// <summary>
/// Verify that if we walk up the tree from a particular leaf, we encounter the expected root hash.
/// </summary>
public static bool VerifyAudit(MerkleHash rootHash, MerkleHash leafHash, List<MerkleProofHash> auditTrail)
{
  Contract(() => auditTrail.Count > 0, "Audit trail cannot be empty.");
  MerkleHash testHash = leafHash;

  // TODO: Inefficient - compute hashes directly.
  foreach (MerkleProofHash auditHash in auditTrail)
  {
    testHash = auditHash.Direction == MerkleProofHash.Branch.Left ?
    MerkleHash.Create(testHash.Value.Concat(auditHash.Hash.Value).ToArray()) :
    MerkleHash.Create(auditHash.Hash.Value.Concat(testHash.Value).ToArray());
  } 

  return rootHash == testHash;
}

获取审计证明作为哈希对进行验证

/// <summary>
/// For demo / debugging purposes, we return the pairs of hashes used to verify the audit proof.
/// </summary>
public static List<Tuple<MerkleHash, MerkleHash>> AuditHashPairs(MerkleHash leafHash, List<MerkleProofHash> auditTrail)
{
  Contract(() => auditTrail.Count > 0, "Audit trail cannot be empty.");
  var auditPairs = new List<Tuple<MerkleHash, MerkleHash>>();
  MerkleHash testHash = leafHash;

  // TODO: Inefficient - compute hashes directly.
  foreach (MerkleProofHash auditHash in auditTrail)
  {
    switch (auditHash.Direction)
    {
      case MerkleProofHash.Branch.Left:
        auditPairs.Add(new Tuple<MerkleHash, MerkleHash>(testHash, auditHash.Hash));
        testHash = MerkleHash.Create(testHash.Value.Concat(auditHash.Hash.Value).ToArray());
        break;

      case MerkleProofHash.Branch.Right:
        auditPairs.Add(new Tuple<MerkleHash, MerkleHash>(auditHash.Hash, testHash));
        testHash = MerkleHash.Create(auditHash.Hash.Value.Concat(testHash.Value).ToArray());
        break;
    }
  }

  return auditPairs;
}

一致性验证验证

public static bool VerifyConsistency(MerkleHash oldRootHash, List<MerkleProofHash> proof)
{
  MerkleHash hash, lhash, rhash;

  if (proof.Count > 1)
  {
    lhash = proof[proof.Count - 2].Hash;
    int hidx = proof.Count - 1;
    hash = rhash = MerkleTree.ComputeHash(lhash, proof[hidx].Hash);
    hidx -= 2;

    while (hidx >= 0)
    {
      lhash = proof[hidx].Hash;
      hash = rhash = MerkleTree.ComputeHash(lhash, rhash);
      --hidx;
    }
  }
  else
  {
    hash = proof[0].Hash;
  }

  return hash == oldRootHash;
}

在叶节点列表中寻找叶节点

protected MerkleNode FindLeaf(MerkleHash leafHash)
{
  // TODO: We can improve the search for the leaf hash by maintaining a sorted list of leaf hashes.
  // We use First because a tree with an odd number of leaves will duplicate the last leaf
  // and will therefore have the same hash.
  return leaves.FirstOrDefault(l => l.Hash == leafHash);
}

根据自定义的节点要求来创建MerkleNode

// Override in derived class to extend the behavior.
// Alternatively, we could implement a factory pattern.

protected virtual MerkleNode CreateNode(MerkleHash hash)
{
  return new MerkleNode(hash);
}

protected virtual MerkleNode CreateNode(MerkleNode left, MerkleNode right)
{
  return new MerkleNode(left, right);
}

其他

FixOddNumberLeaves

在比特币中，一棵树总是有偶数个叶节点。如果树中的最后一个节点为左子节点，那么最后一个节点的内容将被复制到右子节点，他们的父节点的哈希值将通过两个左子节点的哈希值串联得到，你可以使FixOddNumberLeaves来创建这个行为

/// <summary>
/// If we have an odd number of leaves, add a leaf that
/// is a duplicate of the last leaf hash so that when we add the leaves of the new tree,
/// we don't change the root hash of the current tree.
/// This method should only be used if you have a specific reason that you need to balance
/// the last node with it's right branch, for example as a pre-step to computing an audit trail
/// on the last leaf of an odd number of leaves in the tree.
/// </summary>
public void FixOddNumberLeaves()
{
  if ((leaves.Count & 1) == 1)
  {
    var lastLeaf = leaves.Last();
    var l = AppendLeaf(lastLeaf.Hash);
  }
}

计算两个给定哈希值的哈希

public static MerkleHash ComputeHash(MerkleHash left, MerkleHash right)
{
  return MerkleHash.Create(left.Value.Concat(right.Value).ToArray());
}

MerkleProof 类

这个类被用于审计证明，将获得的左右分支与校验哈希相关联，其中子哈希顺序的正确性是获取父哈希的必要条件。

namespace Clifton.Blockchain
{
  public class MerkleProofHash
  {
    public enum Branch
    {
      Left,
      Right,
      OldRoot, // used for linear list of hashes to compute the old root in a consistency proof.
    }

    public MerkleHash Hash { get; protected set; }
    public Branch Direction { get; protected set; }

    public MerkleProofHash(MerkleHash hash, Branch direction)
    {
      Hash = hash;
      Direction = direction;
    }

    public override string ToString()
    {
      return Hash.ToString();
    }
  }
}

单元测试

一共有14个单元测试，其中的一些看起来都很蠢，大部分的测试是是很细微的。其中最有趣的单元测试是一致性证明的单元测试，我下面也只会对它进行说明。

一致性证明单元测试

这个单元测试会创建具有3-100个叶节点的树，测试会为编号从1到n-1的叶节点获取旧的根哈希值（n也就是测试树的叶节点数）。

[TestMethod]
public void ConsistencyTest()
{
  // Start with a tree with 2 leaves:
  MerkleTree tree = new MerkleTree();
  var startingNodes = tree.AppendLeaves(new MerkleHash[]
  {
    MerkleHash.Create("1"),
    MerkleHash.Create("2"),
  });

  MerkleHash firstRoot = tree.BuildTree();

  List<MerkleHash> oldRoots = new List<MerkleHash>() { firstRoot };

  // Add a new leaf and verify that each time we add a leaf, we can get a consistency check
  // for all the previous leaves.
  for (int i = 2; i < 100; i++)
  {
    tree.AppendLeaf(MerkleHash.Create(i.ToString())); //.Text=i.ToString();
    tree.BuildTree();

    // After adding a leaf, verify that all the old root hashes exist.
    oldRoots.ForEachWithIndex((oldRootHash, n) =>
    {
      List<MerkleProofHash> proof = tree.ConsistencyProof(n+2);
      MerkleHash hash, lhash, rhash;

      if (proof.Count > 1)
      {
        lhash = proof[proof.Count - 2].Hash;
        int hidx = proof.Count - 1;
        hash = rhash = MerkleTree.ComputeHash(lhash, proof[hidx].Hash);
        hidx -= 2;

        while (hidx >= 0)
        {
          lhash = proof[hidx].Hash;
          hash = rhash = MerkleTree.ComputeHash(lhash, rhash);
          --hidx;
        }
      }
      else
      {
        hash = proof[0].Hash;
      }

      Assert.IsTrue(hash == oldRootHash, "Old root hash not found for index " + i + " m = " + (n+2).ToString());
    });

    // Then we add this root hash as the next old root hash to check.
    oldRoots.Add(tree.RootNode.Hash);
  }
}

单调哈希和区块链

哈希树是证明数据完整性的一个高效的手段。单调（顺序排列）也是使用哈希链的一个恰当手段。相应的一个例子就是Holochain。"…一个由权威的哈希链提供支持的单调DHT"。Holochain是“…一个共享式的DHT，其中的每条数据都被植入到一方或者多方的签名哈希链当中。它是一个可验证的DHT，数据如果没有经过每个节点共享的验证规则验证就无法继续传播。”因此，可以将单调哈希链和哈希树两项技术结合起来。正如Arthur通过Slack写给我的一样：

Merkle Tree可以在holochain中使用来让其中的每个条目在共享的链空间中实现私有数据（而不仅仅是加密）。我的意思是...假设你有一个具有下面6个字段的数据结构：

1. Sender_ID
2. Sender_Previous_Hash
3. Receiver_ID
4. Receiver_Previous_Hash
5. Number_Credits_Sent
6. Transaction_Payload

如果你使用这六个字段作为Merkle Tree的六个叶节点，你可以让交易的双方把完整的交易内容提交给私有链，然后双方都拿出其部分的Merkle Tree进行签名然后提交给共享的DHT，其中并不包括节点#6。这是因为Merkle Tree的证明使用前五个节点的内容来管理货币就足够了，这种方法允许他们拥有数字资产或者其他总是隐藏在自己的本地链当中的其他类型的数据，这些内容只有自己才能看到而且不会影响互信的加密货币的管理。

在区块链上，交易中所有的数据都会进入链中，要保证数据的私有你必须对数据进行加密，还要祈祷你使用的秘钥和加密方式不会在这个永久、公开的记录中收到影响。而在Holochain上，由于共享的DHT机制为本地链的数据校验提供了支持，所以你可以利用Merkle证明来采用公开/私有数据混合的方式来保证你的私有数据的私有性。

比特币、以太坊以及其他的区块链也是一个单调哈希链（区块组成的链）的例子，其中包含着交易信息组成的Merkle Tree。

结论

写这篇文章是“只有你能教别人的时候你才真正学会了”的一个体现。需要耗费大量的研究和思考才能理解这些算法，尤其是一致性证明。更关键的是，理解“为什么”并没有那么直观明显。在最后我为自己做了一个Q&A的环节，我也将这些问题写在这里，这样你也可以看到我尝试去解答的问题。其中的一些答案可能不是很好，有些可能会引向一些并不重要的复杂细节，所以记住以下只是我开始写这篇文章之前调研的一个快照。

在你阅读我下面的可能很奇怪的调查过程之前，这里还有我的一个看法：我认为区块链以及它的变体在未来一定会成为分布式管理的核心组件。而Merkle Tree和类似的哈希树正是这些技术的基石。

数据同步Q&A

• 作为一个客户端，你正在下载一个大文件，例如10GB，以大约4096byte大小的块为单位进行传播。
• 你需要一个可信的权威的服务器来告诉你每个传输块是不是有效的。
• 你可以从服务器获取根哈希并随着每一个传输块的到来去填充Merkle Tree。
• 但是在验证根节点之前你必须完成所有数据的下载。
• 与之相对应的，你可以要求服务器发送给你每个哈希块相对根哈希的路径以便你自行校验。这使得你可以验证每个传输块是否损坏以便从其他地方重新请求。除此之外你也不需要维护所有块的哈希值而是自己构建了一个Merkle Tree。这使得验证一个块的速度更快并且减少了Merkle Tree所需的内存。
• 服务器作为可信机构并没有保存数据，他只需要具有包含所有叶节点的Merkle Tree即可。

关于数据同步，你基本上有三种选择：

1. 下载所有的数据然后去验证根哈希值。
2. 从可信的机构下载Merkle Tree以便你可以使用不完整的记录集进行测试。
3. 让服务器在特定的叶节点进行审计证明。

验证Q&A

Q1：为什么不直接问服务器某个叶节点是否存在于树的指定节点下呢？

Q2：为什么服务器甚至要花时间来存储叶节点哈希值的Merkle Tree？

A1：你可以这样做，但是SHA256哈希值只有32个字节，虽然这代表了大量的唯一哈希值（2^256），1.15*10^77，但是在树上各个节点的哈希值再次重复哈希最终得到的根哈希值可以提供很强的校验。除此之外，审计校验验证了各个记录与其他叶节点之间的关系。我们不单单想知道某个叶节点是否存在，我们还想知道它是否存在于树的某个特定位置。

A2：假设你有一个大数据集，它被分成了很多小块。首先你很可能不会维护整个数据集，而是只维护你负责的部分。如果这一块的内容发生了变化，你只需要知道左右分支的哈希值即可重新计算根哈希。

这样做的副作用也只是你需要为你的块保留左右分支，而不是整个Merkle Tree（包括了很多你不关心或者不知道的块对应的叶节点哈希值）

要同步的时候，另一个用户会要求你验证根哈希值。如果根哈希值不同，则会请求左右子节点的哈希值，如果不匹配则会一直迭代到识别出更改的块对应的节点为止。此时你只需要将更改的块发送给其他用户进行同步即可。

异常检测Q&A

Q3：如果两个以上的用户同时修改一个块的内容会发生什么？（基于分布式系统的一个想法，你可能希望跨多个peers复制数据来提供弹性）

A1：只有一个peer有权更改数据。

A2：哈希值可以打上时间戳，只接受最近的更改，其他的更改会被丢弃。

A3：差异合并（自动或者手动干预）。

A4：你只接受了最旧的更改，并丢弃最近更新的其他所有内容。

A5：从技术上来说，一个块的内容永远不会改变。如果需要更改，应当将其作为新的交易提交以便实现可审计的变动追踪（不同于Merkle审计）。因为只有具有特定的权限才可以被允许修改交易。

A6：某些阻塞机制可能被用于防止数据被同时修改。比方说，你可能在提交变更的时候收到一个“修改权限秘钥”，如果你的修改权限秘钥和当前的修改权限秘钥不匹配，你的修改请求会被拒绝并且会要求你同步数据以获得新的修改秘钥。

A7：这一点的突出点在于，如果块的大小很小，那么你的修改和其他人冲突的可能性就会很小。

为什么要证明的Q&A

Q4：为什么服务器应当向你发送哈希审计跟踪来校验一个块而不是直接返回给你正确或者错误？

A：这是你验证服务器是否可信的手段——如果它只是说“是”，你怎么知道你可以信任它？通过向你发送左右部分的哈希来告诉你“我是这样检验你的请求的”，而其他伪造的服务器并不能发送任何审计跟踪哈希，因为他们会给出一个不同的根哈希值。

参考文献

1. Ralph Merkle
2. Merkle Tree
3. Method of providing digital signatures, patent US 4309569 A
4. Bitcoin Developer Guide - Block Chain
5. Cassandra's AntiEntropy service
6. Merkling in Ethereum
7. Consistency Proofs and Audit Proofs
8. bitcoin block
9. RFC6962 - Certificate Transparency
10. IBM, Maersk Reveal Blockchain Solution for Global Supply Chain
11. Google's DeepMind plans bitcoin-style health record tracking for hospitals
12. The Blockchain: Capital Markets Use Cases
13. Holochain (website)
14. Holochain (github)

备注：本文以及任何相关的源代码和文件都是遵循The Code Project Open License (CPOL)协议的。

本文的版权归 ArrayZoneYour 所有，如需转载请联系作者。