用 Python 從 0 開始創建一個區塊鏈

對數字貨幣的崛起感到新奇的我們，並且想知道其背後的技術——區塊鏈是怎樣實現的。本文通過 Python 構建一個區塊鏈可以加深對區塊鏈的理解。

準備工作

本文要求讀者對 Python 有基本的理解，能讀寫基本的 Python，並且需要對 HTTP 請求有基本的了解。

我們知道區塊鏈是由區塊的記錄構成的不可變、有序的鏈結構，記錄可以是交易、文件或任何你想要的數據，重要的是它們是通過哈希值（hashes）鏈接起來的。

如果你還不是很了解哈希，可以查看這篇文章https://learncryptography.com/hash-functions/what-are-hash-functions。

環境準備：

環境準備，確保已經安裝 Python3.6+、pip、Flask、requests。

安裝方法：

pip install Flask==0.12.2 requests==2.18.4

同時還需要一個 HTTP 客戶端，比如 Postman、cURL 或其他客戶端。

參考源代碼（原代碼在我翻譯的時候，無法運行，我 fork 了一份，修復了其中的錯誤，並添加了翻譯，感謝 star）。

開始創建 Blockchain

新建一個文件 blockchain.py，本文所有的代碼都寫在這一個文件中，可以隨時參考源代碼。

Blockchain 類

首先創建一個 Blockchain 類，在構造函數中創建了兩個列表，一個用於儲存區塊鏈，一個用於儲存交易。

以下是 Blockchain 類的框架：

class Blockchain(object):
    def __init__(self):
        self.chain = []
        self.current_transactions = []
    def new_block(self):
        # Creates a new Block and adds it to the chain
        pass
    def new_transaction(self):
        # Adds a new transaction to the list of transactions
        pass
    @staticmethod
    def hash(block):
        # Hashes a Block
        pass
    @property
    def last_block(self):
        # Returns the last Block in the chain
        pass
 

Blockchain 類用來管理鏈條，它能存儲交易、加入新塊等，下面我們來進一步完善這些方法。

塊結構

每個區塊包含屬性：索引（index）、Unix 時間戳（timestamp）、交易列表（transactions）、工作量證明（稍後解釋）以及前一個區塊的 Hash 值。

以下是一個區塊的結構：

block = {
    ‘index‘: 1,
    ‘timestamp‘: 1506057125.900785,
    ‘transactions‘: [
        {
            ‘sender‘: "8527147fe1f5426f9dd545de4b27ee00",
            ‘recipient‘: "a77f5cdfa2934df3954a5c7c7da5df1f",
            ‘amount‘: 5,
        }
    ],
    ‘proof‘: 324984774000,
    ‘previous_hash‘: "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"
}

到這裏區塊鏈的概念就清楚了，每個新的區塊都包含上一個區塊的 Hash，這是關鍵的一點，它保障了區塊鏈的不可變性。

如果攻擊者破壞了前面的某個區塊，那麽後面所有區塊的Hash都會變得不正確。不理解的話，慢慢消化，可參考{% post_link whatbc 區塊鏈技術原理 %}。

加入交易

接下來我們需要添加一個交易，來完善下 new_transaction 方法：

class Blockchain(object):
    ...
    def new_transaction(self, sender, recipient, amount):
        """
        生成新交易信息，信息將加入到下一個待挖的區塊中
        :param sender: <str> Address of the Sender
        :param recipient: <str> Address of the Recipient
        :param amount: <int> Amount
        :return: <int> The index of the Block that will hold this transaction
        """
        self.current_transactions.append({
            ‘sender‘: sender,
            ‘recipient‘: recipient,
            ‘amount‘: amount,
        })
        return self.last_block[‘index‘] + 1

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

創建新塊

當 Blockchain 實例化後，我們需要構造一個創世塊（沒有前區塊的第一個區塊），並且給它加上一個工作量證明。每個區塊都需要經過工作量證明，俗稱挖礦，稍後會繼續講解。

為了構造創世塊，我們還需要完善 new_block()，new_transaction() 和hash() 方法：

