🌏

Python、Streamlit と FastAPI で、ブロックチェーンを実装してみた♪①秘密鍵作成～ブロードキャスト

に公開
Python
ブロックチェーン
FastAPI
Streamlit
pytest
tech

今回から、ちょっと本格的なブロックチェーンを実装していこーと思います！
フロントエンドに Streamlit、バックエンドに FastAPI を使って、実際にノードを3つ用意して、通信もできるようします。
前回の簡易版に比べると、本格的になったと言えるようなものに仕上げよーと思います。
基本、Python オンリー、クラスを使わず、データベースも使わず、データは JSON ファイルのみで実装していきます。
ブロックチェーン自体は仮完成しているんですが、けっこーややこしいので、どこまで実装するかは、毎回、判断していこーと思います。
それでは、れっつらごー♪

実装する予定のもの

  • 通信機能（ノード3つ）
  • 秘密鍵（Web3 的）ログイン機能
  • ブロックチェーン（トランザクション、発行、送信）
  • ログ保存
  • テストファイルをできる限り作成

今回やること

ブロックチェーンを開始し、鍵データを作成、ブロードキャストして保存するところまで実装します！

環境

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

ディレクトリ構成

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

必要な関数を実装

秘密鍵作成からブロードキャストまでに、必要な関数を実装していきます。
同じディレクトリに pytest でテスト・ファイルも作成します。
ライブラリをインストールします。

mikoto.py
import binascii
import datetime as dt
import json
from ecdsa import SECP256k1, SigningKey, VerifyingKey
import requests

秘密鍵、公開鍵を作成し、文字列化し、また鍵に戻す関数などを実装します。

mikoto.py
def make_key_data(name: str, url: str) -> dict:
    secret_key = make_secret_key()
    secret_key_str = key_to_str(secret_key)
    public_key = make_public_key(secret_key)
    public_key_str = key_to_str(public_key)
    key_data = {
        'name': name,
        'url': url,
        'secret_key_str': secret_key_str,
        'public_key_str': public_key_str
    }
    return key_data

def str_to_public_key(public_key_str: str) -> VerifyingKey:
    return VerifyingKey.from_string(binascii.unhexlify(public_key_str), curve=SECP256k1)

def str_to_secret_key(secret_key_str: str) -> SigningKey:
    return SigningKey.from_string(binascii.unhexlify(secret_key_str), curve=SECP256k1)

def key_to_str(key: SigningKey | VerifyingKey) -> str:
    return key.to_string().hex()

def make_public_key(secret_key: SigningKey) -> VerifyingKey:
    return secret_key.verifying_key

def make_secret_key() -> SigningKey:
    return SigningKey.generate(curve=SECP256k1)

JSON ファイルの読み書きをする関数を作ります。

mikoto.py
def load_json(path: str) -> list | dict:
    with open(path, mode='r') as f:
        data = json.load(f)
    return data

def save_json(data: list | dict, path: str) -> None:
    with open(path, mode='w') as f:
        json.dump(data, f)
    return None

シグネチャ（電子署名）を作成する関数、データを送信する関数、データを検証する関数、最初のノードの初期化関数、ブロードキャストする際の URL リストを取得する関数を実装します。

mikoto.py
def get_url_list(base_path='json/') -> list:
    key_data_list = load_json(base_path + 'key_data_list.json')
    url_list = [key_data['url'] for key_data in key_data_list]
    return url_list

def make_first_node(save_dir: str = 'json/') -> None:
    """
    実行コマンド
    python -c "import mikoto; mikoto.make_first_node()"
    """
    name = input('name: ')
    url = input('url: ')
    key_data = make_key_data(name, url)
    save_json(key_data, save_dir+'.my_data.json')
    block_chain = [
        {
            'time': dt.datetime.now().isoformat(),
            'transactions': [],
            'hash': 'mikoto_block_chain',
            'nonce': 0
        }
    ]
    save_json(block_chain, save_dir+'block_chain.json')
    key_data_list = []
    key_data.pop('secret_key_str')
    key_data_list.append(key_data)
    save_json(key_data_list, save_dir+'key_data_list.json')
    transaction_pool = []
    save_json(transaction_pool, save_dir+'transaction_pool.json')
    print("made first_node")
    return None

def verify_data(data: dict, public_key_str:str) -> bool:
    """ need key: signature """
    try:
        public_key = str_to_public_key(public_key_str)
        signature = binascii.unhexlify(data['signature'])
        copy_data = data.copy()
        copy_data.pop('signature')
        public_key.verify(signature, json.dumps(copy_data).encode('utf-8'))
        return True
    except:
        return False

