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 的区块链,并且实现了节点同步。
详情访问:
Python中如何实现区块链节点同步时的验证机制?
你可能慢慢的写后,发现 python 性能很差
在区块链节点同步时,验证机制的核心是确保接收到的区块和交易数据符合网络共识规则。这里的关键是独立验证每个新区块,而不是盲目信任对等节点。
一个基本的验证流程包括:
- 结构验证:检查区块数据结构(如哈希值格式、默克尔根)是否符合协议规范。
- 工作量证明验证:验证区块头的哈希值是否满足当前难度的要求(对于PoW链)。
- 交易验证:验证区块内所有交易的签名、输入输出是否有效,防止双花。
- 连续性验证:检查该区块是否正确地链接到本地的已知最长链上(即检查
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

