☰
Python实现区块链时间胶囊:AES加密与链上时间锁实战
2026/10/18 14:34:23 网站建设 项目流程

简介:这是一份基于Python与区块链实现“时间胶囊”功能的完整项目源码,适合有一定Python基础、希望学习去中心化存储与智能合约开发的开发者。项目以Solidity智能合约为核心,定义时间锁与解锁规则,结合Vue.js前端和JavaScript辅助脚本,实现信息加密上链、定时解密、记录查询以及本地元数据管理,并提供连接以太坊网络的部署配置与迁移流程。资源包共24个文件,包含6个Vue组件、9个JavaScript脚本、2个Solidity合约,另有JSON配置、HTML入口和说明文档等,整体仅159KB,结构精简、模块划分清晰,便于快速定位代码和二次开发。目前已有203人浏览学习,可作为区块链时间戳服务、加密存储等场景的参考范例,帮助开发者深入理解区块链的不可篡改性、共识机制以及智能合约从编写到上线的完整设计思路。

1. Python-区块链上的时间胶囊:一条链锁住未来才能读到的消息

你有没有这种时刻:想给一年后的自己留段话,又怕半个月就忍不住翻出来看;想在某个确定时间向全网公开一段数据,又不信任第三方服务器替你保守秘密。这个 Python 项目把时间胶囊的核心逻辑搬上了一条从零手写的极简区块链——正文用 AES 加密存在本地节点,内容哈希与解锁时间戳写进链上区块,只有链上时间越过解锁点,节点才肯交出解密密钥。整套代码不到十个文件,适合对区块链存证、加密存储与定时解密有好奇心的开发者,拿来当一个可跑的实验台正合适。

2. 整体架构:链上做公证,链下做保险箱

2.1 设计取舍:为什么不在链上存正文

我第一版做过把正文直接写进区块的行为,后面一算账就改主意了:PoW 出块要时间,区块体积涨得快,而且链上数据一旦写入就不能改,连撤回一封写错的消息都做不到。于是定了现在这套"链上做公证,链下做保险箱"的方案——加密后的正文存在 SQLite,链上每个区块只记录一个交易摘要:胶囊 ID、正文哈希、解锁时间戳、创建者地址。这个设计的思路在真实存证类项目里也很常见,链上放证据,链下放内容。

这样设计有三个好处。第一,正文不落链,链上体积稳定,出块速度不会被大文本拖垮。第二,哈希上链保证了正文不可抵赖地存在过:一旦有人改了本地密文,重算哈希对不上,链上证据立刻暴露。第三,时间锁被做成链上状态的一部分,解锁判断依赖区块时间戳而不是某个中心服务器的系统时间,这在单机演示下虽然只是多了一层形式,但为以后接多节点留了余地。

技术选型上,我一开始纠结过要不要用现成的区块链框架。最后决定自己实现极简链,原因是时间胶囊业务对链的核心诉求只有两个:不可篡改哈希、可验证时间戳,这两个点用一条百来行的链就能讲清楚。用重型框架反而把共识、网络、合约全搅在一起,读者很难分清哪一层是时间胶囊的功劳。

提示:项目演示的是"链上公证 + 链下存储"的常见组合,不是把完整消息塞进区块的玩具写法。

2.2 目录结构与数据模型

整个项目我保持精简,方便你按图索骥。目录长这样:

time-capsule-on-chain/ ├── app.py # Flask 入口,注册 API 路由 ├── blockchain.py # 区块定义、区块链管理与 PoW ├── capsule_service.py # 时间胶囊业务逻辑(创建/解锁) ├── crypto_utils.py # AES-CBC 加密与解密工具 ├── storage.py # SQLite 持久化(密文、密钥、胶囊元数据) ├── config.py # 难度目标、数据库路径、服务端口 ├── requirements.txt # flask / pycryptodome └── init_db.py # 初始化数据库建表

storage.py 里只放了一张胶囊表,字段不多但每个都有用。表结构如下:

CREATE TABLE capsules ( capsule_id TEXT PRIMARY KEY, message_cipher BLOB NOT NULL, -- AES 加密后的密文 + IV key BLOB NOT NULL, -- 解密密钥(仅教学演示,生产环境需要拆分保管) unlock_at REAL NOT NULL, -- 解锁时间戳,单位秒 sender TEXT NOT NULL, created_at REAL NOT NULL, tx_hash TEXT NOT NULL -- 链上对应交易的哈希 );

