Zenn
🌏

Python、Streamlit と FastAPI で、ブロックチェーンを実装してみた♪④マイニングを実装

に公開
Python
ブロックチェーン
FastAPI
Streamlit
マイニング
tech

今回は、いよいよ仕上げとなるマイニングを実装して、ブロックチェーンを完成させます！

前回までの記事

https://zenn.dev/animalz/articles/896225dc4b6bef
https://zenn.dev/animalz/articles/2946d829db92c9
https://zenn.dev/animalz/articles/758276666c11f3

環境

Windows10、PyCharam、Python v3.13.2、Streamlit v1.45.1、FastAPI v0.115.12

ディレクトリ構成

（node2 と　node3 は、テスト用ファイルとフォルダ以外同じなので省略してます）

ディレクトリ構成
streamlit_fastapi
├ node1
| ├ json（JSON データ保存用）
| ├ test_data（テスト･データ保存用）
| ├ app.py
| ├ block_chain.log（ログ保存用）
| ├ log.py
| ├ main.py
| ├ mikoto.py
| ├ test_app_main.py
| └ test_mikoto.py
...

今回やること

マイニングを実装し、ブロックチェーンを完成させます！

マイニングを実装する

まず、マイニング用の関数を実装し、そのテストを行い、フロントエンド、バックエンドと実装していきます。
mikokto.py、test_mikoto.py、app.py、main.py の順に実装します。
ブロックチェーンの検証は少しややこしいので、説明しておきます。

  1. 検証中のブロックの前のブロックのハッシュ値を確認します
  2. 検証中のブロックのトランザクションの検証を行います
  3. すでにブロックチェーンに組み込まれているトランザクションと、検証中のトランザクションの重複がないかを確認します
  4. 検証中のトランザクション同士の重複がないか確認します
  5. Proof of Work の nonce を検証します（ハッシュ値の先頭にゼロが POW＿ZWROS 個並んでいるかどうか）
mikoto.py
...
POW_ZEROS = 3

def verify_block_chain(block_chain: list) -> bool:
    # 1. 前ブロックのハッシュ値の確認
    if make_hash_str(block_chain[-2]) != block_chain[-1]['hash']:
        return False
    # 2. トランザクション検証
    new_block_transactions = block_chain[-1]['transactions']
    for transaction in new_block_transactions:
        if transaction['sender'] != 'mikoto_project':
            if not verify_data(transaction, transaction['sender']):
                return False
    # 3. 検証済みとの重複確認(今回は空)
    verified_transactions = []
    for block in block_chain[:-1]:
        verified_transactions += block['transactions']
    for transaction in new_block_transactions:
        if transaction in verified_transactions:
            return False
    # 4. 新ブロック内での重複確認（signature で確認）
    signature_list = []
    for transaction in new_block_transactions:
        if transaction['sender'] != 'mikoto_project':
            signature_list.append(transaction['signature'])
    if len(signature_list) != len(set(signature_list)):
        return False
    # 5. proof_of_work 確認
    copy_new_block = block_chain[-1].copy()
    copy_new_block.pop('time')
    if not make_hash_str(copy_new_block).startswith('0'*POW_ZEROS):
        return False
    return True

def mining(transaction_pool: list, block_chain: list, miner_public_key_str: str, mik: int) -> dict:
    verified_transactions = []
    for block in block_chain:
        verified_transactions += block['transactions']  # 検証済
    for transaction in transaction_pool:
        if transaction in verified_transactions:  # 重複チェック
            transaction_pool.remove(transaction)
            continue
        if transaction['sender'] != 'mikoto_project':
            if not verify_data(transaction, transaction['sender']):
                transaction_pool.remove(transaction)
    mikoto_transaction = make_mikoto_transaction(miner_public_key_str, mik)
    transaction_pool.append(mikoto_transaction)
    previous_block_hash_str = make_hash_str(block_chain[-1])
    new_block = {
        'transactions': transaction_pool,
        'hash': previous_block_hash_str,
        'nonce': 0
    }
    new_block = proof_of_work(new_block)  # nonce, PoW
    new_block['time'] = dt.datetime.now().isoformat()
    keys = ['time', 'transactions', 'hash', 'nonce']
    sorted_new_block = {key: new_block[key] for key in keys}
    return sorted_new_block

def proof_of_work(block: dict, zeros=POW_ZEROS) -> dict:
    while not make_hash_str(block).startswith('0'*zeros):
        block['nonce'] += 1
    return block

def make_hash_str(data: dict) -> str:
    return hashlib.sha256(json.dumps(data).encode('utf-8')).hexdigest()

def get_data(url: str) -> requests.Response:
    return requests.get(url)
...

テストを実装します。

test_mikoto.py
...
co_transaction = mik.make_thanks_transaction(
    co_secret_key_str, co_public_key_str,
    co2_public_key_str, 50
)
co_block_chain = [{
    'time': dt.datetime.now().isoformat(),
    'transactions': [],
    'hash': 'mikoto_block_chain',
    'nonce': 0
}]

def test_verify_block_chain():
    """ test: True case, False case """
    new_block = mik.mining(
        [co_transaction], co_block_chain, co_public_key_str, 100)
    co_block_chain.append(new_block)
    assert mik.verify_block_chain(co_block_chain)
    co_block_chain[-1]['nonce'] -= 1
    assert not mik.verify_block_chain(co_block_chain)