def post_data(url: str, data: dict)-> requests.Response:
    return requests.post(url, json.dumps(data))

def make_signature_str(data: dict, secret_key_str: str) -> str:
    secret_key = str_to_secret_key(secret_key_str)
    return secret_key.sign(json.dumps(data).encode('utf-8')).hex()

ここまでのテストを実装します。
通信に関しては、FastAPI を、まだ実装していないため、pytest の @pytest.mark.skipif を使って skip する設定にしてます。
データは使いまわした方が、コードがシンプルになるので、共通利用のデータを作成しています。

test_mikoto.py
import binascii
from ecdsa import SigningKey, SECP256k1, VerifyingKey
import json
import os
import re
import pytest
import mikoto as mik

online = False
""" True: uvicorn main:app --reload --port 8010 """

""" 共通利用のデータなど """
co_secret_key = SigningKey.generate(curve=SECP256k1)
co_public_key = co_secret_key.verifying_key
co_secret_key_str = co_secret_key.to_string().hex()
co_public_key_str = co_public_key.to_string().hex()
co_name_list = ['Dog', 'Cat', 'Lion']
co_base_url = 'http://127.0.0.1'
co_port_list = [':8010', ':8011', ':8012']

def test_get_url_list():
    """ test: type, url """
    url_list = mik.get_url_list(base_path=r'json/')
    assert isinstance(url_list, list)
    for url in url_list:
        assert re.match('https?://.+', url)

def test_make_first_node():
    """ かなりややこしくなるため省略 """
    pass

def test_verify_data():
    """ test: true case, false case """
    data = {'name': co_name_list[0]}
    signature = co_secret_key.sign(json.dumps(data).encode('utf-8')).hex()
    data['signature'] = signature
    assert mik.verify_data(data, co_public_key_str)
    data['name'] = co_name_list[1]  # name を改ざん
    assert not mik.verify_data(data, co_public_key_str)

@pytest.mark.skipif(online==False, reason="not online")
def test_post_data():
    test_data = {'test1': 'animal', 'test2': 100}
    response = mik.post_data(co_base_url+co_port_list[0]+'/test', test_data)
    assert response.status_code == 200
    assert response.json() == {"message": "test pass"}

def test_make_signature_str():
    """ test: type, len, unhexlify """
    data = {'public_key': co_public_key_str}
    signature_str = mik.make_signature_str(data, co_secret_key_str)
    assert isinstance(signature_str, str)
    assert len(signature_str) == 128
    assert binascii.unhexlify(signature_str)

def test_save_json():
    """ test: save, (str, list, dict), exist """
    base_path = 'test_data/'
    assert not mik.save_json('str', base_path+'test_str.json')
    assert os.path.exists(base_path+'test_str.json')
    assert not mik.save_json(['list'], base_path+'test_list.json')
    assert os.path.exists(base_path+'test_list.json')
    assert not mik.save_json({'dict': 1}, base_path+'test_dict.json')
    assert os.path.exists(base_path+'test_dict.json')

def test_load_json():
    """ test: load, type(str, list, dict)
    先に test_save_json() を実行する必要あり
    """
    base_path = 'test_data/'
    assert mik.load_json(base_path+'test_str.json')
    str_data = mik.load_json(base_path+'test_str.json')
    assert isinstance(str_data, str)
    assert mik.load_json(base_path+'test_list.json')
    list_data = mik.load_json(base_path+'test_list.json')
    assert isinstance(list_data, list)
    assert mik.load_json(base_path+'test_dict.json')
    dict_data = mik.load_json(base_path+'test_dict.json')
    assert isinstance(dict_data, dict)

def test_make_key_data():
    """ test: type, keys, url, str_to_key """
    key_data = mik.make_key_data(co_name_list[0], co_base_url+co_port_list[0])
    assert isinstance(key_data, dict)
    key_list = ['name', 'url', 'secret_key_str', 'public_key_str']
    assert list(key_data.keys()) == key_list
    assert isinstance(key_data['name'], str)
    assert re.match('https?://.+', key_data['url'])
    assert SigningKey.from_string(binascii.unhexlify(key_data['secret_key_str']), curve=SECP256k1)
    assert VerifyingKey.from_string(binascii.unhexlify(key_data['public_key_str']), curve=SECP256k1)

def test_str_to_public_key():
    """ test: type """
    public_key = mik.str_to_public_key(co_public_key_str)
    assert isinstance(public_key, type(co_public_key))

