qa: test using lianad with a multisig descriptor

tests/fixtures.py

@@ -3,7 +3,7 @@ from bip380.descriptors import Descriptor
 from concurrent import futures
 from test_framework.bitcoind import Bitcoind
 from test_framework.lianad import Lianad
-from test_framework.signer import SingleSigner
+from test_framework.signer import SingleSigner, MultiSigner
 from test_framework.utils import (
     EXECUTOR_WORKERS,
 )
@@ -145,3 +145,47 @@ def lianad(bitcoind, directory):
         raise
 
     lianad.cleanup()
+
+
+def multi_expression(thresh, keys):
+    exp = f"multi({thresh},"
+    for i, key in enumerate(keys):
+        exp += f"{key.get_xpub()}/<0;1>/*"
+        if i != len(keys) - 1:
+            exp += ","
+    return exp + ")"
+
+
+@pytest.fixture
+def lianad_multisig(bitcoind, directory):
+    datadir = os.path.join(directory, "lianad")
+    os.makedirs(datadir, exist_ok=True)
+    bitcoind_cookie = os.path.join(bitcoind.bitcoin_dir, "regtest", ".cookie")
+
+    # A 3-of-4 that degrades into a 2-of-5 after 10 blocks
+    signer = MultiSigner(3, 4, 2, 5)
+    csv_value = 10
+    prim_multi, recov_multi = (
+        multi_expression(signer.prim_thresh, signer.prim_hds),
+        multi_expression(signer.recov_thresh, signer.recov_hds),
+    )
+    main_desc = Descriptor.from_str(
+        f"wsh(or_d({prim_multi},and_v(v:{recov_multi},older({csv_value}))))"
+    )
+
+    lianad = Lianad(
+        datadir,
+        signer,
+        main_desc,
+        bitcoind.rpcport,
+        bitcoind_cookie,
+    )
+
+    try:
+        lianad.start()
+        yield lianad
+    except Exception:
+        lianad.cleanup()
+        raise
+
+    lianad.cleanup()

tests/test_framework/signer.py

@@ -19,7 +19,7 @@ def sign_psbt(psbt, hds):
 
     :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 owner's key.
+    :returns: PSBT with a signature in each input for the given keys.
     """
     assert isinstance(psbt, PSBT)
 
@@ -50,8 +50,10 @@ def sign_psbt(psbt, hds):
             logging.debug(
                 f"Adding signature {sig.hex()} for pubkey {pubkey.hex()} (path {der_path})"
             )
-            assert PSBT_IN_PARTIAL_SIG not in psbt_in.map
-            psbt_in.map[PSBT_IN_PARTIAL_SIG] = {pubkey: sig}
+            if PSBT_IN_PARTIAL_SIG not in psbt_in.map:
+                psbt_in.map[PSBT_IN_PARTIAL_SIG] = {pubkey: sig}
+            else:
+                psbt_in.map[PSBT_IN_PARTIAL_SIG][pubkey] = sig
 
     return psbt
 
@@ -68,6 +70,28 @@ class SingleSigner:
         'recovery' key as specified.
 
         :param psbt: PSBT of the transaction to be signed.
-        :returns: PSBT with a signature in each input for the owner's key.
+        :returns: PSBT with a signature in each input for the specified key.
         """
         return sign_psbt(psbt, [self.recovery_hd if recovery else self.primary_hd])
+
+
+class MultiSigner:
+    def __init__(
+        self, primary_thresh, primary_hds_count, recovery_thresh, recovery_hds_count
+    ):
+        self.prim_thresh = primary_thresh
+        self.prim_hds = [
+            BIP32.from_seed(os.urandom(32), network="test")
+            for _ in range(primary_hds_count)
+        ]
+        self.recov_thresh = recovery_thresh
+        self.recov_hds = [
+            BIP32.from_seed(os.urandom(32), network="test")
+            for _ in range(recovery_hds_count)
+        ]
+
+    def sign_psbt(self, psbt, key_indices, recovery=False):
+        """Sign a transaction with the keys at the specified indices."""
+        hds = self.recov_hds if recovery else self.prim_hds
+        hds = [hds[i] for i in key_indices]
+        return sign_psbt(psbt, hds)