这张表是整条链的内容仓库:链负责证明"这条消息在某个时间点存在过",表负责保存"打开这颗胶囊需要的一切"。key 直接明文落库是刻意为之——先跑通流程,再谈怎么把密钥拆成碎片的托管方案。如果你做生产级版本,加密后的密钥应该拆开异地保存,这事我在避坑章节再展开。unlock_at 统一用秒为单位的时间戳,避免毫秒、微秒混用。

2.3 区块与交易:时间锁在账本上长什么样

区块结构不炫技,按最常见字段实现。核心代码:

# blockchain.py import hashlib import json import time class Block: def __init__(self, index, timestamp, transactions, prev_hash, nonce=0): self.index = index self.timestamp = timestamp self.transactions = transactions self.prev_hash = prev_hash self.nonce = nonce def compute_hash(self): block_string = json.dumps({ "index": self.index, "timestamp": self.timestamp, "transactions": self.transactions, "prev_hash": self.prev_hash, "nonce": self.nonce, }, sort_keys=True) return hashlib.sha256(block_string.encode()).hexdigest()

compute_hash 把区块关键字段序列化后做 SHA-256,序列化时强制 sort_keys,保证不同节点对同一区块算出的哈希一致。后续加自定义字段时,记得把新字段也加进 block_string,否则改了字段但哈希不变,链上的人一眼看出你做了手脚。

交易本身是字典,胶囊业务里有两类:

# 创建胶囊交易 { "type": "CREATE_CAPSULE", "capsule_id": "018f...", # 胶囊唯一 ID "content_hash": "ab12...", # 密文 + 密钥的 SHA-256 "unlock_at": 1800000000, # 解锁时间戳(秒) "sender": "dev_user_01" } # 心跳交易(用于推进时间,让测试环境能等到解锁那一刻) { "type": "HEARTBEAT", "node": "node_1", "ts": 1717812345.6 }

content_hash 是"密文 + 密钥"整体哈希,上链那一刻正文指纹就被钉死。unlock_at 写在交易里而不是取区块 timestamp,这样胶囊的解锁时间和"区块何时被打包"解耦——你可以把一条消息锁到三年后,而区块只记录今天写入。这两个时间混为一谈,时间胶囊就变成"出块即解锁"的假胶囊了。

3. 核心链路拆解:从加密到解锁的完整旅程

3.1 创建胶囊:AES-CBC 加密与密钥保存

创建胶囊的服务代码是最好懂的一块:

# capsule_service.py import os import hashlib import time import uuid from crypto_utils import encrypt_message from storage import insert_capsule def create_capsule(message: str, unlock_at: float, sender: str): # 1. 生成 32 字节随机密钥 key = hashlib.sha256(os.urandom(32)).digest() # 2. 用 AES-CBC 加密正文,输出 iv + ciphertext sealed = encrypt_message(message, key) # 3. 计算上链用的内容哈希 content_hash = hashlib.sha256(sealed + key).hexdigest() capsule_id = str(uuid.uuid4()) # 4. 密文、密钥写本地库 insert_capsule(capsule_id, sealed, key, unlock_at, sender, content_hash) # 5. 构造交易并放入待打包池,等待出块 tx = { "type": "CREATE_CAPSULE", "capsule_id": capsule_id, "content_hash": content_hash, "unlock_at": unlock_at, "sender": sender, } return capsule_id, tx

对应的加密工具在 crypto_utils.py:

# crypto_utils.py import os from Crypto.Cipher import AES from Crypto.Util.Padding import pad, unpad def encrypt_message(message: str, key: bytes): iv = os.urandom(16) # 每次随机生成 IV cipher = AES.new(key, AES.MODE_CBC, iv=iv) padded = pad(message.encode("utf-8"), AES.block_size) ciphertext = cipher.encrypt(padded) return iv + ciphertext # 密文 = iv 前缀 + 密文主体 def decrypt_message(sealed: bytes, key: bytes): iv = sealed[:16] ciphertext = sealed[16:] cipher = AES.new(key, AES.MODE_CBC, iv=iv) plaintext_padded = cipher.decrypt(ciphertext) return unpad(plaintext_padded, AES.block_size).decode("utf-8")

参数说明:key 固定 32 字节,对应 AES-256。IV 是 16 字节,CBC 模式下每次加密都必须重新生成,如果复用同一个 IV,两个密文的前 16 字节之间存在可推算关系,这是密码学里的老坑,避坑章节我再讲。sealed 的格式是iv + ciphertext,解密时按长度切分,不额外占字段。