import hashlib
import json
from time import time
class Blockchain(object):
    def __init__(self):
        self.current_transactions = []
        self.chain = []
        # Create the genesis block
        self.new_block(previous_hash=1, proof=100)
    def new_block(self, proof, previous_hash=None):
        """
        生成新塊
        :param proof: <int> The proof given by the Proof of Work algorithm
        :param previous_hash: (Optional) <str> Hash of previous Block
        :return: <dict> New Block
        """
        block = {
            ‘index‘: len(self.chain) + 1,
            ‘timestamp‘: time(),
            ‘transactions‘: self.current_transactions,
            ‘proof‘: proof,
            ‘previous_hash‘: previous_hash or self.hash(self.chain[-1]),
        }
        # Reset the current list of transactions
        self.current_transactions = []
        self.chain.append(block)
        return block
    def new_transaction(self, sender, recipient, amount):
        """
        生成新交易信息，信息將加入到下一個待挖的區塊中
        :param sender: <str> Address of the Sender
        :param recipient: <str> Address of the Recipient
        :param amount: <int> Amount
        :return: <int> The index of the Block that will hold this transaction
        """
        self.current_transactions.append({
            ‘sender‘: sender,
            ‘recipient‘: recipient,
            ‘amount‘: amount,
        })
        return self.last_block[‘index‘] + 1
    @property
    def last_block(self):
        return self.chain[-1]
    @staticmethod
    def hash(block):
        """
        生成塊的 SHA-256 hash值
        :param block: <dict> Block
        :return: <str>
        """
        # We must make sure that the Dictionary is Ordered, or we‘ll have inconsistent hashes
        block_string = json.dumps(block, sort_keys=True).encode()
        return hashlib.sha256(block_string).hexdigest()

通過上面的代碼和註釋可以對區塊鏈有直觀的了解，接下來我們看看區塊是怎麽挖出來的。

理解工作量證明

新的區塊依賴工作量證明算法（PoW）來構造。PoW 的目標是找出一個符合特定條件的數字，這個數字很難計算出來，但容易驗證。這就是工作量證明的核心思想。

為了方便理解，舉個例子：

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

用 Python 實現如下：

from hashlib import sha256
x = 5
y = 0  # y未知
while sha256(f‘{x*y}‘.encode()).hexdigest()[-1] != "0":
    y += 1
print(f‘The solution is y = {y}‘)

結果是：y = 21，因為：

hash(5 * 21) = 1253e9373e...5e3600155e860

在比特幣中，使用稱為 Hashcash 的工作量證明算法，它和上面的問題很類似，礦工們為了爭奪創建區塊的權利而爭相計算結果。

通常，計算難度與目標字符串需要滿足的特定字符的數量成正比，礦工算出結果後，會獲得比特幣獎勵。當然，在網絡上非常容易驗證這個結果。

實現工作量證明

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

import hashlib
import json
from time import time
from uuid import uuid4
class Blockchain(object):
    ...
    def proof_of_work(self, last_proof):
        """
        簡單的工作量證明:
         - 查找一個 p‘ 使得 hash(pp‘) 以4個0開頭
         - p 是上一個塊的證明,  p‘ 是當前的證明
        :param last_proof: <int>
        :return: <int>
        """
        proof = 0
        while self.valid_proof(last_proof, proof) is False:
            proof += 1
        return proof
    @staticmethod
    def valid_proof(last_proof, proof):
        """
        驗證證明: 是否hash(last_proof, proof)以4個0開頭?
        :param last_proof: <int> Previous Proof
        :param proof: <int> Current Proof
        :return: <bool> True if correct, False if not.
        """
        guess = f‘{last_proof}{proof}‘.encode()
        guess_hash = hashlib.sha256(guess).hexdigest()
        return guess_hash[:4] == "0000"

衡量算法復雜度的辦法是修改零開頭的個數。使用 4 個零來用於演示，你會發現多一個零都會大大增加計算出結果所需的時間。

現在 Blockchain 類基本已經完成了，接下來使用 HTTP requests 來進行交互。

Blockchain 作為 API 接口

我們將使用 Python Flask 框架，這是一個輕量 Web 應用框架，它方便將網絡請求映射到 Python 函數，現在我們來讓 Blockchain 運行在 Flask Web 上。

我們將創建三個接口：

• /transactions/new 創建一個交易並添加到區塊
• /mine 告訴服務器去挖掘新的區塊
• /chain 返回整個區塊鏈

創建節點

我們的“Flask 服務器”將扮演區塊鏈網絡中的一個節點，我們先添加一些框架代碼：

import hashlib
import json
from textwrap import dedent
from time import time
from uuid import uuid4
from flask import Flask
class Blockchain(object):
    ...
# Instantiate our Node
app = Flask(__name__)
# Generate a globally unique address for this node
node_identifier = str(uuid4()).replace(‘-‘, ‘‘)
# Instantiate the Blockchain
blockchain = Blockchain()
@app.route(‘/mine‘, methods=[‘GET‘])
def mine():
    return "We‘ll mine a new Block"
@app.route(‘/transactions/new‘, methods=[‘POST‘])
def new_transaction():
    return "We‘ll add a new transaction"
@app.route(‘/chain‘, methods=[‘GET‘])
def full_chain():
    response = {
        ‘chain‘: blockchain.chain,
        ‘length‘: len(blockchain.chain),
    }
    return jsonify(response), 200
if __name__ == ‘__main__‘:
    app.run(host=‘0.0.0.0‘, port=5000)

簡單的說明一下以上代碼：

