🌏

Python、Streamlit と FastAPI で、ブロックチェーンを実装してみた♪②秘密鍵ログイン機能とログ保存

2025/07/26に公開

今回は、Web3 のような秘密鍵（ウォレット？）によるログイン機能を実装し、ログを保存できるようにします。
それでは、さっそく実装していきまっす！

今回やること

秘密鍵によるログイン機能を実装し、ログを保存できるようにします！

環境

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
...

秘密鍵によるログイン機能を実装

ログイン機能の実装には、Streamlit の session_state を使います。
session_state: https://docs.streamlit.io/develop/api-reference/caching-and-state/st.session_state
app.py、mikoto.py、main.py の順で実装していきます。
ログイン機能は、session_state に state と label を設定し、ボタンクリックで状態を変えることで実装します。

app.py

...
def change_state(ss):
    if ss.state is True:
        ss.state = False
        ss.label = 'ログイン'
    else:
        login_data, url = mik.make_login_data('json/.my_data.json')
        try:
            response = mik.post_login_data(url, login_data)
            if response.json() == {"message": "login_data valid"}:
                ss.state = True
                ss.label = 'ログアウト'
            else:
                ss.label = 'ログインできませんでした'
        except:
            ss.label = 'ログインできませんでした'
...
else:
    st.markdown("鍵作成済の場合の表示")
    # 初期化
    ss = st.session_state
    if 'state' not in ss:
        ss.state = False
    if 'label' not in ss:
        ss.label = 'ログイン'

    st.button(ss.label, key='login', on_click=change_state, args=(ss,))

    if ss.state is True:
        st.markdown('ログイン中')
    else:
        st.markdown('ログアウト状態')
...

ログイン･データを作成する関数、公開鍵が登録されているか確認する関数を実装します。

mikoto.py

...
from typing import Tuple
...
def public_key_str_search(public_key_str: str) -> bool:
    key_data_list = load_json('json/key_data_list.json')
    public_key_str_list = [key_data['public_key_str'] for key_data in key_data_list]
    return public_key_str in public_key_str_list

def make_login_data(path: str) -> Tuple[dict, str]:
    my_data = load_json(path)
    login_data = {
        'time': dt.datetime.now().isoformat(),
        'public_key_str': my_data['public_key_str']
    }
    signature = make_signature_str(login_data, my_data['secret_key_str'])
    login_data['signature'] = signature
    return login_data, my_data['url']
...

作った関数のテストを作成します。

test_mikoto.py

...
def test_public_key_str_search():
    """ test: true case, false case """
    my_data = mik.load_json('json/.my_data.json')
    assert mik.public_key_str_search(my_data['public_key_str'])
    assert not mik.public_key_str_search(co_public_key_str)

def test_make_login_data():
    """ test: type, keys """
    login_data, url = mik.make_login_data(('json/.my_data.json'))
    assert isinstance(login_data, dict)
    assert isinstance(url, str)
    key_list = ['time', 'public_key_str', 'signature']
    assert list(login_data.keys()) == key_list
...

テストを実行します。

コマンド･プロンプト

pytest test_mikoto.py

無事にテストをパスしました。（通信用のテストは skip しています）

ログイン･データを受け取る、バックエンドを実装します。
公開鍵が登録されているかを確認してから、データを検証します。

main.py

...
class LoginData(BaseModel):
    time: str
    public_key_str: str
    signature: str
...
@app.post('/login_data')
async def post_login_data(login_data: LoginData):
    login_data = dict(login_data)
    if mik.public_key_str_search(login_data['public_key_str']):
        if mik.verify_data(login_data, login_data['public_key_str']):
            return {"message": "login_data valid"}
        else:
            return  {"message": "login_data invalid"}
    else:
        return {"message": "public_key_str invalid"}
...

動作確認するため、それぞれサーバーを立ち上げます。

コマンド・プロンプト

streamlit run app.py --server.port 8501
uvicorn main:app --reload --port 8010

http://localhost:8501/ にアクセスします。

ログインボタンをクリックします。
ちゃんと、ボタンの表示と、ページの表示内容が変わりました。
（ログアウトボタンをクリックすれば、元の画面に戻ります）

ログを保存する

ログの保存には、logging モジュールを使いますが、コードの見通しをよくするため、log.py ファイルを作成し、そこからインポートして使うことにしました。
最初の urllib3 に関するコードは、requests による urllib3 のデバッグ･ログを非表示にするためのものです。

