Multisig - 複数の鍵で承認する仕組み

#!/usr/bin/env python3
"""Local regtest only. 2-of-3 P2WSH and 3-of-3 MuSig2 with Bitcoin Core 32.1.

Run through run.sh: the container has no network (--network none) and every
key is generated by a throwaway regtest wallet that is deleted afterwards.
"""
import json
import os
import shutil
import subprocess
import time

BIN = os.environ.get('BITCOIN_BIN', '/opt/bitcoin/bin')
DATA = '/tmp/regtest'


def cli(*args, wallet=None):
    cmd = [f'{BIN}/bitcoin-cli', '-regtest', f'-datadir={DATA}']
    if wallet:
        cmd.append(f'-rpcwallet={wallet}')
    r = subprocess.run(cmd + [str(a) for a in args], capture_output=True, text=True)
    if r.returncode:
        raise RuntimeError(r.stderr.strip().splitlines()[-1])
    try:
        return json.loads(r.stdout)
    except json.JSONDecodeError:
        return r.stdout.strip()


def start():
    shutil.rmtree(DATA, ignore_errors=True)
    os.makedirs(DATA)
    subprocess.check_call([f'{BIN}/bitcoind', '-regtest', f'-datadir={DATA}', '-daemon',
                           '-listen=0', '-dnsseed=0', '-fixedseeds=0', '-fallbackfee=0.0001'],
                          stdout=subprocess.DEVNULL)
    for _ in range(300):
        if os.path.exists(f'{DATA}/regtest/.cookie'):
            break
        time.sleep(0.1)
    cli('-rpcwait', 'getblockcount')


# --- helpers to rewrite the witness of input 0 (segwit serialization) ---
def varint(n):
    return bytes([n]) if n < 0xfd else b'\xfd' + n.to_bytes(2, 'little')


def read_varint(b, i):
    return (b[i], i + 1) if b[i] < 0xfd else (int.from_bytes(b[i + 1:i + 3], 'little'), i + 3)


def with_witness(tx_hex, items):
    b = bytes.fromhex(tx_hex)
    assert b[4:6] == b'\x00\x01'  # segwit marker and flag
    n_in, i = read_varint(b, 6)
    assert n_in == 1
    i += 36
    length, i = read_varint(b, i)
    i += length + 4
    n_out, i = read_varint(b, i)
    for _ in range(n_out):
        i += 8
        length, i = read_varint(b, i)
        i += length
    j = i
    count, j = read_varint(b, j)
    for _ in range(count):
        length, j = read_varint(b, j)
        j += length
    witness = varint(len(items)) + b''.join(varint(len(x)) + x for x in map(bytes.fromhex, items))
    return (b[:i] + witness + b[j:]).hex()


def verdict(raw, miner):
    mempool = cli('testmempoolaccept', json.dumps([raw]))[0]
    if mempool['allowed']:
        return 'accepted'
    try:
        cli('generateblock', miner, json.dumps([raw]))
        return 'BLOCK ACCEPTED (unexpected)'
    except RuntimeError as e:  # consensus check when the tx is put in a block
        msg = str(e)
        return 'block rejected: ' + msg[msg.index('(') + 1:msg.index(')')]


def spend(wallet, dest, change):
    opts = json.dumps({'changeAddress': change})
    return cli('-named', 'walletcreatefundedpsbt', 'inputs=[]',
               f'outputs=[{{"{dest}":0.5}}]', f'options={opts}', wallet=wallet)['psbt']


