本文章利用 Python 實(shí)現(xiàn)一個(gè)簡(jiǎn)單的功能較為完善的區(qū)塊鏈系統(tǒng)（包括區(qū)塊鏈結(jié)構(gòu)、賬戶、錢包、轉(zhuǎn)賬），采用的共識(shí)機(jī)制是 POW。

一、區(qū)塊與區(qū)塊鏈結(jié)構(gòu)

Block.py

import hashlib
from datetime import datetime


class Block:
    """
        區(qū)塊鏈結(jié)構(gòu)：
            prev_hash：      父區(qū)塊哈希值
            data：           區(qū)塊內(nèi)容
            timestamp：      區(qū)塊創(chuàng)建時(shí)間
            hash：           區(qū)塊哈希值
    """
    def __init__(self, data, prev_hash):
        # 將傳入的父區(qū)塊哈希值和數(shù)據(jù)保存到變量中
        self.prev_hash = prev_hash
        self.data = data

        # 獲得當(dāng)前的時(shí)間
        self.timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

        # 計(jì)算區(qū)塊哈希值
        # 獲取哈希對(duì)象
        message = hashlib.sha256()
        # 先將數(shù)據(jù)內(nèi)容轉(zhuǎn)為字符串并進(jìn)行編碼，再將它們哈希
        # 注意：update() 方法現(xiàn)在只接受 bytes 類型的數(shù)據(jù)，不接收 str 類型
        message.update(str(self.prev_hash).encode('utf-8'))
        message.update(str(self.prev_hash).encode('utf-8'))
        message.update(str(self.prev_hash).encode('utf-8'))
        # update() 更新 hash 對(duì)象，連續(xù)的調(diào)用該方法相當(dāng)于連續(xù)的追加更新
        # 返回字符串類型的消息摘要
        self.hash = message.hexdigest()

?BlockChain.py

from Block import Block


class BlockChain:
    """
        區(qū)塊鏈結(jié)構(gòu)體
            blocks：         包含區(qū)塊的列表
    """

    def __init__(self):
        self.blocks = []

    def add_block(self, block):
        """
        添加區(qū)塊
        :param block:
        :return:
        """
        self.blocks.append(block)


# 新建區(qū)塊
genesis_block = Block(data="創(chuàng)世區(qū)塊", prev_hash="")
new_block1 = Block(data="張三轉(zhuǎn)給李四一個(gè)比特幣", prev_hash=genesis_block.hash)
new_block2 = Block(data="張三轉(zhuǎn)給王五三個(gè)比特幣", prev_hash=genesis_block.hash)

# 新建一個(gè)區(qū)塊鏈對(duì)象
blockChain = BlockChain()
# 將剛才新建的區(qū)塊加入?yún)^(qū)塊鏈
blockChain.add_block(genesis_block)
blockChain.add_block(new_block1)
blockChain.add_block(new_block2)

# 打印區(qū)塊鏈信息
print("區(qū)塊鏈包含區(qū)塊個(gè)數(shù)為：%d\n" % len(blockChain.blocks))
blockHeight = 0
for block in blockChain.blocks:
    print(f"本區(qū)塊高度為：{blockHeight}")
    print(f"父區(qū)塊哈希：{block.prev_hash}")
    print(f"區(qū)塊內(nèi)容：{block.data}")
    print(f"區(qū)塊哈希：{block.hash}")
    print()
    blockHeight += 1

?測(cè)試結(jié)果?

二、加入工作量證明（POW）

將工作量證明加入到 Block.py 中

import hashlib
from datetime import datetime
from time import time


class Block:
    """
        區(qū)塊鏈結(jié)構(gòu)：
            prev_hash：      父區(qū)塊哈希值
            data：           區(qū)塊內(nèi)容
            timestamp：      區(qū)塊創(chuàng)建時(shí)間
            hash：           區(qū)塊哈希值
            nonce：          隨機(jī)數(shù)
    """
    def __init__(self, data, prev_hash):
        # 將傳入的父區(qū)塊哈希值和數(shù)據(jù)保存到變量中
        self.prev_hash = prev_hash
        self.data = data

        # 獲得當(dāng)前的時(shí)間
        self.timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

        # 設(shè)置隨機(jī)數(shù)、哈希初始值為 None
        self.nonce = None
        self.hash = None

    # 類的 __repr__() 方法定義了實(shí)例化對(duì)象的輸出信息
    def __repr__(self):
        return f"區(qū)塊內(nèi)容：{self.data}\n區(qū)塊哈希值：{self.hash}"