log.py

import logging

logging.getLogger('urllib3.connectionpool').setLevel(logging.WARNING)

def get_logger(name):
    return logging.getLogger(name)

def basic_config(level, filename=None):
    format_str = "%(asctime)s - %(levelname)s - %(name)s - \"%(message)s\""
    level_dict = {
        'debug': logging.DEBUG,
        'info': logging.INFO,
        'warning': logging.WARNING,
        'error': logging.ERROR,
        'critical': logging.CRITICAL
    }
    if filename:
        logging.basicConfig(filename=filename, format=format_str, level=level_dict[level])
    else:
        logging.basicConfig(format=format_str, level=level_dict[level])
    return None

def log_debug(logger, message):
    return logger.debug(message)

def log_error(logger, message):
    return logger.error(message)

ログ表示を指定していきます。
（追加しているのは、log で始まる行が多いですが、位置がわかりやすいように周辺の行も表示しています）

app.py

...
import log
import mikoto as mik

logger = log.get_logger('app')
log.basic_config('debug', 'block_chain.log')
...
def change_state(ss):
    if ss.state is True:
        ss.state = False
        ss.label = 'ログイン'
        log.log_debug(logger, 'logout')
...
            if response.json() == {"message": "login_data valid"}:
                ss.state = True
                ss.label = 'ログアウト'
                log.log_debug(logger, 'login')
            else:
                ss.label = 'ログインできませんでした'
                log.log_error(logger, 'login failed')
        except:
            ss.label = 'ログインできませんでした'
            log.log_error(logger, 'login failed')
...
                st.json({"status_code": res.status_code})
                log.log_debug(logger, f'{res}: {res.json()}')
            except:
                st.json({"message": f"{url}: error"})
                log.log_error(logger, f"{url}: error")
        # 認証率合格なら保存
        if res_200_count / len(url_list) > 0.9:
            mik.save_json(key_data, "json/.my_data.json")
            log.log_debug(logger, 'key_data saved')
        else:
            st.json({"message": "key_data invalid"})
            log.log_error(logger, 'key_data invalid')
...

main.py

...
import log
import mikoto as mik

logger = log.get_logger('main')
log.basic_config('debug', 'block_chain.log')
...
    if not mik.verify_data(key_data, key_data['public_key_str']):
        log.log_error(logger, 'key_data invalid')
        return {"message": "key_data invalid"}
...
    mik.save_json(key_data_list, 'json/key_data_list.json')
    log.log_debug(logger, 'key_data received')
...
        if mik.verify_data(login_data, login_data['public_key_str']):
            log.log_debug(logger, 'login_data valid')
            return {"message": "login_data valid"}
        else:
            log.log_error(logger, 'login_data invalid')
            return  {"message": "login_data invalid"}
    else:
        log.log_error(logger, 'public_key_str invalid')
        return {"message": "public_key_str invalid"}
...

ログ用のテスト･ファイルを作成します。
保存したログの内容に関しては、後に動作確認のときにチェックします。

test_log.py

import logging
import log

def test_get_logger():
    """ test: return, type """
    assert log.get_logger('test') == logging.getLogger('test')
    assert isinstance(log.get_logger('test'), type(logging.getLogger('test')))

def test_basic_config():
    """ test: return=None """
    logger = log.get_logger('test')
    log.basic_config('debug', 'test.log')
    assert not log.log_debug(logger, 'debug')

def test_log_debug():
    """ test: return=None """
    logger = logging.getLogger('test')
    assert not log.log_debug(logger, 'debug')

def test_log_error():
    """ test: return=None """
    logger = logging.getLogger('test')
    assert not log.log_error(logger, 'error')

テストを実行、無事にパスしました。

コマンド･プロンプト

pytest test_log.py

動作確認のため、それぞれサーバーを立ち上げます。

コマンド･プロンプト

streamlit run app.py --server.port 8501
uvicorn main:app --reload --port 8010

http://127.0.0.1:8501にアクセスし、ログインとログアウトを実行し、ログの内容を確認します。
ちゃんと、ログが保存されました。（urllib3 のデバックも非表示になっています）
block_chain.log

参考文献など

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

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

今回やること

環境

ディレクトリ構成

秘密鍵によるログイン機能を実装

ログを保存する

参考文献など

今回のコード

Discussion