0 前言

🔥 優質競賽項目系列，今天要分享的是

python區塊鏈實現 - proof of work工作量證明共識算法

該項目較為新穎，適合作為競賽課題方向，學長非常推薦！

🧿 更多資料, 項目分享：

https://gitee.com/dancheng-senior/postgraduate

1 區塊鏈基礎

學長以比特幣的結構向大家詳解區塊鏈的組成部分

1.1 比特幣內部結構

• previous hash(前一個區塊的hash)
• merkle root(默克爾樹根節點,內部存儲交易數據)
• timestamp(當前區塊生成的時間)
• nonce(曠工計算hash值次數)

1.2 實現的區塊鏈數據結構

• index 當前第幾個區塊
• timestamp 該區塊創建時的時間戳
• data 交易信息
• previousHash 前一個區塊的hash
• hash 當前區塊的hash

1.3 注意點

第一個區塊叫做創世區塊(genesis block),區塊鏈創建的時候默認生產的這里用的是單純的鏈表,不是用默克爾樹存儲

示例代碼

?

from hashlib import sha256//區塊schemaclass Block:def __init__(self,index,timestamp,data,previousHash=""):self.index = indexself.timestamp = timestampself.data = dataself.previousHash = previousHashself.hash = self.calculateHash()//計算當前區塊的hash值def calculateHash(self):plainData = str(self.index)+str(self.timestamp)+str(self.data)return sha256(plainData.encode('utf-8')).hexdigest()def __str__(self):return str(self.__dict__)//區塊鏈schemaclass BlockChain://初始化的時候 創建 創世區塊def __init__(self):self.chain = [self.createGenesisBlock()]//構建創世區塊def createGenesisBlock(self):return Block(0,"01/01/2018","genesis block","0")//獲取最后一個區塊def getLatestBlock(self):return self.chain[len(self.chain)-1]//往區塊鏈里面添加區塊def addBlock(self,newBlock):newBlock.previousHash = self.getLatestBlock().hashnewBlock.hash = newBlock.calculateHash()self.chain.append(newBlock)def __str__(self):return str(self.__dict__)    //校驗區塊鏈是不是有效的 有沒有人被篡改def chainIsValid(self):for index in range(1,len(self.chain)):currentBlock = self.chain[index]previousBlock = self.chain[index-1]if (currentBlock.hash != currentBlock.calculateHash()):return Falseif previousBlock.hash != currentBlock.previousHash:return Falsereturn TruemyCoin = BlockChain()
myCoin.addBlock(Block(1,"02/01/2018","{amount:4}"))
myCoin.addBlock(Block(2,"03/01/2018","{amount:5}"))#print block info 打印區塊鏈信息
print("print block info ####:")
for block in myCoin.chain:print(block)
#check blockchain is valid 檢查區塊鏈是不是有效的
print("before tamper block,blockchain is valid ###")
print(myCoin.chainIsValid())
#tamper the blockinfo  篡改區塊2的數據
myCoin.chain[1].data = "{amount:1002}"
print("after tamper block,blockchain is valid ###")
print(myCoin.chainIsValid())

輸出結果

?

print block info ####:
{'index': 0, 'timestamp': '01/01/2018', 'data': 'genesis block', 'previousHash': '0', 'hash': 'd8d21e5ba33780d5eb77d09d3b407ceb8ade4e5545ef951de1997b209d91e264'}
{'index': 1, 'timestamp': '02/01/2018', 'data': '{amount:4}', 'previousHash': 'd8d21e5ba33780d5eb77d09d3b407ceb8ade4e5545ef951de1997b209d91e264', 'hash': '15426e32db30f4b26aa719ba5e573f372f41e27e4728eb9e9ab0bea8eae63a9d'}
{'index': 2, 'timestamp': '03/01/2018', 'data': '{amount:5}', 'previousHash': '15426e32db30f4b26aa719ba5e573f372f41e27e4728eb9e9ab0bea8eae63a9d', 'hash': '75119e897f21c769acee6e32abcefc5e88e250a1f35cc95946379436050ac2f0'}
before tamper block,blockchain is valid ###
True
after tamper block,blockchain is valid ###
False

1.4 區塊鏈的核心-工作量證明算法

上面學長介紹了區塊鏈的基本結構,我在之前的基礎上來簡單實現一下工作量證明算法(proof of
work),在介紹pow之前先思考一下為什么要工作量證明算法,或者再往前想一步為什么比特幣如何解決信任的問題?

1.4.1 拜占庭將軍問題

比特幣出現之前就有了拜占庭將軍問題,主要思想是,如何在分布式系統環境里去相信其他人發給你的信息？

