Python實例題：基于區塊鏈的去中心化應用平臺（區塊鏈、智能合約）

Python實例題

題目

基于區塊鏈的去中心化應用平臺（區塊鏈、智能合約）

問題描述

開發一個基于區塊鏈的去中心化應用平臺，包含以下功能：

區塊鏈基礎：實現區塊鏈的數據結構和基本操作
共識機制：實現工作量證明 (PoW) 或權益證明 (PoS) 共識
智能合約：設計并實現簡單的智能合約語言和執行環境
去中心化應用：構建基于智能合約的去中心化應用
網絡通信：實現節點間的 P2P 通信協議

解題思路

設計區塊鏈的數據結構和區塊格式
實現共識算法保證區塊鏈的一致性和安全性
開發智能合約虛擬機和編程語言
設計 P2P 網絡協議實現節點間的通信和同步
構建用戶界面和 API 供開發者使用

關鍵代碼框架

# 區塊鏈基礎實現
import hashlib
import json
import time
from typing import List, Dict, Any, Optional
import requestsclass Block:def __init__(self, index: int, transactions: List[Dict[str, Any]], timestamp: float, previous_hash: str, nonce: int = 0):"""初始化區塊參數:index: 區塊索引transactions: 交易列表timestamp: 時間戳previous_hash: 前一個區塊的哈希值nonce: 用于工作量證明的隨機數"""self.index = indexself.transactions = transactionsself.timestamp = timestampself.previous_hash = previous_hashself.nonce = nonceself.hash = self.calculate_hash()def calculate_hash(self) -> str:"""計算區塊的哈希值"""block_string = json.dumps({"index": self.index,"transactions": self.transactions,"timestamp": self.timestamp,"previous_hash": self.previous_hash,"nonce": self.nonce}, sort_keys=True).encode()return hashlib.sha256(block_string).hexdigest()def to_dict(self) -> Dict[str, Any]:"""將區塊轉換為字典形式"""return {"index": self.index,"transactions": self.transactions,"timestamp": self.timestamp,"previous_hash": self.previous_hash,"nonce": self.nonce,"hash": self.hash}class Blockchain:def __init__(self):"""初始化區塊鏈"""self.chain: List[Block] = []self.pending_transactions: List[Dict[str, Any]] = []self.nodes: set = set()# 創建創世區塊self.create_genesis_block()def create_genesis_block(self) -> Block:"""創建創世區塊"""genesis_block = Block(0, [], time.time(), "0")self.chain.append(genesis_block)return genesis_blockdef get_latest_block(self) -> Block:"""獲取最新區塊"""return self.chain[-1]def proof_of_work(self, block: Block) -> int:"""工作量證明算法參數:block: 需要進行工作量證明的區塊返回:有效的nonce值"""difficulty = 4  # 難度級別，前導0的數量prefix = "0" * difficultyblock.nonce = 0computed_hash = block.calculate_hash()while not computed_hash.startswith(prefix):block.nonce += 1computed_hash = block.calculate_hash()return block.noncedef is_chain_valid(self, chain: List[Block]) -> bool:"""驗證區塊鏈的有效性參數:chain: 需要驗證的區塊鏈返回:區塊鏈是否有效"""previous_block = chain[0]current_index = 1while current_index < len(chain):block = chain[current_index]# 驗證哈希值if block.hash != block.calculate_hash():return False# 驗證前一個區塊的哈希值if block.previous_hash != previous_block.hash:return False# 驗證工作量證明difficulty = 4prefix = "0" * difficultyif not block.hash.startswith(prefix):return Falseprevious_block = blockcurrent_index += 1return Truedef add_block(self, block: Block, proof: int) -> bool:"""添加新區塊到區塊鏈參數:block: 要添加的區塊proof: 工作量證明的結果返回:是否成功添加區塊"""previous_hash = self.get_latest_block().hash# 驗證前一個區塊的哈希值if previous_hash != block.previous_hash:return False# 驗證工作量證明difficulty = 4prefix = "0" * difficultyif not block.hash.startswith(prefix) or block.hash != block.calculate_hash():return Falseself.chain.append(block)return Truedef add_transaction(self, sender: str, recipient: str, amount: float, contract_code: Optional[str] = None) -> int:"""添加新交易到待處理交易池參數:sender: 發送方地址recipient: 接收方地址amount: 交易金額contract_code: 智能合約代碼(如果是合約部署交易)返回:交易將被包含的區塊索引"""self.pending_transactions.append({"sender": sender,"recipient": recipient,"amount": amount,"timestamp": time.time(),"contract_code": contract_code})return self.get_latest_block().index + 1def mine_block(self, miner_address: str) -> Block:"""挖礦創建新區塊參數:miner_address: 礦工地址，用于接收挖礦獎勵返回:新創建的區塊"""# 添加挖礦獎勵交易self.add_transaction(sender="0",  # 表示系統獎勵recipient=miner_address,amount=10  # 挖礦獎勵)# 創建新區塊previous_block = self.get_latest_block()new_block = Block(index=previous_block.index + 1,transactions=self.pending_transactions,timestamp=time.time(),previous_hash=previous_block.hash)# 執行工作量證明proof = self.proof_of_work(new_block)# 添加新區塊到區塊鏈if self.add_block(new_block, proof):# 清空待處理交易池self.pending_transactions = []return new_blockelse:return Nonedef register_node(self, address: str) -> None:"""注冊新節點參數:address: 節點地址"""self.nodes.add(address)def resolve_conflicts(self) -> bool:"""共識算法：解決沖突，使用最長鏈規則返回:區塊鏈是否被更新"""neighbours = self.nodesnew_chain = Nonemax_length = len(self.chain)# 獲取所有鄰居節點的區塊鏈for node in neighbours:try: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.is_chain_valid([Block(**block_data) for block_data in chain]):max_length = lengthnew_chain = chainexcept requests.exceptions.RequestException:continue# 如果找到更長的有效鏈，則替換當前鏈if new_chain:self.chain = [Block(**block_data) for block_data in new_chain]return Truereturn False

