用 Python 擼一個區塊鏈

相信你和我一樣對數字貨幣的崛起感到新奇，並且想知道其背後的技術——區塊鏈是怎樣實現的。

但是理解區塊鏈並非易事，至少對於我來說是如此。晦澀難懂的視頻、漏洞百出的教程以及示例的匱乏令我倍受挫折。

我喜歡在實踐中學習，通過寫代碼來學習技術會掌握得更牢固。如果你也這樣做，那麼讀完本文，你將獲得一個可用的區塊鏈以及對區塊鏈的深刻理解。

開始之前...

首先你需要知道區塊鏈是由被稱為區塊的記錄構成的不可變的、有序的鏈式結構，這些記錄可以是交易、文件或任何你想要的數據，最重要的是它們是通過 Hash 連接起來的。

如果你不了解 Hash，這裡有個例子 https://learncryptography.com/hash-functions/what-are-hash-functions

其次，你需要安裝 Python3.6+，Flask，Request

pip install Flask==0.12.2 requests==2.18.4

同時你還需要一個 HTTP 客戶端，比如 Postman，cURL 或任何其它客戶端。

最終的源代碼在這裡：https://github.com/dvf/blockchain

第一步： 打造一個 Blockchain

新建一個文件 blockchain.py，本文所有的代碼都寫在這一個文件中。首先創建一個 Blockchain 類，在構造函數中我們創建了兩個列表，一個用於儲存區塊鏈，一個用於儲存交易。

1 class Blockchain(object): 2 def __init__(self): 3 self.chain = [] 4 self.current_transactions = [] 5 6 def new_block(self): 7 # Creates a new Block and adds it to the chain 8 pass 9 10 def new_transaction(self):11 # Adds a new transaction to the list of transactions12 pass13 14 @staticmethod15 def hash(block):16 # Hashes a Block17 pass18 19 @property20 def last_block(self):21 # Returns the last Block in the chain22 pass

一個區塊有五個基本屬性：index，timestamp（in Unix time），transaction 列表，工作量證明（稍後解釋）以及前一個區塊的 Hash 值。

1 block = { 2 "index": 1, 3 "timestamp": 1506057125.900785, 4 "transactions": [ 5 { 6 "sender": "8527147fe1f5426f9dd545de4b27ee00", 7 "recipient": "a77f5cdfa2934df3954a5c7c7da5df1f", 8 "amount": 5, 9 }10 ],11 "proof": 324984774000,12 "previous_hash": "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"13 }

到這裡，區塊鏈的概念應該比較清楚了：每個新的區塊都會包含上一個區塊的 Hash 值。這一點非常關鍵，它是區塊鏈不可變性的根本保障。如果攻擊者破壞了前面的某個區塊，那麼後面所有區塊的 Hash 都會變得不正確。不理解？慢慢消化~

我們需要一個向區塊添加交易的方法：

1 class Blockchain(object): 2 ... 3 4 def new_transaction(self, sender, recipient, amount): 5 """ 6 Creates a new transaction to go into the next mined Block 7 :param sender: <str> Address of the Sender 8 :param recipient: <str> Address of the Recipient 9 :param amount: <int> Amount10 :return: <int> The index of the Block that will hold this transaction11 """12 13 self.current_transactions.append({14 "sender": sender,15 "recipient": recipient,16 "amount": amount,17 })18 19 return self.last_block["index"] + 1

new_transaction() 方法向列表中添加一個交易記錄，並返回該記錄將被添加到的區塊——下一個待挖掘的區塊——的索引，稍後在用戶提交交易時會有用。

當 Blockchain 實例化後，我們需要創建一個初始的區塊（創世塊），並且給它預設一個工作量證明。

除了添加創世塊的代碼，我們還需要補充 new_block(), new_transaction() 和 hash() 方法：

1 import hashlib 2 import json 3 from time import time 4 5 class Blockchain(object): 6 def __init__(self): 7 self.current_transactions = [] 8 self.chain = [] 9 10 # Create the genesis block11 self.new_block(previous_hash=1, proof=100)12 13 def new_block(self, proof, previous_hash=None):14 block = {15 "index": len(self.chain) + 1,16 "timestamp": time(),17 "transactions": self.current_transactions,18 "proof": proof,19 "previous_hash": previous_hash or self.hash(self.chain[-1]),20 }21 22 # Reset the current list of transactions23 self.current_transactions = []24 25 self.chain.append(block)26 return block27 28 def new_transaction(self, sender, recipient, amount):29 self.current_transactions.append({30 "sender": sender,31 "recipient": recipient,32 "amount": amount,33 })34 35 return self.last_block["index"] + 136 37 @property38 def last_block(self):39 return self.chain[-1]40 41 @staticmethod42 def hash(block):43 block_string = json.dumps(block, sort_keys=True).encode()44 return hashlib.sha256(block_string).hexdigest()

