Python中如何实现区块链节点同步时的验证机制?

关键代码

#验证区块链
def validate(blocks):
    bool = True
    #上一个区块
    previous_index = 0
    previous_hash = 0
    for block in blocks:
        index = block["index"]
        hash = block["hash"]
        if (index > 0):
            #如果 index 是衔接的
            if (previous_index == index-1):
                pass
            else:
                bool = False
            #如果上一个区块的当前 hash 和当前区块的上一个 hash 值能对上
            if (previous_hash == block["previous_hash"]):
                pass
            else:
                bool = False
        if bool:
            #把当前的变为上一个
            previous_index = index
            previous_hash = hash
    if (index == 0):
        previous_index = index
        previous_hash = hash
        pass
return bool

用法: 分别启动 2 个节点:

python3 blockchain -p 8001
python3 blockchain -p 8002

浏览器打开 http://localhost:8001/nodes/add/localhost/8002

返回数据说明 8002 节点添加成功

[
  {
    "ip": "localhost",
    "port": 8002
  }
]

在浏览器打开 http://localhost:8002/say/tom

这是在 8002 节点增加了一个区块。

在浏览器中打开 http://localhost:8001/blocks/all

可以看到 8001 节点只有一个跟块,因为没有同步。

[
  {
    "data": "Genesis Block",
    "hash": "0e7a7f285f4c1ef2856c7b24ce7b11de15cddff7e7c2a733a16aa8c7f78085ae",
    "index": 0,
    "previous_hash": 0,
    "timestamp": 1533901217798
  }
]

然后我们在 8001 节点上同步: http://localhost:8001/blocks/sync

显示:

"synced"

说明同步节点并且验证成功。

接下来查看所有的节点: http://localhost:8001/blocks/all

[
  {
    "data": "Genesis Block",
    "hash": "d476a30fcfefa496c3f01a345b3b6d2a8234390da98cfc3c0899eec8037d437b",
    "index": 0,
    "previous_hash": 0,
    "timestamp": 1533901225899
  },
  {
    "data": "tom",
    "hash": "f3fbe9b7b0e74c47f1e2452b422f50d80dca8f05b7912c98843a2c69a4d48433",
    "index": 1,
    "previous_hash": "d476a30fcfefa496c3f01a345b3b6d2a8234390da98cfc3c0899eec8037d437b",
    "timestamp": 1533901358960
  }
]

说明 8001 的区块链已经验证了 8002 的区块链,并且实现了节点同步。

详情访问:

https://github.com/OpensourceBooks/blockchain/tree/master/6


Python中如何实现区块链节点同步时的验证机制?

6 回复

你可能慢慢的写后,发现 python 性能很差


在区块链节点同步时,验证机制的核心是确保接收到的区块和交易数据符合网络共识规则。这里的关键是独立验证每个新区块,而不是盲目信任对等节点。

一个基本的验证流程包括:

  1. 结构验证:检查区块数据结构(如哈希值格式、默克尔根)是否符合协议规范。
  2. 工作量证明验证:验证区块头的哈希值是否满足当前难度的要求(对于PoW链)。
  3. 交易验证:验证区块内所有交易的签名、输入输出是否有效,防止双花。
  4. 连续性验证:检查该区块是否正确地链接到本地的已知最长链上(即检查previous_hash)。

下面是一个简化的示例,展示在接收到新区块时如何进行基础验证:

import hashlib
import json
from typing import List, Dict

class Block:
    def __init__(self, index: int, previous_hash: str, timestamp: float, transactions: List[Dict], nonce: int, hash: str):
        self.index = index
        self.previous_hash = previous_hash
        self.timestamp = timestamp
        self.transactions = transactions
        self.nonce = nonce
        self.hash = hash