class ProofOfWork:
    """
        工作量證明：
            block：          區(qū)塊
            difficulty：     難度值
    """

    def __init__(self, block, difficult=5):
        self.block = block

        # 定義出塊難度，默認(rèn)為 5，表示有效哈希值以 5 個(gè)零開頭
        self.difficulty = difficult

    def mine(self):
        """
        挖礦函數(shù)
        :return:
        """
        i = 0
        prefix = '0' * self.difficulty

        while True:
            message = hashlib.sha256()
            message.update(str(self.block.prev_hash).encode('utf-8'))
            message.update(str(self.block.data).encode('utf-8'))
            message.update(str(self.block.timestamp).encode('utf-8'))
            message.update(str(i).encode('utf-8'))
            # digest() 返回摘要，作為二進(jìn)制數(shù)據(jù)字符串值
            # hexdigest() 返回摘要，作為十六進(jìn)制數(shù)據(jù)字符串值
            digest = message.hexdigest()
            # str.startswith(prefix) 檢測(cè)字符串是否是以 prefix（字符串）開頭，返回布爾值
            if digest.startswith(prefix):
                # 幸運(yùn)數(shù)字
                self.block.nonce = i
                # 區(qū)塊哈希值為十六進(jìn)制數(shù)據(jù)字符串摘要
                self.block.hash = digest
                return self.block
            i += 1

    def validate(self):
        """
        驗(yàn)證有效性
        :return:
        """
        message = hashlib.sha256()
        message.update(str(self.block.prev_hash).encode('utf-8'))
        message.update(str(self.block.data).encode('utf-8'))
        message.update(str(self.block.timestamp).encode('utf-8'))
        message.update(str(self.block.nonce).encode('utf-8'))
        digest = message.hexdigest()

        prefix = '0' * self.difficulty
        return digest.startswith(prefix)


# ++++++++測(cè)試++++++++
# 定義一個(gè)區(qū)塊
b = Block(data="測(cè)試", prev_hash="")

# 定義一個(gè)工作量證明
w = ProofOfWork(b)

# 開始時(shí)間
start_time = time()
# 挖礦，并統(tǒng)計(jì)函數(shù)執(zhí)行時(shí)間
print("+++開始挖礦+++")
valid_block = w.mine()
# 結(jié)束時(shí)間
end_time = time()
print(f"挖礦花費(fèi)時(shí)間：{end_time - start_time}秒")

# 驗(yàn)證區(qū)塊
print(f"區(qū)塊哈希值是否符合規(guī)則：{w.validate()}")
print(f"區(qū)塊哈希值為：{b.hash}")

測(cè)試結(jié)果

更新 BlockChain.py

from Block import Block, ProofOfWork


class BlockChain:
    """
        區(qū)塊鏈結(jié)構(gòu)體
            blocks：         包含區(qū)塊的列表
    """

    def __init__(self):
        self.blocks = []

    def add_block(self, block):
        """
        添加區(qū)塊
        :param block:
        :return:
        """
        self.blocks.append(block)


# 新建一個(gè)區(qū)塊鏈對(duì)象
blockChain = BlockChain()

# 新建區(qū)塊
block1 = Block(data="創(chuàng)世區(qū)塊", prev_hash="")
w1 = ProofOfWork(block1)
genesis_block = w1.mine()
blockChain.add_block(genesis_block)

block2 = Block(data="張三轉(zhuǎn)給李四一個(gè)比特幣", prev_hash=genesis_block.hash)
w2 = ProofOfWork(block2)
block = w2.mine()
blockChain.add_block(block)

block3 = Block(data="張三轉(zhuǎn)給王五三個(gè)比特幣", prev_hash=block.hash)
w3 = ProofOfWork(block3)
block = w3.mine()
blockChain.add_block(block)