def main():
    start()
    for w in ('miner', 'a', 'b', 'c'):
        cli('-named', 'createwallet', f'wallet_name={w}')
    miner = cli('getnewaddress', wallet='miner')
    cli('generatetoaddress', 101, miner)
    dest = cli('getnewaddress', wallet='miner')

    # One key per signer wallet; only public keys leave each wallet.
    key = {}
    for w in 'abc':
        info = cli('getaddressinfo', cli('getnewaddress', '', 'bech32', wallet=w), wallet=w)
        path = info['hdkeypath'][2:].replace('h', "'")
        key[w] = f"[{info['hdmasterfingerprint']}/{path}]{info['pubkey']}"

    # ---- 1. 2-of-3 P2WSH ----
    desc = cli('getdescriptorinfo', f"wsh(multi(2,{key['a']},{key['b']},{key['c']}))")['descriptor']
    cli('-named', 'createwallet', 'wallet_name=coord', 'disable_private_keys=true', 'blank=true')
    cli('importdescriptors', json.dumps([{'desc': desc, 'timestamp': 'now'}]), wallet='coord')
    addr = cli('deriveaddresses', desc)[0]
    cli('sendtoaddress', addr, 1, wallet='miner')
    cli('generatetoaddress', 1, miner)
    psbt = spend('coord', dest, addr)
    signed = {w: cli('walletprocesspsbt', psbt, wallet=w)['psbt'] for w in 'abc'}

    print('== 2-of-3 P2WSH')
    ws_hex = cli('decodepsbt', psbt)['inputs'][0]['witness_script']['hex']
    asm = cli('decodescript', ws_hex)['asm'].split()
    print('witnessScript:', ' '.join(x if x.startswith('OP_') or len(x) < 8 else x[:6] + '..' for x in asm))
    print('signed by a only     -> complete =', cli('finalizepsbt', signed['a'])['complete'])
    full = {}
    for pair in ('ab', 'ac', 'bc'):
        fin = cli('finalizepsbt', cli('combinepsbt', json.dumps([signed[p] for p in pair])))
        full[pair] = fin['hex']
        print(f'signed by {pair[0]} and {pair[1]}  -> complete = {fin["complete"]},',
              'mempool =', verdict(fin['hex'], miner))
    tx = cli('decoderawtransaction', full['ab'])
    wit = tx['vin'][0]['txinwitness']
    print('witness items (bytes):', [len(x) // 2 for x in wit], '| weight =', tx['weight'])
    sig_a, sig_b, ws = wit[1], wit[2], wit[3]
    cases = [
        ('only sig_a (dummy, sig_a)', ['', sig_a, ws]),
        ('sig_b before sig_a', ['', sig_b, sig_a, ws]),
        ('dummy = 0x01', ['01', sig_a, sig_b, ws]),
        ('no dummy', [sig_a, sig_b, ws]),
    ]
    for label, items in cases:
        print(f'{label:26} ->', verdict(with_witness(full['ab'], items), miner))

    # ---- 2. 3-of-3 MuSig2 key path (BIP-327 aggregation inside Bitcoin Core) ----
    print('== 3-of-3 MuSig2 (tr(musig(...)))')
    pub = cli('getdescriptorinfo', f"tr(musig({key['a']},{key['b']},{key['c']}))")['descriptor']
    master = {}
    for w in 'abc':
        d = [x['desc'] for x in cli('listdescriptors', 'true', wallet=w)['descriptors']
             if x['desc'].startswith('wpkh(') and not x['internal']][0]
        master[w] = d[5:d.index('/')]  # tprv of the throwaway regtest wallet
    for w in 'abc':
        parts = [f"{master[o]}/{key[o].split('/', 1)[1].split(']')[0]}" if o == w else key[o] for o in 'abc']
        priv = f"tr(musig({','.join(parts)}))"
        priv += '#' + cli('getdescriptorinfo', priv)['checksum']
        cli('-named', 'createwallet', f'wallet_name=m{w}', 'blank=true')
        cli('importdescriptors', json.dumps([{'desc': priv, 'timestamp': 'now'}]), wallet=f'm{w}')
    cli('-named', 'createwallet', 'wallet_name=mcoord', 'disable_private_keys=true', 'blank=true')
    cli('importdescriptors', json.dumps([{'desc': pub, 'timestamp': 'now'}]), wallet='mcoord')
    maddr = cli('deriveaddresses', pub)[0]
    cli('sendtoaddress', maddr, 1, wallet='miner')
    cli('generatetoaddress', 1, miner)
    psbt = spend('mcoord', dest, maddr)
    round1 = cli('combinepsbt', json.dumps([cli('walletprocesspsbt', psbt, wallet=f'm{w}')['psbt'] for w in 'abc']))
    inp = cli('decodepsbt', round1)['inputs'][0]
    print('round 1: pubnonces =', len(inp['musig2_pubnonces']), ', partial sigs =', len(inp.get('musig2_partial_sigs', [])))
    round2 = cli('combinepsbt', json.dumps([cli('walletprocesspsbt', round1, wallet=f'm{w}')['psbt'] for w in 'abc']))
    inp = cli('decodepsbt', round2)['inputs'][0]
    print('round 2: partial sigs =', len(inp['musig2_partial_sigs']))
    fin = cli('finalizepsbt', round2)
    mtx = cli('decoderawtransaction', fin['hex'])
    print('complete =', fin['complete'], ', mempool =', verdict(fin['hex'], miner))
    print('witness items (bytes):', [len(x) // 2 for x in mtx['vin'][0]['txinwitness']], '| weight =', mtx['weight'])

    # ---- 3. single-key P2TR with the same shape (1 input, 2 outputs) ----
    print('== single-key P2TR for comparison')
    cli('-named', 'createwallet', 'wallet_name=solo')
    saddr = cli('getnewaddress', '', 'bech32m', wallet='solo')
    cli('sendtoaddress', saddr, 1, wallet='miner')
    cli('generatetoaddress', 1, miner)
    sp = cli('walletprocesspsbt', spend('solo', dest, saddr), wallet='solo')
    stx = cli('decoderawtransaction', cli('finalizepsbt', sp['psbt'])['hex'])
    print('witness items (bytes):', [len(x) // 2 for x in stx['vin'][0]['txinwitness']], '| weight =', stx['weight'])
    for label, a in (('musig2', maddr), ('single', saddr)):
        spk = cli('getaddressinfo', a, wallet='mcoord' if label == 'musig2' else 'solo')['scriptPubKey']
        print(f'{label} scriptPubKey: {spk[:4]} + {len(spk) // 2 - 2}-byte key')
    cli('stop')


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""Local anvil only. 2-of-3 Safe (safe-smart-account v1.5.0) and execTransaction.

Run through run.sh. Keys are Anvil's public test mnemonic; never reuse them.
Owners: two EOAs and StudySigner (an ERC-1271 contract written for this lab).
"""
import json
import os
import socket
import subprocess
import time
import urllib.request

BIN = os.environ.get('ETH_BIN', '/opt/eth')
BUILD = json.load(open(os.environ['BUILD']))['contracts']
MNEMONIC = 'test test test test test test test test test test test junk'  # Anvil default, public
ZERO = '0x' + '00' * 20
SAFE_TX = ('SafeTx(address to,uint256 value,bytes data,uint8 operation,uint256 safeTxGas,'
           'uint256 baseGas,uint256 gasPrice,address gasToken,address refundReceiver,uint256 nonce)')
EXEC = 'execTransaction(address,uint256,bytes,uint8,uint256,uint256,uint256,address,address,bytes)'


def cast(*args):
    return subprocess.check_output([f'{BIN}/cast', *map(str, args)], text=True).strip()


with socket.socket() as s:
    s.bind(('127.0.0.1', 0))
    PORT = s.getsockname()[1]


def rpc(method, *params):
    body = json.dumps({'jsonrpc': '2.0', 'id': 1, 'method': method, 'params': params}).encode()
    req = urllib.request.Request(f'http://127.0.0.1:{PORT}', body, {'Content-Type': 'application/json'})
    with urllib.request.urlopen(req, timeout=10) as r:
        res = json.load(r)
    if 'error' in res:
        raise RuntimeError(res['error'])
    return res['result']


def send(sender, data, to=None, value=0):
    tx = {'from': sender, 'data': data, 'value': hex(value), 'gas': hex(8_000_000), 'gasPrice': hex(2 * 10**9)}
    if to:
        tx['to'] = to
    h = rpc('eth_sendTransaction', tx)
    for _ in range(200):
        receipt = rpc('eth_getTransactionReceipt', h)
        if receipt:
            return receipt
        time.sleep(0.05)
    raise TimeoutError(h)


def call(to, data, sender=None):
    """eth_call; returns (result, None) or (None, revert message)."""
    try:
        return rpc('eth_call', {'to': to, 'data': data, **({'from': sender} if sender else {})}, 'latest'), None
    except RuntimeError as e:
        return None, e.args[0]['message']


def deploy(sender, name, args=''):
    art = next(v for k, v in BUILD.items() if k.endswith(':' + name))
    receipt = send(sender, '0x' + art['bin'] + args)
    assert int(receipt['status'], 16) == 1
    return receipt['contractAddress']


def main():
    node = subprocess.Popen([f'{BIN}/anvil', '--host', '127.0.0.1', '--port', str(PORT), '--chain-id', '31337',
                             '--hardfork', 'prague', '--timestamp', '1700000000', '--accounts', '5', '--quiet'],
                            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    try:
        for _ in range(200):
            try:
                accounts = rpc('eth_accounts')
                break
            except OSError:
                time.sleep(0.05)
        relayer, o1, o2, admin, receiver = accounts
        pk = {a: cast('wallet', 'private-key', '--mnemonic', MNEMONIC, '--mnemonic-index', i)
              for i, a in enumerate(accounts)}

        singleton = deploy(relayer, 'Safe')
        factory = deploy(relayer, 'SafeProxyFactory')
        signer = deploy(relayer, 'StudySigner', cast('abi-encode', 'f(address)', admin)[2:])
        owners = [o1, o2, signer]
        setup = cast('calldata', 'setup(address[],uint256,address,bytes,address,address,uint256,address)',
                     '[' + ','.join(owners) + ']', 2, ZERO, '0x', ZERO, ZERO, 0, ZERO)
        create = cast('calldata', 'createProxyWithNonce(address,bytes,uint256)', singleton, setup, 0)
        safe = '0x' + call(factory, create, relayer)[0][-40:]
        assert int(send(relayer, create, factory)['status'], 16) == 1
        send(relayer, '0x', safe, 10**18)
        threshold = int(call(safe, cast('calldata', 'getThreshold()'))[0], 16)
        print(f'Safe proxy: 3 owners (2 EOAs + 1 ERC-1271 contract), threshold = {threshold}')

        def nonce():
            return int(call(safe, cast('calldata', 'nonce()'))[0], 16)

        def safe_tx_hash(nonce_, operation=0, to=receiver, value=10**17, data='0x'):
            """EIP-712 hash computed off-chain, the same bytes each owner signs."""
            struct = cast('keccak', cast('abi-encode',
                          'f(bytes32,address,uint256,bytes32,uint8,uint256,uint256,uint256,address,address,uint256)',
                          cast('keccak', SAFE_TX), to, value, cast('keccak', data), operation, 0, 0, 0, ZERO, ZERO, nonce_))
            domain = cast('keccak', cast('abi-encode', 'f(bytes32,uint256,address)',
                          cast('keccak', 'EIP712Domain(uint256 chainId,address verifyingContract)'), 31337, safe))
            return cast('keccak', '0x1901' + domain[2:] + struct[2:])

        def sign(owner, h):  # 65 bytes r || s || v with v = 27/28
            return cast('wallet', 'sign', '--no-hash', '--private-key', pk[owner], h)[2:]

        def execute(signatures, dry=True):
            data = cast('calldata', EXEC, receiver, 10**17, '0x', 0, 0, 0, 0, ZERO, ZERO, '0x' + signatures)
            if dry:
                return call(safe, data, relayer)[1]
            return send(relayer, data, safe)

        h0 = safe_tx_hash(0)
        onchain = call(safe, cast('calldata', 'getTransactionHash(address,uint256,bytes,uint8,uint256,uint256,uint256,address,address,uint256)',
                                  receiver, 10**17, '0x', 0, 0, 0, 0, ZERO, ZERO, 0))[0]
        print('off-chain SafeTx hash == getTransactionHash:', h0 == onchain)
        print('same tx with operation=1 (delegatecall) hash:', safe_tx_hash(0, operation=1)[:18] + '..',
              'vs', h0[:18] + '..')

        lo, hi = sorted([o1, o2], key=lambda a: int(a, 16))
        ascending = sign(lo, h0) + sign(hi, h0)
        print('== nonce 0')
        for label, sigs in [('1 signature', sign(lo, h0)),
                            ('same owner twice', sign(lo, h0) * 2),
                            ('descending owners', sign(hi, h0) + sign(lo, h0))]:
            print(f'{label:24} -> {execute(sigs)}')
        before = int(rpc('eth_getBalance', receiver, 'latest'), 16)
        receipt = execute(ascending, dry=False)
        gained = int(rpc('eth_getBalance', receiver, 'latest'), 16) - before
        print(f'{"ascending owners":24} -> status {int(receipt["status"], 16)},',
              f'receiver +{gained / 1e18} ETH, nonce now {nonce()}')
        print(f'{"replay nonce-0 sigs":24} -> {execute(ascending)}')

        print('== nonce 1 (EOA + ERC-1271 owner)')
        h1 = safe_tx_hash(1)
        parts = sorted([lo, signer], key=lambda a: int(a, 16))
        # Contract signature: r = owner address, s = offset of the dynamic part, v = 0.
        contract_sig = '00' * 12 + signer[2:].lower() + format(65 * 2, '064x') + '00'
        sigs = ''.join(sign(lo, h1) if p == lo else contract_sig for p in parts) + format(0, '064x')
        print(f'{"before approve(hash)":24} -> {execute(sigs)}')
        assert int(send(admin, cast('calldata', 'approve(bytes32)', h1), signer)['status'], 16) == 1
        receipt = execute(sigs, dry=False)
        print(f'{"after approve(hash)":24} -> status {int(receipt["status"], 16)}, nonce now {nonce()}')
    finally:
        node.terminate()
        node.wait()


if __name__ == '__main__':
    main()