ビットコインの仕組みをpythonで実装してみる #2 「ビットコインの仕組みとトランザクション」

トランザクションA
入力：相手のUTXO
出力：1 BTC → 自分

自分のUTXO一覧：
- 1 BTC

自分のUTXO一覧：
- 1 BTC
- 0.3 BTC

入力：
  1 BTC（消える）

出力：
  0.4 BTC → 相手
  0.6 BTC → 自分（お釣り）

あなたのUTXO一覧：
- 0.3 BTC（そのまま）
- 0.6 BTC（お釣り）

公開鍵をコピーし、それをハッシュせよ
その値がこの「公開鍵ハッシュ」と一致するか確認せよ
一致したら署名を検証せよ

version + 公開鍵ハッシュ + checksum

Base58Checkをデコード
バージョン除去
公開鍵ハッシュ（20バイト）を取り出し、それを scriptPubKey に埋め込む

import hashlib
import ecdsa
import base58

from address import generate_keypair, hash256

def little_endian(value: int, length: int) -> bytes:
	return value.to_bytes(length, byteorder="little")

def encode_varint(value: int) -> bytes:
	if value < 0xFD:
		return bytes([value])
	if value <= 0xFFFF:
		return b"\xFD" + little_endian(value, 2)
	if value <= 0xFFFFFFFF:
		return b"\xFE" + little_endian(value, 4)
	return b"\xFF" + little_endian(value, 8)

def push_data(data: bytes) -> bytes:
	if len(data) >= 0x4C:
		raise ValueError("Demo only supports short pushdata")
	return bytes([len(data)]) + data

def address_to_pubkey_hash(address: str) -> bytes:
	decoded = base58.b58decode(address)
	payload, checksum = decoded[:-4], decoded[-4:]
	if hash256(payload)[:4] != checksum:
		raise ValueError("Invalid address checksum")
	return payload[1:]

def p2pkh_scriptpubkey(pubkey_hash: bytes) -> bytes:
	return b"\x76\xa9" + b"\x14" + pubkey_hash + b"\x88\xac"

def serialize_txin(prev_txid: str, prev_index: int, script_sig: bytes) -> bytes:
	prev_txid_bytes = bytes.fromhex(prev_txid)[::-1]
	return (
		prev_txid_bytes
		+ little_endian(prev_index, 4)
		+ encode_varint(len(script_sig))
		+ script_sig
		+ little_endian(0xFFFFFFFF, 4)
	)

def serialize_txout(amount_sats: int, script_pubkey: bytes) -> bytes:
	return (
		little_endian(amount_sats, 8)
		+ encode_varint(len(script_pubkey))
		+ script_pubkey
	)

def create_sighash_preimage(
	prev_txid: str,
	prev_index: int,
	script_code: bytes,
	outputs: list[tuple[int, bytes]],
	locktime: int = 0,
	sighash_type: int = 1,
) -> bytes:
	version = little_endian(1, 4)
	txin = serialize_txin(prev_txid, prev_index, script_code)
	inputs = encode_varint(1) + txin

	serialized_outputs = b"".join(
		serialize_txout(amount, script) for amount, script in outputs
	)
	outputs_blob = encode_varint(len(outputs)) + serialized_outputs

	return (
		version
		+ inputs
		+ outputs_blob
		+ little_endian(locktime, 4)
		+ little_endian(sighash_type, 4)
	)

def sign_p2pkh_input(
	private_key: bytes,
	public_key: bytes,
	prev_txid: str,
	prev_index: int,
	script_code: bytes,
	outputs: list[tuple[int, bytes]],
) -> bytes:
	preimage = create_sighash_preimage(
		prev_txid, prev_index, script_code, outputs
	)
	z = hash256(preimage)
	sk = ecdsa.SigningKey.from_string(private_key, curve=ecdsa.SECP256k1)
	signature = sk.sign_digest(z, sigencode=ecdsa.util.sigencode_der_canonize)
	signature += b"\x01"  # SIGHASH_ALL

	return push_data(signature) + push_data(public_key)

def build_raw_transaction(
	prev_txid: str,
	prev_index: int,
	script_sig: bytes,
	outputs: list[tuple[int, bytes]],
	locktime: int = 0,
) -> bytes:
	version = little_endian(1, 4)
	inputs = encode_varint(1) + serialize_txin(prev_txid, prev_index, script_sig)
	serialized_outputs = b"".join(
		serialize_txout(amount, script) for amount, script in outputs
	)
	outputs_blob = encode_varint(len(outputs)) + serialized_outputs
	return version + inputs + outputs_blob + little_endian(locktime, 4)

def build_p2pkh_demo_tx(
	sender_priv: bytes,
	sender_pub: bytes,
	sender_addr: str,
	recipient_addr: str,
	utxo_amount: int,
	send_amount: int,
	fee: int,
	prev_txid: str,
	prev_index: int,
) -> bytes:
	change_amount = utxo_amount - send_amount - fee
	if change_amount < 0:
		raise ValueError("Insufficient funds for demo")

	sender_pubkey_hash = address_to_pubkey_hash(sender_addr)
	recipient_pubkey_hash = address_to_pubkey_hash(recipient_addr)

	outputs = [
		(send_amount, p2pkh_scriptpubkey(recipient_pubkey_hash)),
		(change_amount, p2pkh_scriptpubkey(sender_pubkey_hash)),
	]

	script_code = p2pkh_scriptpubkey(sender_pubkey_hash)
	script_sig = sign_p2pkh_input(
		sender_priv,
		sender_pub,
		prev_txid,
		prev_index,
		script_code,
		outputs,
	)

	return build_raw_transaction(prev_txid, prev_index, script_sig, outputs)

