Bitcoin Core RPC 快速上手

比特幣核心客戶端（Bitcoin Core）提供了一個功能完整的 JSON-RPC 介面，讓開發者能夠直接與比特幣網路互動。這篇文章將帶你從基礎查詢到實際應用，透過可運行的程式碼範例深入理解比特幣 RPC 的運作機制。理解 RPC 介面是開發比特幣應用、構建錢包服務、以及實現節點自動化的核心技能。

為什麼選擇 Bitcoin Core RPC

Bitcoin Core 是比特幣網路中最被廣泛採用的完整節點實現，其 RPC 介面提供了最完整、最權威的比特幣網路數據。與依賴第三方 API 不同，直接使用讓你能夠：

Bitcoin Core RPC - 數據自主權：不依賴外部服務，確保數據的可靠性與隱私性

• 完整功能：存取所有鏈上數據，包括未確認交易、記憶池狀態、節點網路資訊
• 直接驗證：能夠獨立驗證交易、區塊、餘額等資訊
• 開發彈性：可根據需求自訂查詢邏輯與自動化流程

環境建置

安裝 Bitcoin Core

在正式開始之前，你需要運行一個比特幣節點。Bitcoin Core 支援 Linux、macOS、Windows 等主流作業系統。以下是在 Ubuntu/Debian 環境下的安裝步驟：

# 新增比特幣 PPA 並安裝
sudo apt-add-repository ppa:bitcoin/bitcoin
sudo apt-get update
sudo apt-get install bitcoin-daemon bitcoin-cli

# 或者從原始碼編譯（進階用戶）
# git clone https://github.com/bitcoin/bitcoin.git
# cd bitcoin && ./autogen.sh && ./configure && make -j$(nproc)

設定 Bitcoin Core

首次運行 Bitcoin Core 前，需要建立設定檔以開啟 RPC 介面並設定認證：

# 建立比特幣設定目錄
mkdir -p ~/.bitcoin

# 編輯比特幣設定檔
cat > ~/.bitcoin/bitcoin.conf << 'EOF'
# 網路設定
testnet=0  # 使用主網（設為1則使用測試網）
mainnet=1

# RPC 介面設定
server=1
daemon=1

# RPC 認證設定（生成 rpcauth）
# 建議使用 rpcauth 格式，避免明文密碼
# rpcauth 可透過以下方式生成：
# python3 -c "import hashlib, base64; \
#   print('rpcauth=username:' + base64.b64encode(hashlib.sha256('username:password'.encode()).digest()).decode())"

# 或者使用傳統的使用者名稱/密碼格式
rpcuser=bitcoinuser
rpcpassword=your_secure_password_here

# RPC 監聽位址與連接數限制
rpcbind=127.0.0.1
rpcallowip=127.0.0.1
rpcport=8332

# 最小化硬碟空間需求（可選）
prune=550  # 保留最近 550 MB 的區塊資料
EOF

生成安全的 RPCAUTH

比特幣官方建議使用 rpcauth 而非明文密碼。以下是安全的認證生成方式：

#!/usr/bin/env python3
"""
生成 Bitcoin Core rpcauth 認證字串
使用方法: python3 gen_rpcauth.py <username> <password>
"""
import sys
import hashlib
import base64
import secrets

def generate_rpcauth(username, password):
    """生成 rpcauth 格式的認證字串"""
    # 生成隨機鹽值
    salt = secrets.token_hex(16)

    # 使用鹽值進行雙重 SHA-256 哈希
    h = hashlib.sha256((username + salt + password).encode()).digest()
    h = hashlib.sha256(h).digest()

    # 編碼為 base64
    encoded = base64.b64encode(h).decode()

    return f"rpcauth={username}:{salt}${encoded}"

if __name__ == "__main__":
    if len(sys.argv) != 3:
        print(f"用法: {sys.argv[0]} <username> <password>")
        sys.exit(1)

    username, password = sys.argv[1], sys.argv[2]
    result = generate_rpcauth(username, password)
    print(result)
    print("\n將上述輸出添加到 bitcoin.conf 中的 rpcauth= 行")

