#!/usr/bin/env python3 """A-05: 公開dummy鍵のみを使う、Core 31.1の隔離regtest。 実資金には絶対に使わない。既存datadir/財布/設定を受け取らない。 Coreが一時RPC cookieを作成し、bitcoin-cliが内部で利用する。 この認証の作成・利用を含む実行承認を得てから起動する。 依存: Python 3.11+、cryptography、公式Core 31.1。 """ from __future__ import annotations import argparse from contextlib import contextmanager import hashlib import hmac import json import os from pathlib import Path import re import shutil import socket import subprocess import tempfile import time from decimal import Decimal from cryptography.hazmat.primitives.asymmetric import ec from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat ORDER = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141 ALPHABET = '123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz' PUBLIC_LABELS = ('A05-public-dummy-A', 'A05-public-dummy-B', 'A05-public-dummy-C', 'A05-public-dummy-miner') def b58check(payload): raw = payload + hashlib.sha256(hashlib.sha256(payload).digest()).digest()[:4] value, encoded = int.from_bytes(raw, 'big'), '' while value: value, remainder = divmod(value, 58) encoded = ALPHABET[remainder] + encoded return '1' * (len(raw) - len(raw.lstrip(b'\0'))) + encoded def public_key(scalar): return ec.derive_private_key(scalar, ec.SECP256K1()).public_key().public_bytes(Encoding.X962, PublicFormat.CompressedPoint) def dummy_root(label): # すべての入力は公開ラベル。現実のseedを受け取らない。 seed = hashlib.sha256(label.encode('ascii')).digest() digest = hmac.new(b'Bitcoin seed', seed, hashlib.sha512).digest() scalar = int.from_bytes(digest[:32], 'big') if not 0 < scalar < ORDER: raise ValueError('invalid dummy master') return scalar, digest[32:] def extended_key(root, private): scalar, chain = root version = bytes.fromhex('04358394' if private else '043587cf') key = b'\0' + scalar.to_bytes(32, 'big') if private else public_key(scalar) return b58check(version + b'\0' * 9 + chain + key) def child(root, index): scalar, chain = root digest = hmac.new(chain, public_key(scalar) + index.to_bytes(4, 'big'), hashlib.sha512).digest() left = int.from_bytes(digest[:32], 'big') derived = (left + scalar) % ORDER if left >= ORDER or derived == 0: raise ValueError('invalid dummy child') return derived, digest[32:] def dummy_wif(root, branch, index=0): scalar, _ = child(child(root, branch), index) return b58check(b'\xef' + scalar.to_bytes(32, 'big') + b'\x01') def descriptor(roots, branch, private_indices=(), wrapper='wsh', members=(0, 1, 2)): keys = [extended_key(roots[i], i in private_indices) + f'/{branch}/*' for i in members] inner = 'sortedmulti(2,' + ','.join(keys) + ')' return f'sh(wsh({inner}))' if wrapper == 'sh-wsh' else f'wsh({inner})' class LabError(Exception): def __init__(self, message, code=None, checksum_error=False): super().__init__(message) self.code = code self.checksum_error = checksum_error def verify_temp_path(path, parent, prefix): if path.is_symlink() or getattr(path, 'is_junction', lambda: False)(): raise LabError('temporary directory redirected') if path.resolve().parent != parent or not path.name.startswith(prefix): raise LabError('temporary directory boundary mismatch') @contextmanager def isolated_directory(prefix): parent = Path(tempfile.gettempdir()).resolve() path = Path(tempfile.mkdtemp(prefix=prefix, dir=parent)).resolve() verify_temp_path(path, parent, prefix) try: yield path except BaseException: # 停止を確認できない失敗時は保存。cookie等の内容を出力しない。 raise else: verify_temp_path(path, parent, prefix) shutil.rmtree(path) def is_checksum_rejection(error): return error.code == -5 and error.checksum_error def run(bin_dir, output, approved): if not approved: raise LabError('execution authorization flag required') suffix = '.exe' if os.name == 'nt' else '' node_path = bin_dir / ('bitcoind' + suffix) cli_path = bin_dir / ('bitcoin-cli' + suffix) # Windows版ではversion照会でもdatadir初期化が先行し得る。 with isolated_directory('a05-version-') as version_temp: version = subprocess.run([str(node_path), f'-datadir={Path(version_temp).resolve()}', '-regtest', '-version'], capture_output=True, text=True, check=True, timeout=15).stdout.splitlines()[0] if 'v31.1.0' not in version: raise LabError('Core version mismatch') binary_hashes = {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in (node_path, cli_path)} started = time.monotonic() roots = [dummy_root(label) for label in PUBLIC_LABELS] with isolated_directory('a05-regtest-') as temp: datadir = Path(temp).resolve() with socket.socket() as sock: sock.bind(('127.0.0.1', 0)) port = sock.getsockname()[1] common = ['-regtest', f'-datadir={datadir}', f'-rpcport={port}'] flags = ['-server=1', '-daemon=0', '-rpcbind=127.0.0.1', '-rpcallowip=127.0.0.1', '-networkactive=0', '-listen=0', '-maxconnections=0', '-dnsseed=0', '-fixedseeds=0', '-discover=0', '-natpmp=0', '-dns=0', '-listenonion=0', '-dbcache=64', '-printtoconsole=0', '-persistmempool=0'] process = subprocess.Popen([str(node_path), *common, *flags], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) def rpc(method, *args, wallet=None): # CLIのみが内部でcookieを扱う。値をPythonで読まない。 command = [str(cli_path), *common] if wallet: command += [f'-rpcwallet={wallet}'] command += [method, *[json.dumps(a, separators=(',', ':')) if isinstance(a, (dict, list, bool)) else str(a) for a in args]] response = subprocess.run(command, capture_output=True, text=True, timeout=90) if response.returncode: match = re.search(r'error code:\s*(-?\d+)', response.stderr) code = int(match.group(1)) if match else None raise LabError(f'RPC {method} failed ({code})', code, 'checksum' in response.stderr.lower()) value = response.stdout.strip() try: return json.loads(value) except json.JSONDecodeError: return value def check(condition, label): if not condition: raise LabError('assertion: ' + label) def with_checksum(body): return body + '#' + rpc('getdescriptorinfo', body)['checksum'] def make_watch(name, bodies): rpc('createwallet', name, True, True) entries = [{'desc': with_checksum(body), 'timestamp': 0, 'range': [0, 2], 'internal': internal} for body, internal in bodies] imported = rpc('importdescriptors', entries, wallet=name) check(all(row.get('success') for row in imported), 'descriptor import ' + name) def balance(name): balances = rpc('getbalances', wallet=name) return sum((Decimal(str(balances.get(k, {}).get('trusted', 0))) for k in ('mine', 'watchonly')), Decimal(0)) try: deadline = time.monotonic() + 60 while True: try: chain = rpc('getblockchaininfo') break except LabError: if process.poll() is not None or time.monotonic() > deadline: raise LabError('node readiness failed') time.sleep(0.25) check(chain['chain'] == 'regtest', 'regtest chain') isolated = rpc('getnetworkinfo') check(not isolated['networkactive'] and isolated['connections'] == 0, 'initial isolation') receive_body = descriptor(roots, 0) change_body = descriptor(roots, 1) receive_desc, change_desc = map(with_checksum, (receive_body, change_body)) receive = rpc('deriveaddresses', receive_desc, [0, 2]) change = rpc('deriveaddresses', change_desc, [0, 2]) miner_desc = with_checksum('wpkh(' + extended_key(roots[3], False) + '/0/*)') miner = rpc('deriveaddresses', miner_desc, [0, 0])[0] # Miningは公開dummyアドレスへ。鍵を生成する財布を作らない。 rpc('generatetoaddress', 101, receive[0]) # 101個目のcoinbaseは未成熟。最初の50 BTCから0.01 BTCの試験出力を作る。 make_watch('full', [(receive_body, False), (change_body, True)]) make_watch('receiveonly', [(receive_body, False)]) mature = [u for u in rpc('listunspent', 101, wallet='full') if Decimal(str(u['amount'])) == Decimal(50)] check(len(mature) >= 1, 'mature coinbase') source = mature[0] raw = rpc('createrawtransaction', [{'txid': source['txid'], 'vout': source['vout']}], [{receive[1]: 0.01}, {miner: 49.98999}]) keys0 = [dummy_wif(roots[i], 0) for i in (0, 1)] prev = [{'txid': source['txid'], 'vout': source['vout'], 'scriptPubKey': source['scriptPubKey'], 'amount': source['amount'], 'witnessScript': rpc('getaddressinfo', receive[0], wallet='full')['hex']}] funding = rpc('signrawtransactionwithkey', raw, keys0, prev) check(funding['complete'], 'funding signature') fund_txid = rpc('sendrawtransaction', funding['hex']) rpc('generatetoaddress', 1, miner) candidates = [u for u in rpc('listunspent', 1, wallet='full') if u['txid'] == fund_txid and Decimal(str(u['amount'])) == Decimal('0.01')] check(len(candidates) == 1, 'test output found') output_coin = candidates[0] receive_control = [u for u in rpc('listunspent', 1, wallet='receiveonly') if u['txid'] == fund_txid and Decimal(str(u['amount'])) == Decimal('0.01')] check(len(receive_control) == 1, 'receive-only positive control') prev = [{'txid': output_coin['txid'], 'vout': output_coin['vout'], 'scriptPubKey': output_coin['scriptPubKey'], 'amount': output_coin['amount'], 'witnessScript': rpc('getaddressinfo', receive[1], wallet='full')['hex']}] spending = rpc('createrawtransaction', [{'txid': output_coin['txid'], 'vout': output_coin['vout']}], [{miner: 0.004}, {change[0]: 0.00599}]) rows, signed_for_change = [], None for case, indices in [('AB', (0, 1)), ('AC', (0, 2)), ('BC', (1, 2)), ('A_only', (0,)), ('public_only', ())]: # それぞれ新しいwatch walletで構成情報を復元。鍵のseedは公開dummy。 reconstructed = rpc('getdescriptorinfo', descriptor(roots, 0, indices))['descriptor'] observed_addresses = rpc('deriveaddresses', reconstructed, [0, 2]) case_wallet = 'case_' + case.lower() make_watch(case_wallet, [(reconstructed.split('#')[0], False), (change_body, True)]) found = [u for u in rpc('listunspent', 1, wallet=case_wallet) if u['txid'] == output_coin['txid'] and u['vout'] == output_coin['vout']] check(observed_addresses == receive and len(found) == 1, 'independent reconstruction ' + case) check(Decimal(str(found[0]['amount'])) == Decimal('0.01'), 'independent amount ' + case) signed = rpc('signrawtransactionwithkey', spending, [dummy_wif(roots[i], 0, 1) for i in indices], prev) accepted = rpc('testmempoolaccept', [signed['hex']])[0]['allowed'] check(signed['complete'] == (len(indices) == 2), 'signature case ' + case) check(accepted == (len(indices) == 2), 'acceptance case ' + case) rows.append({'case': case, 'address_matches': observed_addresses == receive, 'test_output_found_btc': str(found[0]['amount']), 'signature_complete': signed['complete'], 'mempool_acceptance_tested': True, 'mempool_accepted': accepted}) if case == 'AB': signed_for_change = signed['hex'] for name, body in [('missing_C_substitute_2of2', descriptor(roots, 0, members=(0, 1))), ('wrong_script', descriptor(roots, 0, wrapper='sh-wsh')), ('wrong_branch', descriptor(roots, 2))]: addresses = rpc('deriveaddresses', with_checksum(body), [0, 2]) make_watch(name, [(body, False)]) check(addresses[1] != receive[1] and balance(name) == 0, name) rows.append({'case': name, 'address_matches': False, 'found_balance_btc': str(balance(name)), 'signature_complete': None, 'mempool_accepted': None}) bad = receive_desc[:-1] + ('q' if receive_desc[-1] != 'q' else 'p') check(rpc('deriveaddresses', receive_desc, [0, 2]) == receive, 'checksum positive control before') rejected = False try: rpc('deriveaddresses', bad, [0, 2]) except LabError as error: if not is_checksum_rejection(error): raise rejected = True check(rejected, 'checksum rejection') check(rpc('deriveaddresses', receive_desc, [0, 2]) == receive, 'checksum positive control after') # コイン生成分の残高があるため、changeの個別出力のみを比較する。 change_txid = rpc('sendrawtransaction', signed_for_change) rpc('generatetoaddress', 1, miner) def change_seen(name): return sum((Decimal(str(u['amount'])) for u in rpc('listunspent', 1, wallet=name) if u['txid'] == change_txid), Decimal(0)) full_change, receive_change = change_seen('full'), change_seen('receiveonly') check(full_change == Decimal('0.00599') and receive_change == 0, 'change omission') final = rpc('getnetworkinfo') check(not final['networkactive'] and final['connections'] == 0, 'final isolation') result = {'schema_version': 1, 'status': 'PASS', 'bitcoin_core': version, 'binary_sha256': binary_hashes, 'network': 'regtest', 'range': [0, 2], 'timestamp': 0, 'dummy_labels': PUBLIC_LABELS, 'shared_test_output_found_btc': str(output_coin['amount']), 'receive_only_positive_control': True, 'cases': rows, 'checksum_rejected': rejected, 'change_output_btc': str(full_change), 'receive_only_change_seen_btc': str(receive_change), 'networkactive': final['networkactive'], 'connections': final['connections'], 'duration_seconds': round(time.monotonic() - started, 2), 'limitations': ['Core only; no hardware or vendor interoperability test', 'Missing C tests a 2-of-2 substitution, not every possible recovery source', 'Found test output is separated from generated coinbase balances', 'Independent public watch wallets; signatures use derived dummy WIF, not device seed entry', 'Mempool acceptance test does not broadcast the transaction']} finally: try: rpc('stop') except Exception: process.terminate() try: process.wait(timeout=30) except subprocess.TimeoutExpired: process.kill() process.wait(timeout=10) # 停止と一時directory削除後にだけ成功結果を保存する。 output.write_text(json.dumps(result, indent=2) + '\n', encoding='utf-8') return result if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--bin-dir', required=True, type=Path) parser.add_argument('--output', required=True, type=Path) parser.add_argument('--approved-ephemeral-rpc', action='store_true') args = parser.parse_args() try: result = run(args.bin_dir.resolve(), args.output.resolve(), args.approved_ephemeral_rpc) print(json.dumps({'status': result['status'], 'cases': len(result['cases'])})) except Exception as error: print(json.dumps({'status': 'ERROR', 'error_type': type(error).__name__, 'detail': str(error) if isinstance(error, LabError) else 'sanitized'})) raise SystemExit(1)