前回 ブロックチェーンの作り方 #2 では、トランザクションの送信とノード側での受信処理を実装しました。#1でも書きましたが、ブロックチェーンのノードは、他のノードや利用者を一切信用できず不正にコインを手に入れようとしてくることを前提にプログラムを実装しなければなりません。そのためトランザクションの処理についても不正なトランザクションをトランザクションプールに登録させないように、しっかり検証する必要があります。今回 #3 では、ノード側のプログラムにトランザクションの検証処理を追加していきましょう。
関連記事：開発環境の準備とファイル構成
トランザクションの正当性の検証
トランザクションの正当性を検証するための処理を追加します。トランザクションのデータ型チェックは pydantic に任せられますので、以下の検証処理を追加します。
- コインの枚数が0枚以上であるか
- トランザクション署名が正しいか
BlockChain.py
BlockChain クラスに validate_tx メソッドを追加してこのメソッドに検証処理を実装します。
（略） class BlockChain: （略） （↓メソッドを追加↓） # トランザクションの正当性の検証 def validate_tx(self, tx): # トランザクションのコピー（値渡し）で検証を行う tx_copy = tx.copy() # マイナスのコインは許可しない if (tx_copy['coin'] < 0): print('コインの枚数がマイナス') return False # 送信者の公開鍵をオブジェクト化 public_key_obj = VerifyingKey.from_string(binascii.unhexlify(tx_copy['sender']), curve=SECP256k1) # トランザクションから署名を取出す signature = binascii.unhexlify(tx_copy['signature']) del tx_copy['signature'] # トランザクションをJSONに変換 tx_json = json.dumps(tx_copy).encode('utf-8') # トランザクション署名の検証 try: return public_key_obj.verify(signature, tx_json) except BadSignatureError: print('トランザクションの署名が不正') return False
【コードの要点】
tx_copy = tx.copy()
普段から Python を扱っている方には説明不要かと思いますが tx は辞書型のため、参照の引渡しとなります。そのまま tx の値をいじってしまうと参照元の tx にも影響があるため、トランザクションをコピー（値渡し）して、コピーした tx_copy に対して検証を行います。
トランザクション署名の検証は、トランザクションから署名を削除し、JSONに変換したトランザクションに対して行ってください。
トランザクションの重複チェック
続いていわゆる「二重支払い問題」を解決するため、トランザクションの重複チェックの処理を追加します。
余談ですが、ブロックチェーンの概念となっているサトシナカモトの論文ですが、実はこの二重支払い問題の解決策の提案が本題になっています。
概要 完全な P2P 電子通貨の実現により、金融機関の介在無しに、利用者同士の直接的なオンライン決済が可能となるだろう。電子署名により、P2P 電子通貨の機能の一部は実現可能であるが、その機能の主な利点は、信用が置ける第三者機関が二重支払いを防ぐために必要とされる場合、失われることとなる。本論文では、P2P ネットワークの使用による、二重支払い問題の解決策を提案する。（略）
ビットコイン： P2P 電子通貨システム（日本語）より引用
そしてその解決策は実にシンプルで「全トランザクションを監視する事である」とされています。これを実装するにはブロックチェーンに格納されている全てのトランザクションと、ブロックチェーンに追加しようとしているトランザクションを比較し重複していないことをチェックする必要があります。
BlockChain.py
まだチェーン部分を実装していませんので、今の段階ではトランザクションプールに重複したトランザクションが無いことをチェックします。BlockChain クラスに validate_duplicate_tx メソッドを追加してこのメソッドに重複チェックの処理を実装していきます。
（略） class BlockChain: （略） （↓メソッドを追加↓） # トランザクションの重複チェック（二重支払い問題対策） def validate_duplicate_tx(self, tx): # トランザクションプールに同じトランザクションがないか？ if tx in self.tx_pool['txs']: print('トランザクションプールに同じトランザクションがある') return False # チェックOK return True
【コードの要点】
特に難しい処理はありません。コメントの通りトランザクションプールに同じトランザクションがないかをチェックしているのみです。
実行ファイルへの実装
追加した検証用のメソッドを実行ファイル node.py に実装していきましょう。
node.py
受信したトランザクションをトランザクションプールに追加する前に、トランザクションの検証処理を追加して検証NGの場合は失敗のレスポンスを返すようにします。
（略） # トランザクションの受信 @app.post('/tx-pool') def receiv_tx(tx :Tx): # データモデルインスタンスを辞書（dict）型に変換 tx_dict = tx.model_dump() （↓処理の追加↓） # トランザクションの検証 if bc.validate_tx(tx_dict) and bc.validate_duplicate_tx(tx_dict): # 受信したトランザクションをトランザクションプールに追加 bc.add_tx_pool(tx_dict) # 成功のレスポンス return 'ok' else: # 失敗のレスポンス return 'error'
トランザクションの検証処理の動作確認
send_tx.py を以下のように変更して、それぞれのトランザクション検証処理が正しく動作していることを確認してください。
- コインの枚数をマイナスにする
- 別のユーザーの秘密鍵でトランザクション署名の追加する
トランザクションの重複チェックの動作確認
トランザクションの重複チェックについては、テスト用の実行ファイル duplicate_tx.py を作成して確認します。
BlockChain.py
BlockChain クラスにトランザクションプールを取得するためのメソッド get_tx_pool を追加します。（後ほどマイニングの処理でもこのメソッドを使います）
（略） class BlockChain: （略） （↓メソッドを追加↓） # トランザクションプールの取得 def get_tx_pool(self): # ノードのIPアドレス node_ip = self.get_node_ip() # トランザクションプールのURL url = 'http://' + node_ip + ':8000/tx-pool' # 取得実行 res = requests.get(url) # JSONを辞書(dict)型に変換して返す tx_pool_dict = res.json() return tx_pool_dict
duplicate_tx.py
テスト用の実行ファイル duplicate_tx.py を作成します。
from mod.BlockChain import BlockChain # ブロックチェーンインスタンス bc = BlockChain() # トランザクションプールを取得 tx_pool = bc.get_tx_pool() # 先頭のトランザクションを送信（トランザクションを重複させる） res = bc.send_tx(tx_pool['txs'][0]) print(res.text)
duplicate_tx.py を実行して「error」と表示されることを確認してください。
python blockchain/duplicate_tx.py "error"
node.py 側では以下のログが出力されるはずです。
INFO: 127.0.0.1:57098 - "GET /tx-pool HTTP/1.1" 200 OK
トランザクションプールに同じトランザクションがある
INFO: 127.0.0.1:57099 - "POST /tx-pool HTTP/1.1" 200 OK
今回のまとめ
今回は、ノード側で受信したトランザクションを検証する処理を実装しました。次回はジェネシスブロックとチェーンを実装していきます。
コメント