# 智能合約實現
class Contract:def __init__(self, code: str, owner: str, contract_id: str):"""初始化智能合約參數:code: 合約代碼owner: 合約所有者contract_id: 合約ID"""self.code = codeself.owner = ownerself.contract_id = contract_idself.state = {}  # 合約狀態self.balance = 0  # 合約余額def execute(self, function_name: str, params: Dict[str, Any], caller: str) -> Dict[str, Any]:"""執行智能合約函數參數:function_name: 函數名params: 函數參數caller: 調用者地址返回:執行結果"""# 簡單的合約執行環境# 實際應用中需要更復雜的虛擬機和安全機制try:# 這里簡化處理，實際應執行合約代碼if function_name == "get_balance":return {"success": True, "result": self.balance}elif function_name == "transfer":amount = params.get("amount", 0)recipient = params.get("recipient", "")if amount <= 0 or not recipient:return {"success": False, "error": "Invalid parameters"}if amount > self.balance:return {"success": False, "error": "Insufficient balance"}self.balance -= amount# 這里應該觸發一個交易到接收方return {"success": True, "message": f"Transferred {amount} to {recipient}"}elif function_name == "deploy":# 部署新合約的邏輯passelse:return {"success": False, "error": f"Function {function_name} not found"}except Exception as e:return {"success": False, "error": str(e)}class ContractManager:def __init__(self, blockchain: Blockchain):"""初始化智能合約管理器參數:blockchain: 關聯的區塊鏈"""self.blockchain = blockchainself.contracts: Dict[str, Contract] = {}def deploy_contract(self, code: str, owner: str) -> str:"""部署新智能合約參數:code: 合約代碼owner: 合約所有者返回:合約ID"""# 生成合約IDcontract_id = hashlib.sha256(f"{code}{owner}{time.time()}".encode()).hexdigest()[:16]# 創建合約實例contract = Contract(code, owner, contract_id)# 添加到合約管理器self.contracts[contract_id] = contract# 添加部署交易到區塊鏈self.blockchain.add_transaction(sender=owner,recipient="0",  # 表示系統合約amount=0,contract_code=code)return contract_iddef execute_contract(self, contract_id: str, function_name: str, params: Dict[str, Any], caller: str) -> Dict[str, Any]:"""執行智能合約函數參數:contract_id: 合約IDfunction_name: 函數名params: 函數參數caller: 調用者地址返回:執行結果"""if contract_id not in self.contracts:return {"success": False, "error": f"Contract {contract_id} not found"}contract = self.contracts[contract_id]# 添加交易到區塊鏈tx_index = self.blockchain.add_transaction(sender=caller,recipient=contract_id,amount=0,  # 無代幣轉移，只是調用合約contract_code=None)# 執行合約函數result = contract.execute(function_name, params, caller)# 如果執行成功，記錄到區塊鏈if result["success"]:# 這里可以添加更多的執行結果記錄passreturn result

