From 217dbbbb5e38ab582ee0b3ef37fe9e99d887d7c8 Mon Sep 17 00:00:00 2001 From: Fabian Jahr Date: Wed, 1 Oct 2025 21:23:22 +0200 Subject: [PATCH] test: Add musig failure scenarios Also changes the the non-constant variable NUM_WALLETS to lower case and refactors the success case scenarios to reuse existing code. Co-authored-by: rkrux --- test/functional/wallet_musig.py | 177 ++++++++++++++++++++++++-------- 1 file changed, 136 insertions(+), 41 deletions(-) diff --git a/test/functional/wallet_musig.py b/test/functional/wallet_musig.py index 277ca9276ed..d14a1e2027a 100755 --- a/test/functional/wallet_musig.py +++ b/test/functional/wallet_musig.py @@ -21,39 +21,26 @@ MUSIG_RE = re.compile(r"musig\((.*?)\)") PLACEHOLDER_RE = re.compile(r"\$\d") class WalletMuSigTest(BitcoinTestFramework): - WALLET_NUM = 0 + wallet_num = 0 def set_test_params(self): self.num_nodes = 1 def skip_test_if_missing_module(self): self.skip_if_no_wallet() - def do_test(self, comment, pattern, sighash_type=None, scriptpath=False, nosign_wallets=None, only_one_musig_wallet=False): - self.log.info(f"Testing {comment}") - has_internal = MULTIPATH_TWO_RE.search(pattern) is not None - + # Create wallets and extract keys + def create_wallets_and_keys_from_pattern(self, pat): wallets = [] keys = [] - pat = pattern.replace("$H", H_POINT) - - # Figure out how many wallets are needed and create them - expected_pubnonces = 0 - expected_partial_sigs = 0 for musig in MUSIG_RE.findall(pat): - musig_partial_sigs = 0 for placeholder in PLACEHOLDER_RE.findall(musig): wallet_index = int(placeholder[1:]) - if nosign_wallets is None or wallet_index not in nosign_wallets: - expected_pubnonces += 1 - else: - musig_partial_sigs = None - if musig_partial_sigs is not None: - musig_partial_sigs += 1 if wallet_index < len(wallets): continue - wallet_name = f"musig_{self.WALLET_NUM}" - self.WALLET_NUM += 1 + + wallet_name = f"musig_{self.wallet_num}" + self.wallet_num += 1 self.nodes[0].createwallet(wallet_name) wallet = self.nodes[0].get_wallet_rpc(wallet_name) wallets.append(wallet) @@ -74,32 +61,137 @@ class WalletMuSigTest(BitcoinTestFramework): privkey += ORIGIN_PATH_RE.search(pubkey).group(1) break keys.append((privkey, pubkey)) - if musig_partial_sigs is not None: - expected_partial_sigs += musig_partial_sigs - # Construct and import each wallet's musig descriptor that - # contains the private key from that wallet and pubkeys of the others + return wallets, keys + + # Construct and import each wallet's musig descriptor that + # contains the private key from that wallet and pubkeys of the others + def construct_and_import_musig_descriptor_in_wallets(self, pat, wallets, keys, only_one_musig_wallet=False): for i, wallet in enumerate(wallets): if only_one_musig_wallet and i > 0: continue desc = pat - import_descs = [] for j, (priv, pub) in enumerate(keys): if j == i: desc = desc.replace(f"${i}", priv) else: desc = desc.replace(f"${j}", pub) - import_descs.append({ + import_descs = [{ "desc": descsum_create(desc), "active": True, "timestamp": "now", - }) + }] res = wallet.importdescriptors(import_descs) for r in res: assert_equal(r["success"], True) + def setup_musig_scenario(self, pat): + wallets, keys = self.create_wallets_and_keys_from_pattern(pat) + self.construct_and_import_musig_descriptor_in_wallets(pat, wallets, keys, only_one_musig_wallet=False) + + # Fund address + addr = wallets[0].getnewaddress(address_type="bech32m") + for wallet in wallets[1:]: + assert_equal(addr, wallet.getnewaddress(address_type="bech32m")) + + self.def_wallet.sendtoaddress(addr, 10) + self.generate(self.nodes[0], 1) + + # Create PSBT + utxo = wallets[0].listunspent()[0] + psbt = wallets[0].walletcreatefundedpsbt( + outputs=[{self.def_wallet.getnewaddress(): 5}], + inputs=[utxo], + change_type="bech32m", + changePosition=1 + )["psbt"] + + return wallets, psbt + + def test_failure_case_1(self, comment, pat): + self.log.info(f"Testing {comment}") + wallets, psbt = self.setup_musig_scenario(pat) + + # Only 2 out of 3 participants provide nonces + nonce_psbts = [] + for i in range(2): + proc = wallets[i].walletprocesspsbt(psbt=psbt) + nonce_psbts.append(proc["psbt"]) + + comb_nonce_psbt = self.nodes[0].combinepsbt(nonce_psbts) + + # Attempt to create partial sigs. This should not complete due to the + # missing nonce. + for wallet in wallets[:2]: + proc = wallet.walletprocesspsbt(psbt=comb_nonce_psbt) + assert_equal(proc["complete"], False) + # No partial sigs are created + dec = self.nodes[0].decodepsbt(proc["psbt"]) + # There are still only two nonces + assert_equal(len(dec["inputs"][0].get("musig2_pubnonces", [])), 2) + + def test_failure_case_2(self, comment, pat): + self.log.info(f"Testing {comment}") + wallets, psbt = self.setup_musig_scenario(pat) + nonce_psbts = [w.walletprocesspsbt(psbt=psbt)["psbt"] for w in wallets] + comb_nonce_psbt = self.nodes[0].combinepsbt(nonce_psbts) + + # Only 2 out of 3 provide partial sigs + psig_psbts = [] + for i in range(2): + proc = wallets[i].walletprocesspsbt(psbt=comb_nonce_psbt) + psig_psbts.append(proc["psbt"]) + + comb_psig_psbt = self.nodes[0].combinepsbt(psig_psbts) + + # Finalization fails due to missing partial sig + finalized = self.nodes[0].finalizepsbt(comb_psig_psbt) + assert_equal(finalized["complete"], False) + + # Still only two partial sigs in combined PSBT + dec = self.nodes[0].decodepsbt(comb_psig_psbt) + assert_equal(len(dec["inputs"][0]["musig2_partial_sigs"]), 2) + + def test_failure_case_3(self, comment, pat): + self.log.info(f"Testing {comment}") + wallets, psbt = self.setup_musig_scenario(pat) + nonce_psbts = [w.walletprocesspsbt(psbt=psbt)["psbt"] for w in wallets] + comb_nonce_psbt = self.nodes[0].combinepsbt(nonce_psbts) + + finalized = self.nodes[0].finalizepsbt(comb_nonce_psbt) + assert_equal(finalized["complete"], False) + + dec = self.nodes[0].decodepsbt(comb_nonce_psbt) + assert "musig2_pubnonces" in dec["inputs"][0] + assert "musig2_partial_sigs" not in dec["inputs"][0] + + def test_success_case(self, comment, pattern, sighash_type=None, scriptpath=False, nosign_wallets=None, only_one_musig_wallet=False): + self.log.info(f"Testing {comment}") + has_internal = MULTIPATH_TWO_RE.search(pattern) is not None + + pat = pattern.replace("$H", H_POINT) + wallets, keys = self.create_wallets_and_keys_from_pattern(pat) + self.construct_and_import_musig_descriptor_in_wallets(pat, wallets, keys, only_one_musig_wallet) + + expected_pubnonces = 0 + expected_partial_sigs = 0 + for musig in MUSIG_RE.findall(pat): + musig_partial_sigs = 0 + for placeholder in PLACEHOLDER_RE.findall(musig): + wallet_index = int(placeholder[1:]) + if nosign_wallets is None or wallet_index not in nosign_wallets: + expected_pubnonces += 1 + else: + musig_partial_sigs = None + if musig_partial_sigs is not None: + musig_partial_sigs += 1 + if wallet_index < len(wallets): + continue + if musig_partial_sigs is not None: + expected_partial_sigs += musig_partial_sigs + # Check that the wallets agree on the same musig address addr = None change_addr = None @@ -221,22 +313,25 @@ class WalletMuSigTest(BitcoinTestFramework): def run_test(self): self.def_wallet = self.nodes[0].get_wallet_rpc(self.default_wallet_name) - self.do_test("rawtr(musig(keys/*))", "rawtr(musig($0/<0;1>/*,$1/<1;2>/*,$2/<2;3>/*))") - self.do_test("rawtr(musig(keys/*)) with ALL|ANYONECANPAY", "rawtr(musig($0/<0;1>/*,$1/<1;2>/*,$2/<2;3>/*))", "ALL|ANYONECANPAY") - self.do_test("tr(musig(keys/*)) no multipath", "tr(musig($0/0/*,$1/1/*,$2/2/*))") - self.do_test("tr(musig(keys/*)) 2 index multipath", "tr(musig($0/<0;1>/*,$1/<1;2>/*,$2/<2;3>/*))") - self.do_test("tr(musig(keys/*)) 3 index multipath", "tr(musig($0/<0;1;2>/*,$1/<1;2;3>/*,$2/<2;3;4>/*))") - self.do_test("rawtr(musig/*)", "rawtr(musig($0,$1,$2)/<0;1>/*)") - self.do_test("tr(musig/*)", "tr(musig($0,$1,$2)/<0;1>/*)") - self.do_test("rawtr(musig(keys/*)) without all wallets importing", "rawtr(musig($0/<0;1>/*,$1/<0;1>/*,$2/<0;1>/*))", only_one_musig_wallet=True) - self.do_test("tr(musig(keys/*)) without all wallets importing", "tr(musig($0/<0;1>/*,$1/<0;1>/*,$2/<0;1>/*))", only_one_musig_wallet=True) - self.do_test("tr(H, pk(musig(keys/*)))", "tr($H,pk(musig($0/<0;1>/*,$1/<1;2>/*,$2/<2;3>/*)))", scriptpath=True) - self.do_test("tr(H,pk(musig/*))", "tr($H,pk(musig($0,$1,$2)/<0;1>/*))", scriptpath=True) - self.do_test("tr(H,{pk(musig/*), pk(musig/*)})", "tr($H,{pk(musig($0,$1,$2)/<0;1>/*),pk(musig($3,$4,$5)/0/*)})", scriptpath=True) - self.do_test("tr(H,{pk(musig/*), pk(same keys different musig/*)})", "tr($H,{pk(musig($0,$1,$2)/<0;1>/*),pk(musig($1,$2)/0/*)})", scriptpath=True) - self.do_test("tr(musig/*,{pk(partial keys diff musig-1/*),pk(partial keys diff musig-2/*)})}", "tr(musig($0,$1,$2)/<3;4>/*,{pk(musig($0,$1)/<5;6>/*),pk(musig($1,$2)/7/*)})") - self.do_test("tr(musig/*,{pk(partial keys diff musig-1/*),pk(partial keys diff musig-2/*)})} script-path", "tr(musig($0,$1,$2)/<3;4>/*,{pk(musig($0,$1)/<5;6>/*),pk(musig($1,$2)/7/*)})", scriptpath=True, nosign_wallets=[0]) + self.test_success_case("rawtr(musig(keys/*))", "rawtr(musig($0/<0;1>/*,$1/<1;2>/*,$2/<2;3>/*))") + self.test_success_case("rawtr(musig(keys/*)) with ALL|ANYONECANPAY", "rawtr(musig($0/<0;1>/*,$1/<1;2>/*,$2/<2;3>/*))", "ALL|ANYONECANPAY") + self.test_success_case("tr(musig(keys/*)) no multipath", "tr(musig($0/0/*,$1/1/*,$2/2/*))") + self.test_success_case("tr(musig(keys/*)) 2 index multipath", "tr(musig($0/<0;1>/*,$1/<1;2>/*,$2/<2;3>/*))") + self.test_success_case("tr(musig(keys/*)) 3 index multipath", "tr(musig($0/<0;1;2>/*,$1/<1;2;3>/*,$2/<2;3;4>/*))") + self.test_success_case("rawtr(musig/*)", "rawtr(musig($0,$1,$2)/<0;1>/*)") + self.test_success_case("tr(musig/*)", "tr(musig($0,$1,$2)/<0;1>/*)") + self.test_success_case("rawtr(musig(keys/*)) without all wallets importing", "rawtr(musig($0/<0;1>/*,$1/<0;1>/*,$2/<0;1>/*))", only_one_musig_wallet=True) + self.test_success_case("tr(musig(keys/*)) without all wallets importing", "tr(musig($0/<0;1>/*,$1/<0;1>/*,$2/<0;1>/*))", only_one_musig_wallet=True) + self.test_success_case("tr(H, pk(musig(keys/*)))", "tr($H,pk(musig($0/<0;1>/*,$1/<1;2>/*,$2/<2;3>/*)))", scriptpath=True) + self.test_success_case("tr(H,pk(musig/*))", "tr($H,pk(musig($0,$1,$2)/<0;1>/*))", scriptpath=True) + self.test_success_case("tr(H,{pk(musig/*), pk(musig/*)})", "tr($H,{pk(musig($0,$1,$2)/<0;1>/*),pk(musig($3,$4,$5)/0/*)})", scriptpath=True) + self.test_success_case("tr(H,{pk(musig/*), pk(same keys different musig/*)})", "tr($H,{pk(musig($0,$1,$2)/<0;1>/*),pk(musig($1,$2)/0/*)})", scriptpath=True) + self.test_success_case("tr(musig/*,{pk(partial keys diff musig-1/*),pk(partial keys diff musig-2/*)})}", "tr(musig($0,$1,$2)/<3;4>/*,{pk(musig($0,$1)/<5;6>/*),pk(musig($1,$2)/7/*)})") + self.test_success_case("tr(musig/*,{pk(partial keys diff musig-1/*),pk(partial keys diff musig-2/*)})} script-path", "tr(musig($0,$1,$2)/<3;4>/*,{pk(musig($0,$1)/<5;6>/*),pk(musig($1,$2)/7/*)})", scriptpath=True, nosign_wallets=[0]) + self.test_failure_case_1("missing participant nonce", "tr(musig($0/<0;1>/*,$1/<1;2>/*,$2/<2;3>/*))") + self.test_failure_case_2("insufficient partial signatures", "rawtr(musig($0/<0;1>/*,$1/<1;2>/*,$2/<2;3>/*))") + self.test_failure_case_3("finalize without partial sigs", "rawtr(musig($0/<0;1>/*,$1/<1;2>/*))") if __name__ == '__main__': WalletMuSigTest(__file__).main()