tests/test_misc.py

@@ -1,5 +1,74 @@
+import pytest
+
 from fixtures import *
+from test_framework.serializations import PSBT
+from test_framework.utils import wait_for, RpcError
 
 
-def test_startup(lianad):
-    pass
+def test_multisig(lianad_multisig, bitcoind):
+    """Test using lianad with a descriptor that contains multiple keys for both
+    the primary and recovery paths."""
+    lianad = lianad_multisig
+
+    # Receive 3 coins in different blocks on different addresses.
+    for _ in range(3):
+        addr = lianad.rpc.getnewaddress()["address"]
+        txid = bitcoind.rpc.sendtoaddress(addr, 0.01)
+        bitcoind.generate_block(1, wait_for_mempool=txid)
+    wait_for(lambda: len(lianad.rpc.listcoins()["coins"]) == 3)
+
+    print(lianad.rpc.listcoins())
+    # Create a spend that will create a change output, sign and broadcast it.
+    outpoints = [lianad.rpc.listcoins()["coins"][0]["outpoint"]]
+    destinations = {
+        bitcoind.rpc.getnewaddress(): 200_000,
+    }
+    res = lianad.rpc.createspend(destinations, outpoints, 42)
+    psbt = PSBT.from_base64(res["psbt"])
+    txid = psbt.tx.txid().hex()
+    signed_psbt = lianad.signer.sign_psbt(psbt, range(3))
+    lianad.rpc.updatespend(signed_psbt.to_base64())
+    lianad.rpc.broadcastspend(txid)
+    bitcoind.generate_block(1, wait_for_mempool=txid)
+    wait_for(
+        lambda: lianad.rpc.getinfo()["block_height"] == bitcoind.rpc.getblockcount()
+    )
+
+    # Spend all coins to check we can spend from change too. Re-create some deposits.
+    outpoints = [
+        c["outpoint"]
+        for c in lianad.rpc.listcoins()["coins"]
+        if c["spend_info"] is None
+    ]
+    destinations = {
+        bitcoind.rpc.getnewaddress(): 400_000,
+        lianad.rpc.getnewaddress()["address"]: 300_000,
+        lianad.rpc.getnewaddress()["address"]: 800_000,
+    }
+    res = lianad.rpc.createspend(destinations, outpoints, 42)
+    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"):
+        signed_psbt = lianad.signer.sign_psbt(psbt, range(2))
+        lianad.rpc.updatespend(signed_psbt.to_base64())
+        lianad.rpc.broadcastspend(txid)
+    # We can sign with different keys as long as there are 3 sigs
+    signed_psbt = lianad.signer.sign_psbt(psbt, range(1, 4))
+    lianad.rpc.updatespend(signed_psbt.to_base64())
+    lianad.rpc.broadcastspend(txid)
+
+    # Generate 10 blocks to test the recovery path
+    bitcoind.generate_block(10, wait_for_mempool=txid)
+    wait_for(
+        lambda: lianad.rpc.getinfo()["block_height"] == bitcoind.rpc.getblockcount()
+    )
+
+    # Sweep all coins through the recovery path. It needs 2 signatures out of
+    # 5 keys. Sign with the second and the fifth ones.
+    res = lianad.rpc.createrecovery(bitcoind.rpc.getnewaddress(), 2)
+    reco_psbt = PSBT.from_base64(res["psbt"])
+    txid = reco_psbt.tx.txid().hex()
+    signed_psbt = lianad.signer.sign_psbt(reco_psbt, [1, 4], recovery=True)
+    lianad.rpc.updatespend(signed_psbt.to_base64())
+    lianad.rpc.broadcastspend(txid)