# P2P網絡節點
import socket
import threading
import jsonclass P2PNode:def __init__(self, host: str, port: int, blockchain: Blockchain):"""初始化P2P網絡節點參數:host: 主機地址port: 端口號blockchain: 關聯的區塊鏈"""self.host = hostself.port = portself.blockchain = blockchainself.peers = set()  # 存儲連接的對等節點self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)self.server.bind((self.host, self.port))self.server.listen(10)self.running = False# 注冊自身節點到區塊鏈self.blockchain.register_node(f"{self.host}:{self.port}")def start(self) -> None:"""啟動節點"""self.running = Trueprint(f"Node listening on {self.host}:{self.port}")# 啟動接收線程receive_thread = threading.Thread(target=self._receive_connections)receive_thread.daemon = Truereceive_thread.start()def stop(self) -> None:"""停止節點"""self.running = Falseself.server.close()def connect_to_peer(self, peer_address: str) -> None:"""連接到其他對等節點參數:peer_address: 對等節點地址"""if peer_address == f"{self.host}:{self.port}":returnif peer_address not in self.peers:try:peer_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)host, port = peer_address.split(':')peer_socket.connect((host, int(port)))# 發送握手消息self._send_message(peer_socket, {"type": "handshake","data": {"address": f"{self.host}:{self.port}","chain_length": len(self.blockchain.chain)}})# 添加到對等節點列表self.peers.add(peer_address)# 注冊到區塊鏈self.blockchain.register_node(peer_address)# 啟動消息處理線程thread = threading.Thread(target=self._handle_peer, args=(peer_socket,))thread.daemon = Truethread.start()print(f"Connected to peer: {peer_address}")except Exception as e:print(f"Failed to connect to peer {peer_address}: {e}")def broadcast(self, message: Dict[str, Any]) -> None:"""廣播消息到所有連接的對等節點參數:message: 要廣播的消息"""for peer in list(self.peers):try:host, port = peer.split(':')peer_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)peer_socket.connect((host, int(port)))self._send_message(peer_socket, message)peer_socket.close()except Exception as e:print(f"Failed to send message to peer {peer}: {e}")self.peers.discard(peer)def _receive_connections(self) -> None:"""接收新的連接"""while self.running:try:client_socket, client_address = self.server.accept()print(f"New connection from {client_address}")# 啟動消息處理線程thread = threading.Thread(target=self._handle_peer, args=(client_socket,))thread.daemon = Truethread.start()except Exception as e:if self.running:print(f"Error receiving connection: {e}")def _handle_peer(self, peer_socket: socket.socket) -> None:"""處理來自對等節點的消息參數:peer_socket: 對等節點的套接字"""while self.running:try:message = self._receive_message(peer_socket)if not message:breakself._process_message(message, peer_socket)except Exception as e:print(f"Error handling peer: {e}")break# 關閉連接peer_socket.close()def _send_message(self, socket: socket.socket, message: Dict[str, Any]) -> None:"""發送消息到指定套接字參數:socket: 目標套接字message: 要發送的消息"""socket.send(json.dumps(message).encode() + b'\n')def _receive_message(self, socket: socket.socket) -> Dict[str, Any]:"""從指定套接字接收消息參數:socket: 源套接字返回:接收到的消息"""data = b''while True:chunk = socket.recv(1024)if not chunk:return Nonedata += chunkif b'\n' in data:messages = data.split(b'\n')for msg in messages[:-1]:try:return json.loads(msg.decode())except json.JSONDecodeError:continuedata = messages[-1]def _process_message(self, message: Dict[str, Any], peer_socket: socket.socket) -> None:"""處理接收到的消息參數:message: 接收到的消息peer_socket: 發送消息的套接字"""message_type = message.get("type")data = message.get("data")if message_type == "handshake":# 處理握手消息peer_address = data.get("address")peer_chain_length = data.get("chain_length")if peer_address and peer_address != f"{self.host}:{self.port}":self.peers.add(peer_address)self.blockchain.register_node(peer_address)# 檢查誰的鏈更長，決定是否需要同步if peer_chain_length > len(self.blockchain.chain):# 請求對方的完整區塊鏈self._send_message(peer_socket, {"type": "request_chain","data": {}})elif message_type == "new_block":# 處理新區塊通知block_data = data.get("block")if block_data:new_block = Block(index=block_data["index"],transactions=block_data["transactions"],timestamp=block_data["timestamp"],previous_hash=block_data["previous_hash"],nonce=block_data["nonce"])new_block.hash = block_data["hash"]# 添加新區塊self.blockchain.add_block(new_block, new_block.nonce)# 廣播新區塊給其他節點self.broadcast({"type": "new_block","data": {"block": new_block.to_dict()}})elif message_type == "new_transaction":# 處理新交易通知transaction = data.get("transaction")if transaction:self.blockchain.add_transaction(sender=transaction["sender"],recipient=transaction["recipient"],amount=transaction["amount"],contract_code=transaction.get("contract_code"))# 廣播新交易給其他節點self.broadcast({"type": "new_transaction","data": {"transaction": transaction}})elif message_type == "request_chain":# 處理區塊鏈請求self._send_message(peer_socket, {"type": "chain_response","data": {"chain": [block.to_dict() for block in self.blockchain.chain],"length": len(self.blockchain.chain)}})elif message_type == "chain_response":# 處理區塊鏈響應chain_data = data.get("chain")chain_length = data.get("length")if chain_length > len(self.blockchain.chain):# 驗證并替換當前鏈new_chain = [Block(**block_data) for block_data in chain_data]if self.blockchain.is_chain_valid(new_chain):self.blockchain.chain = new_chainprint("Blockchain updated from peer")

