Python实现一个简单的区块链

依赖：

• datetime
• flask
• hashlib
• json
• typing

启动方法：Run起来，浏览器访问对应URL即可。

• http://127.0.0.1:5000/mine_block → 挖掘一个新的区块
• http://127.0.0.1:5000/get_blockchain → 获取JSON格式的BlockChain信息
• http://127.0.0.1:5000/valid_blockchain → 检查区块链的有效性

参考链接：https://www.geeksforgeeks.org/create-simple-blockchain-using-python

在原文章基础上做了一定的修改和汉化，供大家学习参考。

import datetime
import hashlib
import json
from typing import Dict, Union

from flask import Flask, jsonify


# 区块链类
class MyFistBlockchain:

    # 构造方法
    def __init__(self):
        # 实际上存的是JSON
        self.chain = []
        # 初始化的时候先创建第一个区块并将其哈希设置为“0”
        self.create_new_block(proof=1, previous_hash='0')

    # 将新的区块添加到区块链中，每个块都将附加到具有其散列的前一个块，并通过提供其散列附加到下一个块
    def create_new_block(self, proof: int, previous_hash: str) -> Dict[str, Union[int, str]]:
        block = {'index': len(self.chain) + 1,
                 'timestamp': str(datetime.datetime.now()),
                 'proof': proof,
                 'previous_hash': previous_hash}
        self.chain.append(block)
        return block

    # 返回当前最后一个区块
    def get_previous_block(self) -> Dict[str, Union[int, str]]:
        return self.chain[-1]

    # 证明工作量以成功挖掘区块，工作量不能太水也不用太肝(毕竟这是demo)
    def do_proof_of_work(self, previous_proof: int) -> int:
        new_proof = 1
        check_proof = False
        while check_proof is False:
            hash_operation = hashlib.sha256(str(new_proof ** 2 - previous_proof ** 2).encode()).hexdigest()
            if hash_operation[:5] == '00000':
                check_proof = True
            else:
                new_proof += 1
        return new_proof

    # hash函数，基于SHA-256函数，每个块都将包含自己的哈希以及前一个函数的哈希，因此它不会被篡改
    def hash(self, block: Dict[str, Union[int, str]]) -> str:
        encoded_block = json.dumps(block, sort_keys=True).encode()
        return hashlib.sha256(encoded_block).hexdigest()

    # 验证区块链的有效性，防止对区块链进行任何形式的篡改
    def valid_blockchain(self, chain: list) -> bool:
        previous_block = chain[0]
        block_index = 1
        while block_index < len(chain):
            block = chain[block_index]
            # 每个区块的previous_hash必须与上一个区块的hash对得上
            if block['previous_hash'] != self.hash(previous_block):
                return False
            previous_proof = previous_block['proof']
            proof = block['proof']
            hash_operation = hashlib.sha256(str(proof ** 2 - previous_proof ** 2).encode()).hexdigest()
            if hash_operation[:5] != '00000':
                return False
            previous_block = block
            block_index += 1
        return True


blockchain = MyFistBlockchain()

app = Flask(__name__)


# GET方法挖掘一个新的区块
@app.route('/mine_block', methods=['GET'])
def mine_block():
    previous_block = blockchain.get_previous_block()
    previous_proof = previous_block['proof']
    proof = blockchain.do_proof_of_work(previous_proof)
    previous_hash = blockchain.hash(previous_block)
    block = blockchain.create_new_block(proof, previous_hash)
    response = {'message': 'A block is MINED',
                'index': block['index'],
                'timestamp': block['timestamp'],
                'proof': block['proof'],
                'previous_hash': block['previous_hash']}
    return jsonify(response), 200


# GET方法获取JSON格式的BlockChain信息
@app.route('/get_blockchain', methods=['GET'])
def get_blockchain():
    response = {'chain': blockchain.chain, 'length': len(blockchain.chain)}
    return jsonify(response), 200


# GET方法检查区块链的有效性
@app.route('/valid_blockchain', methods=['GET'])
def valid_blockchain():
    valid_result = blockchain.valid_blockchain(blockchain.chain)
    if valid_result:
        response = {'message': 'The Blockchain is valid.'}
    else:
        response = {'message': 'The Blockchain is not valid.'}
    return jsonify(response), 200


# 以localhost:5000启动Flask
app.run(host='127.0.0.1', port=5000)