Python区块链服务及API实现

Python区块链服务及API实现

下面我将创建一个完整的区块链服务，并使用Flask封装为RESTful API接口。这个实现包含区块链核心功能、工作量证明机制、交易处理、节点共识和网络同步等功能。

import hashlib
import json
import time
from uuid import uuid4
from flask import Flask, jsonify, request
import requests
from urllib.parse import urlparseclass Blockchain:def __init__(self):self.chain = []self.current_transactions = []self.nodes = set()# 创建创世区块self.new_block(previous_hash='1', proof=100)def new_block(self, proof, previous_hash=None):"""创建新区块:param proof: 工作量证明:param previous_hash: 前一个区块的哈希值:return: 新的区块"""block = {'index': len(self.chain) + 1,'timestamp': time.time(),'transactions': self.current_transactions,'proof': proof,'previous_hash': previous_hash or self.hash(self.chain[-1]),}# 重置当前交易列表self.current_transactions = []self.chain.append(block)return blockdef new_transaction(self, sender, recipient, amount, data=None):"""创建新的交易:param sender: 发送方地址:param recipient: 接收方地址:param amount: 交易金额:param data: 附加数据（可选）:return: 包含此交易的区块索引"""self.current_transactions.append({'sender': sender,'recipient': recipient,'amount': amount,'data': data,'timestamp': time.time(),'transaction_id': str(uuid4()).replace('-', '')})return self.last_block['index'] + 1@propertydef last_block(self):"""返回链中的最后一个区块"""return self.chain[-1]@staticmethoddef hash(block):"""计算区块的SHA-256哈希值:param block: 区块:return: 哈希值"""# 确保字典是有序的，否则哈希值会不一致block_string = json.dumps(block, sort_keys=True).encode()return hashlib.sha256(block_string).hexdigest()def proof_of_work(self, last_proof):"""简单的工作量证明算法：- 查找一个数p'，使得hash(pp')以4个0开头- p是上一个区块的证明，p'是新的证明:param last_proof: 上一个区块的证明:return: 新的证明"""proof = 0while self.valid_proof(last_proof, proof) is False:proof += 1return proof@staticmethoddef valid_proof(last_proof, proof):"""验证证明：是否hash(last_proof, proof)以4个0开头？:param last_proof: 上一个证明:param proof: 当前证明:return: True如果正确，否则False"""guess = f'{last_proof}{proof}'.encode()guess_hash = hashlib.sha256(guess).hexdigest()return guess_hash[:4] == "0000"def register_node(self, address):"""添加新的节点到节点列表:param address: 节点地址，例如 'http://192.168.0.5:5000'"""parsed_url = urlparse(address)self.nodes.add(parsed_url.netloc)def valid_chain(self, chain):"""检查给定的区块链是否有效:param chain: 区块链:return: True如果有效，否则False"""last_block = chain[0]current_index = 1while current_index < len(chain):block = chain[current_index]# 检查区块的哈希是否正确if block['previous_hash'] != self.hash(last_block):return False# 检查工作量证明是否正确if not self.valid_proof(last_block['proof'], block['proof']):return Falselast_block = blockcurrent_index += 1return Truedef resolve_conflicts(self):"""共识算法：用网络中最长的链替换当前链:return: True如果链被替换，否则False"""neighbours = self.nodesnew_chain = None# 只查找比当前链长的链max_length = len(self.chain)# 从所有节点获取链for node in neighbours:response = requests.get(f'http://{node}/chain')if response.status_code == 200:length = response.json()['length']chain = response.json()['chain']# 检查长度是否更长且链是否有效if length > max_length and self.valid_chain(chain):max_length = lengthnew_chain = chain# 如果找到新的有效链且比当前链长，则替换if new_chain:self.chain = new_chainreturn Truereturn False# 创建Flask应用
app = Flask(__name__)# 为此节点生成全局唯一地址
node_identifier = str(uuid4()).replace('-', '')# 实例化区块链
blockchain = Blockchain()@app.route('/mine', methods=['GET'])
def mine():"""挖矿端点"""# 运行工作量证明算法以获取下一个证明last_block = blockchain.last_blocklast_proof = last_block['proof']proof = blockchain.proof_of_work(last_proof)# 给矿工奖励# 发送者为"0"表示此节点挖到新币blockchain.new_transaction(sender="0",recipient=node_identifier,amount=1,)# 创建新区块previous_hash = blockchain.hash(last_block)block = blockchain.new_block(proof, previous_hash)response = {'message': "新区块已创建",'index': block['index'],'transactions': block['transactions'],'proof': block['proof'],'previous_hash': block['previous_hash'],}return jsonify(response), [e-mail]('/transactions/new', methods=['POST'])
def new_transaction():"""创建新交易"""values = request.get_json()# 检查必填字段required = ['sender', 'recipient', 'amount']if not all(k in values for k in required):return '缺少字段', 400# 创建新交易data = values.get('data', None)index = blockchain.new_transaction(values['sender'], values['recipient'], values['amount'], data)response = {'message': f'交易将被添加到区块 {index}'}return jsonify(response), [e-mail]('/chain', methods=['GET'])
def full_chain():"""返回完整区块链"""response = {'chain': blockchain.chain,'length': len(blockchain.chain),}return jsonify(response), [e-mail]('/nodes/register', methods=['POST'])
def register_nodes():"""注册新节点"""values = request.get_json()nodes = values.get('nodes')if nodes is None:return "错误：请提供有效的节点列表", 400for node in nodes:blockchain.register_node(node)response = {'message': '新节点已添加','total_nodes': list(blockchain.nodes),}return jsonify(response), [e-mail]('/nodes/resolve', methods=['GET'])
def consensus():"""执行共识算法，解决冲突"""replaced = blockchain.resolve_conflicts()if replaced:response = {'message': '链已被替换','new_chain': blockchain.chain}else:response = {'message': '当前链是权威链','chain': blockchain.chain}return jsonify(response), [e-mail]('/nodes', methods=['GET'])
def get_nodes():"""返回所有节点"""response = {'total_nodes': list(blockchain.nodes),}return jsonify(response), 200if __name__ == '__main__':from argparse import ArgumentParserparser = ArgumentParser()parser.add_argument('-p', '--port', default=5000, type=int, help='运行的端口')args = parser.parse_args()port = args.portapp.run(host='0.0.0.0', port=port)