• 第 15 行：創建一個節點。
• 第 18 行：為節點創建一個隨機的名字。
• 第 21 行：實例 Blockchain 類。
• 第 24–26 行：創建 /mine GET 接口。
• 第 28–30 行：創建 /transactions/new POST 接口,可以給接口發送交易數據。
• 第 32–38 行：創建 /chain 接口, 返回整個區塊鏈。
• 第 40–41 行：服務運行在端口 5000 上。

發送交易

發送到節點的交易數據結構如下：

{
 "sender": "my address",
 "recipient": "someone else‘s address",
 "amount": 5
}

之前已經有添加交易的方法，基於接口來添加交易就很簡單了：

import hashlib
import json
from textwrap import dedent
from time import time
from uuid import uuid4
from flask import Flask, jsonify, request
...
@app.route(‘/transactions/new‘, methods=[‘POST‘])
def new_transaction():
    values = request.get_json()
    # Check that the required fields are in the POST‘ed data
    required = [‘sender‘, ‘recipient‘, ‘amount‘]
    if not all(k in values for k in required):
        return ‘Missing values‘, 400
    # Create a new Transaction
    index = blockchain.new_transaction(values[‘sender‘], values[‘recipient‘], values[‘amount‘])
    response = {‘message‘: f‘Transaction will be added to Block {index}‘}
    return jsonify(response), 201

挖礦

挖礦正是神奇所在，它很簡單，做了以下三件事：

• 計算工作量證明 PoW。
• 通過新增一個交易授予礦工（自己）一個幣。
• 構造新區塊並將其添加到鏈中。

import hashlib
import json
from textwrap import dedent
from time import time
from uuid import uuid4
from flask import Flask, jsonify, request
...
import hashlib
import json
from time import time
from uuid import uuid4
from flask import Flask, jsonify, request
...
@app.route(‘/mine‘, methods=[‘GET‘])
def mine():
    # We run the proof of work algorithm to get the next proof...
    last_block = blockchain.last_block
    last_proof = last_block[‘proof‘]
    proof = blockchain.proof_of_work(last_proof)
    # 給工作量證明的節點提供獎勵.
    # 發送者為 "0" 表明是新挖出的幣
    blockchain.new_transaction(
        sender="0",
        recipient=node_identifier,
        amount=1,
    )
    # Forge the new Block by adding it to the chain
    block = blockchain.new_block(proof)
    response = {
        ‘message‘: "New Block Forged",
        ‘index‘: block[‘index‘],
        ‘transactions‘: block[‘transactions‘],
        ‘proof‘: block[‘proof‘],
        ‘previous_hash‘: block[‘previous_hash‘],
    }
    return jsonify(response), 200

註意交易的接收者是我們自己的服務器節點，我們做的大部分工作都只是圍繞 Blockchain 類方法進行交互。到此，我們的區塊鏈就算完成了，我們來實際運行下。

運行區塊鏈

你可以使用 cURL 或 Postman 去和 API 進行交互。

啟動 server：

$ python blockchain.py
* Runing on http://127.0.0.1:5000/ (Press CTRL+C to quit)

讓我們通過請求 http://localhost:5000/mine 來進行挖礦：

通過 post 請求，添加一個新交易：

如果不是使用 Postman，則用以下的 cURL 語句也是一樣的：

$ curl -X POST -H "Content-Type: application/json" -d ‘{
 "sender": "d4ee26eee15148ee92c6cd394edd974e",
 "recipient": "someone-other-address",
 "amount": 5
}‘ "http://localhost:5000/transactions/new"

在挖了兩次礦之後，就有 3 個塊了，通過請求 http://localhost:5000/chain 可以得到所有的塊信息。

{
  "chain": [
    {
      "index": 1,
      "previous_hash": 1,
      "proof": 100,
      "timestamp": 1506280650.770839,
      "transactions": []
    },
    {
      "index": 2,
      "previous_hash": "c099bc...bfb7",
      "proof": 35293,
      "timestamp": 1506280664.717925,
      "transactions": [
        {
          "amount": 1,
          "recipient": "8bbcb347e0634905b0cac7955bae152b",
          "sender": "0"
        }
      ]
    },
    {
      "index": 3,
      "previous_hash": "eff91a...10f2",
      "proof": 35089,
      "timestamp": 1506280666.1086972,
      "transactions": [
        {
          "amount": 1,
          "recipient": "8bbcb347e0634905b0cac7955bae152b",
          "sender": "0"
        }
      ]
    }
  ],
  "length": 3
}

一致性（共識）

我們已經有了一個基本的區塊鏈可以接受交易和挖礦，但是區塊鏈系統應該是分布式的。

既然是分布式的，那麽我們究竟拿什麽保證所有節點有同樣的鏈呢？這就是一致性問題，我們要想在網絡上有多個節點，就必須實現一個一致性的算法。