# 打印區(qū)塊鏈信息
print("區(qū)塊鏈包含區(qū)塊個(gè)數(shù)為：%d\n" % len(blockChain.blocks))
blockHeight = 0
for block in blockChain.blocks:
    print(f"本區(qū)塊高度為：{blockHeight}")
    print(f"父區(qū)塊哈希：{block.prev_hash}")
    print(f"區(qū)塊內(nèi)容：{block.data}")
    print(f"區(qū)塊哈希：{block.hash}")
    print()
    blockHeight += 1
    

?測(cè)試結(jié)果

三、實(shí)現(xiàn)錢包、賬戶、交易功能

?實(shí)現(xiàn)錢包、賬戶、交易功能要先安裝非對(duì)稱加密算法庫(kù) ecdsa。如果網(wǎng)速慢，引用下面這個(gè)網(wǎng)站

-i https://pypi.tuna.tsinghua.edu.cn/simple

添加錢包、賬戶功能 Wallet.py

import base64
import binascii
from hashlib import sha256
# 導(dǎo)入橢圓曲線算法
from ecdsa import SigningKey, SECP256k1, VerifyingKey


class Wallet:
    """
        錢包
    """

    def __init__(self):
        """
            錢包初始化時(shí)基于橢圓曲線生成一個(gè)唯一的秘鑰對(duì)，代表區(qū)塊鏈上一個(gè)唯一的賬戶
        """
        # 生成私鑰
        self._private_key = SigningKey.generate(curve=SECP256k1)
        # 基于私鑰生成公鑰
        self._public_key = self._private_key.get_verifying_key()

    @property
    def address(self):
        """
            這里通過(guò)公鑰生成地址
        """
        h = sha256(self._public_key.to_pem())
        # 地址先由公鑰進(jìn)行哈希算法，再進(jìn)行 Base64 計(jì)算而成
        return base64.b64encode(h.digest())

    @property
    def pubkey(self):
        """
            返回公鑰字符串
        """
        return self._public_key.to_pem()

    def sign(self, message):
        """
            生成數(shù)字簽名
        """
        h = sha256(message.encode('utf8'))
        # 利用私鑰生成簽名
        # 簽名生成的是一串二進(jìn)制字符串，為了便于查看，這里轉(zhuǎn)換為 ASCII 字符串進(jìn)行輸出
        return binascii.hexlify(self._private_key.sign(h.digest()))


def verify_sign(pubkey, message, signature):
    """
        驗(yàn)證簽名
    """
    verifier = VerifyingKey.from_pem(pubkey)
    h = sha256(message.encode('utf8'))
    return verifier.verify(binascii.unhexlify(signature), h.digest())

實(shí)現(xiàn)轉(zhuǎn)賬功能 Transaction.py

import json


class Transaction:
    """
        交易的結(jié)構(gòu)
    """

    def __init__(self, sender, recipient, amount):
        """
            初始化交易，設(shè)置交易的發(fā)送方、接收方和交易數(shù)量
        """
        # 交易發(fā)送者的公鑰
        self.pubkey = None
        # 交易的數(shù)字簽名
        self.signature = None

        if isinstance(sender, bytes):
            sender = sender.decode('utf-8')
        self.sender = sender        # 發(fā)送方
        if isinstance(recipient, bytes):
            recipient = recipient.decode('utf-8')
        self.recipient = recipient  # 接收方
        self.amount = amount        # 交易數(shù)量

    def set_sign(self, signature, pubkey):
        """
            為了便于驗(yàn)證這個(gè)交易的可靠性，需要發(fā)送方輸入他的公鑰和簽名
        """
        self.signature = signature  # 簽名
        self.pubkey = pubkey  # 發(fā)送方公鑰

    def __repr__(self):
        """
            交易大致可分為兩種，一是挖礦所得，而是轉(zhuǎn)賬交易
            挖礦所得無(wú)發(fā)送方，以此進(jìn)行區(qū)分顯示不同內(nèi)容
        """
        if self.sender:
            s = f"從{self.sender}轉(zhuǎn)自{self.recipient}{self.amount}個(gè)加密貨幣"
        elif self.recipient:
            s = f"{self.recipient}挖礦所得{self.amount}個(gè)加密貨幣"
        else:
            s = "error"
        return s


