Inside Bitcoin

251        tx.vin = [CTxIn(outpoint=COutPoint(hash=int('ee' * 32, 16), n=0), scriptSig=b"")]
252        tx.vout = [CTxOut(nValue=0, scriptPubKey=b"")]
253        psbt = PSBT()
254        psbt.g = PSBTMap({PSBT_GLOBAL_UNSIGNED_TX: tx.serialize()})
255        participant1_keydata = in_pubkey1 + in_fake_agg_pubkey + fake_leaf_hash
256        psbt.i = [PSBTMap({
257                    bytes([PSBT_IN_MUSIG2_PARTICIPANT_PUBKEYS]) + in_fake_agg_pubkey: [in_pubkey1, in_pubkey2],
258                    bytes([PSBT_IN_MUSIG2_PUB_NONCE]) + participant1_keydata: fake_pubnonce,
259                    bytes([PSBT_IN_MUSIG2_PARTIAL_SIG]) + participant1_keydata: fake_partialsig,
260                 })]
261        _, out_pubkey1 = generate_keypair()
262        _, out_pubkey2 = generate_keypair()
263        _, out_fake_agg_pubkey = generate_keypair()
264        psbt.o = [PSBTMap({
265                    bytes([PSBT_OUT_MUSIG2_PARTICIPANT_PUBKEYS]) + out_fake_agg_pubkey: [out_pubkey1, out_pubkey2],
266                 })]
267        res = self.nodes[0].decodepsbt(psbt.to_base64())
268        assert_equal(len(res["inputs"]), 1)
269        res_input = res["inputs"][0]
270        assert_equal(len(res["outputs"]), 1)
271        res_output = res["outputs"][0]
272
273        assert "musig2_participant_pubkeys" in res_input
274        in_participant_pks = res_input["musig2_participant_pubkeys"][0]
275        assert "aggregate_pubkey" in in_participant_pks
276        assert_equal(in_participant_pks["aggregate_pubkey"], in_fake_agg_pubkey.hex())
277        assert "participant_pubkeys" in in_participant_pks
278        assert_equal(in_participant_pks["participant_pubkeys"], [in_pubkey1.hex(), in_pubkey2.hex()])
279
280        assert "musig2_pubnonces" in res_input
281        in_pubnonce = res_input["musig2_pubnonces"][0]
282        assert "participant_pubkey" in in_pubnonce
283        assert_equal(in_pubnonce["participant_pubkey"], in_pubkey1.hex())
284        assert "aggregate_pubkey" in in_pubnonce
285        assert_equal(in_pubnonce["aggregate_pubkey"], in_fake_agg_pubkey.hex())
286        assert "leaf_hash" in in_pubnonce
287        assert_equal(in_pubnonce["leaf_hash"], fake_leaf_hash.hex())
288        assert "pubnonce" in in_pubnonce
289        assert_equal(in_pubnonce["pubnonce"], fake_pubnonce.hex())
290
291        assert "musig2_partial_sigs" in res_input
292        in_partialsig = res_input["musig2_partial_sigs"][0]
293        assert "participant_pubkey" in in_partialsig
294        assert_equal(in_partialsig["participant_pubkey"], in_pubkey1.hex())
295        assert "aggregate_pubkey" in in_partialsig
296        assert_equal(in_partialsig["aggregate_pubkey"], in_fake_agg_pubkey.hex())
297        assert "leaf_hash" in in_partialsig
298        assert_equal(in_partialsig["leaf_hash"], fake_leaf_hash.hex())
299        assert "partial_sig" in in_partialsig
300        assert_equal(in_partialsig["partial_sig"], fake_partialsig.hex())
301
302        assert "musig2_participant_pubkeys" in res_output
303        out_participant_pks = res_output["musig2_participant_pubkeys"][0]
304        assert "aggregate_pubkey" in out_participant_pks
305        assert_equal(out_participant_pks["aggregate_pubkey"], out_fake_agg_pubkey.hex())
306        assert "participant_pubkeys" in out_participant_pks
307        assert_equal(out_participant_pks["participant_pubkeys"], [out_pubkey1.hex(), out_pubkey2.hex()])
308
309    def test_musig2_untrusted_derivation(self):
310        self.log.info("Test MuSig2 aggregate derivation from untrusted PSBT fields")
311        node = self.nodes[0]
312
313        script_pubkey = bytes.fromhex(H_POINT)
314        _, aggregate_pubkey = generate_keypair()
315        _, participant_pubkey = generate_keypair()
316
317        # Both have a matching aggregate fingerprint but cannot derive the script pubkey: 0 derives a different key, hardened(0) cannot be derived at all
318        for index in [0, hardened(0)]:
319            psbt = self.create_psbt(inputs={
320                PSBT_IN_WITNESS_UTXO: CTxOut(nValue=1, scriptPubKey=output_key_to_p2tr_script(script_pubkey)).serialize(),
321                bytes([PSBT_IN_TAP_BIP32_DERIVATION]) + script_pubkey: ser_compact_size(0) + hash160(aggregate_pubkey)[:4] + index.to_bytes(4, "little"),
322                PSBT_IN_TAP_INTERNAL_KEY: script_pubkey,
323                bytes([PSBT_IN_MUSIG2_PARTICIPANT_PUBKEYS]) + aggregate_pubkey: [participant_pubkey],
324            }).to_base64()
325            assert_equal(node.analyzepsbt(psbt)["inputs"][0]["is_final"], False)
326            assert_equal(node.finalizepsbt(psbt)["complete"], False)
327
328    def test_combinepsbt_preserves_proprietary_fields(self):
329        self.log.info("Test that combining PSBTs preserves proprietary fields")
330
331        def proprietary_key(type_byte, identifier, subtype, key_data=b""):
332            return bytes([type_byte]) + ser_compact_size(len(identifier)) + identifier + ser_compact_size(subtype) + key_data
333
334        def proprietary_entry(key, value, identifier, subtype):
335            return {"identifier": identifier.hex(), "subtype": subtype, "key": key.hex(), "value": value.hex()}
336
337        tx = CTransaction()
338        tx.vin = [CTxIn(outpoint=COutPoint(hash=int('aa' * 32, 16), n=0), scriptSig=b"")]
339        tx.vout = [CTxOut(nValue=0, scriptPubKey=b"")]
340
341        global_key_a = proprietary_key(type_byte=PSBT_GLOBAL_PROPRIETARY, identifier=b"gc", subtype=1, key_data=b"\x01")
342        global_key_b = proprietary_key(type_byte=PSBT_GLOBAL_PROPRIETARY, identifier=b"gc", subtype=2, key_data=b"\x02")
343        input_key_a = proprietary_key(type_byte=PSBT_IN_PROPRIETARY, identifier=b"in", subtype=3, key_data=b"\x03")
344        input_key_b = proprietary_key(type_byte=PSBT_IN_PROPRIETARY, identifier=b"in", subtype=4, key_data=b"\x04")
345        output_key_a = proprietary_key(type_byte=PSBT_OUT_PROPRIETARY, identifier=b"out", subtype=5, key_data=b"\x05")
346        output_key_b = proprietary_key(type_byte=PSBT_OUT_PROPRIETARY, identifier=b"out", subtype=6, key_data=b"\x06")
347
348        psbt1 = PSBT(
349            g=PSBTMap({
350                PSBT_GLOBAL_UNSIGNED_TX: tx.serialize(),
351                global_key_a: b"\xaa",
352            }),
353            i=[PSBTMap({
354                input_key_a: b"\xbb",
355            })],
356            o=[PSBTMap({
357                output_key_a: b"\xcc",
358            })],
359        ).to_base64()
360        psbt2 = PSBT(
361            g=PSBTMap({
362                PSBT_GLOBAL_UNSIGNED_TX: tx.serialize(),
363                global_key_b: b"\xdd",
364            }),
365            i=[PSBTMap({
366                input_key_b: b"\xee",
367            })],
368            o=[PSBTMap({
369                output_key_b: b"\xff",
370            })],
371        ).to_base64()
372
373        decoded = self.nodes[0].decodepsbt(self.nodes[0].combinepsbt([psbt1, psbt2]))
374        assert_equal(decoded["proprietary"], [
375            proprietary_entry(key=global_key_a, value=b"\xaa", identifier=b"gc", subtype=1),
376            proprietary_entry(key=global_key_b, value=b"\xdd", identifier=b"gc", subtype=2),
377        ])
378        assert_equal(decoded["inputs"][0]["proprietary"], [
379            proprietary_entry(key=input_key_a, value=b"\xbb", identifier=b"in", subtype=3),
380            proprietary_entry(key=input_key_b, value=b"\xee", identifier=b"in", subtype=4),
381        ])
382        assert_equal(decoded["outputs"][0]["proprietary"], [
383            proprietary_entry(key=output_key_a, value=b"\xcc", identifier=b"out", subtype=5),
384            proprietary_entry(key=output_key_b, value=b"\xff", identifier=b"out", subtype=6),
385        ])
386
387    def test_combinepsbt_global_xpub_origin_conflict(self):
388        self.log.info("Test that combining PSBTs with conflicting origins for the same xpub keeps a single record")
389
390        tx = CTransaction()
391        tx.vin = [CTxIn(outpoint=COutPoint(hash=int('aa' * 32, 16), n=0), scriptSig=b"")]
392        tx.vout = [CTxOut(nValue=0, scriptPubKey=b"")]
393
394        xpub = "tpubD6NzVbkrYhZ4XgiXtGrdW5XDAPFCL9h7we1vwNCpn8tGbBcgfVYjXyhWo4E1xkh56hjod1RhGjxbaTLV3X4FyWuejifB9jusQ46QzG87VKp"
395        xpub_data, xpub_version = base58_to_byte(xpub)
396        xpub_key = bytes([PSBT_GLOBAL_XPUB]) + bytes([xpub_version]) + xpub_data
397
398        def psbt_with_origin(fingerprint):
399            return PSBT(
400                g=PSBTMap({
401                    PSBT_GLOBAL_UNSIGNED_TX: tx.serialize(),
402                    xpub_key: fingerprint,
403                }),
404                i=[PSBTMap({})],
405                o=[PSBTMap({})],
406            ).to_base64()
407
408        combined = self.nodes[0].combinepsbt([psbt_with_origin(b"\x00\x00\x00\x00"), psbt_with_origin(b"\x11\x11\x11\x11")])
409        # The same xpub under both origins would serialize as duplicate keys, making the combined PSBT unparseable
410        decoded = self.nodes[0].decodepsbt(combined)
411        assert_equal(decoded["global_xpubs"], [{"xpub": xpub, "master_fingerprint": "00000000", "path": "m"}])
412
413    def test_combinepsbt_tap_leaf_script_conflict(self):
414        self.log.info("Test that combining PSBTs with conflicting leaf scripts for the same control block keeps a single record")
415
416        tx = CTransaction()
417        tx.vin = [CTxIn(outpoint=COutPoint(hash=int('aa' * 32, 16), n=0), scriptSig=b"")]
418        tx.vout = [CTxOut(nValue=0, scriptPubKey=b"")]
419
420        def psbt_with_leaf_scripts(*records):
421            return PSBT(
422                g=PSBTMap({PSBT_GLOBAL_UNSIGNED_TX: tx.serialize()}),
423                i=[PSBTMap({
424                    bytes([PSBT_IN_TAP_LEAF_SCRIPT]) + control_block: bytes(leaf_script) + bytes([LEAF_VERSION_TAPSCRIPT])
425                    for leaf_script, control_block in records
426                })],
427                o=[PSBTMap({})],
428            ).to_base64()
429
430        def combined_tap_scripts(psbts):
431            return self.nodes[0].decodepsbt(self.nodes[0].combinepsbt(psbts))["inputs"][0]["taproot_scripts"]
432
433        def tap_script(leaf_script, control_blocks):
434            return {"script": leaf_script.hex(), "leaf_ver": LEAF_VERSION_TAPSCRIPT, "control_blocks": [cb.hex() for cb in control_blocks]}
435
436        control_block = bytes([LEAF_VERSION_TAPSCRIPT]) + bytes.fromhex(H_POINT)
437        control_block_with_path = control_block + bytes(32)
438        control_block_with_longer_path = control_block + bytes(64)
439        leaf_script_a = CScript([OP_TRUE])
440        leaf_script_b = CScript([OP_TRUE, OP_TRUE])
441        leaf_script_c = CScript([OP_TRUE, OP_TRUE, OP_TRUE])
442
443        psbt_a = psbt_with_leaf_scripts((leaf_script_a, control_block))
444        psbt_b = psbt_with_leaf_scripts((leaf_script_b, control_block))
445        psbt_c = psbt_with_leaf_scripts((leaf_script_c, control_block))
446
447        # The same control block under two leaf scripts would serialize as duplicate keys
448        assert_equal(combined_tap_scripts([psbt_a, psbt_b]), [tap_script(leaf_script_a, [control_block])])
449        # Reversed, so the leaf script kept is decided by the argument order and not by its content
450        assert_equal(combined_tap_scripts([psbt_b, psbt_a]), [tap_script(leaf_script_b, [control_block])])
451        # A third PSBT conflicting with what the first merge kept is dropped the same way
452        assert_equal(combined_tap_scripts([psbt_a, psbt_b, psbt_c]), [tap_script(leaf_script_a, [control_block])])
453        # Combining a PSBT with itself leaves it untouched
454        assert_equal(self.nodes[0].combinepsbt([psbt_a, psbt_a]), psbt_a)
455
456        # Records that do not conflict are all kept, whether or not they share a leaf script
457        psbt_same_leaf = psbt_with_leaf_scripts((leaf_script_a, control_block_with_path))
458        assert_equal(combined_tap_scripts([psbt_a, psbt_same_leaf]), [tap_script(leaf_script_a, [control_block, control_block_with_path])])
459        psbt_other_leaf = psbt_with_leaf_scripts((leaf_script_b, control_block_with_path))
460        assert_equal(combined_tap_scripts([psbt_a, psbt_other_leaf]), [
461            tap_script(leaf_script_a, [control_block]),
462            tap_script(leaf_script_b, [control_block_with_path]),
463        ])
464
465        # Only the conflicting control block of an incoming leaf script is dropped, not all of them
466        psbt_leaf_a_two_blocks = psbt_with_leaf_scripts((leaf_script_a, control_block), (leaf_script_a, control_block_with_path))
467        psbt_leaf_b_two_blocks = psbt_with_leaf_scripts((leaf_script_b, control_block_with_path), (leaf_script_b, control_block_with_longer_path))
468        assert_equal(combined_tap_scripts([psbt_leaf_a_two_blocks, psbt_leaf_b_two_blocks]), [
469            tap_script(leaf_script_a, [control_block, control_block_with_path]),
470            tap_script(leaf_script_b, [control_block_with_longer_path]),
471        ])
472
473    def test_sighash_mismatch(self):
474        self.log.info("Test sighash type mismatches")
475        self.nodes[0].createwallet("sighash_mismatch")
476        wallet = self.nodes[0].get_wallet_rpc("sighash_mismatch")
477        def_wallet = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
478
479        addr = wallet.getnewaddress(address_type="bech32")
480        def_wallet.sendtoaddress(addr, 5)
481        self.generate(self.nodes[0], 6)
482
483        # Retrieve the descriptors so we can do all of the tests with descriptorprocesspsbt as well
484        descs = wallet.listdescriptors(True)["descriptors"]
485
486        # Make a PSBT
487        psbt = wallet.walletcreatefundedpsbt([], [{def_wallet.getnewaddress(): 1}])["psbt"]
488
489        # Modify the PSBT and insert a sighash field for ALL|ANYONECANPAY on input 0
490        mod_psbt = PSBT.from_base64(psbt)
491        mod_psbt.i[0].map[PSBT_IN_SIGHASH_TYPE] = (SIGHASH_ALL | SIGHASH_ANYONECANPAY).to_bytes(4, byteorder="little")
492        psbt = mod_psbt.to_base64()
493
494        # Mismatching sighash type fails, including when no type is specified
495        for sighash in ["DEFAULT", "ALL", "NONE", "SINGLE", "NONE|ANYONECANPAY", "SINGLE|ANYONECANPAY", None]:
496            assert_raises_rpc_error(-22, "Specified sighash value does not match value stored in PSBT", wallet.walletprocesspsbt, psbt, True, sighash)
497
498        # Matching sighash type succeeds
499        proc = wallet.walletprocesspsbt(psbt, True, "ALL|ANYONECANPAY")
500        assert_equal(proc["complete"], True)