連接 Bitcoin Core RPC

使用命令行工具

Bitcoin Core 附帶的 bitcoin-cli 是最直接的互動方式：

# 啟動比特幣節點（首次運行會下載整個區塊鏈）
bitcoind -daemon

# 等待節點同步完成後，進行基本查詢
bitcoin-cli getblockchaininfo

# 查詢節點網路資訊
bitcoin-cli getnetworkinfo

# 查詢記憶池資訊
bitcoin-cli getmempoolinfo

使用 cURL 直接調用

對於開發應用程式，直接使用 HTTP 調用 RPC 更加靈活：

# 基本 RPC 調用範例
RPC_USER="bitcoinuser"
RPC_PASSWORD="your_secure_password_here"
RPC_URL="http://127.0.0.1:8332"

# 查詢比特幣節點版本
curl -s -u "${RPC_USER}:${RPC_PASSWORD}" \
  -X POST \
  -H "Content-Type: application/json" \
  --data '{"jsonrpc":"2.0","method":"getnetworkinfo","params":[],"id":1}' \
  "${RPC_URL}"

使用 Python 互動庫

以下是 Python 程式碼範例，展示如何與 Bitcoin Core RPC 互動：

#!/usr/bin/env python3
"""
比特幣 RPC 互動範例
依賴: pip install requests
"""
import requests
import json
from typing import Any, Dict, Optional

class BitcoinRPC:
    """比特幣 RPC 客戶端封裝"""

    def __init__(self, host: str = "127.0.0.1", port: int = 8332,
                 user: str = None, password: str = None):
        self.url = f"http://{host}:{port}"
        self.auth = (user, password) if user and password else None

    def call(self, method: str, params: list = None) -> Dict[str, Any]:
        """發送 RPC 請求"""
        payload = {
            "jsonrpc": "2.0",
            "method": method,
            "params": params or [],
            "id": 1
        }

        response = requests.post(
            self.url,
            json=payload,
            auth=self.auth,
            headers={"Content-Type": "application/json"}
        )

        if response.status_code != 200:
            raise Exception(f"HTTP Error: {response.status_code}")

        data = response.json()

        if "error" in data and data["error"]:
            raise Exception(f"RPC Error: {data['error']}")

        return data.get("result", {})

    def get_blockchain_info(self) -> Dict[str, Any]:
        """取得區塊鏈基本資訊"""
        return self.call("getblockchaininfo")

    def get_block(self, block_hash: str, verbosity: int = 2) -> Dict[str, Any]:
        """取得區塊詳情"""
        return self.call("getblock", [block_hash, verbosity])

    def get_block_hash(self, height: int) -> str:
        """根據區塊高度取得區塊雜湊"""
        return self.call("getblockhash", [height])

    def get_transaction(self, txid: str, verbose: bool = True) -> Dict[str, Any]:
        """取得交易詳情"""
        return self.call("getrawtransaction", [txid, verbose])

    def get_address_info(self, address: str) -> Dict[str, Any]:
        """取得地址資訊"""
        return self.call("getaddressinfo", [address])

    def list_unspent(self, min_conf: int = 1, max_conf: int = 9999999,
                     addresses: list = None) -> list:
        """列出未花費交易輸出"""
        params = [min_conf, max_conf]
        if addresses:
            params.append(addresses)
        return self.call("listunspent", params)

    def send_to_address(self, address: str, amount: float,
                        comment: str = "", comment_to: str = "") -> str:
        """發送比特幣到指定地址"""
        return self.call("sendtoaddress", [address, amount, comment, comment_to])


# 使用範例
if __name__ == "__main__":
    rpc = BitcoinRPC(user="bitcoinuser", password="your_secure_password_here")

    try:
        # 取得區塊鏈資訊
        info = rpc.get_blockchain_info()
        print(f"目前區塊高度: {info['blocks']}")
        print(f"目前區塊雜湊: {info['bestblockhash']}")
        print(f"難度: {info['difficulty']:.2f}")

        # 取得最新區塊
        latest_hash = rpc.get_block_hash(info['blocks'])
        latest_block = rpc.get_block(latest_hash)
        print(f"最新區塊包含 {len(latest_block['tx'])} 筆交易")

        # 取得節點網路資訊
        net_info = rpc.call("getnetworkinfo")
        print(f"連接的節點數: {net_info['connections']}")

    except Exception as e:
        print(f"錯誤: {e}")