一組拜占庭將軍分別各率領一支軍隊共同圍困一座城市。為了簡化問題，將各支軍隊的行動策略限定為進攻或撤離兩種。因為部分軍隊進攻部分軍隊撤離可能會造成災難性后果，因此各位將軍必須通過投票來達成一致策略，即所有軍隊一起進攻或所有軍隊一起撤離。因為各位將軍分處城市不同方向，他
系統的問題在于，將軍中可能出現叛徒，他們不僅可能向較為糟糕的策略投票，還可能選擇性地發送投票信息。假設有9位將軍投票，其中1名叛徒。8名忠誠的將軍中出現了4人投進攻，4人投撤離的情況。這時候叛徒可能故意給4名投進攻的將領送信表示投票進攻，而給4名投撤離的將領送信表示投撤離。這樣一來在4名投進攻的將領看來，投票結果是5人投進攻，從而發起進攻；而在4名投撤離的將軍看來則是5人投撤離。這樣各支軍隊的一致協同就遭到了破壞。

1.4.2 解決辦法

拜占庭將軍問題主要問題是,中間人可以攔截消息,進行修改;上述的那些士兵可以理解成比特幣中的一些節點,不是所有節點拿到消息后都是可以直接處理的,先去解決一個數學問題,就是工作量證明,只有擁有特定的計算能力解決了問題之后才能去修改或者校驗(驗證,打包,上鏈)。

上圖就是簡單的工作量證明算法流程,一串數字后面有個x,x之前的數可以理解成交易數據,然后需要找到一個x,讓整個數的hash值的開頭有n個0,如果hash是很均勻的話,那么生成的hash值每一位為0或者1都是等可能的,所以前n個都為0的概率就是2的n次方/2的hash值位數,上圖給出了如果hash值是5個bit的情況下的所有可能

1.4.3 代碼實現

?

from hashlib import sha256import timeclass Block:def __init__(self,index,timestamp,data,previousHash=""):self.index = indexself.timestamp = timestampself.data = dataself.previousHash = previousHashself.nonce = 0 //代表當前計算了多少次hash計算self.hash = self.calculateHash()def calculateHash(self):plainData = str(self.index)+str(self.timestamp)+str(self.data)+str(self.nonce)return sha256(plainData.encode('utf-8')).hexdigest()#挖礦 difficulty代表復雜度 表示前difficulty位都為0才算成功def minerBlock(self,difficulty):while(self.hash[0:difficulty]!=str(0).zfill(difficulty)):self.nonce+=1self.hash = self.calculateHash()def __str__(self):return str(self.__dict__)class BlockChain:def __init__(self):self.chain = [self.createGenesisBlock()]self.difficulty = 5def createGenesisBlock(self):return Block(0,"01/01/2018","genesis block")def getLatestBlock(self):return self.chain[len(self.chain)-1]#添加區塊前需要 做一道計算題😶,坐完后才能把區塊加入到鏈上def addBlock(self,newBlock):newBlock.previousHash = self.getLatestBlock().hashnewBlock.minerBlock(self.difficulty)self.chain.append(newBlock)def __str__(self):return str(self.__dict__)    def chainIsValid(self):for index in range(1,len(self.chain)):currentBlock = self.chain[index]previousBlock = self.chain[index-1]if (currentBlock.hash != currentBlock.calculateHash()):return Falseif previousBlock.hash != currentBlock.previousHash:return Falsereturn TruemyCoin = BlockChain()# 下面打印了每個區塊挖掘需要的時間 比特幣通過一定的機制控制在10分鐘出一個塊 # 其實就是根據當前網絡算力 調整我們上面difficulty值的大小,如果你在# 本地把上面代碼difficulty的值調很大你可以看到很久都不會出計算結果startMinerFirstBlockTime = time.time()print("start to miner first block time :"+str(startMinerFirstBlockTime))myCoin.addBlock(Block(1,"02/01/2018","{amount:4}"))print("miner first block time completed" + ",used " +str(time.time()-startMinerFirstBlockTime) +"s")startMinerSecondBlockTime = time.time()print("start to miner first block time :"+str(startMinerSecondBlockTime))myCoin.addBlock(Block(2,"03/01/2018","{amount:5}"))print("miner second block time completed" + ",used " +str(time.time()-startMinerSecondBlockTime) +"s\n")#print block infoprint("print block info ####:\n")for block in myCoin.chain:print("\n")print(block)#check blockchain is validprint("before tamper block,blockchain is valid ###")print(myCoin.chainIsValid())#tamper the blockinfomyCoin.chain[1].data = "{amount:1002}"print("after tamper block,blockchain is valid ###")print(myCoin.chainIsValid())

