mirror of
https://github.com/bitcoin/bitcoin.git
synced 2025-01-26 11:13:23 -03:00
d93b794709
It is best to store all key origin information (master key fingerprint and all derivation steps) in the multisig descriptor. Being explicit with this information should be beneficial if this approach is used with other wallets/signers (whether hardware or software). There is no harm including all of this with xpubs (if anything it simplifies the test code) and makes this example/docs more complete and safer incase it is referenced by others.
157 lines
8.6 KiB
Python
Executable file
157 lines
8.6 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
# Copyright (c) 2021-2022 The Bitcoin Core developers
|
|
# Distributed under the MIT software license, see the accompanying
|
|
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
|
|
"""Test a basic M-of-N multisig setup between multiple people using descriptor wallets and PSBTs, as well as a signing flow.
|
|
|
|
This is meant to be documentation as much as functional tests, so it is kept as simple and readable as possible.
|
|
"""
|
|
|
|
from test_framework.test_framework import BitcoinTestFramework
|
|
from test_framework.util import (
|
|
assert_approx,
|
|
assert_equal,
|
|
)
|
|
|
|
|
|
class WalletMultisigDescriptorPSBTTest(BitcoinTestFramework):
|
|
def add_options(self, parser):
|
|
self.add_wallet_options(parser, legacy=False)
|
|
|
|
def set_test_params(self):
|
|
self.num_nodes = 3
|
|
self.setup_clean_chain = True
|
|
self.wallet_names = []
|
|
self.extra_args = [["-keypool=100"]] * self.num_nodes
|
|
|
|
def skip_test_if_missing_module(self):
|
|
self.skip_if_no_wallet()
|
|
self.skip_if_no_sqlite()
|
|
|
|
@staticmethod
|
|
def _get_xpub(wallet, internal):
|
|
"""Extract the wallet's xpubs using `listdescriptors` and pick the one from the `pkh` descriptor since it's least likely to be accidentally reused (legacy addresses)."""
|
|
pkh_descriptor = next(filter(lambda d: d["desc"].startswith("pkh(") and d["internal"] == internal, wallet.listdescriptors()["descriptors"]))
|
|
# Keep all key origin information (master key fingerprint and all derivation steps) for proper support of hardware devices
|
|
# See section 'Key origin identification' in 'doc/descriptors.md' for more details...
|
|
return pkh_descriptor["desc"].split("pkh(")[1].split(")")[0]
|
|
|
|
@staticmethod
|
|
def _check_psbt(psbt, to, value, multisig):
|
|
"""Helper function for any of the N participants to check the psbt with decodepsbt and verify it is OK before signing."""
|
|
tx = multisig.decodepsbt(psbt)["tx"]
|
|
amount = 0
|
|
for vout in tx["vout"]:
|
|
address = vout["scriptPubKey"]["address"]
|
|
assert_equal(multisig.getaddressinfo(address)["ischange"], address != to)
|
|
if address == to:
|
|
amount += vout["value"]
|
|
assert_approx(amount, float(value), vspan=0.001)
|
|
|
|
def participants_create_multisigs(self, external_xpubs, internal_xpubs):
|
|
"""The multisig is created by importing the following descriptors. The resulting wallet is watch-only and every participant can do this."""
|
|
for i, node in enumerate(self.nodes):
|
|
node.createwallet(wallet_name=f"{self.name}_{i}", blank=True, descriptors=True, disable_private_keys=True)
|
|
multisig = node.get_wallet_rpc(f"{self.name}_{i}")
|
|
external = multisig.getdescriptorinfo(f"wsh(sortedmulti({self.M},{f','.join(external_xpubs)}))")
|
|
internal = multisig.getdescriptorinfo(f"wsh(sortedmulti({self.M},{f','.join(internal_xpubs)}))")
|
|
result = multisig.importdescriptors([
|
|
{ # receiving addresses (internal: False)
|
|
"desc": external["descriptor"],
|
|
"active": True,
|
|
"internal": False,
|
|
"timestamp": "now",
|
|
},
|
|
{ # change addresses (internal: True)
|
|
"desc": internal["descriptor"],
|
|
"active": True,
|
|
"internal": True,
|
|
"timestamp": "now",
|
|
},
|
|
])
|
|
assert all(r["success"] for r in result)
|
|
yield multisig
|
|
|
|
def run_test(self):
|
|
self.M = 2
|
|
self.N = self.num_nodes
|
|
self.name = f"{self.M}_of_{self.N}_multisig"
|
|
self.log.info(f"Testing {self.name}...")
|
|
|
|
participants = {
|
|
# Every participant generates an xpub. The most straightforward way is to create a new descriptor wallet.
|
|
# This wallet will be the participant's `signer` for the resulting multisig. Avoid reusing this wallet for any other purpose (for privacy reasons).
|
|
"signers": [node.get_wallet_rpc(node.createwallet(wallet_name=f"participant_{self.nodes.index(node)}", descriptors=True)["name"]) for node in self.nodes],
|
|
# After participants generate and exchange their xpubs they will each create their own watch-only multisig.
|
|
# Note: these multisigs are all the same, this just highlights that each participant can independently verify everything on their own node.
|
|
"multisigs": []
|
|
}
|
|
|
|
self.log.info("Generate and exchange xpubs...")
|
|
external_xpubs, internal_xpubs = [[self._get_xpub(signer, internal) for signer in participants["signers"]] for internal in [False, True]]
|
|
|
|
self.log.info("Every participant imports the following descriptors to create the watch-only multisig...")
|
|
participants["multisigs"] = list(self.participants_create_multisigs(external_xpubs, internal_xpubs))
|
|
|
|
self.log.info("Check that every participant's multisig generates the same addresses...")
|
|
for _ in range(10): # we check that the first 10 generated addresses are the same for all participant's multisigs
|
|
receive_addresses = [multisig.getnewaddress() for multisig in participants["multisigs"]]
|
|
all(address == receive_addresses[0] for address in receive_addresses)
|
|
change_addresses = [multisig.getrawchangeaddress() for multisig in participants["multisigs"]]
|
|
all(address == change_addresses[0] for address in change_addresses)
|
|
|
|
self.log.info("Get a mature utxo to send to the multisig...")
|
|
coordinator_wallet = participants["signers"][0]
|
|
self.generatetoaddress(self.nodes[0], 101, coordinator_wallet.getnewaddress())
|
|
|
|
deposit_amount = 6.15
|
|
multisig_receiving_address = participants["multisigs"][0].getnewaddress()
|
|
self.log.info("Send funds to the resulting multisig receiving address...")
|
|
coordinator_wallet.sendtoaddress(multisig_receiving_address, deposit_amount)
|
|
self.generate(self.nodes[0], 1)
|
|
for participant in participants["multisigs"]:
|
|
assert_approx(participant.getbalance(), deposit_amount, vspan=0.001)
|
|
|
|
self.log.info("Send a transaction from the multisig!")
|
|
to = participants["signers"][self.N - 1].getnewaddress()
|
|
value = 1
|
|
self.log.info("First, make a sending transaction, created using `walletcreatefundedpsbt` (anyone can initiate this)...")
|
|
psbt = participants["multisigs"][0].walletcreatefundedpsbt(inputs=[], outputs={to: value}, feeRate=0.00010)
|
|
|
|
psbts = []
|
|
self.log.info("Now at least M users check the psbt with decodepsbt and (if OK) signs it with walletprocesspsbt...")
|
|
for m in range(self.M):
|
|
signers_multisig = participants["multisigs"][m]
|
|
self._check_psbt(psbt["psbt"], to, value, signers_multisig)
|
|
signing_wallet = participants["signers"][m]
|
|
partially_signed_psbt = signing_wallet.walletprocesspsbt(psbt["psbt"])
|
|
psbts.append(partially_signed_psbt["psbt"])
|
|
|
|
self.log.info("Finally, collect the signed PSBTs with combinepsbt, finalizepsbt, then broadcast the resulting transaction...")
|
|
combined = coordinator_wallet.combinepsbt(psbts)
|
|
finalized = coordinator_wallet.finalizepsbt(combined)
|
|
coordinator_wallet.sendrawtransaction(finalized["hex"])
|
|
|
|
self.log.info("Check that balances are correct after the transaction has been included in a block.")
|
|
self.generate(self.nodes[0], 1)
|
|
assert_approx(participants["multisigs"][0].getbalance(), deposit_amount - value, vspan=0.001)
|
|
assert_equal(participants["signers"][self.N - 1].getbalance(), value)
|
|
|
|
self.log.info("Send another transaction from the multisig, this time with a daisy chained signing flow (one after another in series)!")
|
|
psbt = participants["multisigs"][0].walletcreatefundedpsbt(inputs=[], outputs={to: value}, feeRate=0.00010)
|
|
for m in range(self.M):
|
|
signers_multisig = participants["multisigs"][m]
|
|
self._check_psbt(psbt["psbt"], to, value, signers_multisig)
|
|
signing_wallet = participants["signers"][m]
|
|
psbt = signing_wallet.walletprocesspsbt(psbt["psbt"])
|
|
assert_equal(psbt["complete"], m == self.M - 1)
|
|
coordinator_wallet.sendrawtransaction(psbt["hex"])
|
|
|
|
self.log.info("Check that balances are correct after the transaction has been included in a block.")
|
|
self.generate(self.nodes[0], 1)
|
|
assert_approx(participants["multisigs"][0].getbalance(), deposit_amount - (value * 2), vspan=0.001)
|
|
assert_equal(participants["signers"][self.N - 1].getbalance(), value * 2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
WalletMultisigDescriptorPSBTTest().main()
|