mirrors/bitcoin
1
0
Fork 0
mirror of https://github.com/bitcoin/bitcoin.git synced 2024-11-19 09:53:47 +01:00
Code Issues Projects Releases Wiki Activity

tests: Use new wallets for each test in wallet_taproot.py

Browse Source 
To avoid a wallet potentially being able to sign a transaction using
keys from descriptors imported in previous tests, make new wallets for
each test case rather than sharing them.
This commit is contained in:
Andrew Chow 2022-10-28 19:00:58 -04:00
parent 8781a1b6bb
commit 6efcdf6b7f
1 changed files with 70 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

117
test/functional/wallet_taproot.py
View File

@ -5,6 +5,7 @@
"""Test generation and spending of P2TR addresses."""
import random
import uuid
from decimal import Decimal
from test_framework.address import output_key_to_p2tr
@ -229,17 +230,28 @@ class WalletTaprootTest(BitcoinTestFramework):
def do_test_addr(self, comment, pattern, privmap, treefn, keys):
self.log.info("Testing %s address derivation" % comment)
# Create wallets
wallet_uuid = uuid.uuid4().hex
self.nodes[0].createwallet(wallet_name=f"privs_tr_enabled_{wallet_uuid}", descriptors=True, blank=True)
self.nodes[0].createwallet(wallet_name=f"pubs_tr_enabled_{wallet_uuid}", descriptors=True, blank=True, disable_private_keys=True)
self.nodes[0].createwallet(wallet_name=f"addr_gen_{wallet_uuid}", descriptors=True, disable_private_keys=True, blank=True)
privs_tr_enabled = self.nodes[0].get_wallet_rpc(f"privs_tr_enabled_{wallet_uuid}")
pubs_tr_enabled = self.nodes[0].get_wallet_rpc(f"pubs_tr_enabled_{wallet_uuid}")
addr_gen = self.nodes[0].get_wallet_rpc(f"addr_gen_{wallet_uuid}")
desc = self.make_desc(pattern, privmap, keys, False)
desc_pub = self.make_desc(pattern, privmap, keys, True)
assert_equal(self.nodes[0].getdescriptorinfo(desc)['descriptor'], desc_pub)
result = self.addr_gen.importdescriptors([{"desc": desc_pub, "active": True, "timestamp": "now"}])
result = addr_gen.importdescriptors([{"desc": desc_pub, "active": True, "timestamp": "now"}])
assert(result[0]['success'])
address_type = "bech32m" if "tr" in pattern else "bech32"
for i in range(4):
addr_g = self.addr_gen.getnewaddress(address_type='bech32m')
addr_g = addr_gen.getnewaddress(address_type=address_type)
if treefn is not None:
addr_r = self.make_addr(treefn, keys, i)
assert_equal(addr_g, addr_r)
desc_a = self.addr_gen.getaddressinfo(addr_g)['desc']
desc_a = addr_gen.getaddressinfo(addr_g)['desc']
if desc.startswith("tr("):
assert desc_a.startswith("tr(")
rederive = self.nodes[1].deriveaddresses(desc_a)
@ -247,25 +259,37 @@ class WalletTaprootTest(BitcoinTestFramework):
assert_equal(rederive[0], addr_g)
# tr descriptors can be imported
result = self.privs_tr_enabled.importdescriptors([{"desc": desc, "timestamp": "now"}])
result = privs_tr_enabled.importdescriptors([{"desc": desc, "timestamp": "now"}])
assert(result[0]["success"])
result = self.pubs_tr_enabled.importdescriptors([{"desc": desc_pub, "timestamp": "now"}])
result = pubs_tr_enabled.importdescriptors([{"desc": desc_pub, "timestamp": "now"}])
assert(result[0]["success"])
# Cleanup
privs_tr_enabled.unloadwallet()
pubs_tr_enabled.unloadwallet()
addr_gen.unloadwallet()
def do_test_sendtoaddress(self, comment, pattern, privmap, treefn, keys_pay, keys_change):
self.log.info("Testing %s through sendtoaddress" % comment)
# Create wallets
wallet_uuid = uuid.uuid4().hex
self.nodes[0].createwallet(wallet_name=f"rpc_online_{wallet_uuid}", descriptors=True, blank=True)
rpc_online = self.nodes[0].get_wallet_rpc(f"rpc_online_{wallet_uuid}")
desc_pay = self.make_desc(pattern, privmap, keys_pay)
desc_change = self.make_desc(pattern, privmap, keys_change)
desc_pay_pub = self.make_desc(pattern, privmap, keys_pay, True)
desc_change_pub = self.make_desc(pattern, privmap, keys_change, True)
assert_equal(self.nodes[0].getdescriptorinfo(desc_pay)['descriptor'], desc_pay_pub)
assert_equal(self.nodes[0].getdescriptorinfo(desc_change)['descriptor'], desc_change_pub)
result = self.rpc_online.importdescriptors([{"desc": desc_pay, "active": True, "timestamp": "now"}])
result = rpc_online.importdescriptors([{"desc": desc_pay, "active": True, "timestamp": "now"}])
assert(result[0]['success'])
result = self.rpc_online.importdescriptors([{"desc": desc_change, "active": True, "timestamp": "now", "internal": True}])
result = rpc_online.importdescriptors([{"desc": desc_change, "active": True, "timestamp": "now", "internal": True}])
assert(result[0]['success'])
address_type = "bech32m" if "tr" in pattern else "bech32"
for i in range(4):
addr_g = self.rpc_online.getnewaddress(address_type='bech32m')
addr_g = rpc_online.getnewaddress(address_type=address_type)
if treefn is not None:
addr_r = self.make_addr(treefn, keys_pay, i)
assert_equal(addr_g, addr_r)
@ -273,31 +297,46 @@ class WalletTaprootTest(BitcoinTestFramework):
to_amnt = random.randrange(1000000, boring_balance)
self.boring.sendtoaddress(address=addr_g, amount=Decimal(to_amnt) / 100000000, subtractfeefromamount=True)
self.generatetoaddress(self.nodes[0], 1, self.boring.getnewaddress(), sync_fun=self.no_op)
test_balance = int(self.rpc_online.getbalance() * 100000000)
test_balance = int(rpc_online.getbalance() * 100000000)
ret_amnt = random.randrange(100000, test_balance)
# Increase fee_rate to compensate for the wallet's inability to estimate fees for script path spends.
res = self.rpc_online.sendtoaddress(address=self.boring.getnewaddress(), amount=Decimal(ret_amnt) / 100000000, subtractfeefromamount=True, fee_rate=200)
res = rpc_online.sendtoaddress(address=self.boring.getnewaddress(), amount=Decimal(ret_amnt) / 100000000, subtractfeefromamount=True, fee_rate=200)
self.generatetoaddress(self.nodes[0], 1, self.boring.getnewaddress(), sync_fun=self.no_op)
assert(self.rpc_online.gettransaction(res)["confirmations"] > 0)
assert(rpc_online.gettransaction(res)["confirmations"] > 0)
# Cleanup
txid = rpc_online.sendall(recipients=[self.boring.getnewaddress()])["txid"]
self.generatetoaddress(self.nodes[0], 1, self.boring.getnewaddress(), sync_fun=self.no_op)
assert(rpc_online.gettransaction(txid)["confirmations"] > 0)
rpc_online.unloadwallet()
def do_test_psbt(self, comment, pattern, privmap, treefn, keys_pay, keys_change):
self.log.info("Testing %s through PSBT" % comment)
# Create wallets
wallet_uuid = uuid.uuid4().hex
self.nodes[0].createwallet(wallet_name=f"psbt_online_{wallet_uuid}", descriptors=True, disable_private_keys=True, blank=True)
self.nodes[1].createwallet(wallet_name=f"psbt_offline_{wallet_uuid}", descriptors=True, blank=True)
psbt_online = self.nodes[0].get_wallet_rpc(f"psbt_online_{wallet_uuid}")
psbt_offline = self.nodes[1].get_wallet_rpc(f"psbt_offline_{wallet_uuid}")
desc_pay = self.make_desc(pattern, privmap, keys_pay, False)
desc_change = self.make_desc(pattern, privmap, keys_change, False)
desc_pay_pub = self.make_desc(pattern, privmap, keys_pay, True)
desc_change_pub = self.make_desc(pattern, privmap, keys_change, True)
assert_equal(self.nodes[0].getdescriptorinfo(desc_pay)['descriptor'], desc_pay_pub)
assert_equal(self.nodes[0].getdescriptorinfo(desc_change)['descriptor'], desc_change_pub)
result = self.psbt_online.importdescriptors([{"desc": desc_pay_pub, "active": True, "timestamp": "now"}])
result = psbt_online.importdescriptors([{"desc": desc_pay_pub, "active": True, "timestamp": "now"}])
assert(result[0]['success'])
result = self.psbt_online.importdescriptors([{"desc": desc_change_pub, "active": True, "timestamp": "now", "internal": True}])
result = psbt_online.importdescriptors([{"desc": desc_change_pub, "active": True, "timestamp": "now", "internal": True}])
assert(result[0]['success'])
result = self.psbt_offline.importdescriptors([{"desc": desc_pay, "active": True, "timestamp": "now"}])
result = psbt_offline.importdescriptors([{"desc": desc_pay, "active": True, "timestamp": "now"}])
assert(result[0]['success'])
result = self.psbt_offline.importdescriptors([{"desc": desc_change, "active": True, "timestamp": "now", "internal": True}])
result = psbt_offline.importdescriptors([{"desc": desc_change, "active": True, "timestamp": "now", "internal": True}])
assert(result[0]['success'])
address_type = "bech32m" if "tr" in pattern else "bech32"
for i in range(4):
addr_g = self.psbt_online.getnewaddress(address_type='bech32m')
addr_g = psbt_online.getnewaddress(address_type=address_type)
if treefn is not None:
addr_r = self.make_addr(treefn, keys_pay, i)
assert_equal(addr_g, addr_r)
@ -305,13 +344,13 @@ class WalletTaprootTest(BitcoinTestFramework):
to_amnt = random.randrange(1000000, boring_balance)
self.boring.sendtoaddress(address=addr_g, amount=Decimal(to_amnt) / 100000000, subtractfeefromamount=True)
self.generatetoaddress(self.nodes[0], 1, self.boring.getnewaddress(), sync_fun=self.no_op)
test_balance = int(self.psbt_online.getbalance() * 100000000)
test_balance = int(psbt_online.getbalance() * 100000000)
ret_amnt = random.randrange(100000, test_balance)
# Increase fee_rate to compensate for the wallet's inability to estimate fees for script path spends.
psbt = self.psbt_online.walletcreatefundedpsbt([], [{self.boring.getnewaddress(): Decimal(ret_amnt) / 100000000}], None, {"subtractFeeFromOutputs":[0], "fee_rate": 200, "change_type": "bech32m"})['psbt']
res = self.psbt_offline.walletprocesspsbt(psbt=psbt, finalize=False)
psbt = psbt_online.walletcreatefundedpsbt([], [{self.boring.getnewaddress(): Decimal(ret_amnt) / 100000000}], None, {"subtractFeeFromOutputs":[0], "fee_rate": 200, "change_type": address_type})['psbt']
res = psbt_offline.walletprocesspsbt(psbt=psbt, finalize=False)
decoded = self.psbt_offline.decodepsbt(res["psbt"])
decoded = psbt_offline.decodepsbt(res["psbt"])
if pattern.startswith("tr("):
for psbtin in decoded["inputs"]:
assert "non_witness_utxo" not in psbtin
@ -326,7 +365,17 @@ class WalletTaprootTest(BitcoinTestFramework):
rawtx = self.nodes[0].finalizepsbt(res['psbt'])['hex']
txid = self.nodes[0].sendrawtransaction(rawtx)
self.generatetoaddress(self.nodes[0], 1, self.boring.getnewaddress(), sync_fun=self.no_op)
assert(self.psbt_online.gettransaction(txid)['confirmations'] > 0)
assert(psbt_online.gettransaction(txid)['confirmations'] > 0)
# Cleanup
psbt = psbt_online.sendall(recipients=[self.boring.getnewaddress()], options={"psbt": True})["psbt"]
res = psbt_offline.walletprocesspsbt(psbt=psbt, finalize=False)
rawtx = self.nodes[0].finalizepsbt(res['psbt'])['hex']
txid = self.nodes[0].sendrawtransaction(rawtx)
self.generatetoaddress(self.nodes[0], 1, self.boring.getnewaddress(), sync_fun=self.no_op)
assert(psbt_online.gettransaction(txid)['confirmations'] > 0)
psbt_online.unloadwallet()
psbt_offline.unloadwallet()
def do_test(self, comment, pattern, privmap, treefn):
nkeys = len(privmap)
@ -336,21 +385,8 @@ class WalletTaprootTest(BitcoinTestFramework):
self.do_test_psbt(comment, pattern, privmap, treefn, keys[2*nkeys:3*nkeys], keys[3*nkeys:4*nkeys])
def run_test(self):
self.log.info("Creating wallets...")
self.nodes[0].createwallet(wallet_name="privs_tr_enabled", descriptors=True, blank=True)
self.privs_tr_enabled = self.nodes[0].get_wallet_rpc("privs_tr_enabled")
self.nodes[0].createwallet(wallet_name="pubs_tr_enabled", descriptors=True, blank=True, disable_private_keys=True)
self.pubs_tr_enabled = self.nodes[0].get_wallet_rpc("pubs_tr_enabled")
self.nodes[0].createwallet(wallet_name="boring")
self.nodes[0].createwallet(wallet_name="addr_gen", descriptors=True, disable_private_keys=True, blank=True)
self.nodes[0].createwallet(wallet_name="rpc_online", descriptors=True, blank=True)
self.nodes[0].createwallet(wallet_name="psbt_online", descriptors=True, disable_private_keys=True, blank=True)
self.nodes[1].createwallet(wallet_name="psbt_offline", descriptors=True, blank=True)
self.boring = self.nodes[0].get_wallet_rpc("boring")
self.addr_gen = self.nodes[0].get_wallet_rpc("addr_gen")
self.rpc_online = self.nodes[0].get_wallet_rpc("rpc_online")
self.psbt_online = self.nodes[0].get_wallet_rpc("psbt_online")
self.psbt_offline = self.nodes[1].get_wallet_rpc("psbt_offline")
self.log.info("Mining blocks...")
gen_addr = self.boring.getnewaddress()
@ -460,18 +496,5 @@ class WalletTaprootTest(BitcoinTestFramework):
lambda k1: key(k1)
)
self.log.info("Sending everything back...")
txid = self.rpc_online.sendall(recipients=[self.boring.getnewaddress()])["txid"]
self.generatetoaddress(self.nodes[0], 1, self.boring.getnewaddress(), sync_fun=self.no_op)
assert(self.rpc_online.gettransaction(txid)["confirmations"] > 0)
psbt = self.psbt_online.sendall(recipients=[self.boring.getnewaddress()], options={"psbt": True})["psbt"]
res = self.psbt_offline.walletprocesspsbt(psbt=psbt, finalize=False)
rawtx = self.nodes[0].finalizepsbt(res['psbt'])['hex']
txid = self.nodes[0].sendrawtransaction(rawtx)
self.generatetoaddress(self.nodes[0], 1, self.boring.getnewaddress(), sync_fun=self.no_op)
assert(self.psbt_online.gettransaction(txid)['confirmations'] > 0)
if __name__ == '__main__':
WalletTaprootTest().main()