第 5 步只是把交易丢进池子,此刻链上还没有这条时间胶囊的存在证明。真正让它生效的是下一步:PoW 打包。

3.2 打包上链:PoW 与时间戳推进

节点把池子里攒下的交易铸成新区块:

# blockchain.py def proof_of_work(block, difficulty): target = "0" * difficulty while not block.compute_hash().startswith(target): block.nonce += 1 return block.compute_hash() class Blockchain: def __init__(self, difficulty=3): self.chain = [] self.pending_transactions = [] self.difficulty = difficulty self._create_genesis_block() def _create_genesis_block(self): genesis = Block(0, time.time(), [], "0") genesis.hash = genesis.compute_hash() self.chain.append(genesis) def last_block(self): return self.chain[-1] def mine_pending(self): new_block = Block( index=self.last_block().index + 1, timestamp=time.time(), transactions=self.pending_transactions, prev_hash=self.last_block().hash, ) new_block.hash = proof_of_work(new_block, self.difficulty) self.pending_transactions = [] self.chain.append(new_block) return new_block

proof_of_work 每次尝试更新 nonce,直到 block_string 的 SHA-256 以 difficulty 个零开头。difficulty 不是算力调节器,它直接决定期望尝试次数:难度 3 平均要试 4096 次,单机跑几毫秒;难度 5 要试一百万次,足够把演示接口卡成"看起来坏了"。教学项目用 3 比较合适,既能证明 PoW 存在感,又不会让测试等太久。difficulty 在 config.py 读取。

心跳矿工是我为了让单机演示不尴尬专门加的线程,它每隔一秒挖一个只含 HEARTBEAT 交易的新区块。链上的 last_block().timestamp 会持续前进,解锁时间设在十秒后的胶囊,服务端等的时候才能准时打开:

# app.py 内部线程 import threading import time def heartbeat_miner(app): while True: tx = {"type": "HEARTBEAT", "node": app.config["NODE_NAME"], "ts": time.time()} app.blockchain.pending_transactions.append(tx) app.blockchain.mine_pending() time.sleep(1)

没有这个线程,链上时间会停在最后一个交易区块的时间戳上。如果这颗胶囊恰好就是最后那个交易,它的时间戳等于创建时刻,永远满足不了区块时间 >= unlock_at,测试会直接卡死。心跳线程在单节点模型下模拟了网络中持续有新块产生的状态,是能跑通演示的关键。

3.3 解锁读取:时间门槛的校验与解密

解锁执行的是反向流程:先读链,再比时间,时间到了才允许取 key 解密密文:

# capsule_service.py def try_unlock(capsule_id, blockchain): capsule = get_capsule(capsule_id) if capsule is None: raise ValueError("capsule not found") latest_time = blockchain.last_block().timestamp # 核心门槛:链上最新区块时间必须晚于等于解锁时间 if latest_time < capsule["unlock_at"]: raise PermissionError( f"time not reached, {latest_time:.1f} < {capsule['unlock_at']:.1f}" ) # 门槛通过,用库里保存的密钥解开正文 plaintext = decrypt_message(capsule["message_cipher"], capsule["key"]) return plaintext

逻辑很直观:取链上最新区块时间作为"当前链上时间",和创建时写入的 unlock_at 比较,没到就抛 PermissionError,到了就解密返回。这里有个值得注意的点——我比较的是"最新区块时间"而不是"本地系统时间",因为链上时间是所有节点都能验证的公共状态,本地时间可以被单方篡改。但如果你整条链都在一台机器上跑,链上时间来源还是本地系统时钟,这层防线就退化成形式了。

解锁成功之后,我建议顺手写一条 VISIT 交易上链,作为"某时刻已读"的存证。这个设计在数字存证场景里很有用,不需要额外出块,append 进 pending_transactions 即可。

4. 本地跑通:十分钟启动一条时间胶囊链

4.1 环境准备:依赖安装与数据库初始化

先准备环境,项目依赖只有两个:

flask==3.0.3 pycryptodome==3.20.0

我用的是 Python 3.10,低于 3.8 的版本建议先升级,f-string 和类型注解会让你改到手软。安装依赖并建表:

cd time-capsule-on-chain python3 -m venv venv source venv/bin/activate pip install -r requirements.txt python init_db.py

init_db.py 做的事情就是 connect 到 capsules.db,执行建表 SQL。如果数据库文件已存在,重跑会提示表结构未变,不用删库。

4.2 启动服务与心跳矿工

启动入口:

python app.py