核心 RPC 方法詳解

區塊鏈資訊查詢

getblockchaininfo 是了解比特幣網路狀態的首要方法：

# 取得完整區塊鏈資訊
info = rpc.get_blockchain_info()

# 解讀關鍵欄位
print(f"區塊鏈名稱: {info['chain']}")           # main, testnet, signet
print(f"目前高度: {info['blocks']}")           # 最新區塊編號
print(f"總區塊數: {info['headers']}")         # 已下載的區塊頭數量
print(f"難度: {info['difficulty']:,.2f}")     # 當前挖礦難度
print(f"軟分叉狀態: {info['softforks']}")     # 各升級的採用狀態
print(f"警告資訊: {info.get('warnings', [])}") # 網路警告

區塊資料結構詳解

比特幣區塊包含多個重要欄位，了解其結構對於開發區塊鏈應用至關重要：

# 取得特定區塊的完整資訊
block_hash = "0000000000000000000b72d3c9e2f3a1a7c19e4d5c6e7f8a9b0c1d2e3f4a5b6"
block = rpc.get_block(block_hash, verbosity=2)

# 區塊結構分析
block_structure = {
    "hash": block["hash"],                    # 區塊雜湊（工作量證明）
    "confirmations": block["confirmations"],  # 確認數
    "size": block["size"],                    # 區塊大小（bytes）
    "strippedsize": block["strippedsize"],    # 不含見證數據的大小
    "weight": block["weight"],                # 權重單位（WU）
    "height": block["height"],                # 區塊高度
    "version": block["version"],              # 區塊版本號
    "versionHex": block["versionHex"],        # 十六進制版本號
    "merkleroot": block["merkleroot"],       # Merkle 根雜湊
    "tx": block["tx"],                        # 交易 ID 列表
    "time": block["time"],                    # 區塊時間戳
    "mediantime": block["mediantime"],        # 中位時間
    "nonce": block["nonce"],                  # 隨機數
    "bits": block["bits"],                    # 難度目標
    "difficulty": block["difficulty"],         # 難度值
    "chainwork": block["chainwork"],          # 鏈工作總量
    "previousblockhash": block["previousblockhash"],  # 前一區塊雜湊
    "nextblockhash": block.get("nextblockhash")       # 下一區塊雜湊
}

print("區塊結構:", json.dumps(block_structure, indent=2))

交易查詢與驗證

比特幣交易查詢是錢包開發的核心功能：

def get_transaction_details(rpc, txid):
    """取得並分析交易詳情"""
    tx = rpc.get_transaction(txid, verbose=True)

    print(f"交易 ID: {tx['txid']}")
    print(f"雜湊: {tx['hash']}")
    print(f"大小: {tx['size']} bytes")
    print(f"權重: {tx['weight']} WU")
    print(f"版本: {tx['version']}")
    print(f"鎖定時間: {tx['locktime']}")

    # 輸入分析
    print("\n=== 輸入 (Inputs) ===")
    for i, vin in enumerate(tx['vin']):
        print(f"輸入 {i+1}:")
        print(f"  交易 ID: {vin.get('txid', 'coinbase')}")
        print(f"  輸出索引: {vin.get('vout', 'N/A')}")
        if 'prevout' in vin:
            print(f"  地址: {vin['prevout']['scriptPubKey'].get('address', 'N/A')}")
            print(f"  金額: {vin['prevout']['value'] / 1e8} BTC")

    # 輸出分析
    print("\n=== 輸出 (Outputs) ===")
    for i, vout in enumerate(tx['vout']):
        print(f"輸出 {i+1}:")
        print(f"  金額: {vout['value'] / 1e8} BTC")
        print(f"  腳本類型: {vout['scriptPubKey']['type']}")
        if 'addresses' in vout['scriptPubKey']:
            print(f"  地址: {vout['scriptPubKey']['addresses']}")