class TransactionEncoder(json.JSONEncoder):
    """
        定義Json的編碼類，用來(lái)序列化Transaction
    """
    def default(self, obj):
        if isinstance(obj, Transaction):
            return obj.__dict__
        else:
            return json.JSONEncoder.default(self, obj)
            # return super(TransactionEncoder, self).default(obj)

更新 Block.py

import hashlib
import json
from datetime import datetime
from Transaction import Transaction, TransactionEncoder


class Block:
    """
        區(qū)塊結(jié)構(gòu)
            prev_hash:      父區(qū)塊哈希值
            transactions:   交易對(duì)
            timestamp:      區(qū)塊創(chuàng)建時(shí)間
            hash:           區(qū)塊哈希值
            Nonce:          隨機(jī)數(shù)
    """

    def __init__(self, transactions, prev_hash):
        # 將傳入的父哈希值和數(shù)據(jù)保存到類變量中
        self.prev_hash = prev_hash
        # 交易列表
        self.transactions = transactions
        # 獲取當(dāng)前時(shí)間
        self.timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

        # 設(shè)置Nonce和哈希的初始值為None
        self.nonce = None
        self.hash = None

    # 類的 __repr__() 方法定義了實(shí)例化對(duì)象的輸出信息
    def __repr__(self):
        return f"區(qū)塊內(nèi)容：{self.transactions}\n區(qū)塊哈希值：{self.hash}"


class ProofOfWork:
    """
        工作量證明
            block：          區(qū)塊
            difficulty：     難度值
    """

    def __init__(self, block, miner, difficult=5):
        self.block = block
        self.miner = miner

        # 定義工作量難度，默認(rèn)為5，表示有效的哈希值以5個(gè)“0”開頭
        self.difficulty = difficult

        # 添加挖礦獎(jiǎng)勵(lì)
        self.reward_amount = 1

    def mine(self):
        """
            挖礦函數(shù)
        """
        i = 0
        prefix = '0' * self.difficulty

        # 設(shè)置挖礦自動(dòng)生成交易信息，添加挖礦獎(jiǎng)勵(lì)
        t = Transaction(
            sender="",
            recipient=self.miner.address,
            amount=self.reward_amount,
        )
        sig = self.miner.sign(json.dumps(t, cls=TransactionEncoder))
        t.set_sign(sig, self.miner.pubkey)
        self.block.transactions.append(t)

        while True:
            message = hashlib.sha256()
            message.update(str(self.block.prev_hash).encode('utf-8'))
            # 更新區(qū)塊中的交易數(shù)據(jù)
            # message.update(str(self.block.data).encode('utf-8'))
            message.update(str(self.block.transactions).encode('utf-8'))
            message.update(str(self.block.timestamp).encode('utf-8'))
            message.update(str(i).encode("utf-8"))
            digest = message.hexdigest()
            if digest.startswith(prefix):
                self.block.nonce = i
                self.block.hash = digest
                return self.block
            i += 1

    def validate(self):
        """
            驗(yàn)證有效性
        """
        message = hashlib.sha256()
        message.update(str(self.block.prev_hash).encode('utf-8'))
        # 更新區(qū)塊中的交易數(shù)據(jù)
        # message.update(str(self.block.data).encode('utf-8'))
        message.update(json.dumps(self.block.transactions).encode('utf-8'))
        message.update(str(self.block.timestamp).encode('utf-8'))
        message.update(str(self.block.nonce).encode('utf-8'))
        digest = message.hexdigest()

        prefix = '0' * self.difficulty
        return digest.startswith(prefix)

更新 BlockChain.py

from Block import Block, ProofOfWork
from Transaction import Transaction
from Wallet import Wallet, verify_sign


