前回ブロックチェーンの作り方 #4 ではジェネシスブロックとチェーンの土台となる部分を実装しました。今回 #5 ではブロックチェーンの中核を担うマイニング処理を実装していきます。
関連記事：開発環境の準備とファイル構成
マイニングの方式（PoW と PoS）
マイニングの方式（「コンセンサスアルゴリズム」と呼ばれます）には大きく分けて PoW（Proof of Work）プルーフ・オブ・ワークと PoS（Proof of Stake）プルーフ・オブ・ステイクがあり、今回は PoW でマイニングを実装します。
PoW は、誰よりも早くマイニングの成功条件を満たすナンスを計算したマイナーがブロックを追加しその対価として報酬を得る方式のため、マイナーは計算能力の高いコンピューターを所有する必要があり、その計算には膨大な電力が必要になります。現在では温室効果ガスの排出削減が求められているため、マイニングに膨大な電力を必要とするビットコインをはじめとする暗号資産は環境負荷の観点から批判されることが多くあります。
一方 PoS は、コイン（暗号資産）の保有量が多いほどブロック追加し報酬を得る確率が高くなる方式のため、膨大な電力は必要無く環境負荷もほとんどありません。（例えばイーサリアムは、2022年9月にマイニングの方式を PoW から PoS に移行し電力を99%以上も削減できたとされています）
今回は、説明を簡単にするためにあえて高い計算能力が必要な PoW でマイニングを実装していますが、本番のブロックチェーンを開発するのであれば PoS もしくは環境負荷の低いコンセンサスアルゴリズムを採用することを強くオススメします。
マイニング処理
BlockChain.py
マイニング処理に必要なメソッドを BlockChain クラスに実装します。hashlib モジュールと time モジュールを使いますので追加でインポートしておきましょう。
（略） （↓モジュールを追加↓） import hashlib import time class BlockChain: def __init__(self): （略） （↓変数を追加↓） # マイニングの報酬 self.reward_coin = 50 # マイニングの難易度（数が多いほど難しくなる） self.difficulty = 4 （略） （↓メソッドを追加↓） # チェーンの取得 def get_chain(self): # ノードのIPアドレス node_ip = self.get_node_ip() # トランザクションプールのURL url = 'http://' + node_ip + ':8000/chain' # 取得実行 res = requests.get(url) # JSONを辞書(dict)型に変換して返す chain_dict = res.json() return chain_dict # ハッシュ値の生成 def gen_hash(self, target_dict): # 対象をJSONに変換してからハッシュ化（辞書型のままだとエラーになる） target_json = json.dumps(target_dict).encode('utf-8') hash = hashlib.sha256(target_json) # ハッシュ値（16進数）を返す return hash.hexdigest() # 報酬用トランザクションの作成 def make_reward_tx(self, miner_public_key): # 報酬用トランザクションを返す return { 'time': time.time(), 'sender': 'reward', 'to': miner_public_key, 'coin': self.reward_coin, 'signature': 'none' } # ナンスの検証 def validate_nonce(self, hash): # ハッシュ値の先頭 self.difficulty 文字が 0 であればマイニング成功（ナンスが正しい）とする if hash[:self.difficulty] == '0' * self.difficulty: return True else: return False # チェーンの送信 def send_chain(self, chain_dict): # ノードのIPアドレス node_ip = self.get_node_ip() # チェーンの送信先URL url = 'http://' + node_ip + ':8000/chain' # チェーンをJSONに変換（辞書型のままでは送信できない） chain_json = json.dumps(chain_dict).encode('utf-8') # 送信 res = requests.post(url, chain_json) # 送信結果を返す return res
【コードの要点】
これらのメソッドは主にマイニングの実行ファイルから利用します。
gen_hash メソッド
ブロックのハッシュ値を生成するためのメソッドです。ハッシュアルゴリズムは SHA-256 を使います。
make_reward_tx メソッド
マイニングの報酬用トランザクションを作成するためのメソッドです。マイニングの報酬となるコインの枚数は self.reward_coin 変数に指定してください。例として報酬コインを50枚に設定していますが、適切なインセンティブになるように調整してください。
また、このメソッドを利用する時に引数 miner_public_key にマイナーの公開鍵を渡して報酬用トランザクションの to に指定します。こうすることで、この報酬用トランザクションがブロックチェーンに追加されれば、マイナーは報酬を得ることになります。報酬の送信者となる sender は、報酬用トランザクションであることを示す「reward」を指定します。トランザクション署名は必要ありませんので signature には「none」を指定します。報酬用トランザクションはトランザクション署名の検証を行いませんので検証エラーになることはありません。
validate_nonce メソッド
PoW（Proof of Work）は、マイニングの成功条件として計算と時間という観点から生成が難しいがその検証は容易となるデータの計算が求められます。具体的には、マイナーは追加するブロックのハッシュ値の先頭文字が指定した数だけ「0」になるナンスを計算します。
例えばハッシュ値の先頭8文字が「0」になるナンスは、ナンスの値を変えて先頭8文字が「0」になるのハッシュ値が見つかるまでひたすら計算するしかありません。一方のその検証は、計算済みのナンスでハッシュ値を１回求めるだけで済みます。ハッシュ値先頭の「0」の数は、難易度を設定する self.difficulty 変数に指定してください。開発中のため「4」を指定していますが、おそらく数秒で計算が終わってしまいますので、実際の運用では適切な難易度になるように調整してください。
マイニングの報酬と難易度の調整はブロックチェーンの実装と運用において最も重要な部分になります。この記事では割愛しますが、マイニングの報酬と難易度は柔軟に調整できるように実装しておくのが良いでしょう。（詳しくは「マイニング難易度調整」などで検索してださい）
mining.py
マイニングを実行してノードにチェーンを送信するための実行ファイル mining.py を新たに作成します。BlockChain.py と users.py 、time モジュールをインポートしてください。
from mod.BlockChain import BlockChain
from mod.users import users
import time
# ブロックチェーンインスタンス
bc = BlockChain()
# チェーンの取得
chain = bc.get_chain()
# 最後のブロックのハッシュ値を生成
last_block = chain['blocks'][-1]
last_block_hash = bc.gen_hash(last_block)
# ブロックに含めるトランザクション
tx_pool = bc.get_tx_pool()
target_txs = tx_pool['txs']
# マイナーの公開鍵
miner_public_key = users['C']['public_key']
# 報酬用トランザクションを作成
reward_tx = bc.make_reward_tx(miner_public_key)
# ブロックに含めるトランザクションに報酬用トランザクションを追加
target_txs.append(reward_tx)
# ナンスの初期値
nonce = 0
# マイニング処理
mining_start_time = time.time()
while True:
# 新しいブロックを作成
new_block = {
'time': time.time(),
'previous_hash': last_block_hash,
'nonce': nonce,
'txs': target_txs,
}
# 新しいブロックのハッシュ値を生成
new_block_hash = bc.gen_hash(new_block)
# ナンスの検証
if bc.validate_nonce(new_block_hash):
# マイニングが成功したらナンスとハッシュ値とマイニング時間を表示してマイニング処理を終了する
print('mining success.')
print('nonce: ' + str(nonce))
print('hash: ' + new_block_hash)
print('mining time: ' + str(time.time() - mining_start_time) + ' second')
break
else:
# マイニングが失敗している場合はナンスを増やして再度マイニング処理を実行
nonce += 1
# チェーンに新しいブロックを追加
chain['blocks'].append(new_block)
# ノードにチェーンを送信
res = bc.send_chain(chain)
print(res.text)
【コードの要点】
last_block_hash = bc.gen_hash(last_block)
チェーンの最後のブロックのハッシュ値を求めます。これから追加する新しいブロックから見ると「前のブロックのハッシュ値」になりますので、これをマイニング処理で作成する新しいブロック new_block のキー previous_hash に指定します。
target_txs = tx_pool['txs']
ブロックに含めるトランザクションを target_txs にまとめます。今の段階ではトランザクションプール内のすべてのトランザクションを対象にしていますが、不正なトランザクションが混じっている可能性もありますので、次回の記事でトランザクションを検証する処理を追加します。
reward_tx = bc.make_reward_tx(miner_public_key)
マイナーの公開鍵を引数に指定して報酬用トランザクションを作成し、target_txs に追加します。こうすることで、マイニングが成功し報酬用トランザクションを含んだブロックがチェーンに追加されれば、マイナーは報酬を得ることになります。
マイニング処理は whileループの中で新しいブロック new_block を作成し、BlockChain クラスの gen_hash に渡して新しいブロックのハッシュ値を生成します。
次に新しいブロックのハッシュ値を BlockChain クラスの validate_nonce に渡してマイニングが成功したか否かを判定し、マイニングに成功した場合は、成功したナンスの値などの情報を表示して、whileループを抜けてチェーンに新しいブロックを追加してノードに送信します。マイニングに失敗した場合は、ナンスの値に1を加えて再度同じ処理をマイニングに成功するまで繰り返します。
動作確認
mining.py を実行してマイニングが成功したナンスとハッシュ値、マイニングに要した秒数が表示されることを確認してください。ハッシュ値の先頭文字が指定した数だけ「0」になっていると思います。（ノード側にチェーンの受信処理を実装していませんので、レスポンスとして「Method Not Allowed」が返ってくると思います）
python blockchain/mining.py
mining success.
nonce: 62211
hash: 000017ef7eb448eeccc9547fbdc438cc089d2b080ee0d17fd14bfb211a0880eb
mining time: 0.36821413040161133 second
{"detail":"Method Not Allowed"}
上の例では、1秒以下でマイニングに成功していますが、ハッシュ値の先頭の「0」の数を8文字にすると、CPU が Apple M3 Pro のマシンではマイニングに10時間程度かかっています。
python blockchain/mining.py
mining success.
nonce: 6053757078
hash: 000000009cde2291a4bc77b8531c3f2d063a43e1e6a44960c5e23c32ac039427
mining time: 36141.47594499588 second
{"detail":"Method Not Allowed"}
チェーンの受信
続いて、ノード側にチェーンの受信処理を追加していきましょう。
今回作成するブロックチェーンでは、新しいブロックも含めて受信したチェーン上のすべてのブロックとトランザクションの正当性を検証し、問題がなければノードに保持しているチェーンを入れ替える形で実装します。
BlockChain.py
チェーンの受信処理に必要なメソッドを BlockChain クラスに実装します。
（略） class BlockChain: def __init__(self): （略） （↓変数を追加↓） # 全ブロックのトランザクションリスト set_all_block_txs() メソッドでセットする self.all_block_txs = [] （略） # トランザクションの重複チェック（二重支払い問題対策） def validate_duplicate_tx(self, tx): （略） （↓処理を追加↓） # 全ブロックのトランザクションリストに同じトランザクションがないか？ if tx in self.all_block_txs: print('全ブロックのトランザクションリストに同じトランザクションがある') return False （略） （↓メソッドを追加↓） # チェーンの正当性の検証 def validate_chain(self, chain): # 保持しているチェーンよりも長いことをチェック if len(self.chain['blocks']) >= len(chain['blocks']): print('保持しているチェーンより短い') return False # ジェネシスブロックが改ざんされていないことをチェック if self.genesis_block != chain['blocks'][0]: print('ジェネシスブロックが改ざんされている') return False # トランザクションの入れ物（全ブロックのトランザクションをここに入れる） all_block_txs = [] # 各ブロックの正当性の検証 for i in range(len(chain['blocks'])): # ジェネシスブロックはチェック済みなので検証をスキップ if i == 0: continue # チェック対象のブロック block = chain['blocks'][i] # 1つ前のブロック previous_block = chain['blocks'][i-1] # 1つ前のブロックのハッシュ値が正しいことをチェック if block['previous_hash'] != self.gen_hash(previous_block): print('1つ前のブロックのハッシュ値に誤りがある') return False # ナンスのチェック block_hash = self.gen_hash(block) if not self.validate_nonce(block_hash): print('ナンスに誤りがある') return False # 報酬用トランザクションの重複チェック用 reward_duplication = False # 各トランザクションの正当性の検証 for tx in block['txs']: # 報酬用のトランザクションのチェック if tx['sender'] == 'reward': # 報酬用のトランザクションが重複チェック if reward_duplication: print('報酬用のトランザクションが重複している') return False else: reward_duplication = True # マイニングの報酬が正しく設定されていることをチェック if tx['coin'] != self.reward_coin: print('マイニングの報酬が誤っている') return False else: # トランザクションの正当性の検証 if not self.validate_tx(tx): print('トランザクションに異常がある') return False # トランザクションが再利用されていないことをチェック if tx not in all_block_txs: all_block_txs.append(tx) else: print('トランザクションが再利用されている') return False # 検証OK return True # チェーンの入替え def replace_chain(self, chain): self.chain = chain # チェーンを入替えたので全ブロックのトランザクションリストを作り直す self.set_all_block_txs() # ブロックに存在するトランザクションはトランザクションプールから削除する for tx in self.all_block_txs: if tx in self.tx_pool['txs']: self.tx_pool['txs'].remove(tx) # トランザクションプールをファイルに保存する self.save_tx_pool() # チェーンをファイルに保存する self.save_chain() # 全ブロックのトランザクションリストの作成 def set_all_block_txs(self): self.all_block_txs = [] for i in range(len(self.chain['blocks'])): block = self.chain['blocks'][i] for tx in block['txs']: self.all_block_txs.append(tx) # チェーンをファイルに保存する def save_chain(self): pd.to_pickle(self.chain, self.chain_file)
【コードの要点】
self.all_block_txs 変数
チェーン上の全てのトランザクションのリストを格納します。ブロックチェーンの作り方 #3 でも少し触れましたが、「二重支払い問題」を解決するためには、”全てのトランザクションが重複していないこと”をチェックする必要があるため、このリストを使ってチェックを行います。全てのトランザクションのリストは set_all_block_txs メソッドで作成します。
validate_duplicate_tx メソッド
チェーンの受信処理ではありませんが、トランザクションの重複チェックに全ブロックのトランザクションリスト self.all_block_txs に対する重複チェック処理を追加しています。この処理の追加により、ノードがトランザクションを受信した次点でトランザクションの重複を排除することができます。
validate_chain メソッド
マイナーから送られてきたチェーン上の全ブロック、全トランザクションの正当性を検証します。ブロックチェーンの信頼性を担う最も重要な処理になります。処理の詳細はコードのコメントをご参照ください。
replace_chain メソッド
チェーンの正当性が確認できたら、ノードに保持しているチェーンを入れ替えて、チェーン上に存在するトランザクション（すなわちマイナーが新しいブロックに含めたトランザクション）は、トランザクションプールから削除します。
node.py
node.py に、チェーンの受信処理を追加します。
（略） # ブロックチェーンインスタンス bc = BlockChain() # ノードの起動処理 （略） （↓起動処理を追加↓） bc.set_all_block_txs() # 全ブロックのトランザクションリストの作成 （略） （↓データ定義を追加↓） # ブロックデータの定義 class Block(BaseModel): time: float previous_hash: str nonce: int txs: List[Tx] # チェーンデータの定義 class Chain(BaseModel): blocks: List[Block] （略） （↓関数を追加↓） # チェーンの受信 @app.post('/chain') def post_chain(chain :Chain): # データモデルインスタンスを辞書（dict）型に変換 chain_dict = chain.model_dump() # チェーンの検証 if bc.validate_chain(chain_dict): # チェーンの入替え bc.replace_chain(chain_dict) # 成功のレスポンス return 'ok' else: # 失敗のレスポンス return 'error'
【コードの要点】
class Block(BaseModel): class Chain(BaseModel):
pydantic の BaseModel を継承してブロックとチェーンのデータ要素を定義しておきます。
@app.post('/chain')
チェーンを受信するエンドポイントを指定します。エンドポイント名に決まりはありませんが BlockChain クラスの send_chain メソッドのチェーンの送信先URLに対応するようにしてください。この直下の関数 post_chainが受信したチェーンを処理します。
動作確認
node.py を再起動して mining.py を実行してください。レスポンス「ok」が返ってくると思います。
python blockchain/mining.py mining success. nonce: 30012 hash: 0000d83d53530b77a3abc6700526791b3deabb56bd26e4f8d749d2e31ad6cf80 mining time: 0.1790330410003662 second "ok"
ブラウザで「http://127.0.0.1:8000/chain」にアクセスしてブロックが追加されていることを確認してください。
今回のまとめ
今回は、ブロックチェーン開発の山場となるマイニング処理とチェーンの受信処理を実装しました。チェーンを受信するノード側で受信したチェーン上の全てのブロックと全てのトランザクションの正当性を検証していますが、コインを持っていないユーザーが送金できてしまうという問題がまだ残ります。次回は、残高チェックの処理を実装していきます。
コメント