看到Running on http://127.0.0.1:5000就说明服务起来了。app.py 的启动顺序是:初始化区块链(加载创世块),启动心跳矿工线程,最后跑 Flask。你可以通过GET /chain观察链高度,每过一秒应该多一个心跳块:

curl http://127.0.0.1:5000/chain | python3 -m json.tool

返回的 chain 数组持续变长,说明心跳正常。如果高度纹丝不动,大概率是心跳线程抛异常,最常见的是 SQLite 数据库锁问题,避坑章节我会给思路。

4.3 用 curl 走完整流程:创建、提前解锁失败、到点解锁成功

项目对外暴露四个核心接口,我先列一张表方便对照:

方法路径作用返回
GET/chain查看整条链区块数组
POST/capsule/create创建时间胶囊capsule_id
GET/capsule/查询胶囊元数据状态与剩余时间
POST/capsule/ /unlock尝试解锁明文或 403

第一步,创建一个解锁时间设在 10 秒后的胶囊:

curl -X POST http://127.0.0.1:5000/capsule/create \ -H "Content-Type: application/json" \ -d '{"message": "给未来的自己:你已经升职了", "unlock_at": 1800000000}'

回包大致是这个样子:

{ "capsule_id": "a1b2c3d4", "unlock_at": 1800000000, "status": "created" }

注意创建前执行date +%s确认,unlock_at 填的是当前秒数加 10,别把毫秒整进来,否则就是一颗永远打不开的胶囊。第二步,立刻尝试解锁,预期被拒绝:

curl -X POST http://127.0.0.1:5000/capsule/a1b2c3d4/unlock

会拿到 HTTP 403,响应 JSON 里的 message 字段就是 try_unlock 抛出的时间差值提示。第三步,等十秒后再执行同样请求,就会看到明文正文。两次调用同一个接口,只是因为链上时间跨过了门槛,返回就从错误变成内容,这正是整个项目想表达的核心体验。

如果不想干等,可以临时把心跳线程的 sleep(1) 改成 sleep(0.2),或者把胶囊的 unlock_at 设成当前时间戳减 1 立刻解锁。但测试完记得把时间校验的真实性补回——绕过时间为难自己,时间逻辑就测不到了。手写 curl 验证容易漏细节,我更习惯用脚本批量走冒烟:

# smoke_test.py import requests import time base = "http://127.0.0.1:5000" capsule_id = "a1b2c3d4" resp = requests.post(f"{base}/capsule/{capsule_id}/unlock") print("early unlock:", resp.status_code, resp.json()) time.sleep(11) resp = requests.post(f"{base}/capsule/{capsule_id}/unlock") print("late unlock:", resp.status_code, resp.json())

如果第二次仍返回 403,把报错文本里的两个时间戳差值打出来,看看链上时间是不是被心跳推到足够靠后。这个脚本在二次开发时可以直接复用,改参数就能测新功能。

5. 常见问题与避坑:时间戳、难度和加密的翻车现场

调试这个项目时,我花在解 bug 上的时间比写代码多一倍。把几个典型问题整理出来,你少走点弯路。

5.1 时间戳被"偷跑",解锁提前发生

现象:unlock_at 设成当前时间加 3600,结果一分钟后尝试解锁就成功了,按理说还有一个小时的门槛。 原因:区块时间戳来自time.time(),矿工节点可以随意调整本机时间生成一个"未来"的区块,链上时间因此被伪造。单节点调试时把系统时钟往后拨,链上的 last_block 时间戳立刻超过 unlock_at,门槛形同虚设。 解决:把解锁校验从"最新区块时间"改成"最近 N 个区块时间的加权中位数",遍历self.chain[-10:]取时间戳排序后的中位数。单一矿工只能影响一个块,拉不动整体中位数,这是主流公链常见的思路。

