潁州網(wǎng)站建設(shè)最近新聞摘抄
Python實現(xiàn)簡單的區(qū)塊鏈
記錄自己假期所學相關(guān)內(nèi)容
文章中的內(nèi)容,開源代碼地址見文末。
文章目錄
- Python實現(xiàn)簡單的區(qū)塊鏈
- 1、分模塊實現(xiàn)簡單的單節(jié)點區(qū)塊鏈
- 1.1 Transaction類
- 1.2 DaDaMessage類
- 1.3 Block類
- 1.4 Dada_BlockCoin類
- 1.5 主函數(shù)BlockChainApp類
- 1.6 主函數(shù)類中實現(xiàn)了可視化界面,以下為演示效果
- 2、網(wǎng)絡共識區(qū)塊鏈編程實現(xiàn)
- 2.1 DaDaCoinBlockChain類
- 2.2 flask框架部分
- 2.3 初始化三個節(jié)點,分別模擬網(wǎng)絡中的三個礦工
- 2.4 效果演示
- 3、其他
1、分模塊實現(xiàn)簡單的單節(jié)點區(qū)塊鏈
1.1 Transaction類
class Transaction:def __init__(self,payer, # 付款方r ecer, # 收款方money): # 數(shù)字貨幣的數(shù)額self.payer = payerself.recer = recerself.money = moneyself.timestamp = datetime.datetime.now() # 交易時間def __repr__(self):return str(self.payer) + " pays " + str(self.recer) + \" " + str(self.money) + " " + str(self.timestamp)
1.2 DaDaMessage類
class DaDaMessage: # 交易記錄類def __init__(self, data):self.hash = None # 自身的哈希self.prev_hash = None # 上一個信息記錄的哈希self.timestamp = datetime.datetime.now() # 交易時間self.data = data # 交易信息self.payload_hash = self._hash_payload() # 交易后的哈希def _hash_payload(self): # 對于交易時間與交易數(shù)據(jù)進行哈希計算return hashlib.sha256((str(self.timestamp) + str(self.data)).encode()).hexdigest()def _hash_message(self): # 對于交易進行鎖定return hashlib.sha256((str(self.prev_hash) + str(self.payload_hash)).encode()).hexdigest()def seal(self): # 密封self.hash = self._hash_message() # 對應數(shù)據(jù)鎖定,對于交易前的鏈鎖定def validate(self): # 驗證if self.payload_hash != self._hash_payload(): # 判斷是否有人修改raise InvalidMessage("交易數(shù)據(jù)與時間被修改" + str(self))if self.hash != self._hash_message(): # 判斷消息鏈raise InvalidMessage("交易的哈希鏈接被修改" + str(self))return "數(shù)據(jù)正常" + str(self)def __repr__(self): # 返回對象的基本信息mystr = "hash:{}, prev_hash:{}, data:{}".format(self.hash, self.prev_hash, self.data)return mystrdef link(self, Message):self.prev_hash = Message.hash # 鏈接
1.3 Block類
class Block:def __init__(self, *args): # 初始化self.messageList = [] # 存儲多個交易記錄,存放區(qū)塊self.timestamp = None # 存儲多個記錄最終鎖定的時間self.hash = None # 當前的哈希散列self.prev_hash = None # 上一塊的哈希散列if args:for arg in args:self.add_message(arg)# self.messagelist.append(arg)def add_message(self, message): # 增加交易信息# 區(qū)分第一條與后面多條,是否需要鏈接if len(self.messageList) > 0:message.link(self.messageList[-1]) # 鏈接message.seal() # 密封message.validate() # 校驗self.messageList.append(message) # 追加記錄def link(self, block): # 區(qū)塊鏈鏈接self.prev_hash = block.hashdef seal(self): # 密封self.timestamp = datetime.datetime.now() # 密封確定當前時間self.hash = self._hash_block() # 密封當前的哈希值def _hash_block(self): # 密封 上一塊哈希,時間線,交易記錄的最后一個if len(self.messageList) > 0:return hashlib.md5((str(self.prev_hash) +str(self.timestamp) +str(self.messageList[-1].hash)).encode("utf-8")).hexdigest()# else:# return hashlib.sha256((str(self.prev_hash) +# str(self.timestamp) +# str(0)).encode("utf-8")).hexdigest()def validate(self): # 校驗for i, message in enumerate(self.messageList): # 校驗每一個交易記錄message.validate() # 校驗每一條if i > 0 and message.prev_hash != self.messageList[i - 1].hash:raise InvalidBlock("無效block,交易記錄被修改為在第{}條記錄".format(i) + str(self))# print("無效block,交易記錄被修改為在第{}條記錄".format(i))# return str(self) + "數(shù)據(jù)NO"return " " + str(self) + " 數(shù)據(jù)OK"def __repr__(self): # 類的對象描述# return "money block = hash : {}, pre_hash : {}, len : {}, time : {}".\# format(self.hash, self.prev_hash, len(self.messageList), self.timestamp)return "money block \n hash : {} \n pre_hash : {} \n len : {} \n time : {}". \format(self.hash, self.prev_hash, len(self.messageList), self.timestamp)
1.4 Dada_BlockCoin類
class Dada_BlockCoin: # 區(qū)塊鏈def __init__(self): # 初始化self.blockList = []def add_block(self, block): # 增加區(qū)塊if (len(self.blockList) > 0):block.prev_hash = self.blockList[-1].hash # 區(qū)塊鏈的哈希block.seal() # 密封block.validate() # 校驗self.blockList.append(block) # 增加區(qū)塊def validate(self): # 校驗for i, block in enumerate(self.blockList):try:block.validate()except InvalidBlock as e:raise InvalidBlockCoin("區(qū)塊校驗錯誤,區(qū)塊索引{}".format(i))except InvalidMessage as e:print(e)def __repr__(self): # 字符串格式化return "Dada_BlockCoin : {}".format(len(self.blockList)) # 獲取長度
1.5 主函數(shù)BlockChainApp類
class BlockChainApp:def __init__(self, root):self.root = rootself.root.title("區(qū)塊鏈可視化")self.blockchain = Dada_BlockCoin()self.current_block_transactions = []self.selected_block_index = None# 添加滾動條self.canvas_frame = tk.Frame(root)self.canvas_frame.pack(expand=tk.YES, fill=tk.BOTH)self.scrollbar_x = tk.Scrollbar(self.canvas_frame, orient=tk.HORIZONTAL)self.scrollbar_x.pack(side=tk.BOTTOM, fill=tk.X)self.scrollbar_y = tk.Scrollbar(self.canvas_frame)self.scrollbar_y.pack(side=tk.RIGHT, fill=tk.Y)# 區(qū)塊鏈信息顯示框self.canvas = tk.Canvas(self.canvas_frame, width=800, yscrollcommand=self.scrollbar_y.set,xscrollcommand=self.scrollbar_x.set)self.canvas.pack(side=tk.LEFT, expand=tk.YES, fill=tk.BOTH)self.scrollbar_x.config(command=self.canvas.xview)self.scrollbar_y.config(command=self.canvas.yview)self.canvas.config(scrollregion=self.canvas.bbox(tk.ALL))# 交易信息輸入框self.payer_label = tk.Label(root, text="付款方:")self.payer_label.pack()self.payer_entry = tk.Entry(root)self.payer_entry.pack()self.recer_label = tk.Label(root, text="收款方:")self.recer_label.pack()self.recer_entry = tk.Entry(root)self.recer_entry.pack()self.money_label = tk.Label(root, text="金額:")self.money_label.pack()self.money_entry = tk.Entry(root)self.money_entry.pack()# 添加交易按鈕self.add_transaction_btn = tk.Button(root, text="添加交易", command=self.add_transaction)self.add_transaction_btn.pack()# 添加區(qū)塊按鈕self.add_block_btn = tk.Button(root, text="添加區(qū)塊", command=self.add_block)self.add_block_btn.pack()# 模擬篡改按鈕self.tamper_block_btn = tk.Button(root, text="模擬篡改數(shù)據(jù)", command=self.tamper_block)self.tamper_block_btn.pack()# 區(qū)塊查詢框self.query_frame = tk.Frame(root)self.query_frame.pack()self.query_label = tk.Label(self.query_frame, text="查詢區(qū)塊索引:")self.query_label.pack(side=tk.LEFT)self.query_entry = tk.Entry(self.query_frame)self.query_entry.pack(side=tk.LEFT)self.query_btn = tk.Button(self.query_frame, text="查詢區(qū)塊", command=self.query_block)self.query_btn.pack(side=tk.LEFT)# 交易信息顯示框self.transaction_info_text = tk.Text(root, wrap=tk.WORD, width=60, height=10)self.transaction_info_text.pack()def add_transaction(self):payer = self.payer_entry.get()recer = self.recer_entry.get()money = self.money_entry.get()try:money = int(money)except ValueError:messagebox.showerror("錯誤", "金額必須為整數(shù)")returnif payer and recer and money > 0:transaction = Transaction(payer, recer, money)self.current_block_transactions.append(DaDaMessage(transaction))self.update_blockchain_info()messagebox.showinfo("成功", "交易添加成功")else:messagebox.showerror("錯誤", "請輸入有效的交易信息")def add_block(self):if self.current_block_transactions:block = Block(*self.current_block_transactions)self.blockchain.add_block(block)self.current_block_transactions = []self.update_blockchain_info()messagebox.showinfo("成功", "區(qū)塊添加成功")else:messagebox.showerror("錯誤", "當前區(qū)塊沒有交易信息")def tamper_block(self):try:block_index = int(input("請輸入要篡改的區(qū)塊索引:"))except ValueError:messagebox.showerror("錯誤", "請輸入有效的區(qū)塊索引")returnif 0 <= block_index < len(self.blockchain.blockList):block = self.blockchain.blockList[block_index]if len(block.messageList) > 0:# 模擬篡改第一條交易信息block.messageList[0].data = "篡改后的交易信息"try:self.blockchain.validate()except InvalidBlockCoin as e:messagebox.showerror("篡改數(shù)據(jù)", "區(qū)塊鏈數(shù)據(jù)已被篡改,篡改發(fā)生在第{}個區(qū)塊".format(e.args[0]))except InvalidMessage as e:messagebox.showerror("篡改數(shù)據(jù)", "區(qū)塊鏈數(shù)據(jù)已被篡改,篡改發(fā)生在第{}個區(qū)塊的第{}條交易信息".format(e.args[0][0], e.args[0][1]))else:messagebox.showinfo("篡改數(shù)據(jù)", "區(qū)塊鏈數(shù)據(jù)未被篡改")finally:self.update_blockchain_info()else:messagebox.showerror("錯誤", "區(qū)塊中沒有交易信息,無法篡改")else:messagebox.showerror("錯誤", "區(qū)塊索引超出范圍")def query_block(self):try:block_index = int(self.query_entry.get())except ValueError:messagebox.showerror("錯誤", "請輸入有效的區(qū)塊索引")returnif 0 <= block_index < len(self.blockchain.blockList):block = self.blockchain.blockList[block_index]transactions = "\n".join(str(msg.data) for msg in block.messageList)self.transaction_info_text.delete("1.0", tk.END)self.transaction_info_text.insert(tk.END, transactions)self.selected_block_index = block_indexmessagebox.showinfo("成功", "查詢成功")else:messagebox.showerror("錯誤", "區(qū)塊索引超出范圍")def update_blockchain_info(self):self.canvas.delete("all") # 清空畫布上的內(nèi)容x, y = 20, 50 # 區(qū)塊鏈的初始位置block_width, block_height = 250, 150 # 區(qū)塊的寬度和高度for block in self.blockchain.blockList:self.canvas.create_rectangle(x, y, x + block_width, y + block_height, outline="black") # 繪制區(qū)塊框self.canvas.create_text(x + block_width // 2, y + block_height // 2, text=str(block)) # 顯示區(qū)塊信息if block.prev_hash: # 繪制區(qū)塊之間的連接線prev_x, prev_y = x - block_width, y + block_height // 2self.canvas.create_line(prev_x, prev_y, x, y + block_height // 2, fill="red")x += block_width + 50 # 每個區(qū)塊之間留一定的間隔# 設(shè)置Canvas可滾動范圍self.canvas.config(scrollregion=self.canvas.bbox(tk.ALL))if __name__ == "__main__":root = tk.Tk()app = BlockChainApp(root)app.root.mainloop()
1.6 主函數(shù)類中實現(xiàn)了可視化界面,以下為演示效果
- 初始化界面
- 添加交易后添加區(qū)塊
在上述區(qū)塊鏈可視化界面中顯示對應區(qū)塊,并用紅色連線連接,表示區(qū)塊鏈的哈希連接。
- 查詢區(qū)塊信息
輸入?yún)^(qū)塊對應的索引(從0開始),即可查詢對應區(qū)塊的交易信息
2、網(wǎng)絡共識區(qū)塊鏈編程實現(xiàn)
2.1 DaDaCoinBlockChain類
class DaDaCoinBlockChain:def __init__(self): # 初始化self.current_transactions = [] # 交易列表self.chain = [] # 區(qū)塊鏈管理多個區(qū)塊self.nodes = set() # 保存網(wǎng)絡中其他節(jié)點self.new_block(previous_hash="1", proof=100) # 創(chuàng)建創(chuàng)世區(qū)塊def new_block(self,proof: int, # 確定proof為int類型previous_hash: Optional[str] # 上一塊的哈希類型) -> Dict[str, Any]: # 創(chuàng)建一個區(qū)塊,返回一個字典數(shù)據(jù)類型block = {"index": len(self.chain) + 1, # 索引"timestamp": time.time(), # 當前時間"transaction": self.current_transactions, # 交易記錄"proof": proof, # 工作量證明"previous_hash": previous_hash or self.hash(self.chain[-1]) # 前一區(qū)塊哈希}self.current_transactions = [] # 交易記錄加入?yún)^(qū)塊之后清空self.chain.append(block) # 區(qū)塊加入?yún)^(qū)塊鏈return blockdef new_transactions(self, sender: str, recipient: str, amount) -> int: # 創(chuàng)建一個交易self.current_transactions.append({"sender": sender, # 付款方"recipient": recipient, # 收款方"amount": amount # 數(shù)量})return self.last_block["index"] + 1 # 索引標記交易數(shù)量@propertydef last_block(self) -> Dict[str, Any]: # 取得最后一個區(qū)塊return self.chain[-1]@staticmethoddef hash(block: Dict[str, any]) -> str: # 哈希加密,傳遞一個字典,返回字符串blockString = json.dumps(block, sort_keys=True).encode() # 編碼return hashlib.sha256(blockString).hexdigest()def proof_of_work(self, last_proof: int) -> int: # 工作量證明,挖礦過程proof = 0while self.valid_proof(last_proof, proof) is False:proof += 1return proof@staticmethod # 第N個區(qū)塊依賴于N-1個區(qū)塊,簡單挖礦def valid_proof(last_proof: int, proof: int) -> bool: # 驗證證明guess = f'{last_proof * proof}'.encode()guess_hash = hashlib.sha256(guess).hexdigest()# print(str(guess_hash))return guess_hash[-4:] == "1234"def valid_chain(self, chain: List[Dict[str, Any]]) -> bool: # 區(qū)塊鏈校驗# List[Dict[str, Any]]是一個列表,列表的每個元素都是字典last_block = chain[0] # 第一個區(qū)塊curr_index = 1 # 當前的第一個索引while curr_index < len(chain):block = chain[curr_index] # 當前區(qū)塊# 哈希校驗,校驗區(qū)塊鏈的鏈接if block["previous_hash"] != self.hash(last_block):return False# 工作量校驗,挖礦的工作量校驗if not self.valid_proof(last_block["proof"], block["proof"]):return Falselast_block = block # 輪詢curr_index += 1 # 索引自增return Truedef register_node(self, addr: str) -> None: # 加入網(wǎng)絡的其他節(jié)點,用于更新now_url = urlparse(addr) # 解析self.nodes.add(now_url.netloc) # 增加網(wǎng)絡節(jié)點def resolve_conflicts(self) -> bool: # 共識算法# 網(wǎng)絡中的多個節(jié)點,取出最長的neighbours = self.nodes # 取得所有的節(jié)點new_chain = None # 新的區(qū)塊鏈max_length = len(self.chain) # 當前的區(qū)塊鏈長度for node in neighbours:response = requests.get(f"http://{node}/chain") # 訪問網(wǎng)絡節(jié)點if response.status_code == 200:length = response.json()["length"] # 取出長度chain = response.json()["chain"] # 取出區(qū)塊鏈# 如果當前區(qū)塊鏈比我長并且經(jīng)得起校驗,那么就更新if length > max_length and self.valid_chain(chain):max_length = lengthnew_chain = chain # 保存長度與區(qū)塊鏈if new_chain:self.chain = new_chain # 替換區(qū)塊鏈return Truereturn False
2.2 flask框架部分
dadacoin = DaDaCoinBlockChain() # 創(chuàng)建一個網(wǎng)絡節(jié)點
node_id = str(uuid4()).replace("-", "") # 節(jié)點替換,生成密鑰
print("當前錢包地址:", node_id)
app = Flask(__name__) # 初始化flask框架@app.route("/")
def index_page():return "你好,歡迎來到達達幣系統(tǒng)!"@app.route("/chain") # 查看所有的區(qū)塊鏈
def index_chain():response = {"chain": dadacoin.chain, # 區(qū)塊鏈"length": len(dadacoin.chain) # 區(qū)塊鏈長度}return jsonify(response), 200@app.route("/mine") # 挖礦
def index_mine():last_block = dadacoin.last_block # 取得最后一個區(qū)塊last_proof = last_block["proof"] # 取得工作量證明proof = dadacoin.proof_of_work(last_proof) # 挖礦計算# 系統(tǒng)獎勵比特幣,挖礦產(chǎn)生交易dadacoin.new_transactions(sender="0", # 0代表系統(tǒng)獎勵recipient=node_id, # 當前錢包地址amount=10 # 獎勵數(shù)量)block = dadacoin.new_block(proof, None) # 增加一個區(qū)塊response = {"message": "新的區(qū)塊創(chuàng)建","index": block["index"], # 倉建的索引"transaction": block["transaction"], # 交易"proof": block["proof"], # 工作量證明"previous_hash": block["previous_hash"] # 上一塊的哈希}return jsonify(response), 200@app.route("/new_transaction", methods=["POST"]) # 創(chuàng)建一個新的交易
def new_transaction():values = request.get_json() # 抓取網(wǎng)絡傳輸?shù)男畔?/span>required = ["sender", "recipient", "amount"]if not all(key in values for key in required):return "數(shù)據(jù)不完整", 400index = dadacoin.new_transactions(values["sender"],values["recipient"],values["amount"]) # 新增一個交易response = {"message": f"交易加入到區(qū)塊{index}",}return jsonify(response), 200@app.route("/new_node", methods=["POST"]) # 增加網(wǎng)絡節(jié)點
def new_node():values = request.get_json() # 獲取json字符串nodes = values.get("nodes") # 獲取所有節(jié)點if nodes is None:return "節(jié)點為空", 400for node in nodes:dadacoin.register_node(node) # 增加網(wǎng)絡節(jié)點response = {"message": "網(wǎng)絡節(jié)點已經(jīng)增加","nodes": list(dadacoin.nodes) # 查看所有節(jié)點}return jsonify(response), 200@app.route("/node_refresh")
def node_refresh():replaced = dadacoin.resolve_conflicts() # 共識算法進行最長替換message = ""if replaced:message += "區(qū)塊鏈已經(jīng)被替換為最長"else:message += "當前區(qū)塊鏈已經(jīng)是最長無需替換"response = {"message": message,"new--chain": dadacoin.chain}return jsonify(response), 200
2.3 初始化三個節(jié)點,分別模擬網(wǎng)絡中的三個礦工
if __name__ == '__main__':app.run("127.0.0.1", 5000)if __name__ == '__main__':app.run("127.0.0.1", 5001)if __name__ == '__main__':app.run("127.0.0.1", 5002)
2.4 效果演示
分別啟動三個礦工節(jié)點
- 初始化界面
- 初次訪問chain路徑時,顯示創(chuàng)世區(qū)塊
- 訪問mine路徑時,進行挖礦操作,挖出新的區(qū)塊,礦工獲得出塊獎勵
- 添加交易
使用postman進行操作
此時僅僅是將交易添加到區(qū)塊,還沒有挖出區(qū)塊3,再次進行mine操作,可以看到區(qū)塊3中除了出塊獎勵,還有此次交易
- 網(wǎng)絡共識,解決分叉沖突
此時模擬節(jié)點1挖出6個區(qū)塊,節(jié)點2挖出4個區(qū)塊,節(jié)點3挖出3個區(qū)塊。將節(jié)點1和節(jié)點2的信息同步給區(qū)塊3,此時根據(jù)最長鏈原則,應該更新為節(jié)點1的6個區(qū)塊。
使用postman添加節(jié)點
刷新節(jié)點信息
此時節(jié)點3已經(jīng)被最長鏈代替。
3、其他
開源代碼地址: Gitee倉庫
b站參考視頻:參考視頻