def build_p2pkh_demo_tx_details(
	sender_priv: bytes,
	sender_pub: bytes,
	sender_addr: str,
	recipient_addr: str,
	utxo_amount: int,
	send_amount: int,
	fee: int,
	prev_txid: str,
	prev_index: int,
) -> tuple[bytes, int, list[tuple[int, str]]]:
	change_amount = utxo_amount - send_amount - fee
	if change_amount < 0:
		raise ValueError("Insufficient funds for demo")

	sender_pubkey_hash = address_to_pubkey_hash(sender_addr)
	recipient_pubkey_hash = address_to_pubkey_hash(recipient_addr)

	outputs = [
		(send_amount, p2pkh_scriptpubkey(recipient_pubkey_hash)),
		(change_amount, p2pkh_scriptpubkey(sender_pubkey_hash)),
	]

	script_code = p2pkh_scriptpubkey(sender_pubkey_hash)
	script_sig = sign_p2pkh_input(
		sender_priv,
		sender_pub,
		prev_txid,
		prev_index,
		script_code,
		outputs,
	)

	raw_tx = build_raw_transaction(prev_txid, prev_index, script_sig, outputs)
	output_labels = [
		(send_amount, recipient_addr),
		(change_amount, sender_addr),
	]
	return raw_tx, change_amount, output_labels

def build_demo_state(
	me_to_other_amount: int,
	other_to_me_amount: int,
	fee: int,
	me_utxo_amount: int = 100_000,
) -> dict:
	if me_to_other_amount + fee > me_utxo_amount:
		raise ValueError("Me UTXO amount is insufficient for tx1")
	if other_to_me_amount + fee > me_to_other_amount:
		raise ValueError("Other amount is insufficient for tx2")

	me_priv, me_pub, me_addr = generate_keypair()
	other_priv, other_pub, other_addr = generate_keypair()

	prev_txid_1 = "00" * 32
	prev_index_1 = 0
	raw_tx_1, change_amount_1, outputs_1 = build_p2pkh_demo_tx_details(
		me_priv,
		me_pub,
		me_addr,
		other_addr,
		me_utxo_amount,
		me_to_other_amount,
		fee,
		prev_txid_1,
		prev_index_1,
	)
	txid_1 = hash256(raw_tx_1)[::-1].hex()

	prev_txid_2 = txid_1
	prev_index_2 = 0
	raw_tx_2, change_amount_2, outputs_2 = build_p2pkh_demo_tx_details(
		other_priv,
		other_pub,
		other_addr,
		me_addr,
		me_to_other_amount,
		other_to_me_amount,
		fee,
		prev_txid_2,
		prev_index_2,
	)
	txid_2 = hash256(raw_tx_2)[::-1].hex()

	utxos_t0 = [
		{
			"owner": "me",
			"txid": prev_txid_1,
			"index": prev_index_1,
			"amount": me_utxo_amount,
			"address": me_addr,
		},
	]
	utxos_t1 = [
		{
			"owner": "other",
			"txid": txid_1,
			"index": 0,
			"amount": outputs_1[0][0],
			"address": outputs_1[0][1],
		},
		{
			"owner": "me",
			"txid": txid_1,
			"index": 1,
			"amount": change_amount_1,
			"address": me_addr,
		},
	]
	utxos_t2 = [
		{
			"owner": "me",
			"txid": txid_2,
			"index": 0,
			"amount": outputs_2[0][0],
			"address": outputs_2[0][1],
		},
		{
			"owner": "other",
			"txid": txid_2,
			"index": 1,
			"amount": change_amount_2,
			"address": other_addr,
		},
		{
			"owner": "me",
			"txid": txid_1,
			"index": 1,
			"amount": change_amount_1,
			"address": me_addr,
		},
	]

	return {
		"me_address": me_addr,
		"other_address": other_addr,
		"tx1": {
			"raw_hex": raw_tx_1.hex(),
			"txid": txid_1,
			"input": {
				"txid": prev_txid_1,
				"index": prev_index_1,
				"amount": me_utxo_amount,
				"address": me_addr,
			},
			"outputs": [
				{"amount": outputs_1[0][0], "address": outputs_1[0][1]},
				{"amount": change_amount_1, "address": me_addr},
			],
		},
		"tx2": {
			"raw_hex": raw_tx_2.hex(),
			"txid": txid_2,
			"input": {
				"txid": prev_txid_2,
				"index": prev_index_2,
				"amount": me_to_other_amount,
				"address": other_addr,
			},
			"outputs": [
				{"amount": outputs_2[0][0], "address": outputs_2[0][1]},
				{"amount": change_amount_2, "address": other_addr},
			],
		},
		"utxo_timeline": {
			"t0": utxos_t0,
			"t1": utxos_t1,
			"t2": utxos_t2,
		},
	}