class BlockChain:
    """
        區(qū)塊鏈結(jié)構(gòu)體
            blocks:        包含的區(qū)塊列表
    """

    def __init__(self):
        self.blocks = []

    def add_block(self, block):
        """
            添加區(qū)塊
        """
        self.blocks.append(block)

    def print_list(self):
        print(f"區(qū)塊鏈包含個(gè)數(shù)為：{len(self.blocks)}")
        for block in self.blocks:
            height = 0
            print(f"區(qū)塊鏈高度為：{height}")
            print(f"父區(qū)塊為：{block.prev_hash}")
            print(f"區(qū)塊內(nèi)容為：{block.transactions}")
            print(f"區(qū)塊哈希值為：{block.hash}")
            height += 1
            print()

為了方便我們對(duì)區(qū)塊鏈進(jìn)行操作，我們可以在 BlockChain.py 中補(bǔ)充一些方法

# 傳入用戶和區(qū)塊鏈，返回用戶的“余額”
def get_balance(user, blockchain):
    balance = 0
    for block in blockchain.blocks:
        for t in block.transactions:
            if t.sender == user.address.decode():
                balance -= t.amount
            elif t.recipient == user.address.decode():
                balance += t.amount
    return balance


# user生成創(chuàng)世區(qū)塊（新建區(qū)塊鏈），并添加到區(qū)塊鏈中
def generate_genesis_block(user):
    blockchain = BlockChain()
    new_block = Block(transactions=[], prev_hash="")
    w = ProofOfWork(new_block, user)
    genesis_block = w.mine()
    blockchain.add_block(genesis_block)
    # 返回創(chuàng)世區(qū)塊
    return blockchain


# 用戶之間進(jìn)行交易并記入交易列表
def add_transaction(sender, recipient, amount):
    # 新建交易
    new_transaction = Transaction(
        sender=sender.address,
        recipient=recipient.address,
        amount=amount
    )
    # 生成數(shù)字簽名
    sig = sender.sign(str(new_transaction))
    # 傳入付款方的公鑰和簽名
    new_transaction.set_sign(sig, sender.pubkey)
    return new_transaction


# 驗(yàn)證交易，若驗(yàn)證成功則加入交易列表
def verify_new_transaction(new_transaction, transactions):
    if verify_sign(new_transaction.pubkey,
                   str(new_transaction),
                   new_transaction.signature
                   ):
        # 驗(yàn)證交易簽名沒(méi)問(wèn)題，加入交易列表
        print("交易驗(yàn)證成功")
        transactions.append(new_transaction)
    else:
        print("交易驗(yàn)證失敗")


# 礦工將全部驗(yàn)證成功的交易列表打包出塊
def generate_block(miner, transactions, blockchain):
    new_block = Block(transactions=transactions,
                      prev_hash=blockchain.blocks[len(blockchain.blocks) - 1].hash)
    print("生成新的區(qū)塊...")
    # 挖礦
    w = ProofOfWork(new_block, miner)
    block = w.mine()
    print("將新區(qū)塊添加到區(qū)塊鏈中")
    blockchain.add_block(block)

進(jìn)行測(cè)試

# 新建交易列表
transactions = []

# 創(chuàng)建 3 個(gè)用戶
alice = Wallet()
tom = Wallet()
bob = Wallet()

print("alice創(chuàng)建創(chuàng)世區(qū)塊...")
blockchain = generate_genesis_block(alice)
print()

print(f"alice 的余額為{get_balance(alice, blockchain)}個(gè)比特幣")
print(f"tom 的余額為{get_balance(tom, blockchain)}個(gè)比特幣")
print(f"bob 的余額為{get_balance(bob, blockchain)}個(gè)比特幣")
print()

# 打印區(qū)塊鏈信息
blockchain.print_list()

print("新增交易：alice 轉(zhuǎn)賬 0.5 比特幣給 tom")
nt = add_transaction(alice, tom, 0.5)
print()
verify_new_transaction(nt, transactions)
print(f"礦工 bob 將全部驗(yàn)證成功的交易列表打包出塊...")
generate_block(bob, transactions, blockchain)
print("添加完成\n")

# 打印區(qū)塊鏈信息
blockchain.print_list()

測(cè)試結(jié)果

文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-503018.html

到了這里，關(guān)于【Python】實(shí)現(xiàn)一個(gè)簡(jiǎn)單的區(qū)塊鏈系統(tǒng)的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！