# 使用範例
# txid = "你的交易ID"
# get_transaction_details(rpc, txid)

記憶池監控

比特幣記憶池（Memory Pool）是未確認交易的臨時儲存區，監控記憶池狀態對於費用估算和交易優先級判斷非常重要：

def analyze_mempool(rpc):
    """分析記憶池狀態"""
    mempool = rpc.call("getmempoolinfo")

    print("=== 記憶池狀態 ===")
    print(f"交易數量: {mempool['size']}")
    print(f"記憶池大小: {mempool['bytes'] / 1e6:.2f} MB")
    print(f"總費用: {mempool['totalfee'] / 1e8:.4f} BTC")
    print(f"最低費用率: {mempool['mempoolminfee'] / 1e5:.4f} sat/vB")
    print(f"未確認總金額: {mempool['totalvalue'] / 1e8:.2f} BTC")

    # 取得記憶池中的大額交易
    raw_mempool = rpc.call("getrawmempool", [False, True])

    # 按費用率排序，找出最值得打包的交易
    sorted_txs = sorted(
        raw_mempool.items(),
        key=lambda x: x[1]['fees']['base'] / x[1]['vsize'],
        reverse=True
    )

    print("\n=== 費用率最高的 5 筆交易 ===")
    for txid, tx_data in sorted_txs[:5]:
        fee_rate = tx_data['fees']['base'] / tx_data['vsize']
        print(f"交易: {txid[:16]}...")
        print(f"  費用率: {fee_rate:.2f} sat/vB")
        print(f"  費用: {tx_data['fees']['base'] / 1e8:.6f} BTC")
        print(f"  大小: {tx_data['vsize']} vBytes")

# 使用範例
# analyze_mempool(rpc)

錢包操作

建立與管理錢包

Bitcoin Core 內建錢包功能，支援多錢包管理：

def wallet_operations(rpc):
    """錢包操作範例"""

    # 建立新錢包
    new_wallet = rpc.call("createwallet", ["my_wallet", True, True])
    print(f"錢包建立成功: {new_wallet}")

    # 列出所有錢包
    wallets = rpc.call("listwallets")
    print(f"現有錢包: {wallets}")

    # 取得錢包資訊
    wallet_info = rpc.call("getwalletinfo")
    print(f"餘額: {wallet_info['balance'] / 1e8} BTC")
    print(f"未確認餘額: {wallet_info['unconfirmed_balance'] / 1e8} BTC")
    print(f"錢包金鑰池大小: {wallet_info['keypoolsize']}")

    # 取得新地址
    new_address = rpc.call("getnewaddress", ["legacy", "bech32"])
    print(f"新地址 (bech32): {new_address}")

    # 取得地址標籤列表
    labels = rpc.call("listlabels")
    print(f"地址標籤: {labels}")

# 注意：錢包操作需要載入錢包
# rpc.call("loadwallet", ["wallet_name"])

UTXO 管理與交易構造

比特幣基於 UTXO（未花費交易輸出）模型，正確管理 UTXO 是開發比特幣應用的核心技能：