def test_str_to_secret_key():
    """ test: type """
    secret_key = mik.str_to_secret_key(co_secret_key_str)
    assert isinstance(secret_key, type(co_secret_key))

def test_key_to_str_public_key():
    """ test: type, len """
    public_key_str = mik.key_to_str(co_public_key)
    assert isinstance(public_key_str, str)
    assert len(public_key_str) == 128

def test_key_to_str_secret_key():
    """ test: type, len """
    secret_key_str = mik.key_to_str(co_secret_key)
    assert isinstance(secret_key_str, str)
    assert len(secret_key_str) == 64

def test_make_public_key():
    """ test: type """
    public_key = mik.make_public_key(co_secret_key)
    assert isinstance(public_key, type(co_secret_key.verifying_key))

def test_make_secret_key():
    """ test: type (curve=SECP256k1)"""
    secret_key = mik.make_secret_key()
    secret_key_type = type(co_secret_key)
    assert isinstance(secret_key, secret_key_type)

テストを実行します。
ファイル内の1つの関数のみ実行する場合は、:: をつけて関数名で実行できます。

コマンド・プロンプト
pytest test_mikoto.py
# pytest test_mikoto.py::test_make_secret_key

結果、すべてのテストをパスし、通信関連のものは、ちゃんとスキップされました。

通信機能を実装

続いて、フロントエンドを Streamlit、バックエンドを FastAPI で実装していきます。
鍵データがない場合、鍵を作成する画面を表示させ、鍵データがある場合の表示は、次回以降に実装します。
鍵データを作成してから、ブロードキャストし、認証率が90%を越えた場合、鍵データを保存します。
（無効だった場合の、バックエンドの処理は、省略してます）
フロントエンドを、Streamlit で実装します。

app.py
import os
import streamlit as st
import mikoto as mik

st.markdown("### みことプロジェクト")

if not os.path.exists('json/.my_data.json'):
    name = st.text_input('ニックネームを入力してください')
    add_url = st.text_input('URL を入力してください')

    if st.button('鍵を作成する', key='make_key'):
        key_data = mik.make_key_data(name, add_url)
        secret_key_str = key_data['secret_key_str']
        send_data = key_data.copy()
        send_data.pop('secret_key_str')
        signature = mik.make_signature_str(send_data, secret_key_str)
        key_data["signature"] = signature

        url_list = mik.get_url_list()
        url_list.append(add_url)
        res_200_count = 0
        for url in url_list:
            try:
                res = mik.post_data(url+'/key_data', key_data)
                if res.status_code == 200:
                    res_200_count += 1
                st.json({"status_code": res.status_code})
                st.json(res.json())
            except:
                st.json({"message": f"{url}: error"})
        # 認証率合格なら保存
        if res_200_count / len(url_list) > 0.9:
            mik.save_json(key_data, "json/.my_data.json")
        else:
            st.json({"message": "key_data invalid"})
else:
    st.markdown("鍵作成済の場合の表示")

# streamlit run app.py --server.port 8501

バックエンドで氣をつけるところは、pydantic の BaseModel を使っているので、受け取ったデータを、まず辞書にすることくらいかと思います。
バックエンドを、FastAPI で実装します。

main.py
from fastapi import FastAPI
from pydantic import BaseModel
import mikoto as mik

class Test(BaseModel):
    test1: str
    test2: int

class KeyData(BaseModel):
    name: str
    url: str
    public_key_str: str
    signature: str

app = FastAPI()

@app.get('/')
async def index():
    return "Welcome to Mikoto Project!"

@app.post('/test')
async def post_test(data: Test) -> dict:
    return {"message": "test pass"}

@app.post('/key_data')
async def post_key_data(key_data: KeyData):
    key_data = dict(key_data)
    print(key_data)
    if not mik.verify_data(key_data, key_data['public_key_str']):
        return {"message": "key_data invalid"}
    key_data.pop('signature')
    key_data_list = mik.load_json('json/key_data_list.json')
    key_data_list.append(key_data)
    mik.save_json(key_data_list, 'json/key_data_list.json')
    return {"message": "key_data received"}

# uvicorn main:app --reload --port 8010

次に、通信用のテストを作成します。
if 文などがあり、ややこしいので、入力やボタンのテストは手動で行います。

test_app_main.py
from fastapi.testclient import TestClient
from streamlit.testing.v1 import AppTest
from main import app

""" app.py(Streamlit) """
def test_app():
    at = AppTest.from_file("app.py").run()
    assert at.markdown[0].value == "### みことプロジェクト"

""" main.py(FastAPI) """
client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == "Welcome to Mikoto Project!"

