ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

用Python从零实现一个简单区块链:哈希、工作量证明与Flask接口

用Python从零实现一个简单区块链:哈希、工作量证明与Flask接口 提到区块链很多人第一反应是比特币、是去中心化账本、是各种白皮书里绕来绕去的经济学模型。但从写代码的角度看一个最小可运行的区块链其实非常朴素——它就是一串用哈希串起来的结构化数据只是额外加了“谁都不能偷偷篡改”的约束。我做了不少年 Python 开发每次有人问我理解区块链最快的方式是什么我都说别先啃共识算法先拿 Python 写一个简单的区块链 demo。这篇文章就把这个 demo 完整落地一遍把它拆开揉碎讲清楚。这篇文章的目标读者是“有一定 Python 基础但没深入碰过区块链”的同学也可能是准备面试想手写一个最小实现。我会从区块结构、哈希计算、工作量证明、链校验一直写到 Flask 接口层让你能在本地跑起一条能挖矿、能转账、能校验的微型链。不需要你有密码学基础只要会 Python 的基本语法就能跟着走下来。1. 动手前想清楚一个“简单区块链”到底包含什么1.1 别只追概念先看一个区块长什么样网上关于区块链的科普图大多是“一串方块连着一串方块”看着很简单但实际落到代码里每个区块可以理解为这样一个数据结构index块高度也就是第几个块timestamp出块时间transactions块里打包的数据在加密货币里是一笔笔转账在溯源系统里可能是商品流转记录previous_hash上一个区块的哈希值这是“链”字的核心没有它每个块就是一堆独立数据nonce随机数挖矿时反复调整它来改变哈希值hash当前区块自己的哈希值用 Python 表达很简单可以用类也可以用普通字典。我的建议是定义一个Block类但字段保持原生类型千万不要在这里引入类嵌套类、字典套对象这类复杂结构否则后面做json.dumps时会给自己挖坑。1.2 链是怎么“连”起来的很多人误以为“链”是靠指针或者列表顺序连起来的。其实单看self.chain []它就是一个普通 Python 列表真正把区块“锁”在一起的是previous_hash这个字段。每个新区块生成时必须拿到上一个区块的哈希值写进自己的头部。于是形成一条引用链第 N 块的哈希由第 N 块的完整内容算出来第 N1 块的 previous_hash必须等于第 N 块的 hash一旦有人篡改第 N 块里的任何字段第 N 块的哈希就变了第 N1 块存着的 previous_hash 就对不上第 N1 块的校验也会失败再往后全部连环失效这种设计的巧妙之处在于不需要什么高深权限系统只要全网节点都保存了完整链条你就不可能“只改自己手里这一份”因为别人手里的旧链你看不见也够不着。这也是为什么后来大家总说“区块链防篡改不是靠密码而是靠冗余和校验”。1.3 完整的数据流向挖矿视角在进入代码之前最好先有一个全局流程否则看代码容易一头扎进细节里客户端提交一条交易数据先放进内存里的待确认区unconfirmed_transactions挖矿节点从待确认区取走这批数据组装成新区块新区块要附带工作量证明也就是找到一个符合条件的 nonce校验通过后把新区块追加到链尾清空待确认区整个系统继续等下一批数据我见过不少初学者跳过步骤 5导致区块越挖越多但待确认区永远不空数据被反复打包。理解这个流程再写代码逻辑会顺很多。2. 环境准备从零搭建最小可运行工程2.1 Python 版本和模块选型我用的 Python 版本是 3.10实际上 3.8 以上都能跑关键是不依赖任何第三方库。核心区块链逻辑只需要四个标准库hashlib计算 SHA-256 哈希json把区块序列化成字符串再算哈希time记录时间戳itertools可选用但循环里用不到因为我们要手动维护 nonce这里我特别想劝一句新手不要一上来就装几十个依赖区块链最核心的机制就是哈希运算和链表校验根本不需要 numpy、pandas、web3 这些花哨的东西。依赖越少越能看清本质。到最后做接口演示时我会额外用 Flask这算是唯一一个第三方库。你也可以用 FastAPI、Django甚至不用接口直接命令行测试都行。Flask 只是因为轻量、上手快。2.2 工程目录与代码骨架我的建议是拆成两个文件职责清楚simple_blockchain/ ├── blockchain.py # 核心逻辑区块、链、挖矿、校验 └── app.py # Flask 接口层负责接收请求和展示数据如果你完全不想碰 Flask只跑核心逻辑也可以只留blockchain.py在里面写几行测试脚本就能看到效果。我个人倾向隔离接口层因为这个分层思路以后做真实项目同样适用——核心逻辑不应该和 Web 框架绑死。2.3 动手前的三个设计决策写之前有三件事明确下来能避免后面改代码改到怀疑人生。第一哈希用什么算法。我用 SHA-256密码学哈希函数输出固定 64 位十六进制字符串。Python 里就是hashlib.sha256(...).hexdigest()没有任何改造成本。第二数据序列化怎么做。对字符串做哈希前必须有确定的字节流这里统一用json.dumps并且sort_keysTrue保证字段顺序稳定。很多人哈希对不上多半就是序列化时字段顺序、空格、编码不一致造成的。第三交易结构写成什么。为了演示直观我把每笔交易定义成字典sender 是发送方、recipient 是接收方、amount 是金额。它也可以换成任意业务数据比如“溯源系统里的某件商品从这个仓库到了那个仓库”本质上没有任何区别。3. 核心代码实现区块、哈希与工作量证明3.1 定义 Block 类字段、nonce 与 compute_hash直接看代码import hashlib import json import time class Block: def __init__(self, index, transactions, timestamp, previous_hash, nonce0): self.index index self.transactions transactions self.timestamp timestamp self.previous_hash previous_hash self.nonce nonce self.hash None def compute_hash(self): block_data { index: self.index, transactions: self.transactions, timestamp: self.timestamp, previous_hash: self.previous_hash, nonce: self.nonce, } block_string json.dumps(block_data, sort_keysTrue, ensure_asciiFalse) return hashlib.sha256(block_string.encode(utf-8)).hexdigest()注意几个地方self.hash初始化为None等到区块真正确定下来之后再赋值。为什么不在一创建就赋值因为挖矿过程中 nonce 会变哈希每次都在变提前赋值只会让语义混乱。后面你会看到创世区块和挖出来的区块拿到hash值的时机不同。compute_hash()方法里的block_data手动构建不直接用self.__dict__这是一个很重要的习惯。很多人图省事写成json.dumps(self.__dict__)但__dict__里包含hash字段会出现两个问题一是序列化结果不确定二是每次调用可能带上已经设置的哈希导致算出来的东西不稳定。手动指定要参与哈希的字段反而更稳字段顺序、范围都可控。ensure_asciiFalse表示 JSON 里可以保留中文原文不然中文会被转成\uXXXX形式。这本身不影响哈希正确性但会让调试时看的输出不够直观。如果最终算出的哈希跟别人对不上多半编码或空格差异就是一大利器所以我在演示里直接统一。3.2 实现 Blockchain 类创世区块、添加区块、校验链接下来是最核心的Blockchain类class Blockchain: def __init__(self): self.chain [] self.unconfirmed_transactions [] self.difficulty 4 self.create_genesis_block() def create_genesis_block(self): genesis Block(0, [], time.time(), 0 * 64) genesis.hash genesis.compute_hash() self.chain.append(genesis) property def last_block(self): return self.chain[-1] def proof_of_work(self, block): block.nonce 0 computed_hash block.compute_hash() while not computed_hash.startswith(0 * self.difficulty): block.nonce 1 computed_hash block.compute_hash() return computed_hash def add_block(self, block, proof): if self.last_block.hash ! block.previous_hash: return False if not self.is_valid_proof(block, proof): return False block.hash proof self.chain.append(block) return True def is_valid_proof(self, block, block_hash): return ( block_hash.startswith(0 * self.difficulty) and block_hash block.compute_hash() ) def is_chain_valid(self, chain): for i in range(1, len(chain)): current chain[i] previous chain[i - 1] if current.previous_hash ! previous.hash: return False if current.hash ! current.compute_hash(): return False if not current.hash.startswith(0 * self.difficulty): return False return True def mine_block(self): if not self.unconfirmed_transactions: return None new_block Block( indexself.last_block.index 1, transactionsself.unconfirmed_transactions, timestamptime.time(), previous_hashself.last_block.hash, ) proof self.proof_of_work(new_block) if self.add_block(new_block, proof): self.unconfirmed_transactions [] return new_block return None讲几个关键点。创世区块是整条链的源头它没有 previous_hash 可参考所以0 * 64这种占位符是从业者常用的约定。理论上任何固定字符串都行但社区习惯用 64 个 0因为 SHA-256 输出就是 64 位十六进制这个占位符看着和真实哈希长度接近不容易混淆。difficulty 4是工作量证明的核心参数它意味着新区块哈希必须以前 4 个字符都是0开头。这里用的是十六进制字符串所以每一位可能的取值是 0-9、a-f共 16 种。前 4 位全是 0 的概率就是 1/(16^4)也就是大约 1/65536。因此平均情况下挖一个难度 4 的块需要尝试 6 万多次哈希。这个量级在普通笔记本上大概一两秒内能算完拿来演示再合适不过。mine_block是整个系统的主流程组装候选区块后调用proof_of_work拿回一个满足难度条件的哈希再交给add_block做合法性校验。注意清空unconfirmed_transactions的时机必须在成功上链后清。如果校验失败交易数据应该保留在待确认区等下次继续尝试而不是直接丢进黑洞。3.3 工作量证明PoW难度如何设定这里专门聊聊难度因为这是初学者最容易踩坑的地方。难度 1 时平均尝试 16 次就能出块几乎一瞬间难度 2平均 256 次难度 3平均 4096 次难度 4平均 65536 次难度 5平均 104 万次难度 6平均 1677 万次。到了难度 6 以上普通笔记本挖一个块可能需要几十分钟甚至更久。所以教学场景我强烈建议用难度 4。选难度 4 的原因有两个一方面它不会太快你能肉眼看到脚本卡个半秒到一两秒有一种“真的在挖矿”的感觉另一方面它又不会太慢演示和调试都不至于让人失去耐心。如果是在区块链上实现真正的业务难度当然要根据全网算力动态调整但那个不属于“简单区块链”的范畴。咱们先把这个 demo 跑通再谈扩展。4. 用 Flask 把区块链变成可交互服务4.1 为什么需要应用层接口到目前为止我们只有一个 Python 类跑测试也只能在脚本里手动调用这当然能说明原理但太不直观。为了让这条链具备“可交互”的感觉也为了让你后续能改成多节点、能对接前端页面最合适的做法是加一个 HTTP 接口层。Flask 的好处是 40 行代码就能把接口写完而且它天然支持 JSON接口返回的数据可以直接拿去调试。这里不需要 URL 规范化、权限认证、分页那些工程化内容我们只做三件事提交交易、挖矿、查链另外再加一个校验接口真正确保这条链没有被篡改过才是完整演示。4.2 挖矿、查询链、验证接口的设计新建app.pyfrom flask import Flask, jsonify, request from blockchain import Blockchain app Flask(__name__) blockchain Blockchain() app.route(/new_transaction, methods[POST]) def new_transaction(): tx_data request.get_json() required_fields [sender, recipient, amount] for field in required_fields: if not tx_data or tx_data.get(field) is None: return jsonify({message: 交易数据不完整}), 400 blockchain.unconfirmed_transactions.append(tx_data) return jsonify({message: 交易已暂存等待打包}), 201 app.route(/mine, methods[GET]) def mine(): block blockchain.mine_block() if block: response { message: 新区块挖矿成功, index: block.index, hash: block.hash, previous_hash: block.previous_hash, transactions: block.transactions, } return jsonify(response), 200 return jsonify({message: 没有待确认的交易无法挖矿}), 400 app.route(/chain, methods[GET]) def full_chain(): chain_data [block.__dict__ for block in blockchain.chain] return jsonify(chain_data), 200 app.route(/validate, methods[GET]) def validate(): is_valid blockchain.is_chain_valid(blockchain.chain) return jsonify({valid: is_valid, count: len(blockchain.chain)}), 200 if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)简单介绍一下这几个接口。/new_transaction接收 POST 请求把这笔交易放进unconfirmed_transactions。注意存放的是字典不是Block对象因为区块是在挖矿时组装出来的没必要提前封装。接口只做基础校验比如 sender、recipient、amount 是否存在。/mine是 GET因为大多教程为了方便演示都直接用浏览器访问。real API 设计里挖矿更适合 POST 或者后台任务但这里是教学演示能少一步算一步。/chain返回整条链的数据用block.__dict__转成可 JSON 序列化的字典。这里有个细节block.hash如果是None转成 JSON 之后会变成null看起来不太舒服但不会报错。你完全可以在展示时把它替换成空字符串或其他标识。/validate是校验接口直接把整条链丢给is_chain_valid方法返回值一目了然。4.3 本地启动与接口测试需要先安装 Flaskpip install flask然后启动服务python app.py如果看到 Flask 默认的启动日志说明服务已经跑起来了。接着就可以拿 curl 或者浏览器直接操作。我在下一节里演示完整的交互流程。5. 完整测试流程与实验结果分析5.1 用 curl 或 Python requests 模拟“转账”先提交两笔交易curl -X POST http://127.0.0.1:5000/new_transaction \ -H Content-Type: application/json \ -d {sender:alice,recipient:bob,amount:5} curl -X POST http://127.0.0.1:5000/new_transaction \ -H Content-Type: application/json \ -d {sender:bob,recipient:carol,amount:2}返回结果分别是{message:交易已暂存等待打包} {message:交易已暂存等待打包}然后触发挖矿curl http://127.0.0.1:5000/mine第一次挖矿大概会停顿不到一秒钟然后返回{ message: 新区块挖矿成功, index: 1, hash: 00004f7a2b3c..., previous_hash: 创世区块的哈希, transactions: [ {sender: alice, recipient: bob, amount: 5}, {sender: bob, recipient: carol, amount: 2} ] }注意返回的 hash 开头是0000这就是难度 4 在起作用。再连续触发两次挖矿但在这之前分别再提交两笔不同的交易这样链上就会有多个区块它们的哈希彼此依赖这才是一条完整的链。5.2 验证链完整性与恶意篡改的效果演示请求:curl http://127.0.0.1:5000/validate返回{count: 4, valid: true}说明区块链是完好的。现在关键实验来了手动改到链上数据。写一个临时脚本import json import urllib.request # 获取链 resp urllib.request.urlopen(http://127.0.0.1:5000/chain) chain json.load(resp) # 篡改第 1 个区块的金额 chain[1][transactions][0][amount] 999 print(chain[1])但这只改了本地表示服务器内存里的链还是原来的。真要在服务端做篡改实验可以临时写一个测试脚本直接导入blockchain.pyfrom blockchain import Blockchain blockchain Blockchain() # 模拟添加一个区块 blockchain.unconfirmed_transactions.append( {sender: alice, recipient: bob, amount: 5} ) block 1 blockchain.mine_block() # 看看校验结果 print(blockchain.is_chain_valid(blockchain.chain)) # True # 篡改区块里的交易金额 blockchain.chain[1].transactions[0][amount] 999 print(blockchain.is_chain_valid(blockchain.chain)) # False第二个is_chain_valid返回False。原因很清楚第 1 个区块的交易变了重新计算出的哈希和区块里保存的hash对不上同时后面所有区块的previous_hash都指向原哈希所以整条链校验从第 1 个区块开始就断裂。这是区块链防篡改机制最直观的体现。这个实验我建议你一定要亲手跑一遍因为它的冲击力比任何图解都强。改了一笔数据整个链崩了你能清晰感受到“哈希链”不是理论而是真实存在于每个字段之间的约束。如果你还想多走一步可以试试“修复”这条链也就是把第 1 个区块的哈希改成新值然后顺藤摸瓜修改第 2 个区块的previous_hash再重算第 2 个哈希一路改到尾。理论上确实能伪造一条自洽的链但真实场景下节点不认你的链因为还有多个副本做对照。这个思想也顺带解释了为什么区块链需要“分布式多节点”。6. 常见问题与排查技巧实录6.1 哈希对不上的几个典型原因我见过太多人写的简单区块链第一次跑is_chain_valid就是 False排查了半天基本都是这几类。第一序列化不一致。json.dumps默认会在冒号后面加空格如果你在另一个地方手工拼字符串来算哈希算出来的字节流不同哈希当然不同。解决办法统一用一个方法构建区块字符串比如compute_hash()里的json.dumps(block_data, sort_keysTrue)。第二忽略ensure_ascii。中文交易数据在我们这个实现里无所谓因为同一个 Python 进程内调用是稳定的但如果不同端比如 Node.js 前端用另一种字符串拼接方式发过来编码差异就会导致哈希结果不一致。建议统一用 UTF-8。第三重新计算时算错字段。有些人重算哈希只包了index和transactions把nonce、previous_hash、timestamp丢在外面自然怎么对都对不上。建议把核心字段列成一份清单像我上面代码里那样全部塞进block_data。第四把hash字段自己包含进去了。json.dumps(self.__dict__)会把已经算出来的哈希当作输入的一部分形成循环依赖。我特别强调过这点这大概是我在答疑时遇到最多的初学者坑。6.2 JSON 序列化报错Object of type ... is not JSON serializable在使用 Flask 返回区块时如果交易内容不是纯字典、数组、字符串、数字而是自定义对象jsonify会报一个非常有代表性的错误TypeError: Object of type Block is not JSON serializable解决办法有三个思路尽量让transactions里的数据保持基础类型这是最推荐的因为区块哈希计算本身也要经过json.dumps接口返回前把自定义对象整体转成字典比如block.__dict__给 Flask 配置一个 JSONEncoder 子类处理自定义对象但这是工程化做法教学演示里没必要顺便提醒不要在transactions里放datetime对象。json.dumps默认也处理不了最好转换成时间戳或 ISO 格式字符串。6.3 挖矿很卡或长时间没有反应如果你把difficulty调到 5 以上脚本卡住几十秒甚至几分钟都很正常。平均尝试次数是指数上涨的难度 4 是 6 万多次难度 5 就跳到 104 万次。本地 Python 用while循环跑哈希性能上限大概每秒钟几十万次左右所以不是程序坏了。另外proof_of_work这个方法里没有设置最大尝试次数。在理想教学场景里这样没问题但如果你把难度改成 8 再加一个漏洞它可能永远找不到目标。真实工程里会给max_attempts或者引入超时机制。如果你的mine_block一直返回None多半不是挖矿问题而是unconfirmed_transactions为空。此时挖矿接口会报“没有待确认的交易无法挖矿”。这是预期行为不是 bug。6.4 时间戳和字段类型引发的校验失败有个比较隐蔽的问题如果你把链导出成 JSON然后再加载回内存timestamp这个浮点数有时会发生变化。比如原来 Python 里的1234567890.12345经过 JSON 序列化和反序列化之后变成1234567890.12345看起来一样但底层二进制可能不一样重新compute_hash()就对不上了。我在实际测试中踩过这坑解决方案很简单如果要做持久化或者跨语言通信建议把时间戳统一存成整数或者把字段类型固定为字符串格式。这也是为什么有些企业级区块链会把时间戳设计成长整型毫秒数而不是浮点数秒能少很多坑。再补充一个和字段类型有关的点金额不要用浮点数。我演示代码里用了整数 5、2没毛病但如果是{amount: 0.1}浮点存储和 JSON 序列化都可能带来位级误差。现实中做计算类业务金额一律用整数“最小单位”或者用decimal类型别问问就是吃过亏。6.5 为什么改了数据后链还能“看起来正常”有人会说我改了区块后重新执行一遍compute_hash()并把结果写回block.hash后面区块的previous_hash也不检查那校验不就通过了吗确实如果完全不设置约束任何链都能被改成合法的。但注意两点第一在is_chain_valid里我们会遍历比较每个区块挨个检查当前区块的哈希是否等于重算值同时检查previous_hash是否等于前一个区块的哈希这两条规则能把那种“只改自己”的行为拦下来。你要是把全链每个区块都改掉并重算确实能伪造自洽链但这相当于你把整条链重写了和“篡改一笔交易”已经不是一个量级的工作。第二真实区块链还有分布式多节点。你改了你自己电脑上的整条链其他节点手里的链和你不一样网络共识会拒绝你这才是最终防线。所以理解哈希校验这条防线之后再去聊共识和分布式才能把逻辑串起来。7. 后续还能怎么扩展如果你已经跑通了上面所有流程恭喜你对区块链核心机制的理解超过了大多数只背概念的人。我在实际使用中强烈建议你把整套代码放进版本管理然后把下面几个方向逐个试一遍。第一个方向是改成多节点。把 Flask 服务同时启动两份端口不同再写一个简单的节点发现和链同步请求让两个节点可以互换数据。虽然很粗糙但只要把“最长链优先”的规则写进去你就能亲手摸到“共识”这个词的感觉。第二个方向是把交易改成真实业务数据。比如做一个简单的溯源 demo每笔交易表示“商品从 A 仓库流转到 B 仓库”链可校验、不可篡改这两个特点刚好用得上。我这个 demo 里只把交易的发送方、接收方、金额写死换成任意 JSON 数据都不影响。第三个方向是试试改变量证明算法比如换成下面这种形式def proof_of_work_v2(block): block.nonce 0 while not ( block.compute_hash().startswith(block.previous_hash[-6:]) ): block.nonce 1 return block.compute_hash()这就是把难度条件从“前缀 0”换成“跟在上一区块哈希尾部相似”虽然很粗糙但对理解“难度条件可自定义”非常有帮助。你很快会发现只要双方事先约定好同一条规则任何看似古怪的证明方式都能当作工作量证明来用。第四个方向是给区块加签名也就是让每笔交易附带私钥签名的验证逻辑。这一步会用到ecdsa或cryptography库但和区块链本身的实现完全解耦。加不了也不影响主体流程思路上了解一下就好。到最后回过头来看这个“简单的区块链”确实很小也就两百多行但它包含了区块结构、哈希链、工作量证明、链校验、API 交互这些核心组件。别小看它很多面试题、毕业设计、甚至一遍偏工程化的落地讨论都是从这个最小模型向外延伸的。甚至有这种体量的代码打底你再去读比特币或者以太坊的白皮书会发现很多抽象概念都有了具象的落脚点。可以的话把它运行起来亲手改一改亲眼看到校验失败的输出那些黑话才会真正变成你的本事。
RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进