def list_and_analyze_utxo(rpc, min_amount_sats=10000):
    """列出並分析未花費交易輸出"""
    # 取得錢包中所有 UTXO
    unspent = rpc.list_unspent(0, 9999999)

    print(f"=== UTXO 分析 (共 {len(unspent)} 筆) ===\n")

    total_sats = 0
    utxo_by_type = {}

    for utxo in unspent:
        amount_sats = int(utxo['amount'] * 1e8)
        total_sats += amount_sats

        # 按腳本類型分類
        script_type = utxo['scriptPubKey']['type']
        if script_type not in utxo_by_type:
            utxo_by_type[script_type] = {'count': 0, 'total': 0}

        utxo_by_type[script_type]['count'] += 1
        utxo_by_type[script_type]['total'] += amount_sats

        # 顯示詳細資訊（針對大額 UTXO）
        if amount_sats >= min_amount_sats:
            print(f"UTXO: {utxo['txid'][:16]}...[{utxo['vout']}]")
            print(f"  金額: {amount_sats:,} sats ({amount_sats/1e8:.8f} BTC)")
            print(f"  類型: {script_type}")
            print(f"  地址: {utxo.get('address', 'N/A')}")
            print(f"  確認數: {utxo['confirmations']}")
            print()

    print(f"=== 總計 ===")
    print(f"總金額: {total_sats:,} sats ({total_sats/1e8:.8f} BTC)")
    print(f"\n=== 按類型分類 ===")
    for script_type, data in utxo_by_type.items():
        print(f"{script_type}: {data['count']} 筆, 共 {data['total']:,} sats")

交易構造與廣播

手動構造比特幣交易是一個複雜但強大的技能：

def create_and_send_transaction(rpc, to_address, amount_btc, fee_rate_sat_vb=10):
    """
    創建並發送交易
    注意：這是一個簡化範例，生產環境需要更完善的錯誤處理
    """
    # 步驟 1：收集可用 UTXO
    unspent = rpc.list_unspent(1, 9999999)

    if not unspent:
        raise Exception("沒有可用的 UTXO")

    # 步驟 2：選擇 UTXO（簡單策略：使用第一個足夠的 UTXO）
    selected_utxo = None
    target_sats = int(amount_btc * 1e8)
    fee_rate = fee_rate_sat_vb

    for utxo in unspent:
        utxo_sats = int(utxo['amount'] * 1e8)
        # 估算交易大小（簡化：每個輸入 148 bytes，每個輸出 34 bytes）
        estimated_vsize = 148 + 34 * 2  # 1 輸入 2 輸出
        estimated_fee = estimated_vsize * fee_rate

        if utxo_sats >= target_sats + estimated_fee:
            selected_utxo = utxo
            break

    if not selected_utxo:
        raise Exception("沒有足夠的 UTXO")

    # 步驟 3：構造交易輸入
    utxo_sats = int(selected_utxo['amount'] * 1e8)
    tx_inputs = [{
        "txid": selected_utxo['txid'],
        "vout": selected_utxo['vout']
    }]

    # 步驟 4：構造交易輸出
    # 計算實際費用（需要先創建原始交易才能知道準確大小）
    send_amount_sats = target_sats
    change_address = rpc.call("getrawchangeaddress", ["bech32"])
    change_sats = utxo_sats - send_amount_sats  # 暫不扣手續費

    tx_outputs = {
        to_address: send_amount_sats / 1e8,
        change_address: change_sats / 1e8
    }

    # 步驟 5：建立原始交易
    raw_tx = rpc.call("createrawtransaction", [tx_inputs, tx_outputs])
    print(f"原始交易: {raw_tx}")

    # 步驟 6：估算實際費用並調整
    # 這裡需要簽名後才能知道實際大小
    # 簡化處理：直接使用較高的費用率
    finalized_fee_rate = fee_rate_sat_vb * 1.5  # 增加 50% 作為緩衝

    # 步驟 7：簽名交易
    # 需要錢包解鎖
    signed_tx = rpc.call("signrawtransactionwithwallet", [raw_tx])
    print(f"簽名後交易: {signed_tx['hex']}")

    if not signed_tx['complete']:
        print(f"簽名警告: {signed_tx['errors']}")

    # 步驟 8：發送交易
    txid = rpc.call("sendrawtransaction", [signed_tx['hex']])
    print(f"交易已發送，TXID: {txid}")

    return txid

費用估算與交易優化

動態費用估算

比特幣網路費用波動劇烈，正確估算費用對於優化交易成本至關重要：