上面的代碼應該很直觀，我們基本上有了區塊鏈的雛形。但此時你肯定很想知道一個區塊究竟是怎樣被創建或挖掘出來的。

新的區塊來自工作量證明（PoW）演算法。PoW 的目標是計算出一個符合特定條件的數字，這個數字對於所有人而言必須在計算上非常困難，但易於驗證。這就是工作量證明的核心思想。

舉個例子：

假設一個整數 x 乘以另一個整數 y 的積的 Hash 值必須以 0 結尾，即 hash(x * y) = ac23dc...0。設 x = 5，求 y？

1 from hashlib import sha2562 x = 53 y = 0 # We don"t know what y should be yet...4 while sha256(f"{x*y}".encode()).hexdigest()[-1] != "0":5 y += 16 print(f"The solution is y = {y}")

結果是 y = 21 // hash(5 * 21) = 1253e9373e...5e3600155e860

在比特幣中，工作量證明演算法被稱為 Hashcash，它和上面的問題很相似，只不過計算難度非常大。這就是礦工們為了爭奪創建區塊的權利而爭相計算的問題。通常，計算難度與目標字元串需要滿足的特定字元的數量成正比，礦工算出結果後，就會獲得一定數量的比特幣獎勵（通過交易）。

網路要驗證結果，當然非常容易。

讓我們來實現一個 PoW 演算法，和上面的例子非常相似，規則是：尋找一個數 p，使得它與前一個區塊的 proof 拼接成的字元串的 Hash 值以 4 個零開頭。

1 import hashlib 2 import json 3 from time import time 4 from uuid import uuid4 5 6 class Blockchain(object): 7 ... 8 9 def proof_of_work(self, last_proof):10 proof = 011 while self.valid_proof(last_proof, proof) is False:12 proof += 113 14 return proof15 16 @staticmethod17 def valid_proof(last_proof, proof):18 guess = f"{last_proof}{proof}".encode()19 guess_hash = hashlib.sha256(guess).hexdigest()20 return guess_hash[:4] == "0000"

衡量演算法複雜度的辦法是修改零的個數。4 個零足夠用於演示了，你會發現哪怕多一個零都會大大增加計算出結果所需的時間。

我們的 Blockchain 基本已經完成了，接下來我們將使用 HTTP requests 來與之交互。

第二步：作為 API 的 Blockchain

我們將使用 Flask 框架，它十分輕量並且很容易將網路請求映射到 Python 函數。

我們將創建三個介面：

/transactions/new 創建一個交易並添加到區塊

/mine 告訴伺服器去挖掘新的區塊

/chain 返回整個區塊鏈

我們的伺服器將扮演區塊鏈網路中的一個節點。我們先添加一些常規代碼：

1 import hashlib 2 import json 3 from textwrap import dedent 4 from time import time 5 from uuid import uuid4 6 from flask import Flask, jsonify, request 7 8 class Blockchain(object): 9 ...10 11 # Instantiate our Node12 app = Flask(__name__)13 14 # Generate a globally unique address for this node15 node_identifier = str(uuid4()).replace("-", "")16 17 # Instantiate the Blockchain18 blockchain = Blockchain()19 20 @app.route("/mine", methods=["GET"])21 def mine():22 return "We"ll mine a new Block"23 24 @app.route("/transactions/new", methods=["POST"])25 def new_transaction():26 return "We"ll add a new transaction"27 28 @app.route("/chain", methods=["GET"])29 def full_chain():30 response = {31 "chain": blockchain.chain,32 "length": len(blockchain.chain),33 }34 return jsonify(response), 20035 36 if __name__ == "__main__":37 app.run(host="127.0.0.1", port=5000)

這是用戶發起交易時發送到伺服器的請求：

