nik/liana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

qa: abstract the signer from the Lianad class

Browse Source 
That's prep work to introduce a multisig
This commit is contained in:
Antoine Poinsot 2023-01-12 17:10:31 +01:00
parent e22e30dc8d
commit 3c82173f46
No known key found for this signature in database
GPG Key ID: E13FC145CD3F4304
6 changed files with 90 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
tests/fixtures.py
View File

@ -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,

54
tests/test_framework/lianad.py
View File

@ -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.

73
tests/test_framework/signer.py Normal file
View File

@ -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])

4
tests/test_framework/utils.py
View File

@ -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)

4
tests/test_rpc.py
View File

@ -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

6
tests/test_spend.py
View File

@ -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