def estimate_smart_fee(rpc, target_confirm_blocks=6):
    """智慧費用估算"""
    # 使用 Bitcoin Core 的費用估算功能
    fee_estimate = rpc.call("estimatesmartfee", [target_confirm_blocks, "CONSERVATIVE"])

    if 'feerate' in fee_estimate:
        feerate_sat_vb = fee_estimate['feerate'] * 1e5  # 轉換為 sat/vB
        print(f"預計確認時間: {fee_estimate.get('blocks', 'N/A')} 區塊")
        print(f"建議費用率: {feerate_sat_vb:.2f} sat/vB")

        # 估算不同交易大小的費用
        for size in [250, 500, 1000, 2000]:
            fee_sats = int(size * feerate_sat_vb)
            print(f"  {size} vByte 交易: ~{fee_sats:,} sats ({fee_sats/1e8:.6f} BTC)")

        return feerate_sat_vb

    return None


def analyze_fee_market(rpc):
    """分析費用市場狀態"""
    mempool = rpc.call("getmempoolinfo")
    print("=== 費用市場分析 ===")

    # 取得不同費用率的交易統計
    fee_rate_buckets = [1, 5, 10, 20, 50, 100, 200]

    for i, bucket in enumerate(fee_rate_buckets):
        if i == len(fee_rate_buckets) - 1:
            above = rpc.call("getmempool", ["", True, f">{bucket}"])
            print(f">{bucket} sat/vB: {len(above)} 交易")
        else:
            next_bucket = fee_rate_buckets[i+1]
            # 取得該費用率區間的交易
            at_rate = rpc.call("getmempool", ["", True, f"{bucket}-{next_bucket}"])
            print(f"{bucket}-{next_bucket} sat/vB: {len(at_rate)} 交易")

鏈上數據分析範例

比特幣網路健康監控

def network_health_check(rpc):
    """比特幣網路健康狀態檢查"""
    print("=== Bitcoin Core 網路健康檢查 ===\n")

    # 1. 節點同步狀態
    blockchain = rpc.get_blockchain_info()
    sync_progress = (blockchain['blocks'] / blockchain['headers']) * 100
    print(f"1. 同步進度: {sync_progress:.2f}%")

    if blockchain['blocks'] < blockchain['headers']:
        print(f"   落後: {blockchain['headers'] - blockchain['blocks']} 區塊")

    # 2. 網路連接狀態
    netinfo = rpc.call("getnetworkinfo")
    print(f"2. P2P 連接數: {netinfo['connections']}")
    print(f"   版本: {netinfo['version']} ({netinfo['subversion']})")

    # 3. 記憶池狀態
    mempool = rpc.call("getmempoolinfo")
    print(f"3. 記憶池: {mempool['size']} 筆交易 ({mempool['bytes']/1e6:.2f} MB)")
    print(f"   最低費用率: {mempool['mempoolminfee']/1e5:.2f} sat/vB")

    # 4. 難度與算力估算
    print(f"4. 當前難度: {blockchain['difficulty']:,.2f}")
    # 難度每 2016 區塊調整，目標區塊時間 10 分鐘
    hashrate_eh = (blockchain['difficulty'] * 2**32) / 600 / 1e18
    print(f"   估算算力: {hashrate_eh:.2f} EH/s")

    # 5. 錢包餘額
    try:
        wallet = rpc.call("getwalletinfo")
        print(f"5. 錢包餘額: {wallet['balance']/1e8:.8f} BTC")
    except:
        print("5. 錢包未載入或不可用")

    # 6. 警告檢查
    warnings = blockchain.get('warnings', [])
    if warnings:
        print(f"\n⚠️  警告:")
        for w in warnings:
            print(f"   {w}")
    else:
        print(f"\n✅ 無網路警告")

比特幣供應量與減半追蹤