1 {2 "sender": "my address",3 "recipient": "someone else"s address",4 "amount": 55 }

我們已經有了向區塊添加交易的方法，因此剩下的部分就很簡單了：

1 @app.route("/transactions/new", methods=["POST"]) 2 def new_transaction(): 3 values = request.get_json() 4 5 # Check that the required fields are in the POST"ed data 6 required = ["sender", "recipient", "amount"] 7 if not all(k in values for k in required): 8 return "Missing values", 400 9 10 # Create a new Transaction11 index = blockchain.new_transaction(values["sender"], values["recipient"], values["amount"])12 13 response = {"message": f"Transaction will be added to Block {index}"}14 return jsonify(response), 201

挖掘端正是奇蹟發生的地方，它只做三件事：計算 PoW；通過新增一個交易授予礦工一定數量的比特幣；構造新的區塊並將其添加到區塊鏈中。

1 @app.route("/mine", methods=["GET"]) 2 def mine(): 3 # We run the proof of work algorithm to get the next proof... 4 last_block = blockchain.last_block 5 last_proof = last_block["proof"] 6 proof = blockchain.proof_of_work(last_proof) 7 8 # We must receive a reward for finding the proof. 9 # The sender is "0" to signify that this node has mined a new coin.10 blockchain.new_transaction(11 sender="0",12 recipient=node_identifier,13 amount=1,14 )15 16 # Forge the new Block by adding it to the chain17 block = blockchain.new_block(proof)18 19 response = {20 "message": "New Block Forged",21 "index": block["index"],22 "transactions": block["transactions"],23 "proof": block["proof"],24 "previous_hash": block["previous_hash"],25 }26 return jsonify(response), 200

需注意交易的接收者是我們自己的伺服器節點，目前我們做的大部分事情都只是圍繞 Blockchain 類進行交互。到此，我們的區塊鏈就算完成了。

第三步：交互演示

使用 Postman 演示，略。

第四步：一致性

這真的很棒，我們已經有了一個基本的區塊鏈可以添加交易和挖礦。但是，整個區塊鏈系統必須是分散式的。既然是分散式的，那麼我們究竟拿什麼保證所有節點運行在同一條鏈上呢？這就是一致性問題，我們要想在網路中添加新的節點，就必須實現保證一致性的演算法。

在實現一致性演算法之前，我們需要找到一種方式讓一個節點知道它相鄰的節點。每個節點都需要保存一份包含網路中其它節點的記錄。讓我們新增幾個介面：

1. /nodes/register 接收以 URL 的形式表示的新節點的列表

2. /nodes/resolve 用於執行一致性演算法，用於解決任何衝突，確保節點擁有正確的鏈

1 ... 2 from urllib.parse import urlparse 3 ... 4 5 class Blockchain(object): 6 def __init__(self): 7 ... 8 self.nodes = set() 9 ...10 11 def register_node(self, address):12 parsed_url = urlparse(address)13 self.nodes.add(parsed_url.netloc)

注意到我們用 set 來儲存節點，這是一種避免重複添加節點的簡便方法。

前面提到的衝突是指不同的節點擁有的鏈存在差異，要解決這個問題，我們規定最長的合規的鏈就是最有效的鏈，換句話說，只有最長且合規的鏈才是實際存在的鏈。

讓我們再添加兩個方法，一個用於添加相鄰節點，另一個用於解決衝突。

1 ... 2 import requests 3 4 class Blockchain(object) 5 ... 6 7 def valid_chain(self, chain): 8 last_block = chain[0] 9 current_index = 110 11 while current_index < len(chain):12 block = chain[current_index]13 print(f"{last_block}")14 print(f"{block}")15 print("
-----------
")16 # Check that the hash of the block is correct17 if block["previous_hash"] != self.hash(last_block):18 return False19 20 # Check that the Proof of Work is correct21 if not self.valid_proof(last_block["proof"], block["proof"]):22 return False23 24 last_block = block25 current_index += 126 27 return True28 29 def resolve_conflicts(self):30 neighbours = self.nodes31 new_chain = None32 33 # We"re only looking for chains longer than ours34 max_length = len(self.chain)35 36 # Grab and verify the chains from all the nodes in our network37 for node in neighbours:38 response = requests.get(f"http://{node}/chain")39 40 if response.status_code == 200:41 length = response.json()["length"]42 chain = response.json()["chain"]43 44 # Check if the length is longer and the chain is valid45 if length > max_length and self.valid_chain(chain):46 max_length = length47 new_chain = chain48 49 # Replace our chain if we discovered a new, valid chain longer than ours50 if new_chain:51 self.chain = new_chain52 return True53 54 return False

現在你可以新開一台機器，或者在本機上開啟不同的網路介面來模擬多節點的網路，或者邀請一些朋友一起來測試你的區塊鏈。

我希望本文能激勵你創造更多新東西。我之所以對數字貨幣入迷，是因為我相信區塊鏈會很快改變我們看待事物的方式，包括經濟、政府、檔案管理等。

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！