用python实现区块链简单原理

本是古典何须时尚 2023-07-07 12:55 15阅读 0赞

# 1.设置好区块链的对应方法 #

#一个区块应该包括的内容
    # { 
    # 'index':0, #索引
    # 'timestamp':"", #时间戳
    # 'trascations':[ #交易信息
    # {"sender":"", #交易发送者
    # recipient":"",#交易接收者
    # "amout":""} #转账金额
    # ],
    # 'proof':"",#工作量证明
    # 'previous_hash':"" #上一个交易的hash值
    # }
    import hashlib
    import json
    from time import time
    from urllib.parse import urlparse
    
    import requests
    
    
    class Blokchain:
        def __init__(self):
            self.chain=[]
            #准备好一个存储当前交易记录的列表
            self.current_trascations=[]
            #准备实现一个创世纪的区块，其参数值可以是任意的
            self.new_block(proof=1,previous_hash=1)
            #准备好变量保存节点信息
            self.nodes=set()
        #注册节点,保存好节点的网络信息
        def register_node(self,address:str):
            parsed_url=urlparse(address)
            self.nodes.add(parsed_url.netloc)
        #解决冲突，实现共识机制
        def resolve_conflics(self):
            neighbours=self.nodes
            max_length=len(self.chain)
            new_chain=None
            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=length
                        new_chain=chain
            if new_chain:
                self.chain=new_chain
                return True
            return False
    
        #添加一个新的区块
        def new_block(self,proof,previous_hash=None):
            #准备一个新的区块
            block={ 
            'index':len(self.chain)+1,
            'timestamp':time(),
            'trascations':self.current_trascations,
            'proof':proof,
            'previous_hash':previous_hash or self.hash(self.last_block)
            }
            #清空新的交易
            self.current_trascations=[]
            #把区块加入链条中
            self.chain.append(block)
            return block
        #创建一个新的交易
        def new_trascations(self,sender,recipient,amout):
            ''' :param sender: 交易的发送者 :param recipient: 交易的接收者 :param amout: 交易的金额 :return: 整型，返回创建新区块链的索引。 '''
            self.current_trascations.append({ 
                'sender':sender,
                'recipient':recipient,
                'amout':amout
            })
            return self.last_block['index']+1
    
        #计算一个区块的hash值，使用静态方法
        @staticmethod
        def hash(block):
            #把节点传入的信息进行转化为字节串，并编码,这样hash函数才能够认识
            block_string=json.dumps(block,sort_keys=True).encode()
            #进行hash计算得到的是一个对象，进行hesdidest()方法得到hash摘要
            return  hashlib.sha256(block_string).hexdigest()
        #获取一个最后一个区块。当做属性来处理
        @property
        def last_block(self):
            return self.chain[-1]
        #实现工作量证明，为了简化操作，我们就用上一个区块链的（hash值）工作量证明来验证
        def proof_of_work(self,last_proof):
            proof=0
            while self.valid_proof(last_proof,proof) is False:
                proof+=1
            print(proof)
            return proof   #返回工作量证明
    
        #验证工作量的合法性
        def valid_proof(self,last_proof,proof):
            #把last_proof和proof拼接在一起，进行编码
            guess=f'{last_proof}{proof}'.encode()
            guess_hash=hashlib.sha256(guess).hexdigest()
            print(guess_hash)
            #说明工作强度，返回一个bool值
            return guess_hash[0:4]=='0000'
        #判断链条是否为有效链条
        def valid_chain(self, chain):
            last_block=chain[0]
            current_index=1
            while current_index<len(chain):
                block=chain[current_index]
                #如果上一块链的hash值不等于，则说明是假的
                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
    
    
    if __name__=="__main__":
        testPow=Blokchain()
        testPow.proof_of_work(100)

# 2通过flask 实现网络节点的调用 #

''' 设置网络模拟，区块链中各个节点的运行原理 '''
    from uuid import uuid4
    
    from blockchain import Blokchain
    import flask
    import requests
    from flask import Flask, jsonify, request
    
    block_chain=Blokchain()
    #初始化一个Flask
    app=Flask(__name__)
    #实现创建一个新的交易
    @app.route('/transction/new',methods=['POST'])
    def new_transction():
        values=request.get_json()
        #判断获取的信息在不在范围内
        required=['sender','recipient','amout']
        if values==None:
            return 'Miss values !', 400
        #如果请求不在范围require的列表内，则返回错误
        if not all(k in values for k in required):
            return 'Miss values !',400
        index=block_chain.new_trascations(values['sender'],values['recipient'],values['amout'])
        response={ 'message':f'Transction will be added to Block {index}'}
        return jsonify(response),201
    #引入一个计算地址的库，保存好地址
    node_identifier=str(uuid4().replace('-',''))
    
    #实现挖矿（实现交易的打包）
    @app.route('/mine',methods=['GET'])
    def mine():
        last_block=block_chain.last_block
        last_proof=last_block['proof']
    
        proof=block_chain.proof_of_work(last_proof)
        # 完成一笔交易，给自己一点奖励
        block_chain.new_trascations(sender='0',
                                    recipient=node_identifier,
                                    amout=1)
        block=block_chain.new_block(proof,None)#给区块新建一个区块
        #打包好区块，广播给各个节点
        response={ 
            "message":'Now block forgod',
            'index':block['index'],
            'transactions':block['transactions'],
            'proof':block['proof'],
            'previous_hash':block['previous_hash']
        }
        return jsonify(response),200
    
    #实现增加一个节点
    @app.route('/chain',methods=['GET'])
    def chain():
        response={ 
            'chain':block_chain.chain,
            'length':len(block_chain.chain)
        }
        return jsonify(response),200
    #接收用户的节点注册:所接收的json字符串格式为{"nodes",["http:127.XXX]}
    @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
        response={ 
            "message":"New nodes have been added",
            "total_nodes":list(block_chain.nodes)
        }
        return jsonify(response),201
    #设置达成共识的请求，解决冲突
    @app.route('/nodes/resolve',methods=['GET'])
    def consensus():
        replace=block_chain.resolve_conflics()
        if replace:
            response={ 
                'message':"our chain was replaced",
                'chain':block_chain.chain
            }
        else:
            response = { 
                'message': "our chain was authoritative",
                'chain': block_chain.chain
            }
        return jsonify(response),200
    if __name__=='__main__':
        app.run(host='0.0.0.0', port=6000)