diff --git a/tests/fixtures.py b/tests/fixtures.py index 4a6b6ebb..5ab987c9 100644 --- a/tests/fixtures.py +++ b/tests/fixtures.py @@ -3,6 +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.utils import ( EXECUTOR_WORKERS, ) @@ -118,18 +119,19 @@ def lianad(bitcoind, directory): os.makedirs(datadir, exist_ok=True) bitcoind_cookie = os.path.join(bitcoind.bitcoin_dir, "regtest", ".cookie") - owner_hd = BIP32.from_seed(os.urandom(32), network="test") - recovery_hd = BIP32.from_seed(os.urandom(32), network="test") - owner_xpub, recovery_xpub = owner_hd.get_xpub(), recovery_hd.get_xpub() + signer = SingleSigner() + primary_xpub, recovery_xpub = ( + signer.primary_hd.get_xpub(), + signer.recovery_hd.get_xpub(), + ) csv_value = 10 main_desc = Descriptor.from_str( - f"wsh(or_d(pk({owner_xpub}/<0;1>/*),and_v(v:pkh({recovery_xpub}/<0;1>/*),older({csv_value}))))" + f"wsh(or_d(pk({primary_xpub}/<0;1>/*),and_v(v:pkh({recovery_xpub}/<0;1>/*),older({csv_value}))))" ) lianad = Lianad( datadir, - owner_hd, - recovery_hd, + signer, main_desc, bitcoind.rpcport, bitcoind_cookie, diff --git a/tests/test_framework/lianad.py b/tests/test_framework/lianad.py index 51a0007b..eb273c31 100644 --- a/tests/test_framework/lianad.py +++ b/tests/test_framework/lianad.py @@ -2,7 +2,6 @@ import logging import os import shutil -from bip32.utils import coincurve from bip380.descriptors import Descriptor from bip380.miniscript import SatisfactionMaterial from test_framework.utils import ( @@ -15,11 +14,9 @@ from test_framework.utils import ( ) from test_framework.serializations import ( PSBT, - sighash_all_witness, CTxInWitness, CScriptWitness, PSBT_IN_BIP32_DERIVATION, - PSBT_IN_WITNESS_SCRIPT, PSBT_IN_PARTIAL_SIG, PSBT_IN_FINAL_SCRIPTWITNESS, ) @@ -29,8 +26,7 @@ class Lianad(TailableProc): def __init__( self, datadir, - owner_hd, - recovery_hd, + signer, multi_desc, bitcoind_rpc_port, bitcoind_cookie_path, @@ -40,8 +36,7 @@ class Lianad(TailableProc): self.datadir = datadir self.prefix = os.path.split(datadir)[-1] - self.owner_hd = owner_hd - self.recovery_hd = recovery_hd + self.signer = signer self.multi_desc = multi_desc self.receive_desc, self.change_desc = multi_desc.singlepath_descriptors() @@ -65,51 +60,6 @@ class Lianad(TailableProc): f.write(f"cookie_path = '{bitcoind_cookie_path}'\n") f.write(f"addr = '127.0.0.1:{bitcoind_rpc_port}'\n") - def sign_psbt(self, psbt, recovery=False): - """Sign a transaction. - - This will fill the 'partial_sigs' field of all inputs. Uses either the 'primary' - '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. - """ - assert isinstance(psbt, PSBT) - - # Which key to sign the transaction with. - hd = self.recovery_hd if recovery else self.owner_hd - - # Sign each input. - for i, psbt_in in enumerate(psbt.i): - # First, gather the needed information from the PSBT input. - # 'hd_keypaths' is of the form {pubkey: (fingerprint (4 bytes), derivation path (n * 4 bytes))} - fing_der = next(iter(psbt_in.map[PSBT_IN_BIP32_DERIVATION].values())) - raw_der_path = fing_der[4:] - der_path = [ - int.from_bytes(raw_der_path[i : i + 4], byteorder="little", signed=True) - for i in range(0, len(raw_der_path), 4) - ] - script_code = psbt_in.map[PSBT_IN_WITNESS_SCRIPT] - - # Now sign the transaction. - sighash = sighash_all_witness(script_code, psbt, i) - privkey = coincurve.PrivateKey(hd.get_privkey_from_path(der_path)) - pubkey = privkey.public_key.format() - assert pubkey in psbt_in.map[PSBT_IN_BIP32_DERIVATION].keys(), ( - der_path, - fing_der, - pubkey, - psbt_in.map[PSBT_IN_BIP32_DERIVATION].keys(), - ) - sig = privkey.sign(sighash, hasher=None) + b"\x01" - 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} - - return psbt - def finalize_psbt(self, psbt): """Create a valid witness for all inputs in the PSBT. This will fail if the PSBT input does not contain enough material. diff --git a/tests/test_framework/signer.py b/tests/test_framework/signer.py new file mode 100644 index 00000000..b14a13d2 --- /dev/null +++ b/tests/test_framework/signer.py @@ -0,0 +1,73 @@ +import logging +import os + +from bip32 import BIP32 +from bip32.utils import coincurve +from test_framework.serializations import ( + PSBT, + sighash_all_witness, + PSBT_IN_BIP32_DERIVATION, + PSBT_IN_WITNESS_SCRIPT, + PSBT_IN_PARTIAL_SIG, +) + + +def sign_psbt(psbt, hds): + """Sign a transaction. + + This will fill the 'partial_sigs' 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 owner's key. + """ + assert isinstance(psbt, PSBT) + + # Sign each input. + for i, psbt_in in enumerate(psbt.i): + # First, gather the needed information from the PSBT input. + # 'hd_keypaths' is of the form {pubkey: (fingerprint (4 bytes), derivation path (n * 4 bytes))} + fing_der = next(iter(psbt_in.map[PSBT_IN_BIP32_DERIVATION].values())) + raw_der_path = fing_der[4:] + der_path = [ + int.from_bytes(raw_der_path[i : i + 4], byteorder="little", signed=True) + for i in range(0, len(raw_der_path), 4) + ] + script_code = psbt_in.map[PSBT_IN_WITNESS_SCRIPT] + + # Now sign the transaction for all the given keys. + for hd in hds: + sighash = sighash_all_witness(script_code, psbt, i) + privkey = coincurve.PrivateKey(hd.get_privkey_from_path(der_path)) + pubkey = privkey.public_key.format() + assert pubkey in psbt_in.map[PSBT_IN_BIP32_DERIVATION].keys(), ( + der_path, + fing_der, + pubkey, + psbt_in.map[PSBT_IN_BIP32_DERIVATION].keys(), + ) + sig = privkey.sign(sighash, hasher=None) + b"\x01" + 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} + + return psbt + + +class SingleSigner: + def __init__(self): + self.primary_hd = BIP32.from_seed(os.urandom(32), network="test") + self.recovery_hd = BIP32.from_seed(os.urandom(32), network="test") + + def sign_psbt(self, psbt, recovery=False): + """Sign a transaction. + + This will fill the 'partial_sigs' field of all inputs. Uses either the 'primary' + '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. + """ + return sign_psbt(psbt, [self.recovery_hd if recovery else self.primary_hd]) diff --git a/tests/test_framework/utils.py b/tests/test_framework/utils.py index e506e676..75862282 100644 --- a/tests/test_framework/utils.py +++ b/tests/test_framework/utils.py @@ -67,7 +67,7 @@ def spend_coins(lianad, bitcoind, coins): } res = lianad.rpc.createspend(destinations, [c["outpoint"] for c in coins], 1) - signed_psbt = lianad.sign_psbt(PSBT.from_base64(res["psbt"])) + 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) @@ -77,7 +77,7 @@ def spend_coins(lianad, bitcoind, coins): def sign_and_broadcast(lianad, bitcoind, psbt, recovery=False): """Sign a PSBT, finalize it, extract the transaction and broadcast it.""" - signed_psbt = lianad.sign_psbt(psbt, recovery) + 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) diff --git a/tests/test_rpc.py b/tests/test_rpc.py index 0b45e178..83f035c6 100644 --- a/tests/test_rpc.py +++ b/tests/test_rpc.py @@ -262,7 +262,7 @@ def test_broadcast_spend(lianad, bitcoind): # We can't broadcast an unsigned transaction with pytest.raises(RpcError, match="Failed to finalize the spend transaction.*"): lianad.rpc.broadcastspend(txid) - signed_psbt = lianad.sign_psbt(PSBT.from_base64(res["psbt"])) + signed_psbt = lianad.signer.sign_psbt(PSBT.from_base64(res["psbt"])) lianad.rpc.updatespend(signed_psbt.to_base64()) # Now we've signed and stored it, the daemon will take care of finalizing @@ -372,7 +372,7 @@ def test_listtransactions(lianad, bitcoind): def sign_and_broadcast(psbt): txid = psbt.tx.txid().hex() - psbt = lianad.sign_psbt(psbt) + psbt = lianad.signer.sign_psbt(psbt) lianad.rpc.updatespend(psbt.to_base64()) lianad.rpc.broadcastspend(txid) return txid diff --git a/tests/test_spend.py b/tests/test_spend.py index 2d9959ff..93fb7596 100644 --- a/tests/test_spend.py +++ b/tests/test_spend.py @@ -27,7 +27,7 @@ def test_spend_change(lianad, bitcoind): assert len(spend_psbt.tx.vout) == 3 # Sign and broadcast this first Spend transaction. - signed_psbt = lianad.sign_psbt(spend_psbt) + signed_psbt = lianad.signer.sign_psbt(spend_psbt) lianad.rpc.updatespend(signed_psbt.to_base64()) spend_txid = signed_psbt.tx.txid().hex() lianad.rpc.broadcastspend(spend_txid) @@ -48,7 +48,7 @@ def test_spend_change(lianad, bitcoind): spend_psbt = PSBT.from_base64(res["psbt"]) # We can sign and broadcast it. - signed_psbt = lianad.sign_psbt(spend_psbt) + signed_psbt = lianad.signer.sign_psbt(spend_psbt) lianad.rpc.updatespend(signed_psbt.to_base64()) spend_txid = signed_psbt.tx.txid().hex() lianad.rpc.broadcastspend(spend_txid) @@ -85,7 +85,7 @@ def test_coin_marked_spent(lianad, bitcoind): def sign_and_broadcast(psbt): txid = psbt.tx.txid().hex() - psbt = lianad.sign_psbt(psbt) + psbt = lianad.signer.sign_psbt(psbt) lianad.rpc.updatespend(psbt.to_base64()) lianad.rpc.broadcastspend(txid) return txid