0 前言

🔥 優(yōu)質(zhì)競賽項(xiàng)目系列，今天要分享的是

python區(qū)塊鏈實(shí)現(xiàn) - proof of work工作量證明共識算法

該項(xiàng)目較為新穎，適合作為競賽課題方向，學(xué)長非常推薦！

🧿 更多資料, 項(xiàng)目分享：

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

1 區(qū)塊鏈基礎(chǔ)

學(xué)長以比特幣的結(jié)構(gòu)向大家詳解區(qū)塊鏈的組成部分

1.1 比特幣內(nèi)部結(jié)構(gòu)

• previous hash(前一個區(qū)塊的hash)
• merkle root(默克爾樹根節(jié)點(diǎn),內(nèi)部存儲交易數(shù)據(jù))
• timestamp(當(dāng)前區(qū)塊生成的時間)
• nonce(曠工計(jì)算hash值次數(shù))

1.2 實(shí)現(xiàn)的區(qū)塊鏈數(shù)據(jù)結(jié)構(gòu)

• index 當(dāng)前第幾個區(qū)塊
• timestamp 該區(qū)塊創(chuàng)建時的時間戳
• data 交易信息
• previousHash 前一個區(qū)塊的hash
• hash 當(dāng)前區(qū)塊的hash

1.3 注意點(diǎn)

第一個區(qū)塊叫做創(chuàng)世區(qū)塊(genesis block),區(qū)塊鏈創(chuàng)建的時候默認(rèn)生產(chǎn)的這里用的是單純的鏈表,不是用默克爾樹存儲

示例代碼

?

from hashlib import sha256//區(qū)塊schemaclass Block:def __init__(self,index,timestamp,data,previousHash=""):self.index = indexself.timestamp = timestampself.data = dataself.previousHash = previousHashself.hash = self.calculateHash()//計(jì)算當(dāng)前區(qū)塊的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__)//區(qū)塊鏈schemaclass BlockChain://初始化的時候 創(chuàng)建 創(chuàng)世區(qū)塊def __init__(self):self.chain = [self.createGenesisBlock()]//構(gòu)建創(chuàng)世區(qū)塊def createGenesisBlock(self):return Block(0,"01/01/2018","genesis block","0")//獲取最后一個區(qū)塊def getLatestBlock(self):return self.chain[len(self.chain)-1]//往區(qū)塊鏈里面添加區(qū)塊def addBlock(self,newBlock):newBlock.previousHash = self.getLatestBlock().hashnewBlock.hash = newBlock.calculateHash()self.chain.append(newBlock)def __str__(self):return str(self.__dict__)    //校驗(yàn)區(qū)塊鏈?zhǔn)遣皇怯行У?有沒有人被篡改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 打印區(qū)塊鏈信息
print("print block info ####:")
for block in myCoin.chain:print(block)
#check blockchain is valid 檢查區(qū)塊鏈?zhǔn)遣皇怯行У?/span>
print("before tamper block,blockchain is valid ###")
print(myCoin.chainIsValid())
#tamper the blockinfo  篡改區(qū)塊2的數(shù)據(jù)
myCoin.chain[1].data = "{amount:1002}"
print("after tamper block,blockchain is valid ###")
print(myCoin.chainIsValid())

輸出結(jié)果

?

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 區(qū)塊鏈的核心-工作量證明算法

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

1.4.1 拜占庭將軍問題

比特幣出現(xiàn)之前就有了拜占庭將軍問題,主要思想是,如何在分布式系統(tǒng)環(huán)境里去相信其他人發(fā)給你的信息？

一組拜占庭將軍分別各率領(lǐng)一支軍隊(duì)共同圍困一座城市。為了簡化問題，將各支軍隊(duì)的行動策略限定為進(jìn)攻或撤離兩種。因?yàn)椴糠周婈?duì)進(jìn)攻部分軍隊(duì)撤離可能會造成災(zāi)難性后果，因此各位將軍必須通過投票來達(dá)成一致策略，即所有軍隊(duì)一起進(jìn)攻或所有軍隊(duì)一起撤離。因?yàn)楦魑粚④姺痔幊鞘胁煌较?#xff0c;他
系統(tǒng)的問題在于，將軍中可能出現(xiàn)叛徒，他們不僅可能向較為糟糕的策略投票，還可能選擇性地發(fā)送投票信息。假設(shè)有9位將軍投票，其中1名叛徒。8名忠誠的將軍中出現(xiàn)了4人投進(jìn)攻，4人投撤離的情況。這時候叛徒可能故意給4名投進(jìn)攻的將領(lǐng)送信表示投票進(jìn)攻，而給4名投撤離的將領(lǐng)送信表示投撤離。這樣一來在4名投進(jìn)攻的將領(lǐng)看來，投票結(jié)果是5人投進(jìn)攻，從而發(fā)起進(jìn)攻；而在4名投撤離的將軍看來則是5人投撤離。這樣各支軍隊(duì)的一致協(xié)同就遭到了破壞。

1.4.2 解決辦法

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

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

1.4.3 代碼實(shí)現(xiàn)

