本文章利用 Python 实现一个简单的功能较为完善的区块链系统（包括区块链结构、账户、钱包、转账），采用的共识机制是 POW。
一、区块与区块链结构
Block.py
import hashlib
from datetime import datetime
class Block:
"""
区块链结构：
prev_hash： 父区块哈希值
data： 区块内容
timestamp： 区块创建时间
hash： 区块哈希值
"""
def __init__(self, data, prev_hash):
# 将传入的父区块哈希值和数据保存到变量中
self.prev_hash = prev_hash
self.data = data
# 获得当前的时间
self.timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# 计算区块哈希值
# 获取哈希对象
message = hashlib.sha256()
# 先将数据内容转为字符串并进行编码，再将它们哈希
# 注意：update() 方法现在只接受 bytes 类型的数据，不接收 str 类型
message.update(str(self.prev_hash).encode('utf-8'))
message.update(str(self.prev_hash).encode('utf-8'))
message.update(str(self.prev_hash).encode('utf-8'))
# update() 更新 hash 对象，连续的调用该方法相当于连续的追加更新
# 返回字符串类型的消息摘要
self.hash = message.hexdigest()
BlockChain.py
from Block import Block
class BlockChain:
"""
区块链结构体
blocks： 包含区块的列表
"""
def __init__(self):
self.blocks = []
def add_block(self, block):
"""
添加区块
:param block:
:return:
"""
self.blocks.append(block)
# 新建区块
genesis_block = Block(data="创世区块", prev_hash="")
new_block1 = Block(data="张三转给李四一个比特币", prev_hash=genesis_block.hash)
new_block2 = Block(data="张三转给王五三个比特币", prev_hash=genesis_block.hash)
# 新建一个区块链对象
blockChain = BlockChain()
# 将刚才新建的区块加入区块链
blockChain.add_block(genesis_block)
blockChain.add_block(new_block1)
blockChain.add_block(new_block2)
# 打印区块链信息
print("区块链包含区块个数为：%d\n" % len(blockChain.blocks))
blockHeight = 0
for block in blockChain.blocks:
print(f"本区块高度为：{blockHeight}")
print(f"父区块哈希：{block.prev_hash}")
print(f"区块内容：{block.data}")
print(f"区块哈希：{block.hash}")
print()
blockHeight += 1
测试结果
二、加入工作量证明（POW）
将工作量证明加入到 Block.py 中
import hashlib
from datetime import datetime
from time import time
class Block:
"""
区块链结构：
prev_hash： 父区块哈希值
data： 区块内容
timestamp： 区块创建时间
hash： 区块哈希值
nonce： 随机数
"""
def __init__(self, data, prev_hash):
# 将传入的父区块哈希值和数据保存到变量中
self.prev_hash = prev_hash
self.data = data
# 获得当前的时间
self.timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# 设置随机数、哈希初始值为 None
self.nonce = None
self.hash = None
# 类的 __repr__() 方法定义了实例化对象的输出信息
def __repr__(self):
return f"区块内容：{self.data}\n区块哈希值：{self.hash}"
class ProofOfWork:
"""
工作量证明：
block： 区块
difficulty： 难度值
"""
def __init__(self, block, difficult=5):
self.block = block
# 定义出块难度，默认为 5，表示有效哈希值以 5 个零开头
self.difficulty = difficult
def mine(self):
"""
挖矿函数
:return:
"""
i = 0
prefix = '0' * self.difficulty
while True:
message = hashlib.sha256()
message.update(str(self.block.prev_hash).encode('utf-8'))
message.update(str(self.block.data).encode('utf-8'))
message.update(str(self.block.timestamp).encode('utf-8'))
message.update(str(i).encode('utf-8'))
# digest() 返回摘要，作为二进制数据字符串值
# hexdigest() 返回摘要，作为十六进制数据字符串值
digest = message.hexdigest()
# str.startswith(prefix) 检测字符串是否是以 prefix（字符串）开头，返回布尔值
if digest.startswith(prefix):
# 幸运数字
self.block.nonce = i
# 区块哈希值为十六进制数据字符串摘要
self.block.hash = digest
return self.block
i += 1
def validate(self):
"""
验证有效性
:return:
"""
message = hashlib.sha256()
message.update(str(self.block.prev_hash).encode('utf-8'))
message.update(str(self.block.data).encode('utf-8'))
message.update(str(self.block.timestamp).encode('utf-8'))
message.update(str(self.block.nonce).encode('utf-8'))
digest = message.hexdigest()
prefix = '0' * self.difficulty
return digest.startswith(prefix)
# ++++++++测试++++++++
# 定义一个区块
b = Block(data="测试", prev_hash="")
# 定义一个工作量证明
w = ProofOfWork(b)
# 开始时间
start_time = time()
# 挖矿，并统计函数执行时间
print("+++开始挖矿+++")
valid_block = w.mine()
# 结束时间
end_time = time()
print(f"挖矿花费时间：{end_time - start_time}秒")
# 验证区块
print(f"区块哈希值是否符合规则：{w.validate()}")
print(f"区块哈希值为：{b.hash}")
测试结果
更新 BlockChain.py
from Block import Block, ProofOfWork
class BlockChain:
"""
区块链结构体
blocks： 包含区块的列表
"""
def __init__(self):
self.blocks = []
def add_block(self, block):
"""
添加区块
:param block:
:return:
"""
self.blocks.append(block)
# 新建一个区块链对象
blockChain = BlockChain()
# 新建区块
block1 = Block(data="创世区块", prev_hash="")
w1 = ProofOfWork(block1)
genesis_block = w1.mine()
blockChain.add_block(genesis_block)
block2 = Block(data="张三转给李四一个比特币", prev_hash=genesis_block.hash)
w2 = ProofOfWork(block2)
block = w2.mine()
blockChain.add_block(block)
block3 = Block(data="张三转给王五三个比特币", prev_hash=block.hash)
w3 = ProofOfWork(block3)
block = w3.mine()
blockChain.add_block(block)
# 打印区块链信息
print("区块链包含区块个数为：%d\n" % len(blockChain.blocks))
blockHeight = 0
for block in blockChain.blocks:
print(f"本区块高度为：{blockHeight}")
print(f"父区块哈希：{block.prev_hash}")
print(f"区块内容：{block.data}")
print(f"区块哈希：{block.hash}")
print()
blockHeight += 1
测试结果
三、实现钱包、账户、交易功能
实现钱包、账户、交易功能要先安装非对称加密算法库 ecdsa。如果网速慢，引用下面这个网站
-i https://pypi.tuna.tsinghua.edu.cn/simple
添加钱包、账户功能 Wallet.py
import base64
import binascii
from hashlib import sha256
# 导入椭圆曲线算法
from ecdsa import SigningKey, SECP256k1, VerifyingKey
class Wallet:
"""
钱包
"""
def __init__(self):
"""
钱包初始化时基于椭圆曲线生成一个唯一的秘钥对，代表区块链上一个唯一的账户
"""
# 生成私钥
self._private_key = SigningKey.generate(curve=SECP256k1)
# 基于私钥生成公钥
self._public_key = self._private_key.get_verifying_key()
@property
def address(self):
"""
这里通过公钥生成地址
"""
h = sha256(self._public_key.to_pem())
# 地址先由公钥进行哈希算法，再进行 Base64 计算而成
return base64.b64encode(h.digest())
@property
def pubkey(self):
"""
返回公钥字符串
"""
return self._public_key.to_pem()
def sign(self, message):
"""
生成数字签名
"""
h = sha256(message.encode('utf8'))
# 利用私钥生成签名
# 签名生成的是一串二进制字符串，为了便于查看，这里转换为 ASCII 字符串进行输出
return binascii.hexlify(self._private_key.sign(h.digest()))
def verify_sign(pubkey, message, signature):
"""
验证签名
"""
verifier = VerifyingKey.from_pem(pubkey)
h = sha256(message.encode('utf8'))
return verifier.verify(binascii.unhexlify(signature), h.digest())
实现转账功能 Transaction.py
import json
class Transaction:
"""
交易的结构
"""
def __init__(self, sender, recipient, amount):
"""
初始化交易，设置交易的发送方、接收方和交易数量
"""
# 交易发送者的公钥
self.pubkey = None
# 交易的数字签名
self.signature = None
if isinstance(sender, bytes):
sender = sender.decode('utf-8')
self.sender = sender # 发送方
if isinstance(recipient, bytes):
recipient = recipient.decode('utf-8')
self.recipient = recipient # 接收方
self.amount = amount # 交易数量
def set_sign(self, signature, pubkey):
"""
为了便于验证这个交易的可靠性，需要发送方输入他的公钥和签名
"""
self.signature = signature # 签名
self.pubkey = pubkey # 发送方公钥
def __repr__(self):
"""
交易大致可分为两种，一是挖矿所得，而是转账交易
挖矿所得无发送方，以此进行区分显示不同内容
"""
if self.sender:
s = f"从{self.sender}转自{self.recipient}{self.amount}个加密货币"
elif self.recipient:
s = f"{self.recipient}挖矿所得{self.amount}个加密货币"
else:
s = "error"
return s
class TransactionEncoder(json.JSONEncoder):
"""
定义Json的编码类，用来序列化Transaction
"""
def default(self, obj):
if isinstance(obj, Transaction):
return obj.__dict__
else:
return json.JSONEncoder.default(self, obj)
# return super(TransactionEncoder, self).default(obj)
更新 Block.py
import hashlib
import json
from datetime import datetime
from Transaction import Transaction, TransactionEncoder
class Block:
"""
区块结构
prev_hash: 父区块哈希值
transactions: 交易对
timestamp: 区块创建时间
hash: 区块哈希值
Nonce: 随机数
"""
def __init__(self, transactions, prev_hash):
# 将传入的父哈希值和数据保存到类变量中
self.prev_hash = prev_hash
# 交易列表
self.transactions = transactions
# 获取当前时间
self.timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# 设置Nonce和哈希的初始值为None
self.nonce = None
self.hash = None
# 类的 __repr__() 方法定义了实例化对象的输出信息
def __repr__(self):
return f"区块内容：{self.transactions}\n区块哈希值：{self.hash}"
class ProofOfWork:
"""
工作量证明
block： 区块
difficulty： 难度值
"""
def __init__(self, block, miner, difficult=5):
self.block = block
self.miner = miner
# 定义工作量难度，默认为5，表示有效的哈希值以5个“0”开头
self.difficulty = difficult
# 添加挖矿奖励
self.reward_amount = 1
def mine(self):
"""
挖矿函数
"""
i = 0
prefix = '0' * self.difficulty
# 设置挖矿自动生成交易信息，添加挖矿奖励
t = Transaction(
sender="",
recipient=self.miner.address,
amount=self.reward_amount,
)
sig = self.miner.sign(json.dumps(t, cls=TransactionEncoder))
t.set_sign(sig, self.miner.pubkey)
self.block.transactions.append(t)
while True:
message = hashlib.sha256()
message.update(str(self.block.prev_hash).encode('utf-8'))
# 更新区块中的交易数据
# message.update(str(self.block.data).encode('utf-8'))
message.update(str(self.block.transactions).encode('utf-8'))
message.update(str(self.block.timestamp).encode('utf-8'))
message.update(str(i).encode("utf-8"))
digest = message.hexdigest()
if digest.startswith(prefix):
self.block.nonce = i
self.block.hash = digest
return self.block
i += 1
def validate(self):
"""
验证有效性
"""
message = hashlib.sha256()
message.update(str(self.block.prev_hash).encode('utf-8'))
# 更新区块中的交易数据
# message.update(str(self.block.data).encode('utf-8'))
message.update(json.dumps(self.block.transactions).encode('utf-8'))
message.update(str(self.block.timestamp).encode('utf-8'))
message.update(str(self.block.nonce).encode('utf-8'))
digest = message.hexdigest()
prefix = '0' * self.difficulty
return digest.startswith(prefix)
更新 BlockChain.py
from Block import Block, ProofOfWork
from Transaction import Transaction
from Wallet import Wallet, verify_sign
class BlockChain:
"""
区块链结构体
blocks: 包含的区块列表
"""
def __init__(self):
self.blocks = []
def add_block(self, block):
"""
添加区块
"""
self.blocks.append(block)
def print_list(self):
print(f"区块链包含个数为：{len(self.blocks)}")
for block in self.blocks:
height = 0
print(f"区块链高度为：{height}")
print(f"父区块为：{block.prev_hash}")
print(f"区块内容为：{block.transactions}")
print(f"区块哈希值为：{block.hash}")
height += 1
print()
为了方便我们对区块链进行操作，我们可以在 BlockChain.py 中补充一些方法
# 传入用户和区块链，返回用户的“余额”
def get_balance(user, blockchain):
balance = 0
for block in blockchain.blocks:
for t in block.transactions:
if t.sender == user.address.decode():
balance -= t.amount
elif t.recipient == user.address.decode():
balance += t.amount
return balance
# user生成创世区块（新建区块链），并添加到区块链中
def generate_genesis_block(user):
blockchain = BlockChain()
new_block = Block(transactions=[], prev_hash="")
w = ProofOfWork(new_block, user)
genesis_block = w.mine()
blockchain.add_block(genesis_block)
# 返回创世区块
return blockchain
# 用户之间进行交易并记入交易列表
def add_transaction(sender, recipient, amount):
# 新建交易
new_transaction = Transaction(
sender=sender.address,
recipient=recipient.address,
amount=amount
)
# 生成数字签名
sig = sender.sign(str(new_transaction))
# 传入付款方的公钥和签名
new_transaction.set_sign(sig, sender.pubkey)
return new_transaction
# 验证交易，若验证成功则加入交易列表
def verify_new_transaction(new_transaction, transactions):
if verify_sign(new_transaction.pubkey,
str(new_transaction),
new_transaction.signature
):
# 验证交易签名没问题，加入交易列表
print("交易验证成功")
transactions.append(new_transaction)
else:
print("交易验证失败")
# 矿工将全部验证成功的交易列表打包出块
def generate_block(miner, transactions, blockchain):
new_block = Block(transactions=transactions,
prev_hash=blockchain.blocks[len(blockchain.blocks) - 1].hash)
print("生成新的区块...")
# 挖矿
w = ProofOfWork(new_block, miner)
block = w.mine()
print("将新区块添加到区块链中")
blockchain.add_block(block)
进行测试
# 新建交易列表
transactions = []
# 创建 3 个用户
alice = Wallet()
tom = Wallet()
bob = Wallet()
print("alice创建创世区块...")
blockchain = generate_genesis_block(alice)
print()
print(f"alice 的余额为{get_balance(alice, blockchain)}个比特币")
print(f"tom 的余额为{get_balance(tom, blockchain)}个比特币")
print(f"bob 的余额为{get_balance(bob, blockchain)}个比特币")
print()
# 打印区块链信息
blockchain.print_list()
print("新增交易：alice 转账 0.5 比特币给 tom")
nt = add_transaction(alice, tom, 0.5)
print()
verify_new_transaction(nt, transactions)
print(f"矿工 bob 将全部验证成功的交易列表打包出块...")
generate_block(bob, transactions, blockchain)
print("添加完成\n")
# 打印区块链信息
blockchain.print_list()
测试结果
相关文章
您阅读本篇文章共花了：
发表评论