def test_post_test():
    """ test: status, message """
    test_data = {'test1': 'animal', 'test2': 100}
    response = client.post(
        "/test",
        json=test_data,
    )
    assert response.status_code == 200
    assert response.json() == {"message": "test pass"}

実際に、動作確認をします。
初期ノードを作成し、そこから鍵データとテストファイル、テストデータ以外のファイルを、次のノードへコピーし、次のノードを登録していきます。
具体的には、app.py、main.py、mikoto.py と、json フォルダ内の block_chain.json、key_data_list.json、transaction_pool.json になります。

1ノードにつき、2ポート、サーバー立ち上げに使うので、3ノードで計6ポート使います。
8500番台がフロントエンド・Streamlit で、8010番台がバックエンド・FastAPI になります。
node1：http://127.0.0.1:8501、http://127.0.0.1:8010
node2：http://127.0.0.1:8502、http://127.0.0.1:8011
node3：http://127.0.0.1:8503、http://127.0.0.1:8012

  1. node1 で初期データ作成
  2. node2 に必要なファイル（前述）をコピー
  3. フロントエンドからキー作成
  4. node3 に必要なファイル（前述）をコピー
  5. フロントエンドからキー作成

の順で、動作確認します。

まず、node1 で初期データを作成します。

コマンド・プロンプト
python -c "import mikoto; mikoto.make_first_node()"

json フォルダに、ファイルが作成されます。

必要なファイルを、node2 にコピーします。

node1 と node2 のサーバーを立ち上げます。（それぞれのディレクトリから）

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

node2 の Streamlit（http://127.0.0.1:8502）にアクセスして鍵を作成します。
（画像をとり忘れましたが、node3 のときの画像が後にあります）
URL は、バックエンドの URL を入力します。
node2 に json/.my_data.json が作成され、key_data_list に dog が追加されます。

json/.my_data.json
{"name": "dog", "url": "http://127.0.0.1:8011", "secret_key_str": "611ceaa68b703dce8fb59a365f4b579bd51ace24e91d7067b40d0362f57cb7b6", "public_key_str": "a5a9fd9c283bab07ae575074a8be53263220fe8284ed5f08bfefe9d6c8575ba48b5441daf09ea585ba40c52412d121fa9e185b4a1fec89b05e7fa07fe592c147", "signature": "db5cb4359a1cc049b0875cb13be51ccefab6100913697d2808d5d09b4027da6383d0f3615a7bfb3cbe40b9ebfff590bbdb088f509a18ebfcb804f5b536f76aed"}

[{"name": "animal", "url": "http://127.0.0.1:8010", "public_key_str": "f2bb0f7849572b85b700be3efdd0d673d3a45aa21847d82809eb0156661b5ed764adda2817cb392bdaf08ce9fa5f49b26b6d2d38eafa068933ae7e505a0bbf27"}, {"name": "dog", "url": "http://127.0.0.1:8011", "public_key_str": "a5a9fd9c283bab07ae575074a8be53263220fe8284ed5f08bfefe9d6c8575ba48b5441daf09ea585ba40c52412d121fa9e185b4a1fec89b05e7fa07fe592c147"}]

同様に、node3 を作成し、鍵を作成します。
サーバーを立ち上げます。

コマンド・プロンプト
streamlit run app.py --server.port 8503
uvicorn main:app --reload --port 8012

鍵を作成します。
無事に作成されたようです。

json/key_data_list.json は、どのノードも同じ内容に更新されました。

json/key_data_list.json
[{"name": "animal", "url": "http://127.0.0.1:8010", "public_key_str": "f2bb0f7849572b85b700be3efdd0d673d3a45aa21847d82809eb0156661b5ed764adda2817cb392bdaf08ce9fa5f49b26b6d2d38eafa068933ae7e505a0bbf27"}, {"name": "dog", "url": "http://127.0.0.1:8011", "public_key_str": "a5a9fd9c283bab07ae575074a8be53263220fe8284ed5f08bfefe9d6c8575ba48b5441daf09ea585ba40c52412d121fa9e185b4a1fec89b05e7fa07fe592c147"}, {"name": "cat", "url": "http://127.0.0.1:8012", "public_key_str": "cc5373b75e7305ac2f7993c02b4cc0fd51003d0e526a4868acc4ad7d5fc8ef1d09b3f77312792b4a89b3e4a6863d875ad9302d637c7ac39aed6faa338b20e074"}]

参考文献など

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_1 です。

今回は以上となります、ありがとうございましたー♪

Discussion