def test_mining():
    transaction1 = mik.make_thanks_transaction(
        co_secret_key_str, co_public_key_str,
        co2_public_key_str, 50
    )
    transaction_pool = [transaction1]
    block_chain = [{
        'time': dt.datetime.now().isoformat(),
        'transactions': [],
        'hash': 'mikoto_block_chain',
        'nonce': 0
    }]
    new_block = mik.mining(transaction_pool, block_chain, co_public_key_str, 100)
    assert isinstance(new_block, dict)
    key_list = ['time', 'transactions', 'hash', 'nonce']
    assert list(new_block.keys()) == key_list

def test_proof_of_work():
    block = {'transactions': [], 'nonce': 0}
    block = mik.proof_of_work(block, 3)
    assert hashlib.sha256(json.dumps(block).encode('utf-8')).hexdigest()[:3] == '0'*3

def test_make_hash_str():
    dict_data1 = {'a': 1, 'b': 2}
    dict_data2 = {'a': 1, 'b': 3}
    dict_hash_str1 = mik.make_hash_str(dict_data1)
    dict_hash_str2 = mik.make_hash_str(dict_data2)
    assert isinstance(dict_hash_str1, str)
    assert len(dict_hash_str1) == 64
    assert dict_hash_str1 != dict_hash_str2

@pytest.mark.skipif(online==False, reason="not online")
def test_get_data():
    response = mik.get_data(co_base_url + co_port_list[0])
    assert response.status_code == 200
    assert response.json() == "Welcome to Mikoto Project!"
...

テストを実行、すべてパスできました。

つづいて、フロントエンドを実装します。
セレクト･ボックスで、マイニングを選択できるようにします。

app.py
...
    if ss.state is True:
        st.markdown('ログイン中')
        menu = ['発行または送信', 'マイニング']
...
        if choice == 'マイニング':
            st.button('マイニングを実行')
            block_chain = mik.load_json('json/block_chain.json')
            transaction_pool = mik.load_json('json/transaction_pool.json')
            new_block = mik.mining(
                transaction_pool,
                block_chain,
                my_data['public_key_str'],
                100)
            block_chain.append(new_block)
            
            for url in url_list:
                try:
                    res = mik.post_data(url + '/block_chain', block_chain)
                    st.json(res.json())
                    log.log_debug(logger, f'{res}: {res.json()}')
                except:
                    st.json({"message": f"{url}: error"})
                    log.log_error(logger, f"{url}: error")
...

最後に、バックエンドを実装します。

main.py
...
from typing import List
...
class Block(BaseModel):
    time: str
    transactions: list = None
    hash: str
    nonce: int
...
@app.post('/block_chain')
async def post_block_chain(block_chain: List[Block]):
    block_chain = [dict(block) for block in block_chain]
    if not mik.verify_block_chain(block_chain):
        return {"message": "block_chain invalid"}
    mik.save_json([], 'json/transaction_pool.json')
    block_chain = mik.load_json('json/block_chain.json')
    block_chain.append(new_block_chain)
    mik.save_json(block_chain, 'json/block_chain.json')
    return {"message": "block_chain received"}
...

動作確認します。
今回のファイルの更新を、node2、node3 に反映させ、すべてのノードの transaction_pool を空にし、block_chain をファースト・ブロックに戻し、ログ・ファイルも削除して、動作確認を行います。
各自、100 MIK を発行、それぞれのノードが送信し（内容省略します）、animal がマイニング実行します。
前回と同じになるので、画像はマイニング以降にします。
すべてのノードのサーバーを立ち上げます。

コマンド・プロンプト
# node1
streamlit run app.py --server.port 8501
uvicorn main:app --reload --port 8010
# node2
streamlit run app.py --server.port 8502
uvicorn main:app --reload --port 8011
# node3
streamlit run app.py --server.port 8503
uvicorn main:app --reload --port 8012

マイニングまでの操作を実行後、http://127.0.0.1:8501にアクセスします。

マイニングを実行します。
無事に実行できたようです。

transaction_pool は、すべて空になっています。
新しいブロックが、無事にブロックチェーンに追加されました。
block_chain.json
（ファースト・ブロック～発行まで）

（それ以降）

block_chain.log（ node1 ）

参考文献など

Satoshi Nakamoto (2008), ビットコイン：P2P 電子通貨システム,
https://bitcoin.org/files/bitcoin-paper/bitcoin_jp.pdf, 論文
モヤっとデータサイエンティスト (2022), 『Python によるブロックチェーン開発教本』, 電子書籍
安田恒 (2023), 『ブロックチェーンを作る！』, 秀和システム, 書籍

今回のコード

GitHub: https://github.com/Animalyzm/mikoto_project
今回のコードは、block_chain/streamlit_fastapi です。
Git のコミット･メッセージは、block_chain_streamlit_fastapi_4_mining です。
新しく使う場合は、json フォルダの中身すべてと、block_chain.log を削除してください。

4回にわたって、ブロックチェーンを実装してきました。
これで、とりあえず完成にしよーと思います。
まだ、実用するには、保有残高の計算や、非承認時の処理など、いろいろあるとは思いますが、前回よりは、本格的なブロックチェーンを実装できたのではないかと思います。

前回の実装

https://zenn.dev/animalz/articles/c82f20acccd30f
https://zenn.dev/animalz/articles/928dd981c2d1a5
https://zenn.dev/animalz/articles/f2e56b071909b6

今回は、以上になります！
どうもありがとうございました―♪

Discussion