mirrors/bitcoin
1
0
Fork 0
mirror of https://github.com/bitcoin/bitcoin.git synced 2024-11-20 10:38:42 +01:00
Code Issues Projects Releases Wiki Activity

test: Add Wallet Unlock Context Manager

Browse Source 
Add Context Manager to manage wallet locking/unlocking with passphrase
This commit is contained in:
Brandon Odiwuor 2023-10-09 15:30:40 +03:00
parent db283a6b6f
commit 004903ebad
7 changed files with 140 additions and 125 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
test/functional/test_framework/wallet_util.py
View File

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

60
test/functional/wallet_createwallet.py
View File

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

25
test/functional/wallet_descriptor.py
View File

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

37
test/functional/wallet_dump.py
View File

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

22
test/functional/wallet_encryption.py
View File

@ -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__':

48
test/functional/wallet_fundrawtransaction.py
View File

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

54
test/functional/wallet_keypool.py
View File

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