def median_time_past(self, n=10): timestamps = sorted(b.timestamp for b in self.chain[-n:]) return timestamps[len(timestamps) // 2]

解锁接口里把latest_time换成median_time_past(),整体侵入很小。从那以后我凡涉及时间门槛,一律先问一句"这个时间是单点给的还是多点共识出来的"。

5.2 毫秒和秒对不上,永远等不到解锁

现象:创建胶囊接口一直正常,但解锁永远是 403,把错误信息里的两个时间打出来看,一个带小数,另一个是整数,差了上千倍。 原因:测试脚本用了Date.now()取毫秒时间戳,后端约定的是秒。毫秒值进入 unlock_at 之后,链上秒级区块时间永远追不上,成为一颗永远不会解开的胶囊。 解决:入口处做时间规范化。我在 create_capsule 函数第一行加了归一化:如果 unlock_at 大于 100000000000(约 公元 5138 年),就自动除以 1000。

def normalize_timestamp(ts): return ts / 1000.0 if ts > 100000000000 else ts

这个坑太常见了,前端、接口、数据库四处都有可能埋雷。建议接口层统一收一发时间转换,别指望每个调用方都守规矩。

5.3 AES 解密报错 ValueError: Padding is incorrect

现象:某些胶囊创建后立即解锁报 padding 错误,或者密文被篡改一比特后解密照样失败。 原因:CBC 模式不提供真实性校验,密文任何一比特被改,解出来的明文 padding 都大概率不合法。另一个高频原因是 IV 复用——我改版时把 IV 固定成常量,结果相同明文在相同 IV 下产生了相同密文前缀,这属于密码学上的信息泄露。 解决:IV 每次随机生成并随密文存储。如果还要防篡改,把 AES 模式从 CBC 换成 GCM,GCM 自带认证标签,密文被改一比特都能在校验时发现。教学项目里我保留了 CBC 但强制随机 IV,GCM 可以作为你的第一个升级方向去练手。

5.4 难度配太高,接口像死了一样

现象:启动后 curl 创建胶囊,请求挂起一百多秒才回来,然后整个 Flask 假死。 原因:我把难度从 3 改到 6 做压测,每次出块要尝试约 1600 万次 nonce。Flask 默认单线程,请求被挖矿卡死,心跳线程也跟着堵塞。 解决:不要把 Flask 请求线程和 PoW 计算绑在一起。临时方案是难度回滚到 3,或者创建胶囊接口单独用难度 2;正式方案是把挖矿丢给后台进程,接口只负责往交易池塞数据。你如果坚持高难度验证,至少先确认请求不是同步等挖矿返回。

5.5 多节点分叉后,同一胶囊出现两个版本

现象:把它扩展成双节点时遇到的——两个节点各自在本地挖出了高度相同的块,分叉后各自包含同一个胶囊 ID 的不同 content_hash,解锁时读到的版本取决于哪个节点响应。 原因:单机演示项目没有对等网络同步和分叉处理,两个节点天然以自己本地链为准。最直接的 last_block 读取方式在分叉场景下不可信。 解决:加一条最长链优先规则,同步时比较双方链上累计难度,只保留累计难度大的一方;在胶囊创建后记录"所在区块高度",解锁时沿这个高度向上游走确认没有分叉回退。这是完整的改造工作量,想做多节点建议先加这条规则再加通信。

6. 二次开发:把时间胶囊改造成定时红包

最有意思的改造方向,是把时间胶囊的"解锁正文"换成"解锁转账",做成一条链上的定时红包。两者底层逻辑完全一致:都是"时间锁 + 条件释放"。

6.1 把解锁动作换成账户转账

胶囊里原来存的是密文和密钥,现在换成 recipient 地址和 amount 金额。CREATE_CAPSULE 交易改成:

{ "type": "CREATE_TBOX", "tbox_id": "uuid", "amount": 100, "recipient": "addr_xxx", "sender": "addr_yyy", "unlock_at": 1800000000 }

6.2 加余额账本和 TRANSFER 交易

存储层加一张 accounts 表保存余额。try_unlock 变成 try_transfer:时间到后从 sender 余额扣 amount,给 recipient 加上,并写一条 TRANSFER 交易进链。注意余额扣减必须先校验时间,否则红包可以提前拆开。

def try_transfer(tbox_id, blockchain): if blockchain.median_time_past() < tbox["unlock_at"]: raise PermissionError("time not reached") transfer_or_create_transaction( sender=tbox["sender"], recipient=tbox["recipient"], amount=tbox["amount"], reason=tbox_id, )

6.3 回归验证与个人习惯

改完先跑四组用例:未到期转账拒绝、到点转账成功、余额不足拒绝、同一红包解两次幂等。测试定时红包时,把解锁时间设在time.time() + 2,配合 sleep 等它到点,比手动改系统时间靠谱得多,因为改系统时间会把 PoW 的时间戳也带偏。如果你想把时间锁玩出更多花样,把这套源码拉到本地跑一遍冒烟脚本,改起来比想象中快。从那以后我每次写时间敏感逻辑,都会强制过一遍三问:时间单位是秒还是毫秒、源头是链上时间还是本机时间、解锁条件用的是中位数还是单块时间。希望帮到你。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询