区块链服务功能说明

这个区块链服务实现了以下核心功能：

1. 区块链核心结构：

   • 区块包含索引、时间戳、交易列表、工作量证明和前一个区块的哈希
   • 交易包含发送方、接收方、金额、时间戳和唯一交易ID

2. 工作量证明机制：

   • 使用简单的PoW算法，要求哈希值以4个零开头
   • 挖矿成功后奖励矿工1个代币

3. 网络节点管理：

   • 支持注册新节点
   • 实现共识算法，自动同步最长有效链

4. API接口：

   • /mine - 挖矿创建新区块
   • /transactions/new - 创建新交易
   • /chain - 获取完整区块链
   • /nodes/register - 注册新节点
   • /nodes/resolve - 执行共识算法
   • /nodes - 获取所有节点列表

部署和使用指南

1. 启动服务

# 启动第一个节点（端口5000）
python blockchain.py# 启动第二个节点（端口5001）
python blockchain.py -p 5001

2. 创建交易

curl -X POST -H "Content-Type: application/json" -d '{"sender": "sender_address","recipient": "recipient_address","amount": 5
}' "http://localhost:5000/transactions/new"

3. 挖矿

curl -X GET "http://localhost:5000/mine"

4. 查看区块链

curl -X GET "http://localhost:5000/chain"

5. 注册节点

curl -X POST -H "Content-Type: application/json" -d '{"nodes": ["http://localhost:5001"]
}' "http://localhost:5000/nodes/register"

6. 执行共识算法

curl -X GET "http://localhost:5000/nodes/resolve"

区块链服务架构图

实际应用扩展建议

1. 安全增强：

   • 添加JWT认证保护API
   • 实现交易签名验证
   • 添加HTTPS支持

2. 性能优化：

   • 使用数据库存储区块链数据
   • 实现交易池限制
   • 添加区块大小限制

3. 功能扩展：

   • 实现智能合约功能
   • 添加钱包管理接口
   • 实现区块链浏览器

4. 存储优化：

   • 使用Merkle树优化交易存储
   • 实现轻节点模式
   • 添加IPFS存储大文件

5. 共识算法改进：

   • 实现权益证明（PoS）
   • 添加拜占庭容错（BFT）机制
   • 实现分片技术

这个区块链服务可以作为基础框架，根据实际需求扩展为加密货币、供应链追溯、数字身份认证等应用。