?

from hashlib import sha256import timeclass Block:def __init__(self,index,timestamp,data,previousHash=""):self.index = indexself.timestamp = timestampself.data = dataself.previousHash = previousHashself.nonce = 0 //代表當(dāng)前計(jì)算了多少次hash計(jì)算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代表復(fù)雜度 表示前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]#添加區(qū)塊前需要 做一道計(jì)算題😶,坐完后才能把區(qū)塊加入到鏈上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()# 下面打印了每個區(qū)塊挖掘需要的時間 比特幣通過一定的機(jī)制控制在10分鐘出一個塊 # 其實(shí)就是根據(jù)當(dāng)前網(wǎng)絡(luò)算力 調(diào)整我們上面difficulty值的大小,如果你在# 本地把上面代碼difficulty的值調(diào)很大你可以看到很久都不會出計(jì)算結(jié)果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 快速實(shí)現(xiàn)一個區(qū)塊鏈

2.1 什么是區(qū)塊鏈

區(qū)塊鏈?zhǔn)且粋€不可變得，有序的被稱之為塊的記錄鏈，它們可以包含交易、文件或者任何你喜歡的數(shù)據(jù)，但最重要的是，它們用hash連接在一起。

2.2 一個完整的快包含什么

一個索引，一個時間戳，一個事物列表，一個校驗(yàn)， 一個前快的散鏈表

2.3 什么是挖礦

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

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

2.4 工作量證明算法：

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

2.5 實(shí)現(xiàn)代碼

?

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 來儲存節(jié)點(diǎn)，避免重復(fù)添加節(jié)點(diǎn)....self.chain = []self.current_transactions = []#創(chuàng)建創(chuàng)世區(qū)塊self.new_block(previous_hash=1,proof=100)def reister_node(self,address):"""在節(jié)點(diǎn)列表中添加一個新節(jié)點(diǎn):param address::return:"""prsed_url = urlparse(address)self.nodes.add(prsed_url.netloc)def valid_chain(self,chain):"""確定一個給定的區(qū)塊鏈?zhǔn)欠裼行?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)# 獲取并驗(yàn)證網(wǎng)絡(luò)中的所有節(jié)點(diǎn)的鏈for node in neighbours:response = requests.get(f'http://{node}/chain')if response.status_code == 200:length = response.json()['length']chain = response.json()['chain']# 檢查長度是否長，鏈?zhǔn)欠裼行?/span>if length > max_length and self.valid_chain(chain):max_length = lengthnew_chain = chain# 如果發(fā)現(xiàn)一個新的有效鏈比當(dāng)前的長，就替換當(dāng)前的鏈if new_chain:self.chain = new_chainreturn Truereturn Falsedef new_block(self,proof,previous_hash=None):"""創(chuàng)建一個新的塊并將其添加到鏈中:param proof: 由工作證明算法生成證明:param previous_hash: 前一個區(qū)塊的hash值:return: 新區(qū)塊"""block = {'index':len(self.chain)+1,'timestamp':time(),'transactions':self.current_transactions,'proof':proof,'previous_hash':previous_hash or self.hash(self.chain[-1]),}# 重置當(dāng)前交易記錄self.current_transactions = []self.chain.append(block)return blockdef new_transaction(self,sender,recipient,amount):# 將新事務(wù)添加到事務(wù)列表中"""Creates a new transaction to go into the next mined Block:param sender:發(fā)送方的地址:param recipient:收信人地址:param amount:數(shù)量:return:保存該事務(wù)的塊的索引"""self.current_transactions.append({'sender':sender,'recipient':recipient,'amount':amount,})return  self.last_block['index'] + 1@staticmethoddef hash(block):"""給一個區(qū)塊生成 SHA-256 值:param block::return:"""# 必須確保這個字典（區(qū)塊）是經(jīng)過排序的，否則將會得到不一致的散列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):# 驗(yàn)證證明guess =  f'{last_proof}{proof}'.encode()guess_hash = hashlib.sha256(guess).hexdigest()return guess_hash[:4] =="0000"# 實(shí)例化節(jié)點(diǎn)app = Flask(__name__)# 為該節(jié)點(diǎn)生成一個全局惟一的地址node_identifier = str(uuid4()).replace('-','')# 實(shí)例化Blockchain類blockchain = Blockchain()# 進(jìn)行挖礦請求@app.route('/mine',methods=['GET'])def mine():# 運(yùn)行工作算法的證明來獲得下一個證明。last_block = blockchain.last_blocklast_proof = last_block['proof']proof = blockchain.proof_of_work(last_proof)# 必須得到一份尋找證據(jù)的獎賞。blockchain.new_transaction(sender="0",recipient=node_identifier,amount=1,)# 通過將其添加到鏈中來構(gòu)建新的塊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# 創(chuàng)建交易請求@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#創(chuàng)建一個新的事物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# 添加節(jié)點(diǎn)@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)

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

學(xué)長通過請求 http://localhost:5000/mine進(jìn)行采礦

3 最后

🧿 更多資料, 項(xiàng)目分享：

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