diff --git a/test/functional/test_framework/wallet_util.py b/test/functional/test_framework/wallet_util.py index 319f120297e..44811918bf9 100755 --- a/test/functional/test_framework/wallet_util.py +++ b/test/functional/test_framework/wallet_util.py @@ -122,3 +122,22 @@ def generate_keypair(compressed=True, wif=False): if wif: privkey = bytes_to_wif(privkey.get_bytes(), compressed) return privkey, pubkey + +class WalletUnlock(): + """ + A context manager for unlocking a wallet with a passphrase and automatically locking it afterward. + """ + + MAXIMUM_TIMEOUT = 999000 + + def __init__(self, wallet, passphrase, timeout=MAXIMUM_TIMEOUT): + self.wallet = wallet + self.passphrase = passphrase + self.timeout = timeout + + def __enter__(self): + self.wallet.walletpassphrase(self.passphrase, self.timeout) + + def __exit__(self, *args): + _ = args + self.wallet.walletlock() diff --git a/test/functional/wallet_createwallet.py b/test/functional/wallet_createwallet.py index eb83e11f360..8e07021e036 100755 --- a/test/functional/wallet_createwallet.py +++ b/test/functional/wallet_createwallet.py @@ -12,7 +12,7 @@ from test_framework.util import ( assert_equal, assert_raises_rpc_error, ) -from test_framework.wallet_util import generate_keypair +from test_framework.wallet_util import generate_keypair, WalletUnlock EMPTY_PASSPHRASE_MSG = "Empty string given as passphrase, wallet will not be encrypted." @@ -108,24 +108,24 @@ class CreateWalletTest(BitcoinTestFramework): w4.encryptwallet('pass') assert_raises_rpc_error(-4, "Error: This wallet has no available keys", w4.getnewaddress) assert_raises_rpc_error(-4, "Error: This wallet has no available keys", w4.getrawchangeaddress) - # Now set a seed and it should work. Wallet should also be encrypted - w4.walletpassphrase("pass", 999000) - if self.options.descriptors: - w4.importdescriptors([{ - 'desc': descsum_create('wpkh(tprv8ZgxMBicQKsPcwuZGKp8TeWppSuLMiLe2d9PupB14QpPeQsqoj3LneJLhGHH13xESfvASyd4EFLJvLrG8b7DrLxEuV7hpF9uUc6XruKA1Wq/0h/*)'), - 'timestamp': 'now', - 'active': True - }, - { - 'desc': descsum_create('wpkh(tprv8ZgxMBicQKsPcwuZGKp8TeWppSuLMiLe2d9PupB14QpPeQsqoj3LneJLhGHH13xESfvASyd4EFLJvLrG8b7DrLxEuV7hpF9uUc6XruKA1Wq/1h/*)'), - 'timestamp': 'now', - 'active': True, - 'internal': True - }]) - else: - w4.sethdseed() - w4.getnewaddress() - w4.getrawchangeaddress() + with WalletUnlock(w4, "pass"): + # Now set a seed and it should work. Wallet should also be encrypted + if self.options.descriptors: + w4.importdescriptors([{ + 'desc': descsum_create('wpkh(tprv8ZgxMBicQKsPcwuZGKp8TeWppSuLMiLe2d9PupB14QpPeQsqoj3LneJLhGHH13xESfvASyd4EFLJvLrG8b7DrLxEuV7hpF9uUc6XruKA1Wq/0h/*)'), + 'timestamp': 'now', + 'active': True + }, + { + 'desc': descsum_create('wpkh(tprv8ZgxMBicQKsPcwuZGKp8TeWppSuLMiLe2d9PupB14QpPeQsqoj3LneJLhGHH13xESfvASyd4EFLJvLrG8b7DrLxEuV7hpF9uUc6XruKA1Wq/1h/*)'), + 'timestamp': 'now', + 'active': True, + 'internal': True + }]) + else: + w4.sethdseed() + w4.getnewaddress() + w4.getrawchangeaddress() self.log.info("Test blank creation with privkeys disabled and then encryption") self.nodes[0].createwallet(wallet_name='w5', disable_private_keys=True, blank=True) @@ -142,23 +142,23 @@ class CreateWalletTest(BitcoinTestFramework): self.nodes[0].createwallet(wallet_name='wblank', disable_private_keys=False, blank=True, passphrase='thisisapassphrase') wblank = node.get_wallet_rpc('wblank') assert_raises_rpc_error(-13, "Error: Please enter the wallet passphrase with walletpassphrase first.", wblank.signmessage, "needanargument", "test") - wblank.walletpassphrase("thisisapassphrase", 999000) - assert_raises_rpc_error(-4, "Error: This wallet has no available keys", wblank.getnewaddress) - assert_raises_rpc_error(-4, "Error: This wallet has no available keys", wblank.getrawchangeaddress) + with WalletUnlock(wblank, "thisisapassphrase"): + assert_raises_rpc_error(-4, "Error: This wallet has no available keys", wblank.getnewaddress) + assert_raises_rpc_error(-4, "Error: This wallet has no available keys", wblank.getrawchangeaddress) self.log.info('Test creating a new encrypted wallet.') # Born encrypted wallet is created (has keys) self.nodes[0].createwallet(wallet_name='w6', disable_private_keys=False, blank=False, passphrase='thisisapassphrase') w6 = node.get_wallet_rpc('w6') assert_raises_rpc_error(-13, "Error: Please enter the wallet passphrase with walletpassphrase first.", w6.signmessage, "needanargument", "test") - w6.walletpassphrase("thisisapassphrase", 999000) - w6.signmessage(w6.getnewaddress('', 'legacy'), "test") - w6.keypoolrefill(1) - # There should only be 1 key for legacy, 3 for descriptors - walletinfo = w6.getwalletinfo() - keys = 4 if self.options.descriptors else 1 - assert_equal(walletinfo['keypoolsize'], keys) - assert_equal(walletinfo['keypoolsize_hd_internal'], keys) + with WalletUnlock(w6, "thisisapassphrase"): + w6.signmessage(w6.getnewaddress('', 'legacy'), "test") + w6.keypoolrefill(1) + # There should only be 1 key for legacy, 3 for descriptors + walletinfo = w6.getwalletinfo() + keys = 4 if self.options.descriptors else 1 + assert_equal(walletinfo['keypoolsize'], keys) + assert_equal(walletinfo['keypoolsize_hd_internal'], keys) # Allow empty passphrase, but there should be a warning resp = self.nodes[0].createwallet(wallet_name='w7', disable_private_keys=False, blank=False, passphrase='') assert_equal(resp["warnings"], [EMPTY_PASSPHRASE_MSG] if self.options.descriptors else [EMPTY_PASSPHRASE_MSG, LEGACY_WALLET_MSG]) diff --git a/test/functional/wallet_descriptor.py b/test/functional/wallet_descriptor.py index 6af01f8cfd1..9935d313f89 100755 --- a/test/functional/wallet_descriptor.py +++ b/test/functional/wallet_descriptor.py @@ -16,6 +16,7 @@ from test_framework.util import ( assert_equal, assert_raises_rpc_error ) +from test_framework.wallet_util import WalletUnlock class WalletDescriptorTest(BitcoinTestFramework): @@ -129,11 +130,10 @@ class WalletDescriptorTest(BitcoinTestFramework): # Encrypt wallet 0 send_wrpc.encryptwallet('pass') - send_wrpc.walletpassphrase("pass", 999000) - addr = send_wrpc.getnewaddress() - info2 = send_wrpc.getaddressinfo(addr) - assert info1['hdmasterfingerprint'] != info2['hdmasterfingerprint'] - send_wrpc.walletlock() + with WalletUnlock(send_wrpc, "pass"): + addr = send_wrpc.getnewaddress() + info2 = send_wrpc.getaddressinfo(addr) + assert info1['hdmasterfingerprint'] != info2['hdmasterfingerprint'] assert 'hdmasterfingerprint' in send_wrpc.getaddressinfo(send_wrpc.getnewaddress()) info3 = send_wrpc.getaddressinfo(addr) assert_equal(info2['desc'], info3['desc']) @@ -143,14 +143,13 @@ class WalletDescriptorTest(BitcoinTestFramework): send_wrpc.getnewaddress() self.log.info("Test that unlock is needed when deriving only hardened keys in an encrypted wallet") - send_wrpc.walletpassphrase("pass", 999000) - send_wrpc.importdescriptors([{ - "desc": "wpkh(tprv8ZgxMBicQKsPd7Uf69XL1XwhmjHopUGep8GuEiJDZmbQz6o58LninorQAfcKZWARbtRtfnLcJ5MQ2AtHcQJCCRUcMRvmDUjyEmNUWwx8UbK/0h/*h)#y4dfsj7n", - "timestamp": "now", - "range": [0,10], - "active": True - }]) - send_wrpc.walletlock() + with WalletUnlock(send_wrpc, "pass"): + send_wrpc.importdescriptors([{ + "desc": "wpkh(tprv8ZgxMBicQKsPd7Uf69XL1XwhmjHopUGep8GuEiJDZmbQz6o58LninorQAfcKZWARbtRtfnLcJ5MQ2AtHcQJCCRUcMRvmDUjyEmNUWwx8UbK/0h/*h)#y4dfsj7n", + "timestamp": "now", + "range": [0,10], + "active": True + }]) # Exhaust keypool of 100 for _ in range(100): send_wrpc.getnewaddress(address_type='bech32') diff --git a/test/functional/wallet_dump.py b/test/functional/wallet_dump.py index 8c68d03f971..eaedbbed419 100755 --- a/test/functional/wallet_dump.py +++ b/test/functional/wallet_dump.py @@ -12,6 +12,7 @@ from test_framework.util import ( assert_equal, assert_raises_rpc_error, ) +from test_framework.wallet_util import WalletUnlock def read_dump(file_name, addrs, script_addrs, hd_master_addr_old): @@ -173,26 +174,26 @@ class WalletDumpTest(BitcoinTestFramework): # encrypt wallet, restart, unlock and dump self.nodes[0].encryptwallet('test') - self.nodes[0].walletpassphrase("test", 999000) - # Should be a no-op: - self.nodes[0].keypoolrefill() - self.nodes[0].dumpwallet(wallet_enc_dump) + with WalletUnlock(self.nodes[0], "test"): + # Should be a no-op: + self.nodes[0].keypoolrefill() + self.nodes[0].dumpwallet(wallet_enc_dump) - found_comments, found_legacy_addr, found_p2sh_segwit_addr, found_bech32_addr, found_script_addr, found_addr_chg, found_addr_rsv, _ = \ - read_dump(wallet_enc_dump, addrs, [multisig_addr], hd_master_addr_unenc) - assert '# End of dump' in found_comments # Check that file is not corrupt - assert_equal(dump_time_str, next(c for c in found_comments if c.startswith('# * Created on'))) - assert_equal(dump_best_block_1, next(c for c in found_comments if c.startswith('# * Best block'))) - assert_equal(dump_best_block_2, next(c for c in found_comments if c.startswith('# mined on'))) - assert_equal(found_legacy_addr, test_addr_count) # all keys must be in the dump - assert_equal(found_p2sh_segwit_addr, test_addr_count) # all keys must be in the dump - assert_equal(found_bech32_addr, test_addr_count) # all keys must be in the dump - assert_equal(found_script_addr, 1) - assert_equal(found_addr_chg, 90 * 2) # old reserve keys are marked as change now - assert_equal(found_addr_rsv, 90 * 2) + found_comments, found_legacy_addr, found_p2sh_segwit_addr, found_bech32_addr, found_script_addr, found_addr_chg, found_addr_rsv, _ = \ + read_dump(wallet_enc_dump, addrs, [multisig_addr], hd_master_addr_unenc) + assert '# End of dump' in found_comments # Check that file is not corrupt + assert_equal(dump_time_str, next(c for c in found_comments if c.startswith('# * Created on'))) + assert_equal(dump_best_block_1, next(c for c in found_comments if c.startswith('# * Best block'))) + assert_equal(dump_best_block_2, next(c for c in found_comments if c.startswith('# mined on'))) + assert_equal(found_legacy_addr, test_addr_count) # all keys must be in the dump + assert_equal(found_p2sh_segwit_addr, test_addr_count) # all keys must be in the dump + assert_equal(found_bech32_addr, test_addr_count) # all keys must be in the dump + assert_equal(found_script_addr, 1) + assert_equal(found_addr_chg, 90 * 2) # old reserve keys are marked as change now + assert_equal(found_addr_rsv, 90 * 2) - # Overwriting should fail - assert_raises_rpc_error(-8, "already exists", lambda: self.nodes[0].dumpwallet(wallet_enc_dump)) + # Overwriting should fail + assert_raises_rpc_error(-8, "already exists", lambda: self.nodes[0].dumpwallet(wallet_enc_dump)) # Restart node with new wallet, and test importwallet self.restart_node(0) diff --git a/test/functional/wallet_encryption.py b/test/functional/wallet_encryption.py index e8381ba8f2e..b30634010d5 100755 --- a/test/functional/wallet_encryption.py +++ b/test/functional/wallet_encryption.py @@ -11,6 +11,7 @@ from test_framework.util import ( assert_raises_rpc_error, assert_equal, ) +from test_framework.wallet_util import WalletUnlock class WalletEncryptionTest(BitcoinTestFramework): @@ -59,19 +60,17 @@ class WalletEncryptionTest(BitcoinTestFramework): assert_raises_rpc_error(-14, "wallet passphrase entered was incorrect", self.nodes[0].walletpassphrase, passphrase + "wrong", 10) # Test walletlock - self.nodes[0].walletpassphrase(passphrase, 999000) - sig = self.nodes[0].signmessage(address, msg) - assert self.nodes[0].verifymessage(address, sig, msg) - self.nodes[0].walletlock() + with WalletUnlock(self.nodes[0], passphrase): + sig = self.nodes[0].signmessage(address, msg) + assert self.nodes[0].verifymessage(address, sig, msg) assert_raises_rpc_error(-13, "Please enter the wallet passphrase with walletpassphrase first", self.nodes[0].signmessage, address, msg) # Test passphrase changes self.nodes[0].walletpassphrasechange(passphrase, passphrase2) assert_raises_rpc_error(-14, "wallet passphrase entered was incorrect", self.nodes[0].walletpassphrase, passphrase, 10) - self.nodes[0].walletpassphrase(passphrase2, 999000) - sig = self.nodes[0].signmessage(address, msg) - assert self.nodes[0].verifymessage(address, sig, msg) - self.nodes[0].walletlock() + with WalletUnlock(self.nodes[0], passphrase2): + sig = self.nodes[0].signmessage(address, msg) + assert self.nodes[0].verifymessage(address, sig, msg) # Test timeout bounds assert_raises_rpc_error(-8, "Timeout cannot be negative.", self.nodes[0].walletpassphrase, passphrase2, -10) @@ -97,10 +96,9 @@ class WalletEncryptionTest(BitcoinTestFramework): self.nodes[0].walletpassphrasechange(passphrase2, passphrase_with_nulls) # walletpassphrasechange should not stop at null characters assert_raises_rpc_error(-14, "wallet passphrase entered was incorrect", self.nodes[0].walletpassphrase, passphrase_with_nulls.partition("\0")[0], 10) - self.nodes[0].walletpassphrase(passphrase_with_nulls, 999000) - sig = self.nodes[0].signmessage(address, msg) - assert self.nodes[0].verifymessage(address, sig, msg) - self.nodes[0].walletlock() + with WalletUnlock(self.nodes[0], passphrase_with_nulls): + sig = self.nodes[0].signmessage(address, msg) + assert self.nodes[0].verifymessage(address, sig, msg) if __name__ == '__main__': diff --git a/test/functional/wallet_fundrawtransaction.py b/test/functional/wallet_fundrawtransaction.py index ca4feefb2b6..295cbee843a 100755 --- a/test/functional/wallet_fundrawtransaction.py +++ b/test/functional/wallet_fundrawtransaction.py @@ -25,7 +25,7 @@ from test_framework.util import ( find_vout_for_address, get_fee, ) -from test_framework.wallet_util import generate_keypair +from test_framework.wallet_util import generate_keypair, WalletUnlock ERR_NOT_ENOUGH_PRESET_INPUTS = "The preselected coins total amount does not cover the transaction target. " \ "Please allow other inputs to be automatically selected or include more coins manually" @@ -581,19 +581,18 @@ class RawTransactionsTest(BitcoinTestFramework): wallet.encryptwallet("test") if self.options.descriptors: - wallet.walletpassphrase("test", 999000) - wallet.importdescriptors([{ - 'desc': descsum_create('wpkh(tprv8ZgxMBicQKsPdYeeZbPSKd2KYLmeVKtcFA7kqCxDvDR13MQ6us8HopUR2wLcS2ZKPhLyKsqpDL2FtL73LMHcgoCL7DXsciA8eX8nbjCR2eG/0h/*h)'), - 'timestamp': 'now', - 'active': True - }, - { - 'desc': descsum_create('wpkh(tprv8ZgxMBicQKsPdYeeZbPSKd2KYLmeVKtcFA7kqCxDvDR13MQ6us8HopUR2wLcS2ZKPhLyKsqpDL2FtL73LMHcgoCL7DXsciA8eX8nbjCR2eG/1h/*h)'), - 'timestamp': 'now', - 'active': True, - 'internal': True - }]) - wallet.walletlock() + with WalletUnlock(wallet, "test"): + wallet.importdescriptors([{ + 'desc': descsum_create('wpkh(tprv8ZgxMBicQKsPdYeeZbPSKd2KYLmeVKtcFA7kqCxDvDR13MQ6us8HopUR2wLcS2ZKPhLyKsqpDL2FtL73LMHcgoCL7DXsciA8eX8nbjCR2eG/0h/*h)'), + 'timestamp': 'now', + 'active': True + }, + { + 'desc': descsum_create('wpkh(tprv8ZgxMBicQKsPdYeeZbPSKd2KYLmeVKtcFA7kqCxDvDR13MQ6us8HopUR2wLcS2ZKPhLyKsqpDL2FtL73LMHcgoCL7DXsciA8eX8nbjCR2eG/1h/*h)'), + 'timestamp': 'now', + 'active': True, + 'internal': True + }]) # Drain the keypool. wallet.getnewaddress() @@ -619,9 +618,8 @@ class RawTransactionsTest(BitcoinTestFramework): assert_raises_rpc_error(-4, "Transaction needs a change address, but we can't generate it.", wallet.fundrawtransaction, rawtx) # Refill the keypool. - wallet.walletpassphrase("test", 999000) - wallet.keypoolrefill(8) #need to refill the keypool to get an internal change address - wallet.walletlock() + with WalletUnlock(wallet, "test"): + wallet.keypoolrefill(8) #need to refill the keypool to get an internal change address assert_raises_rpc_error(-13, "walletpassphrase", wallet.sendtoaddress, self.nodes[0].getnewaddress(), 1.2) @@ -634,16 +632,16 @@ class RawTransactionsTest(BitcoinTestFramework): assert fundedTx["changepos"] != -1 # Now we need to unlock. - wallet.walletpassphrase("test", 999000) - signedTx = wallet.signrawtransactionwithwallet(fundedTx['hex']) - wallet.sendrawtransaction(signedTx['hex']) - self.generate(self.nodes[1], 1) + with WalletUnlock(wallet, "test"): + signedTx = wallet.signrawtransactionwithwallet(fundedTx['hex']) + wallet.sendrawtransaction(signedTx['hex']) + self.generate(self.nodes[1], 1) - # Make sure funds are received at node1. - assert_equal(oldBalance+Decimal('51.10000000'), self.nodes[0].getbalance()) + # Make sure funds are received at node1. + assert_equal(oldBalance+Decimal('51.10000000'), self.nodes[0].getbalance()) - # Restore pre-test wallet state - wallet.sendall(recipients=[df_wallet.getnewaddress(), df_wallet.getnewaddress(), df_wallet.getnewaddress()]) + # Restore pre-test wallet state + wallet.sendall(recipients=[df_wallet.getnewaddress(), df_wallet.getnewaddress(), df_wallet.getnewaddress()]) wallet.unloadwallet() self.generate(self.nodes[1], 1) diff --git a/test/functional/wallet_keypool.py b/test/functional/wallet_keypool.py index 0ba8a46bae1..d2341fb12e0 100755 --- a/test/functional/wallet_keypool.py +++ b/test/functional/wallet_keypool.py @@ -9,6 +9,7 @@ from decimal import Decimal from test_framework.test_framework import BitcoinTestFramework from test_framework.util import assert_equal, assert_raises_rpc_error +from test_framework.wallet_util import WalletUnlock class KeyPoolTest(BitcoinTestFramework): def add_options(self, parser): @@ -85,9 +86,8 @@ class KeyPoolTest(BitcoinTestFramework): assert_raises_rpc_error(-12, "Error: Keypool ran out, please call keypoolrefill first", nodes[0].getnewaddress) # put six (plus 2) new keys in the keypool (100% external-, +100% internal-keys, 1 in min) - nodes[0].walletpassphrase("test", 999000) - nodes[0].keypoolrefill(6) - nodes[0].walletlock() + with WalletUnlock(nodes[0], 'test'): + nodes[0].keypoolrefill(6) wi = nodes[0].getwalletinfo() if self.options.descriptors: assert_equal(wi['keypoolsize_hd_internal'], 24) @@ -131,29 +131,29 @@ class KeyPoolTest(BitcoinTestFramework): nodes[0].getnewaddress() assert_raises_rpc_error(-12, "Keypool ran out", nodes[0].getnewaddress) - nodes[0].walletpassphrase("test", 999000) - nodes[0].keypoolrefill(100) - wi = nodes[0].getwalletinfo() - if self.options.descriptors: - assert_equal(wi['keypoolsize_hd_internal'], 400) - assert_equal(wi['keypoolsize'], 400) - else: - assert_equal(wi['keypoolsize_hd_internal'], 100) - assert_equal(wi['keypoolsize'], 100) + with WalletUnlock(nodes[0], 'test'): + nodes[0].keypoolrefill(100) + wi = nodes[0].getwalletinfo() + if self.options.descriptors: + assert_equal(wi['keypoolsize_hd_internal'], 400) + assert_equal(wi['keypoolsize'], 400) + else: + assert_equal(wi['keypoolsize_hd_internal'], 100) + assert_equal(wi['keypoolsize'], 100) - if not self.options.descriptors: - # Check that newkeypool entirely flushes the keypool - start_keypath = nodes[0].getaddressinfo(nodes[0].getnewaddress())['hdkeypath'] - start_change_keypath = nodes[0].getaddressinfo(nodes[0].getrawchangeaddress())['hdkeypath'] - # flush keypool and get new addresses - nodes[0].newkeypool() - end_keypath = nodes[0].getaddressinfo(nodes[0].getnewaddress())['hdkeypath'] - end_change_keypath = nodes[0].getaddressinfo(nodes[0].getrawchangeaddress())['hdkeypath'] - # The new keypath index should be 100 more than the old one - new_index = int(start_keypath.rsplit('/', 1)[1][:-1]) + 100 - new_change_index = int(start_change_keypath.rsplit('/', 1)[1][:-1]) + 100 - assert_equal(end_keypath, "m/0'/0'/" + str(new_index) + "'") - assert_equal(end_change_keypath, "m/0'/1'/" + str(new_change_index) + "'") + if not self.options.descriptors: + # Check that newkeypool entirely flushes the keypool + start_keypath = nodes[0].getaddressinfo(nodes[0].getnewaddress())['hdkeypath'] + start_change_keypath = nodes[0].getaddressinfo(nodes[0].getrawchangeaddress())['hdkeypath'] + # flush keypool and get new addresses + nodes[0].newkeypool() + end_keypath = nodes[0].getaddressinfo(nodes[0].getnewaddress())['hdkeypath'] + end_change_keypath = nodes[0].getaddressinfo(nodes[0].getrawchangeaddress())['hdkeypath'] + # The new keypath index should be 100 more than the old one + new_index = int(start_keypath.rsplit('/', 1)[1][:-1]) + 100 + new_change_index = int(start_change_keypath.rsplit('/', 1)[1][:-1]) + 100 + assert_equal(end_keypath, "m/0'/0'/" + str(new_index) + "'") + assert_equal(end_change_keypath, "m/0'/1'/" + str(new_change_index) + "'") # create a blank wallet nodes[0].createwallet(wallet_name='w2', blank=True, disable_private_keys=True) @@ -170,9 +170,9 @@ class KeyPoolTest(BitcoinTestFramework): else: res = w2.importmulti([{'desc': desc, 'timestamp': 'now'}]) assert_equal(res[0]['success'], True) - w1.walletpassphrase("test", 999000) - res = w1.sendtoaddress(address=address, amount=0.00010000) + with WalletUnlock(w1, 'test'): + res = w1.sendtoaddress(address=address, amount=0.00010000) self.generate(nodes[0], 1) destination = addr.pop()