# 去中心化應用示例
class DApp:def __init__(self, node: P2PNode, contract_manager: ContractManager, wallet_address: str):"""初始化去中心化應用參數:node: P2P網絡節點contract_manager: 智能合約管理器wallet_address: 錢包地址"""self.node = nodeself.contract_manager = contract_managerself.wallet_address = wallet_addressdef deploy_token_contract(self, name: str, symbol: str, total_supply: int) -> str:"""部署代幣合約參數:name: 代幣名稱symbol: 代幣符號total_supply: 總供應量返回:合約ID"""# 代幣合約代碼contract_code = f"""// 簡單的ERC20風格代幣合約contract {name} {{string public name = "{name}";string public symbol = "{symbol}";uint256 public totalSupply = {total_supply};mapping(address => uint256) public balanceOf;constructor() {{balanceOf[msg.sender] = totalSupply;}}function transfer(address to, uint256 value) public returns (bool) {{require(balanceOf[msg.sender] >= value, "Insufficient balance");balanceOf[msg.sender] -= value;balanceOf[to] += value;return true;}}function balanceOf(address account) public view returns (uint256) {{return balanceOf[account];}}}}"""# 部署合約contract_id = self.contract_manager.deploy_contract(contract_code, self.wallet_address)print(f"代幣合約已部署，合約ID: {contract_id}")return contract_iddef transfer_tokens(self, contract_id: str, recipient: str, amount: int) -> Dict[str, Any]:"""轉移代幣參數:contract_id: 合約IDrecipient: 接收方地址amount: 轉移數量返回:交易結果"""# 執行合約函數result = self.contract_manager.execute_contract(contract_id=contract_id,function_name="transfer",params={"to": recipient, "value": amount},caller=self.wallet_address)return resultdef get_balance(self, contract_id: str, address: str) -> int:"""獲取代幣余額參數:contract_id: 合約IDaddress: 查詢地址返回:余額"""# 執行合約函數result = self.contract_manager.execute_contract(contract_id=contract_id,function_name="balanceOf",params={"account": address},caller=self.wallet_address)if result["success"]:return result["result"]else:return 0# 主程序示例
def main():# 創建區塊鏈blockchain = Blockchain()# 創建智能合約管理器contract_manager = ContractManager(blockchain)# 創建P2P節點node = P2PNode(host="localhost", port=5000, blockchain=blockchain)node.start()# 連接到其他節點（如果有）# node.connect_to_peer("localhost:5001")# 創建錢包地址wallet1 = hashlib.sha256("user1".encode()).hexdigest()wallet2 = hashlib.sha256("user2".encode()).hexdigest()# 創建去中心化應用dapp = DApp(node, contract_manager, wallet1)# 部署代幣合約token_contract_id = dapp.deploy_token_contract("MyToken", "MTK", 1000000)# 挖礦以確認合約部署print("Mining to confirm contract deployment...")block = blockchain.mine_block(wallet1)print(f"Block mined: {block.index}")# 轉移代幣print(f"Transferring tokens from {wallet1} to {wallet2}")result = dapp.transfer_tokens(token_contract_id, wallet2, 1000)print(f"Transfer result: {result}")# 挖礦以確認交易print("Mining to confirm transaction...")block = blockchain.mine_block(wallet1)print(f"Block mined: {block.index}")# 查詢余額balance1 = dapp.get_balance(token_contract_id, wallet1)balance2 = dapp.get_balance(token_contract_id, wallet2)print(f"{wallet1} balance: {balance1}")print(f"{wallet2} balance: {balance2}")# 顯示區塊鏈信息print(f"Blockchain length: {len(blockchain.chain)}")for block in blockchain.chain:print(f"Block {block.index}: {block.hash[:10]}...")for tx in block.transactions:print(f"  Transaction: {tx['sender'][:10]}... -> {tx['recipient'][:10]}... ({tx['amount']})")# 停止節點node.stop()if __name__ == "__main__":main()