def analyze_supply_halving(rpc):
    """分析比特幣供應量與減半週期"""
    info = rpc.get_blockchain_info()
    current_height = info['blocks']

    # 比特幣減半高度
    halving_intervals = [
        (0, 50, "創世區塊"),
        (210000, 25, "第一次減半"),
        (420000, 12.5, "第二次減半"),
        (630000, 6.25, "第三次減半"),
        (840000, 3.125, "第四次減半")
    ]

    print("=== 比特幣供應分析 ===\n")
    print(f"當前區塊高度: {current_height:,}")

    # 計算當前區塊獎勵
    current_reward = 50
    for height, reward, desc in halving_intervals:
        if current_height >= height:
            current_reward = reward
            last_halving_height = height

    print(f"當前區塊獎勵: {current_reward} BTC")

    # 計算已發行的比特幣
    # 這是一個近似計算（每個區塊獎勵遞減）
    total_mined = 0
    prev_height = 0
    for height, reward, desc in halving_intervals:
        if current_height >= height:
            blocks = min(current_height, height) - prev_height
            total_mined += blocks * reward
            prev_height = height
            if current_height >= height:
                print(f"{desc}: {height:,} 區塊 - 獎勵 {reward} BTC")

    # 加上當前區塊的部分獎勵
    remaining_blocks = current_height - prev_height
    total_mined += remaining_blocks * current_reward

    print(f"\n預估已開採比特幣: {total_mined:,.2f} BTC")
    print(f"比特幣總量上限: 21,000,000 BTC")
    print(f"流通比例: {total_mined/21000000*100:.2f}%")

    # 預測下一次減半
    next_halving_height = ((current_height // 210000) + 1) * 210000
    blocks_until_halving = next_halving_height - current_height
    days_until_halving = blocks_until_halving * 10 / 1440  # 假設平均 10 分鐘/區塊

    print(f"\n下一次減半:")
    print(f"  預計高度: {next_halving_height:,}")
    print(f"  剩餘區塊: {blocks_until_halving:,}")
    print(f"  預估天數: {days_until_halving:.0f} 天")

安全性考量

RPC 介面安全最佳實踐

比特幣 RPC 介面直接控制比特幣節點和錢包，安全性至關重要：

# 1. 永遠不要將 RPC 暴露在公網
# 錯誤示範
rpcbind=0.0.0.0        # 允許任何 IP 訪問
rpcallowip=*           # 允許任何 IP

# 正確做法
rpcbind=127.0.0.1       # 只允許本機訪問
# 如需遠程訪問，使用 SSH 隧道或 VPN

# 2. 使用強密碼
# 生成強密碼
openssl rand -base64 32

# 3. 限制 RPC 使用者權限
# 在 bitcoin.conf 中
# 只允許讀取的方法（錢包唯讀模式）
# wallet=              # 空值表示不載入任何錢包

# 4. 啟用交易簽名隔離
# 對於需要簽名的操作，使用硬體錢包或離線簽名

防火牆與網路隔離

# 使用 UFW 防火牆
sudo ufw enable
sudo ufw allow ssh
sudo ufw allow from 127.0.0.1 to any port 8332 proto tcp
sudo ufw status

# 或者使用 iptables
sudo iptables -A INPUT -i lo -j ACCEPT
sudo iptables -A INPUT -p tcp --dport 8332 -s 127.0.0.1 -j ACCEPT
sudo iptables -A INPUT -p tcp --dport 8332 -j DROP

失敗模式與故障排除

常見錯誤與解決方案

# 常見 RPC 錯誤處理範例
def handle_rpc_errors(func):
    """RPC 錯誤處理裝飾器"""
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except requests.exceptions.ConnectionError:
            raise Exception("無法連接比特幣節點，請確認 bitcoind 是否正在運行")
        except requests.exceptions.Timeout:
            raise Exception("RPC 請求逾時，節點可能正在落後或忙碌")
        except Exception as e:
            error_msg = str(e)
            # 解析比特幣 RPC 錯誤碼
            if "-4" in error_msg or "Insufficient funds" in error_msg:
                raise Exception("餘額不足")
            elif "-5" in error_msg or "Invalid address" in error_msg:
                raise Exception("無效的比特幣地址")
            elif "-6" in error_msg or "Insufficient funds" in error_msg:
                raise Exception("費用不足")
            elif "-8" in error_msg:
                raise Exception("參數錯誤，請檢查輸入格式")
            elif "-26" in error_msg or "dust" in error_msg:
                raise Exception("金額過小（dust），無法發送")
            elif "-25" in error_msg:
                raise Exception("交易被拒絕，可能金額過低")
            else:
                raise Exception(f"比特幣 RPC 錯誤: {error_msg}")

    return wrapper


# 節點健康檢查
def check_node_health(rpc):
    """檢查節點健康狀態"""
    try:
        # 測試基本 RPC 連接
        info = rpc.get_blockchain_info()

        # 檢查是否完全同步
        if info['blocks'] < info['headers']:
            return False, f"節點正在同步: {info['blocks']}/{info['headers']}"

        # 檢查節點連接數
        netinfo = rpc.call("getnetworkinfo")
        if netinfo['connections'] < 1:
            return False, "節點沒有 P2P 連接"

        return True, "節點正常"

    except Exception as e:
        return False, str(e)

進階應用

建立比特幣價格監控系統

#!/usr/bin/env python3
"""
比特幣區塊鏈監控系統範例
監控新區塊、交易和錢包餘額變化
"""
import time
import threading
from datetime import datetime

class BitcoinMonitor:
    """比特幣區塊鏈監控器"""

    def __init__(self, rpc):
        self.rpc = rpc
        self.last_block_height = 0
        self.running = False
        self.last_mempool_size = 0

    def start(self, callback=None):
        """啟動監控"""
        self.running = True
        self.last_block_height = self.rpc.get_blockchain_info()['blocks']
        self.last_mempool_size = self.rpc.get_mempoolinfo()['size']

        # 啟動監控執行緒
        monitor_thread = threading.Thread(target=self._monitor_loop, args=(callback,))
        monitor_thread.daemon = True
        monitor_thread.start()

    def stop(self):
        """停止監控"""
        self.running = False

    def _monitor_loop(self, callback):
        """監控循環"""
        while self.running:
            try:
                # 檢查新區塊
                current_height = self.rpc.get_blockchain_info()['blocks']

                if current_height > self.last_block_height:
                    new_blocks = current_height - self.last_block_height

                    for height in range(self.last_block_height + 1, current_height + 1):
                        block_hash = self.rpc.get_block_hash(height)
                        block = self.rpc.get_block(block_hash)

                        if callback:
                            callback('new_block', {
                                'height': height,
                                'hash': block_hash,
                                'tx_count': len(block['tx']),
                                'timestamp': block['time']
                            })

                    self.last_block_height = current_height

                # 檢查記憶池變化
                mempool = self.rpc.get_mempoolinfo()
                if mempool['size'] != self.last_mempool_size:
                    if callback:
                        callback('mempool_change', {
                            'old_size': self.last_mempool_size,
                            'new_size': mempool['size']
                        })
                    self.last_mempool_size = mempool['size']

            except Exception as e:
                print(f"監控錯誤: {e}")

            time.sleep(10)  # 每 10 秒檢查一次

# 使用範例
def on_event(event_type, data):
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(f"[{timestamp}] {event_type}: {data}")

# monitor = BitcoinMonitor(rpc)
# monitor.start(callback=on_event)
# time.sleep(60)
# monitor.stop()

總結

Bitcoin Core RPC 提供了比特幣網路最完整、最權威的互動介面。透過本文的範例，你應該能夠：

• 正確建置 Bitcoin Core 環境並設定安全的 RPC 認證
• 使用多種方式（CLI、cURL、Python）與比特幣節點互動
• 查詢區塊、交易、記憶池等鏈上數據
• 執行錢包操作，包括地址管理和 UTXO 處理
• 構造並發送比特幣交易
• 估算費用並優化交易成本
• 監控比特幣網路狀態
• 實作基本的安全性最佳實踐

比特幣 RPC 介面功能強大，本文的範例僅涵蓋基礎應用場景。更進階的應用包括：構建比特幣區塊瀏覽器、開發比特幣錢包服務、實現交易加速器、創建比特幣支付閘道等。建議讀者進一步研究 Bitcoin Core 的完整 RPC 文件，探索更多應用可能。

相關主題

• Bitcoin Core 節點運作
• 記憶池與交易費用
• UTXO 模型詳解
• 比特幣腳本語言入門