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文章來源地址http://www.zghlxwxcb.cn/news/detail-831886.html

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(曠工計算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ū)塊schema
    class Block:
         
        def __init__(self,index,timestamp,data,previousHash=""):
            
            self.index = index
            self.timestamp = timestamp
            self.data = data
            self.previousHash = previousHash
            self.hash = self.calculateHash()
            
        //計算當(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ū)塊鏈schema
    class 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().hash
            newBlock.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 False
                if previousBlock.hash != currentBlock.previousHash:
                    return False
            return True


myCoin = 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)一支軍隊共同圍困一座城市。為了簡化問題，將各支軍隊的行動策略限定為進(jìn)攻或撤離兩種。因?yàn)椴糠周婈犨M(jìn)攻部分軍隊撤離可能會造成災(zāi)難性后果，因此各位將軍必須通過投票來達(dá)成一致策略，即所有軍隊一起進(jìn)攻或所有軍隊一起撤離。因?yàn)楦魑粚④姺痔幊鞘胁煌较?，?br> 系統(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人投撤離。這樣各支軍隊的一致協(xié)同就遭到了破壞。

1.4.2 解決辦法

拜占庭將軍問題主要問題是,中間人可以攔截消息,進(jìn)行修改;上述的那些士兵可以理解成比特幣中的一些節(jié)點(diǎn),不是所有節(jié)點(diǎn)拿到消息后都是可以直接處理的,先去解決一個數(shù)學(xué)問題,就是工作量證明,只有擁有特定的計算能力解決了問題之后才能去修改或者校驗(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 sha256
    import time
    class Block:
         
        def __init__(self,index,timestamp,data,previousHash=""):
            
            self.index = index
            self.timestamp = timestamp
            self.data = data
            self.previousHash = previousHash
            self.nonce = 0 //代表當(dāng)前計算了多少次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代表復(fù)雜度 表示前difficulty位都為0才算成功
        def minerBlock(self,difficulty):
            while(self.hash[0:difficulty]!=str(0).zfill(difficulty)):
                self.nonce+=1
                self.hash = self.calculateHash()
        
        def __str__(self):
            return str(self.__dict__)


    class BlockChain:
        
        def __init__(self):
            self.chain = [self.createGenesisBlock()]
            self.difficulty = 5
    
        def createGenesisBlock(self):
            return Block(0,"01/01/2018","genesis block")
        
        def getLatestBlock(self):
            return self.chain[len(self.chain)-1]
        #添加區(qū)塊前需要 做一道計算題??,坐完后才能把區(qū)塊加入到鏈上
        def addBlock(self,newBlock):
            newBlock.previousHash = self.getLatestBlock().hash
            newBlock.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 False
                if previousBlock.hash != currentBlock.previousHash:
                    return False
            return True
           

    myCoin = BlockChain()
    
    # 下面打印了每個區(qū)塊挖掘需要的時間 比特幣通過一定的機(jī)制控制在10分鐘出一個塊 
    # 其實(shí)就是根據(jù)當(dāng)前網(wǎng)絡(luò)算力 調(diào)整我們上面difficulty值的大小,如果你在
    # 本地把上面代碼difficulty的值調(diào)很大你可以看到很久都不會出計算結(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 info
    print("print block info ####:\n")
    for block in myCoin.chain:
        print("\n")
        print(block)
        
    #check blockchain is valid
    print("before tamper block,blockchain is valid ###")
    print(myCoin.chainIsValid())
    
    #tamper the blockinfo
    myCoin.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、計算工作量證明poW
2、通過新增一個交易賦予礦工（自已）一個幣
3、構(gòu)造新區(qū)塊并將其添加到鏈中

2.4 工作量證明算法：

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

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

?

    import hashlib
    import json
    import requests
    from textwrap import dedent
    from time import time
    from uuid import uuid4
    from urllib.parse import urlparse
    from flask import Flask, jsonify, request
    
    class 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 = 1
    
            while 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 False
    
                last_block = block
                current_index += 1
            return True

        def ressolve_conflicts(self):
            """
            共識算法
            :return:
            """
            neighbours = self.nodes
            new_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 = length
                        new_chain = chain
    
            # 如果發(fā)現(xiàn)一個新的有效鏈比當(dāng)前的長，就替換當(dāng)前的鏈
            if new_chain:
                self.chain = new_chain
                return True
            return False
    
        def 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 block
    
        def 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

        @staticmethod
        def 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()
    
        @property
        def last_block(self):
            # 返回鏈中的最后一個塊
            return self.chain[-1]

        def proof_of_work(self,last_proof):
            # 工作算法的簡單證明
            proof = 0
            while self.valid_proof(last_proof,proof)is False:
                proof +=1
            return proof
    
        @staticmethod
        def 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_block
        last_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", 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
    
    if __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

到了這里，關(guān)于挑戰(zhàn)杯 python區(qū)塊鏈實(shí)現(xiàn) - proof of work工作量證明共識算法的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！