qa: functional tests lianad using Taproot descriptors

We introduce Taproot support in the test framework through a global
toggle. A few modifications are made to some tests to adapt them under
Taproot (notably the hardcoded fees / amounts).

This is based on my introduction of a quick and dirty support for
TapMiniscript in my python-bip380 library:
darosior/python-bip380#23. In addition to this i
didn't want to implement a signer in the Python test suite so here we
introduce a simple Rust program based on our "hot signer" which will
sign a PSBT with an xpriv provided through its stdin and output the
signed PSBT on its stdout. Eventually it would be nicer to have a Python
signer instead of having to call a program.

The whole test suite should pass under both Taproot and P2WSH. Only a
single test is skipped for now under Taproot since it needs a finalizer
in the test suite.

I also caught a bug in the RBF tests which i fixed in place.

tests/README.md

@@ -44,6 +44,16 @@ From the root of the repository:
 pytest tests/
 ```
 
+For running the tests under Taproot a `bitcoind` version 26.0 or superior must be used. It can be
+pointed to using the `BITCOIND_PATH` variable. For now, one must also compile the `taproot_signer`
+Rust program:
+```
+(cd tests/tools/taproot_signer && cargo build --release)
+```
+
+Then the test suite can be run by using Taproot descriptors instead of P2WSH descriptors by setting
+the `USE_TAPROOT` environment variable to `1`.
+
 ### Tips and tricks
 #### Logging
 

tests/fixtures.py

@@ -5,10 +5,9 @@
 from test_framework.bitcoind import Bitcoind
 from test_framework.lianad import Lianad
 from test_framework.signer import SingleSigner, MultiSigner
-from test_framework.utils import (
-    EXECUTOR_WORKERS,
-)
+from test_framework.utils import EXECUTOR_WORKERS, USE_TAPROOT
 
+import hashlib
 import os
 import pytest
 import shutil
@@ -120,22 +119,36 @@ def xpub_fingerprint(hd):
     return _pubkey_to_fingerprint(hd.pubkey).hex()
 
 
+def single_key_desc(prim_fg, prim_xpub, reco_fg, reco_xpub, csv_value, is_taproot):
+    if is_taproot:
+        return f"tr([{prim_fg}]{prim_xpub}/<0;1>/*,and_v(v:pk([{reco_fg}]{reco_xpub}/<0;1>/*),older({csv_value})))"
+    else:
+        return f"wsh(or_d(pk([{prim_fg}]{prim_xpub}/<0;1>/*),and_v(v:pkh([{reco_fg}]{reco_xpub}/<0;1>/*),older({csv_value}))))"
+
+
 @pytest.fixture
 def lianad(bitcoind, directory):
     datadir = os.path.join(directory, "lianad")
     os.makedirs(datadir, exist_ok=True)
     bitcoind_cookie = os.path.join(bitcoind.bitcoin_dir, "regtest", ".cookie")
 
-    signer = SingleSigner()
+    signer = SingleSigner(is_taproot=USE_TAPROOT)
     (prim_fingerprint, primary_xpub), (reco_fingerprint, recovery_xpub) = (
         (xpub_fingerprint(signer.primary_hd), signer.primary_hd.get_xpub()),
         (xpub_fingerprint(signer.recovery_hd), signer.recovery_hd.get_xpub()),
     )
     csv_value = 10
-    # NOTE: origins are the actual xpub themselves which is incorrect but make it
+    # NOTE: origins are the actual xpub themselves which is incorrect but makes it
     # possible to differentiate them.
     main_desc = Descriptor.from_str(
-        f"wsh(or_d(pk([{prim_fingerprint}]{primary_xpub}/<0;1>/*),and_v(v:pkh([{reco_fingerprint}]{recovery_xpub}/<0;1>/*),older({csv_value}))))"
+        single_key_desc(
+            prim_fingerprint,
+            primary_xpub,
+            reco_fingerprint,
+            recovery_xpub,
+            csv_value,
+            is_taproot=USE_TAPROOT,
+        )
     )
 
     lianad = Lianad(
@@ -156,8 +169,19 @@ def lianad(bitcoind, directory):
     lianad.cleanup()
 
 
-def multi_expression(thresh, keys):
-    exp = f"multi({thresh},"
+def unspendable_internal_xpub(xpubs):
+    """Deterministic, unique, unspendable internal key.
+    See See https://delvingbitcoin.org/t/unspendable-keys-in-descriptors/304/21
+    """
+    chaincode = hashlib.sha256(b"".join(xpub.pubkey for xpub in xpubs)).digest()
+    bip341_nums = bytes.fromhex(
+        "0250929b74c1a04954b78b4b6035e97a5e078a5a0f28ec96d547bfee9ace803ac0"
+    )
+    return BIP32(chaincode, pubkey=bip341_nums, network="test")
+
+
+def multi_expression(thresh, keys, is_taproot):
+    exp = f"multi_a({thresh}," if is_taproot else f"multi({thresh},"
     for i, key in enumerate(keys):
         # NOTE: origins are the actual xpub themselves which is incorrect but make it
         # possible to differentiate them.
@@ -168,6 +192,21 @@ def multi_expression(thresh, keys):
     return exp + ")"
 
 
+def multisig_desc(multi_signer, csv_value, is_taproot):
+    prim_multi, recov_multi = (
+        multi_expression(3, multi_signer.prim_hds, is_taproot),
+        multi_expression(2, multi_signer.recov_hds[csv_value], is_taproot),
+    )
+    if is_taproot:
+        all_xpubs = [
+            hd for hd in multi_signer.prim_hds + multi_signer.recov_hds[csv_value]
+        ]
+        internal_key = unspendable_internal_xpub(all_xpubs).get_xpub()
+        return f"tr([00000000]{internal_key}/<0;1>/*,{{{prim_multi},and_v(v:{recov_multi},older({csv_value}))}})"
+    else:
+        return f"wsh(or_d({prim_multi},and_v(v:{recov_multi},older({csv_value}))))"
+
+
 @pytest.fixture
 def lianad_multisig(bitcoind, directory):
     datadir = os.path.join(directory, "lianad")
@@ -176,13 +215,9 @@ def lianad_multisig(bitcoind, directory):
 
     # A 3-of-4 that degrades into a 2-of-5 after 10 blocks
     csv_value = 10
-    signer = MultiSigner(4, {csv_value: 5})
-    prim_multi, recov_multi = (
-        multi_expression(3, signer.prim_hds),
-        multi_expression(2, signer.recov_hds[csv_value]),
-    )
+    signer = MultiSigner(4, {csv_value: 5}, is_taproot=USE_TAPROOT)
     main_desc = Descriptor.from_str(
-        f"wsh(or_d({prim_multi},and_v(v:{recov_multi},older({csv_value}))))"
+        multisig_desc(signer, csv_value, is_taproot=USE_TAPROOT)
     )
 
     lianad = Lianad(
@@ -203,6 +238,28 @@ def lianad_multisig(bitcoind, directory):
     lianad.cleanup()
 
 
+def multipath_desc(multi_signer, csv_values, is_taproot):
+    prim_multi = multi_expression(3, multi_signer.prim_hds, is_taproot)
+    first_recov_multi = multi_expression(
+        3, multi_signer.recov_hds[csv_values[0]], is_taproot
+    )
+    second_recov_multi = multi_expression(
+        1, multi_signer.recov_hds[csv_values[1]], is_taproot
+    )
+    if is_taproot:
+        all_xpubs = [
+            hd
+            for hd in multi_signer.prim_hds
+            + multi_signer.recov_hds[csv_values[0]]
+            + multi_signer.recov_hds[csv_values[1]]
+        ]
+        internal_key = unspendable_internal_xpub(all_xpubs).get_xpub()
+        # On purpose we use a single leaf instead of 3 different ones. It shouldn't be an issue.
+        return f"tr([00000000]{internal_key}/<0;1>/*,or_d({prim_multi},or_i(and_v(v:{first_recov_multi},older({csv_values[0]})),and_v(v:{second_recov_multi},older({csv_values[1]})))))"
+    else:
+        return f"wsh(or_d({prim_multi},or_i(and_v(v:{first_recov_multi},older({csv_values[0]})),and_v(v:{second_recov_multi},older({csv_values[1]})))))"
+
+
 @pytest.fixture
 def lianad_multipath(bitcoind, directory):
     datadir = os.path.join(directory, "lianad")
@@ -211,12 +268,11 @@ def lianad_multipath(bitcoind, directory):
 
     # A 3-of-4 that degrades into a 3-of-5 after 10 blocks and into a 1-of-10 after 20 blocks.
     csv_values = [10, 20]
-    signer = MultiSigner(4, {csv_values[0]: 5, csv_values[1]: 10})
-    prim_multi = multi_expression(3, signer.prim_hds)
-    first_recov_multi = multi_expression(3, signer.recov_hds[csv_values[0]])
-    second_recov_multi = multi_expression(1, signer.recov_hds[csv_values[1]])
+    signer = MultiSigner(
+        4, {csv_values[0]: 5, csv_values[1]: 10}, is_taproot=USE_TAPROOT
+    )
     main_desc = Descriptor.from_str(
-        f"wsh(or_d({prim_multi},or_i(and_v(v:{first_recov_multi},older({csv_values[0]})),and_v(v:{second_recov_multi},older({csv_values[1]})))))"
+        multipath_desc(signer, csv_values, is_taproot=USE_TAPROOT)
     )
 
     lianad = Lianad(

tests/test_chain.py

@@ -9,6 +9,7 @@
     COIN,
     sign_and_broadcast,
     sign_and_broadcast_psbt,
+    USE_TAPROOT,
 )
 from test_framework.serializations import PSBT
 
@@ -353,6 +354,9 @@ def test_rescan_and_recovery(lianad, bitcoind):
     sign_and_broadcast(lianad, bitcoind, reco_psbt, recovery=True)
 
 
+@pytest.mark.skipif(
+    USE_TAPROOT, reason="Needs a finalizer implemented in the Python test framework."
+)
 def test_conflicting_unconfirmed_spend_txs(lianad, bitcoind):
     """Test we'll update the spending txid of a coin if a conflicting spend enters our mempool."""
     # Get an (unconfirmed, on purpose) coin to be spent by 2 different txs.

tests/test_framework/signer.py

@@ -1,5 +1,6 @@
 import logging
 import os
+import subprocess
 
 from bip32 import BIP32
 from bip32.utils import coincurve
@@ -9,10 +10,16 @@
     PSBT_IN_BIP32_DERIVATION,
     PSBT_IN_WITNESS_SCRIPT,
     PSBT_IN_PARTIAL_SIG,
+    PSBT_IN_TAP_KEY_SIG,
+    PSBT_IN_TAP_SCRIPT_SIG,
+    PSBT_IN_TAP_LEAF_SCRIPT,
+    PSBT_IN_TAP_BIP32_DERIVATION,
+    PSBT_IN_TAP_INTERNAL_KEY,
+    PSBT_IN_TAP_MERKLE_ROOT,
 )
 
 
-def sign_psbt(psbt, hds):
+def sign_psbt_wsh(psbt, hds):
     """Sign a transaction.
 
     This will fill the 'partial_sigs' field of all inputs.
@@ -58,12 +65,47 @@ def sign_psbt(psbt, hds):
     return psbt
 
 
+def sign_psbt_taproot(psbt, hds):
+    """Sign a transaction.
+
+    This will fill the 'tap_script_sig' / 'tap_key_sig' field of all inputs.
+
+    :param psbt: PSBT of the transaction to be signed.
+    :param hds: the BIP32 objects to sign the transaction with.
+    :returns: PSBT with a signature in each input for the given keys.
+    """
+    assert isinstance(psbt, PSBT)
+
+    # This file is under tests/test_framework/ and we want tests/tools/taproot_signer/target/release/taproot_signer.
+    bin_path = os.path.join(
+        os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+        "tools",
+        "taproot_signer",
+        "target",
+        "release",
+        "taproot_signer",
+    )
+    if not os.path.exists(bin_path):
+        raise Exception(
+            "Please compile the Taproot signer under tests/tools using 'cargo bin --release'."
+        )
+
+    psbt_str = psbt.to_base64()
+    for hd in hds:
+        xprv = hd.get_xpriv()
+        proc = subprocess.run([bin_path, psbt_str, xprv], capture_output=True, check=True)
+        psbt_str = proc.stdout.decode("utf-8")
+
+    return PSBT.from_base64(psbt_str)
+
+
 class SingleSigner:
     """Assumes a simple 1-primary path 1-recovery path Liana descriptor."""
 
-    def __init__(self):
+    def __init__(self, is_taproot):
         self.primary_hd = BIP32.from_seed(os.urandom(32), network="test")
         self.recovery_hd = BIP32.from_seed(os.urandom(32), network="test")
+        self.is_taproot = is_taproot
 
     def sign_psbt(self, psbt, recovery=False):
         """Sign a transaction.
@@ -75,13 +117,17 @@ def sign_psbt(self, psbt, recovery=False):
         :returns: PSBT with a signature in each input for the specified key.
         """
         assert isinstance(recovery, bool)
-        return sign_psbt(psbt, [self.recovery_hd if recovery else self.primary_hd])
+        if self.is_taproot:
+            return sign_psbt_taproot(
+                psbt, [self.recovery_hd if recovery else self.primary_hd]
+            )
+        return sign_psbt_wsh(psbt, [self.recovery_hd if recovery else self.primary_hd])
 
 
 class MultiSigner:
     """A signer that has multiple keys and may have multiple recovery path."""
 
-    def __init__(self, primary_hds_count, recovery_hds_counts):
+    def __init__(self, primary_hds_count, recovery_hds_counts, is_taproot):
         self.prim_hds = [
             BIP32.from_seed(os.urandom(32), network="test")
             for _ in range(primary_hds_count)
@@ -91,6 +137,7 @@ def __init__(self, primary_hds_count, recovery_hds_counts):
             self.recov_hds[timelock] = [
                 BIP32.from_seed(os.urandom(32), network="test") for _ in range(count)
             ]
+        self.is_taproot = is_taproot
 
     def sign_psbt(self, psbt, key_indices):
         """Sign a transaction with the keys at the specified indices.
@@ -106,4 +153,6 @@ def sign_psbt(self, psbt, key_indices):
             ]
         else:
             hds = [self.prim_hds[i] for i in key_indices]
-        return sign_psbt(psbt, hds)
+        if self.is_taproot:
+            return sign_psbt_taproot(psbt, hds)
+        return sign_psbt_wsh(psbt, hds)

tests/test_framework/utils.py

@@ -24,6 +24,7 @@
 BITCOIND_PATH = os.getenv("BITCOIND_PATH", DEFAULT_BITCOIND_PATH)
 OLD_LIANAD_PATH = os.getenv("OLD_LIANAD_PATH", None)
 IS_NOT_BITCOIND_24 = bool(int(os.getenv("IS_NOT_BITCOIND_24", True)))
+USE_TAPROOT = bool(int(os.getenv("USE_TAPROOT", False)))  # TODO: switch to True in a couple releases.
 
 
 COIN = 10**8
@@ -55,6 +56,21 @@ def get_txid(hex_tx):
     return tx.txid().hex()
 
 
+def sign_and_broadcast(lianad, bitcoind, psbt, recovery=False):
+    """Sign a PSBT, finalize it, extract the transaction and broadcast it."""
+    signed_psbt = lianad.signer.sign_psbt(psbt, recovery)
+    # Under Taproot i didn't bother implementing a finalizer in the test suite.
+    if USE_TAPROOT:
+        lianad.rpc.updatespend(signed_psbt.to_base64())
+        txid = signed_psbt.tx.txid().hex()
+        lianad.rpc.broadcastspend(txid)
+        lianad.rpc.delspendtx(txid)
+        return txid
+    finalized_psbt = lianad.finalize_psbt(signed_psbt)
+    tx = finalized_psbt.tx.serialize_with_witness().hex()
+    return bitcoind.rpc.sendrawtransaction(tx)
+
+
 def spend_coins(lianad, bitcoind, coins):
     """Spend these coins, no matter how.
     This will create a single transaction spending them all at once at the minimum
@@ -68,21 +84,8 @@ def spend_coins(lianad, bitcoind, coins):
         bitcoind.rpc.getnewaddress(): total_value - 11 - 31 - 300 * len(coins)
     }
     res = lianad.rpc.createspend(destinations, [c["outpoint"] for c in coins], 1)
-
-    signed_psbt = lianad.signer.sign_psbt(PSBT.from_base64(res["psbt"]))
-    finalized_psbt = lianad.finalize_psbt(signed_psbt)
-    tx = finalized_psbt.tx.serialize_with_witness().hex()
-    bitcoind.rpc.sendrawtransaction(tx)
-
-    return tx
-
-
-def sign_and_broadcast(lianad, bitcoind, psbt, recovery=False):
-    """Sign a PSBT, finalize it, extract the transaction and broadcast it."""
-    signed_psbt = lianad.signer.sign_psbt(psbt, recovery)
-    finalized_psbt = lianad.finalize_psbt(signed_psbt)
-    tx = finalized_psbt.tx.serialize_with_witness().hex()
-    return bitcoind.rpc.sendrawtransaction(tx)
+    txid = sign_and_broadcast(lianad, bitcoind, PSBT.from_base64(res["psbt"]))
+    return bitcoind.rpc.getrawtransaction(txid)
 
 
 def sign_and_broadcast_psbt(lianad, psbt):

tests/test_misc.py

@@ -66,9 +66,7 @@ def receive_and_send(lianad, bitcoind):
     psbt = PSBT.from_base64(res["psbt"])
     txid = psbt.tx.txid().hex()
     # If we sign only with two keys it won't be able to finalize
-    with pytest.raises(
-        RpcError, match="Miniscript Error: could not satisfy at index 0"
-    ):
+    with pytest.raises(RpcError, match="ould not satisfy.* at index 0"):
         signed_psbt = lianad.signer.sign_psbt(psbt, range(2))
         lianad.rpc.updatespend(signed_psbt.to_base64())
         lianad.rpc.broadcastspend(txid)