註冊節點

在實現一致性算法之前，我們需要找到一種方式讓一個節點知道它相鄰的節點。

每個節點都需要保存一份包含網絡中其他節點的記錄，因此讓我們新增幾個接口：

• /nodes/register 接收 URL 形式的新節點列表。
• /nodes/resolve 執行一致性算法，解決任何沖突，確保節點擁有正確的鏈。

我們修改下 Blockchain 的 init 函數並提供一個註冊節點方法：

...
from urllib.parse import urlparse
...
class Blockchain(object):
    def __init__(self):
        ...
        self.nodes = set()
        ...
    def register_node(self, address):
        """
        Add a new node to the list of nodes
        :param address: <str> Address of node. Eg. ‘http://192.168.0.5:5000‘
        :return: None
        """
        parsed_url = urlparse(address)
        self.nodes.add(parsed_url.netloc)

我們用 set 來儲存節點，這是一種避免重復添加節點的簡單方法。

實現共識算法

前面提到，沖突是指不同的節點擁有不同的鏈，為了解決這個問題，規定最長的、有效的鏈才是最終的鏈，換句話說，網絡中有效最長鏈才是實際的鏈。

我們使用以下的算法，來達到網絡中的共識：

...
import requests
class Blockchain(object)
    ...
    def valid_chain(self, chain):
        """
        Determine if a given blockchain is valid
        :param chain: <list> A blockchain
        :return: <bool> True if valid, False if not
        """
        last_block = chain[0]
        current_index = 1
        while current_index < len(chain):
            block = chain[current_index]
            print(f‘{last_block}‘)
            print(f‘{block}‘)
            print("\n-----------\n")
            # Check that the hash of the block is correct
            if block[‘previous_hash‘] != self.hash(last_block):
                return False
            # Check that the Proof of Work is correct
            if not self.valid_proof(last_block[‘proof‘], block[‘proof‘]):
                return False
            last_block = block
            current_index += 1
        return True
    def resolve_conflicts(self):
        """
        共識算法解決沖突
        使用網絡中最長的鏈.
        :return: <bool> True 如果鏈被取代, 否則為False
        """
        neighbours = self.nodes
        new_chain = None
        # We‘re only looking for chains longer than ours
        max_length = len(self.chain)
        # Grab and verify the chains from all the nodes in our network
        for node in neighbours:
            response = requests.get(f‘http://{node}/chain‘)
            if response.status_code == 200:
                length = response.json()[‘length‘]
                chain = response.json()[‘chain‘]
                # Check if the length is longer and the chain is valid
                if length > max_length and self.valid_chain(chain):
                    max_length = length
                    new_chain = chain
        # Replace our chain if we discovered a new, valid chain longer than ours
        if new_chain:
            self.chain = new_chain
            return True
        return False

第一個方法 valid_chain() 用來檢查是否是有效鏈，遍歷每個塊驗證 hash 和 proof。

第二個方法 resolve_conflicts() 用來解決沖突，遍歷所有的鄰居節點，並用上一個方法檢查鏈的有效性， 如果發現有效更長鏈，就替換掉自己的鏈。

讓我們添加兩個路由，一個用來註冊節點，一個用來解決沖突。

@app.route(‘/nodes/register‘, methods=[‘POST‘])
def register_nodes():
    values = request.get_json()
    nodes = values.get(‘nodes‘)
    if nodes is None:
        return "Error: Please supply a valid list of nodes", 400
    for node in nodes:
        blockchain.register_node(node)
    response = {
        ‘message‘: ‘New nodes have been added‘,
        ‘total_nodes‘: list(blockchain.nodes),
    }
    return jsonify(response), 201
@app.route(‘/nodes/resolve‘, methods=[‘GET‘])
def consensus():
    replaced = blockchain.resolve_conflicts()
    if replaced:
        response = {
            ‘message‘: ‘Our chain was replaced‘,
            ‘new_chain‘: blockchain.chain
        }
    else:
        response = {
            ‘message‘: ‘Our chain is authoritative‘,
            ‘chain‘: blockchain.chain
        }
    return jsonify(response), 200

你可以在不同的機器運行節點，或在一臺機機開啟不同的網絡端口來模擬多節點的網絡。

這裏在同一臺機器開啟不同的端口演示，在不同的終端運行以下命令，就啟動了兩個節點：

• http://localhost:5000
• http://localhost:5001

pipenv run python blockchain.py
pipenv run python blockchain.py -p 5001

然後在節點 2 上挖兩個塊，確保是更長的鏈，然後在節點 1 上訪問接口 /nodes/resolve，這時節點 1 的鏈會通過共識算法被節點 2 的鏈取代。

好啦，你可以邀請朋友們一起來測試你的區塊鏈。

http://www.aibbt.com/a/19369.html

用Python從零開始創建區塊鏈