class BlockchainNode:
    def __init__(self):
        self.chain: List[Block] = []  # 本地存储的链
        self.difficulty = 4  # 工作量证明难度,要求哈希开头有4个零

    def validate_block(self, new_block: Block) -> bool:
        """验证一个新区块是否有效"""
        # 1. 验证工作量证明:检查哈希是否满足难度要求
        if not new_block.hash.startswith('0' * self.difficulty):
            print(f"无效的工作量证明: {new_block.hash}")
            return False

        # 2. 验证哈希值是否正确计算(防止数据篡改)
        calculated_hash = self.calculate_block_hash(new_block)
        if calculated_hash != new_block.hash:
            print(f"哈希值不匹配: 计算值={calculated_hash}, 接收值={new_block.hash}")
            return False

        # 3. 验证前一个哈希是否匹配本地最新区块
        if self.chain:
            latest_block = self.chain[-1]
            if new_block.previous_hash != latest_block.hash:
                print(f"前序哈希不匹配: 期望={latest_block.hash}, 接收={new_block.previous_hash}")
                return False
            # 验证区块高度是否连续
            if new_block.index != latest_block.index + 1:
                print(f"区块高度不连续: 期望={latest_block.index + 1}, 接收={new_block.index}")
                return False
        else:
            # 如果是创世区块,检查其特定属性(这里简化为检查索引为0)
            if new_block.index != 0:
                print("无效的创世区块索引")
                return False

        # 4. 这里可以添加交易验证(验证签名、余额等)
        if not self.validate_transactions(new_block.transactions):
            print("交易验证失败")
            return False

        print(f"区块 #{new_block.index} 验证通过")
        return True

    def calculate_block_hash(self, block: Block) -> str:
        """计算区块的哈希值"""
        block_data = {
            'index': block.index,
            'previous_hash': block.previous_hash,
            'timestamp': block.timestamp,
            'transactions': block.transactions,
            'nonce': block.nonce
        }
        block_string = json.dumps(block_data, sort_keys=True).encode()
        return hashlib.sha256(block_string).hexdigest()

    def validate_transactions(self, transactions: List[Dict]) -> bool:
        """简化版的交易验证(实际中需要检查签名、UTXO等)"""
        # 这里只是一个占位实现,真实场景需要:
        # 1. 验证数字签名
        # 2. 检查输入是否未花费(防双花)
        # 3. 验证输入总额 >= 输出总额
        for tx in transactions:
            if not self._validate_single_transaction(tx):
                return False
        return True

    def _validate_single_transaction(self, transaction: Dict) -> bool:
        # 实际交易验证逻辑应在此实现
        # 示例:至少检查交易有必要的字段
        required_fields = {'sender', 'recipient', 'amount', 'signature'}
        if not all(field in transaction for field in required_fields):
            return False
        # 这里应验证签名,示例中跳过
        return True

    def add_block(self, new_block: Block):
        """验证通过后,将区块添加到本地链"""
        if self.validate_block(new_block):
            self.chain.append(new_block)
            print(f"区块 #{new_block.index} 已添加到链上")
        else:
            print(f"区块 #{new_block.index} 验证失败,已拒绝")

# 示例用法
if __name__ == "__main__":
    node = BlockchainNode()

    # 假设我们接收到了一个区块(模拟数据)
    fake_transactions = [
        {"sender": "Alice", "recipient": "Bob", "amount": 10, "signature": "sig123"}
    ]
    new_block = Block(
        index=1,
        previous_hash="0" * 64,  # 假设的创世区块哈希
        timestamp=1625097600.0,
        transactions=fake_transactions,
        nonce=12345,
        hash="0000abc123..."  # 符合难度的哈希
    )

    node.add_block(new_block)

关键点说明:

  • 独立验证:每个节点都必须独立运行完整的验证,不依赖其他节点的声明。
  • 分步验证:验证是顺序且严格的,任何一步失败都会立即拒绝整个区块。
  • 交易验证:这是最复杂的部分,在比特币等UTXO模型中需要检查引用的输出是否存在且未花费;在以太坊等账户模型中需要检查账户余额和nonce。
  • 共识规则:验证的具体内容(如难度调整、区块奖励、gas限制等)由网络的共识规则严格定义。

在实际的区块链实现(如比特币核心或Geth)中,验证过程会更加复杂,涉及更多细节(如脚本执行、智能合约状态变更等),但核心原则不变:只接受通过所有规则检查的区块

总结:实现同步验证就是严格实现共识规则。

然后呢?然后用 go 重写一次是吧~

只是教学级的,性能不是问题。。

技术很炫酷,如果尝试着把区块链技术运用到其他某些方向,可能会给公众创造更大的价值,我有一些不太完善的想法想和你交流。

你可以加我微信 cr4fun

回到顶部