輸出

2 快速實現一個區塊鏈

2.1 什么是區塊鏈

區塊鏈是一個不可變得，有序的被稱之為塊的記錄鏈，它們可以包含交易、文件或者任何你喜歡的數據，但最重要的是，它們用hash連接在一起。

2.2 一個完整的快包含什么

一個索引，一個時間戳，一個事物列表，一個校驗， 一個前快的散鏈表

2.3 什么是挖礦

挖礦其實非常簡單就做了以下三件事：

1、計算工作量證明poW
2、通過新增一個交易賦予礦工（自已）一個幣
3、構造新區塊并將其添加到鏈中

2.4 工作量證明算法：

使用該算法來證明是如何在區塊上創建和挖掘新的區塊，pow的目標是計算出一個符合特定條件的數字，這個數字對于所有人而言必須在計算上非常困難，但易于驗證，這就是工作證明背后的核心思想計算難度與目標字符串需要滿足的特定字符串成正比。

2.5 實現代碼

?

import hashlibimport jsonimport requestsfrom textwrap import dedentfrom time import timefrom uuid import uuid4from urllib.parse import urlparsefrom flask import Flask, jsonify, requestclass Blockchain(object):def __init__(self):...self.nodes = set()# 用 set 來儲存節點，避免重復添加節點....self.chain = []self.current_transactions = []#創建創世區塊self.new_block(previous_hash=1,proof=100)def reister_node(self,address):"""在節點列表中添加一個新節點:param address::return:"""prsed_url = urlparse(address)self.nodes.add(prsed_url.netloc)def valid_chain(self,chain):"""確定一個給定的區塊鏈是否有效:param chain::return:"""last_block = chain[0]current_index = 1while current_index<len(chain):block = chain[current_index]print(f'{last_block}')print(f'{block}')print("\n______\n")# 檢查block的散列是否正確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 ressolve_conflicts(self):"""共識算法:return:"""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 Falsedef new_block(self,proof,previous_hash=None):"""創建一個新的塊并將其添加到鏈中:param proof: 由工作證明算法生成證明:param previous_hash: 前一個區塊的hash值:return: 新區塊"""block = {'index':len(self.chain)+1,'timestamp':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):# 將新事務添加到事務列表中"""Creates a new transaction to go into the next mined Block:param sender:發送方的地址:param recipient:收信人地址:param amount:數量:return:保存該事務的塊的索引"""self.current_transactions.append({'sender':sender,'recipient':recipient,'amount':amount,})return  self.last_block['index'] + 1@staticmethoddef hash(block):"""給一個區塊生成 SHA-256 值:param block::return:"""# 必須確保這個字典（區塊）是經過排序的，否則將會得到不一致的散列block_string = json.dumps(block,sort_keys=True).encode()return hashlib.sha256(block_string).hexdigest()@propertydef last_block(self):# 返回鏈中的最后一個塊return self.chain[-1]def proof_of_work(self,last_proof):# 工作算法的簡單證明proof = 0while self.valid_proof(last_proof,proof)is False:proof +=1return proof@staticmethoddef valid_proof(last_proof,proof):# 驗證證明guess =  f'{last_proof}{proof}'.encode()guess_hash = hashlib.sha256(guess).hexdigest()return guess_hash[:4] =="0000"# 實例化節點app = Flask(__name__)# 為該節點生成一個全局惟一的地址node_identifier = str(uuid4()).replace('-','')# 實例化Blockchain類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)# 必須得到一份尋找證據的獎賞。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': "New Block Forged",'index': block['index'],'transactions': block['transactions'],'proof': block['proof'],'previous_hash': block['previous_hash'],}return jsonify(response), 200# 創建交易請求@app.route('/transactions/new',methods=['POST'])def new_transactions():values = request.get_json()# 檢查所需要的字段是否位于POST的data中required = ['seder','recipient','amount']if not all(k in values for k in request):return 'Missing values',400#創建一個新的事物index = blockchain.new_transaction(values['sender'], values['recipient'], values['amount'])response = {'message': f'Transaction will be added to Block {index}'}return jsonify(response), 201# 獲取所有快信息@app.route('/chain',methods=['GET'])def full_chain():response = {'chain':blockchain.chain,'length':len(blockchain.chain),}return jsonify(response),200# 添加節點@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", 400for 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), 200if __name__ == '__main__':app.run(host='0.0.0.0',port=5000)

代碼弄好啟動你的項目以后打開Postman 完成以下操作

學長通過請求 http://localhost:5000/mine進行采礦

3 最后

🧿 更多資料, 項目分享：

https://gitee.com/dancheng-senior/postgraduate