SomberNight
45458c2f89
If the user tries to open a wallet for a different chain (mainnet vs testnet), try to show a reasonable error message. See previous attempt at this: https://github.com/spesmilo/electrum/commit/c13e05770150c5210783c3d42d3d2b1a683f18b4, which added `wallet.test_addresses_sanity()`. However there are many codepaths where "random" exceptions might get raised before the Wallet object is even instantiated. See [discussion there](https://github.com/spesmilo/electrum/commit/c13e05770150c5210783c3d42d3d2b1a683f18b4#commitcomment-28017341): > should we actually fix that? > if yes, it would be better to write the network type in storage Indeed now I think we should do that. At the time I was concerned it would not help against altcoin forks if we put "mainnet" or "testnet" in the DB. Now I realise we should just put the genesis block hash in the DB instead. Many of the reports in https://github.com/spesmilo/electrum/issues/6526 are likely due to users trying to open a mainnet wallet in testnet mode or vice-versa. fixes https://github.com/spesmilo/electrum/issues/9134 same issue in wizard 2fa two-step wallet-creation flow
376 lines
20 KiB
Python
376 lines
20 KiB
Python
import asyncio
|
|
from collections import defaultdict
|
|
import os
|
|
from typing import Optional, Iterable
|
|
from unittest import mock
|
|
|
|
from electrum.commands import Commands
|
|
from electrum.daemon import Daemon
|
|
from electrum.simple_config import SimpleConfig
|
|
from electrum.wallet import Abstract_Wallet
|
|
from electrum.lnworker import LNWallet, LNPeerManager
|
|
from electrum.lnwatcher import LNWatcher
|
|
from electrum import util
|
|
from electrum.utils.memory_leak import count_objects_in_memory
|
|
from electrum import constants
|
|
|
|
from . import ElectrumTestCase, as_testnet, restore_wallet_from_text__for_unittest
|
|
|
|
|
|
class DaemonTestCase(ElectrumTestCase):
|
|
config: 'SimpleConfig'
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.config = SimpleConfig({'electrum_path': self.electrum_path})
|
|
self.config.NETWORK_OFFLINE = True
|
|
|
|
self.wallet_dir = os.path.dirname(self.config.get_wallet_path())
|
|
assert "wallets" == os.path.basename(self.wallet_dir)
|
|
|
|
async def asyncSetUp(self):
|
|
await super().asyncSetUp()
|
|
self.daemon = Daemon(config=self.config, listen_jsonrpc=False)
|
|
assert self.daemon.network is None
|
|
|
|
async def asyncTearDown(self):
|
|
await self.daemon.stop()
|
|
await super().asyncTearDown()
|
|
|
|
def _restore_wallet_from_text(self, text, *, password: Optional[str], encrypt_file: bool = None, **kwargs) -> str:
|
|
"""Returns path for created wallet."""
|
|
basename = util.get_new_wallet_name(self.wallet_dir)
|
|
path = os.path.join(self.wallet_dir, basename)
|
|
wallet_dict = restore_wallet_from_text__for_unittest(
|
|
text,
|
|
path=path,
|
|
password=password,
|
|
encrypt_file=encrypt_file,
|
|
config=self.config,
|
|
**kwargs,
|
|
)
|
|
# We return the path instead of the wallet object, as extreme
|
|
# care would be needed to use the wallet object directly:
|
|
# Unless the daemon knows about it, daemon._load_wallet might create a conflicting wallet object
|
|
# for the same fs path, and there would be two wallet objects contending for the same file.
|
|
return path
|
|
|
|
|
|
class TestUnifiedPassword(DaemonTestCase):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.config.WALLET_SHOULD_USE_SINGLE_PASSWORD = True
|
|
|
|
def _run_post_unif_sanity_checks(self, paths: Iterable[str], *, password: str):
|
|
for path in paths:
|
|
w = self.daemon.load_wallet(path, password)
|
|
self.assertIsNotNone(w)
|
|
w.check_password(password)
|
|
self.assertTrue(w.has_storage_encryption())
|
|
if w.can_have_keystore_encryption():
|
|
self.assertTrue(w.has_keystore_encryption())
|
|
if w.has_seed():
|
|
self.assertIsInstance(w.get_seed(password), str)
|
|
can_be_unified, is_unified, wallet_paths_can_unlock = self.daemon.check_password_for_directory(
|
|
old_password=password,
|
|
wallet_dir=self.wallet_dir,
|
|
)
|
|
self.assertEqual((True, True, len(paths)), (can_be_unified, is_unified, len(wallet_paths_can_unlock)))
|
|
|
|
# "cannot unify pw" tests --->
|
|
|
|
async def test_cannot_unify_two_std_wallets_both_have_ks_and_sto_enc(self):
|
|
path1 = self._restore_wallet_from_text("9dk", password="123456", encrypt_file=True)
|
|
path2 = self._restore_wallet_from_text("x8", password="asdasd", encrypt_file=True)
|
|
with open(path1, "rb") as f:
|
|
raw1_before = f.read()
|
|
with open(path2, "rb") as f:
|
|
raw2_before = f.read()
|
|
|
|
can_be_unified, is_unified, _ = self.daemon.check_password_for_directory(old_password="123456", wallet_dir=self.wallet_dir)
|
|
self.assertEqual((False, False), (can_be_unified, is_unified))
|
|
is_unified = self.daemon.update_password_for_directory(old_password="123456", new_password="123456")
|
|
self.assertFalse(is_unified)
|
|
# verify that files on disk haven't changed:
|
|
with open(path1, "rb") as f:
|
|
raw1_after = f.read()
|
|
with open(path2, "rb") as f:
|
|
raw2_after = f.read()
|
|
self.assertEqual(raw1_before, raw1_after)
|
|
self.assertEqual(raw2_before, raw2_after)
|
|
|
|
async def test_cannot_unify_mixed_wallets(self):
|
|
path1 = self._restore_wallet_from_text("9dk", password="123456", encrypt_file=True)
|
|
path2 = self._restore_wallet_from_text("9dk", password="asdasd", encrypt_file=False)
|
|
path3 = self._restore_wallet_from_text("9dk", password=None)
|
|
with open(path1, "rb") as f:
|
|
raw1_before = f.read()
|
|
with open(path2, "rb") as f:
|
|
raw2_before = f.read()
|
|
with open(path3, "rb") as f:
|
|
raw3_before = f.read()
|
|
|
|
can_be_unified, is_unified, _ = self.daemon.check_password_for_directory(old_password="123456", wallet_dir=self.wallet_dir)
|
|
self.assertEqual((False, False), (can_be_unified, is_unified))
|
|
is_unified = self.daemon.update_password_for_directory(old_password="123456", new_password="123456")
|
|
self.assertFalse(is_unified)
|
|
# verify that files on disk haven't changed:
|
|
with open(path1, "rb") as f:
|
|
raw1_after = f.read()
|
|
with open(path2, "rb") as f:
|
|
raw2_after = f.read()
|
|
with open(path3, "rb") as f:
|
|
raw3_after = f.read()
|
|
self.assertEqual(raw1_before, raw1_after)
|
|
self.assertEqual(raw2_before, raw2_after)
|
|
self.assertEqual(raw3_before, raw3_after)
|
|
|
|
# "can unify pw" tests --->
|
|
|
|
async def test_can_unify_two_std_wallets_both_have_ks_and_sto_enc(self):
|
|
path1 = self._restore_wallet_from_text("9dk", password="123456", encrypt_file=True)
|
|
path2 = self._restore_wallet_from_text("x8", password="123456", encrypt_file=True)
|
|
can_be_unified, is_unified, _ = self.daemon.check_password_for_directory(old_password="123456", wallet_dir=self.wallet_dir)
|
|
self.assertEqual((True, True), (can_be_unified, is_unified))
|
|
is_unified = self.daemon.update_password_for_directory(old_password="123456", new_password="123456")
|
|
self.assertTrue(is_unified)
|
|
self._run_post_unif_sanity_checks([path1, path2], password="123456")
|
|
|
|
async def test_can_unify_two_std_wallets_one_has_ks_enc_other_has_both_enc(self):
|
|
path1 = self._restore_wallet_from_text("9dk", password="123456", encrypt_file=True)
|
|
path2 = self._restore_wallet_from_text("x8", password="123456", encrypt_file=False)
|
|
with open(path2, "rb") as f:
|
|
raw2_before = f.read()
|
|
|
|
can_be_unified, is_unified, _ = self.daemon.check_password_for_directory(old_password="123456", wallet_dir=self.wallet_dir)
|
|
self.assertEqual((True, False), (can_be_unified, is_unified))
|
|
is_unified = self.daemon.update_password_for_directory(old_password="123456", new_password="123456")
|
|
self.assertTrue(is_unified)
|
|
self._run_post_unif_sanity_checks([path1, path2], password="123456")
|
|
# verify that file at path2 changed:
|
|
with open(path2, "rb") as f:
|
|
raw2_after = f.read()
|
|
self.assertNotEqual(raw2_before, raw2_after)
|
|
|
|
async def test_can_unify_two_std_wallets_one_without_password(self):
|
|
path1 = self._restore_wallet_from_text("9dk", password=None)
|
|
path2 = self._restore_wallet_from_text("x8", password="123456", encrypt_file=True)
|
|
can_be_unified, is_unified, _ = self.daemon.check_password_for_directory(old_password="123456", wallet_dir=self.wallet_dir)
|
|
self.assertEqual((True, False), (can_be_unified, is_unified))
|
|
is_unified = self.daemon.update_password_for_directory(old_password="123456", new_password="123456")
|
|
self.assertTrue(is_unified)
|
|
self._run_post_unif_sanity_checks([path1, path2], password="123456")
|
|
|
|
@as_testnet
|
|
async def test_can_unify_large_folder_yet_to_be_unified(self):
|
|
paths = []
|
|
# seed
|
|
paths.append(self._restore_wallet_from_text("9dk", password="123456", encrypt_file=True))
|
|
paths.append(self._restore_wallet_from_text("9dk", password="123456", encrypt_file=False))
|
|
paths.append(self._restore_wallet_from_text("9dk", password=None))
|
|
# xpub
|
|
xpub = "vpub5UqWay427dCjkpE3gPKLnkBUqDRoBed1328uNrLDoTyKo6HFSs9agfDMy1VXbVtcuBVRiAZQsPPsPdu1Ge8m8qvNZPyzJ4ecPsf6U1ieW4x"
|
|
paths.append(self._restore_wallet_from_text(xpub, password="123456", encrypt_file=True))
|
|
paths.append(self._restore_wallet_from_text(xpub, password="123456", encrypt_file=False))
|
|
paths.append(self._restore_wallet_from_text(xpub, password=None))
|
|
# xprv
|
|
xprv = "vprv9FrABTX8HFeSYL9aaMnLRcEkHBbJnBu9foDJaTvcF8SLvHx6uKqL8rtt7kTd66V4QPLfWPaCJMVZa3h9zuzLr7YFZd1uoEevqqyxp66oSbN"
|
|
paths.append(self._restore_wallet_from_text(xprv, password="123456", encrypt_file=True))
|
|
paths.append(self._restore_wallet_from_text(xprv, password="123456", encrypt_file=False))
|
|
paths.append(self._restore_wallet_from_text(xprv, password=None))
|
|
# WIFs
|
|
wifs= "p2wpkh:cRyfp9nJ8soK1bBUJAcWbMrsJZxKJpe7HBSxz5uXVbwydvUxz9zT p2wpkh:cV6J6T2AG4oXAXdYHAV6dbzR41QnGumDSVvWrmj2yYpos81RtyBK"
|
|
paths.append(self._restore_wallet_from_text(wifs, password="123456", encrypt_file=True))
|
|
paths.append(self._restore_wallet_from_text(wifs, password="123456", encrypt_file=False))
|
|
paths.append(self._restore_wallet_from_text(wifs, password=None))
|
|
# addrs
|
|
addrs = "tb1qq2tmmcngng78nllq2pvrkchcdukemtj5s6l0zu tb1qm7ckcjsed98zhvhv3dr56a22w3fehlkxyh4wgd"
|
|
paths.append(self._restore_wallet_from_text(addrs, password="123456", encrypt_file=True))
|
|
paths.append(self._restore_wallet_from_text(addrs, password="123456", encrypt_file=False))
|
|
paths.append(self._restore_wallet_from_text(addrs, password=None))
|
|
# do unification
|
|
can_be_unified, is_unified, _ = self.daemon.check_password_for_directory(old_password="123456", wallet_dir=self.wallet_dir)
|
|
self.assertEqual((True, False), (can_be_unified, is_unified))
|
|
is_unified = self.daemon.update_password_for_directory(old_password="123456", new_password="123456")
|
|
self.assertTrue(is_unified)
|
|
self._run_post_unif_sanity_checks(paths, password="123456")
|
|
|
|
# misc --->
|
|
|
|
async def test_wallet_objects_are_properly_garbage_collected_after_check_pw_for_dir(self):
|
|
orig_cb_count = util.callback_mgr.count_all_callbacks()
|
|
# GC sanity-check:
|
|
mclasses = [Abstract_Wallet, LNWallet, LNWatcher, LNPeerManager]
|
|
objmap = count_objects_in_memory(mclasses)
|
|
for mcls in mclasses:
|
|
self.assertEqual(len(objmap[mcls]), 0, msg=f"too many lingering objs of type={mcls}")
|
|
# restore some wallets
|
|
paths = []
|
|
paths.append(self._restore_wallet_from_text("9dk", password="123456", encrypt_file=True))
|
|
paths.append(self._restore_wallet_from_text("9dk", password="123456", encrypt_file=False))
|
|
paths.append(self._restore_wallet_from_text("9dk", password=None))
|
|
paths.append(self._restore_wallet_from_text("9dk", password="123456", encrypt_file=True, passphrase="hunter2"))
|
|
paths.append(self._restore_wallet_from_text("9dk", password="999999", encrypt_file=False, passphrase="hunter2"))
|
|
paths.append(self._restore_wallet_from_text("9dk", password=None, passphrase="hunter2"))
|
|
# test unification
|
|
can_be_unified, is_unified, paths_succeeded = self.daemon.check_password_for_directory(old_password="123456", wallet_dir=self.wallet_dir)
|
|
self.assertEqual((False, False, 5), (can_be_unified, is_unified, len(paths_succeeded)))
|
|
# gc
|
|
try:
|
|
async with util.async_timeout(5):
|
|
while True:
|
|
objmap = count_objects_in_memory(mclasses)
|
|
if sum(len(lst) for lst in objmap.values()) == 0:
|
|
break # all "mclasses"-type objects have been GC-ed
|
|
await asyncio.sleep(0.01)
|
|
except asyncio.TimeoutError:
|
|
for mcls in mclasses:
|
|
self.assertEqual(len(objmap[mcls]), 0, msg=f"too many lingering objs of type={mcls}")
|
|
# also check callbacks have been cleaned up:
|
|
self.assertEqual(orig_cb_count, util.callback_mgr.count_all_callbacks())
|
|
|
|
|
|
class TestCommandsWithDaemon(DaemonTestCase):
|
|
TESTNET = True
|
|
SEED = "bitter grass shiver impose acquire brush forget axis eager alone wine silver"
|
|
|
|
async def test_wp_command_with_inmemory_wallet_has_password(self):
|
|
cmds = Commands(config=self.config, daemon=self.daemon)
|
|
wallet = restore_wallet_from_text__for_unittest(
|
|
self.SEED,
|
|
path=None,
|
|
password="123456",
|
|
config=self.config)['wallet']
|
|
self.assertEqual(self.SEED, await cmds.getseed(wallet=wallet, password="123456"))
|
|
|
|
async def test_wp_command_with_inmemory_wallet_no_password(self):
|
|
cmds = Commands(config=self.config, daemon=self.daemon)
|
|
wallet = restore_wallet_from_text__for_unittest(
|
|
self.SEED,
|
|
path=None,
|
|
config=self.config)['wallet']
|
|
self.assertEqual(self.SEED, await cmds.getseed(wallet=wallet))
|
|
|
|
async def test_wp_command_with_diskfile_wallet_has_password(self):
|
|
cmds = Commands(config=self.config, daemon=self.daemon)
|
|
wpath = self._restore_wallet_from_text(self.SEED, password="123456", encrypt_file=True)
|
|
basename = os.path.basename(wpath)
|
|
await cmds.load_wallet(wallet_path=wpath, password="123456")
|
|
wallet = self.daemon.get_wallet(wpath)
|
|
self.assertIsInstance(wallet, Abstract_Wallet)
|
|
self.assertEqual(self.SEED, await cmds.getseed(wallet_path=wpath, password="123456"))
|
|
self.assertEqual(self.SEED, await cmds.getseed(wallet_path=basename, password='123456'))
|
|
self.assertEqual(self.SEED, await cmds.getseed(wallet=wallet, password="123456"))
|
|
|
|
async def test_wp_command_with_diskfile_wallet_no_password(self):
|
|
cmds = Commands(config=self.config, daemon=self.daemon)
|
|
wpath = self._restore_wallet_from_text(self.SEED, password=None)
|
|
basename = os.path.basename(wpath)
|
|
await cmds.load_wallet(wallet_path=wpath, password=None)
|
|
wallet = self.daemon.get_wallet(wpath)
|
|
self.assertIsInstance(wallet, Abstract_Wallet)
|
|
self.assertEqual(self.SEED, await cmds.getseed(wallet_path=wpath))
|
|
self.assertEqual(self.SEED, await cmds.getseed(wallet_path=basename))
|
|
self.assertEqual(self.SEED, await cmds.getseed(wallet=wallet))
|
|
|
|
|
|
class TestLoadWallet(DaemonTestCase):
|
|
|
|
async def test_simple_load(self):
|
|
path1 = self._restore_wallet_from_text("9dk", password=None)
|
|
wallet1 = self.daemon.load_wallet(path1, password=None)
|
|
await self.daemon._stop_wallet(path1)
|
|
|
|
async def test_password_checks_for_no_password(self):
|
|
real_password = None
|
|
path1 = self._restore_wallet_from_text("9dk", password=real_password)
|
|
# load_wallet will not validate the password arg unless needed for storage.decrypt():
|
|
wallet1 = self.daemon.load_wallet(path1, password="garbage")
|
|
await self.daemon._stop_wallet(path1)
|
|
# unless force_check_password is set:
|
|
with self.assertRaises(util.InvalidPassword):
|
|
wallet1 = self.daemon.load_wallet(path1, password="garbage", force_check_password=True)
|
|
|
|
wallet1 = self.daemon.load_wallet(path1, password=real_password)
|
|
await self.daemon._stop_wallet(path1)
|
|
|
|
wallet1 = self.daemon.load_wallet(path1, password=real_password, force_check_password=True)
|
|
await self.daemon._stop_wallet(path1)
|
|
|
|
# load_wallet will not validate the password arg if wallet is already loaded, unless force_check_password
|
|
wallet1 = self.daemon.load_wallet(path1, password=real_password)
|
|
wallet1 = self.daemon.load_wallet(path1, password="garbage")
|
|
with self.assertRaises(util.InvalidPassword):
|
|
wallet1 = self.daemon.load_wallet(path1, password="garbage", force_check_password=True)
|
|
|
|
|
|
async def test_password_checks_for_ks_enc(self):
|
|
real_password = "1234"
|
|
path1 = self._restore_wallet_from_text("9dk", password=real_password, encrypt_file=False)
|
|
# load_wallet will not validate the password arg unless needed for storage.decrypt():
|
|
wallet1 = self.daemon.load_wallet(path1, password="garbage")
|
|
await self.daemon._stop_wallet(path1)
|
|
# unless force_check_password is set:
|
|
with self.assertRaises(util.InvalidPassword):
|
|
wallet1 = self.daemon.load_wallet(path1, password="garbage", force_check_password=True)
|
|
|
|
wallet1 = self.daemon.load_wallet(path1, password=real_password)
|
|
await self.daemon._stop_wallet(path1)
|
|
|
|
wallet1 = self.daemon.load_wallet(path1, password=real_password, force_check_password=True)
|
|
await self.daemon._stop_wallet(path1)
|
|
|
|
# load_wallet will not validate the password arg if wallet is already loaded, unless force_check_password
|
|
wallet1 = self.daemon.load_wallet(path1, password=real_password)
|
|
wallet1 = self.daemon.load_wallet(path1, password="garbage")
|
|
with self.assertRaises(util.InvalidPassword):
|
|
wallet1 = self.daemon.load_wallet(path1, password="garbage", force_check_password=True)
|
|
|
|
|
|
async def test_password_checks_for_sto_enc(self):
|
|
real_password = "1234"
|
|
path1 = self._restore_wallet_from_text("9dk", password=real_password, encrypt_file=True)
|
|
|
|
with self.assertRaises(util.InvalidPassword):
|
|
wallet1 = self.daemon.load_wallet(path1, password="garbage")
|
|
|
|
with self.assertRaises(util.InvalidPassword):
|
|
wallet1 = self.daemon.load_wallet(path1, password="garbage", force_check_password=True)
|
|
|
|
wallet1 = self.daemon.load_wallet(path1, password=real_password)
|
|
await self.daemon._stop_wallet(path1)
|
|
|
|
wallet1 = self.daemon.load_wallet(path1, password=real_password, force_check_password=True)
|
|
await self.daemon._stop_wallet(path1)
|
|
|
|
# load_wallet will not validate the password arg if wallet is already loaded, unless force_check_password
|
|
wallet1 = self.daemon.load_wallet(path1, password=real_password)
|
|
wallet1 = self.daemon.load_wallet(path1, password="garbage")
|
|
with self.assertRaises(util.InvalidPassword):
|
|
wallet1 = self.daemon.load_wallet(path1, password="garbage", force_check_password=True)
|
|
|
|
async def test_mainnet_testnet_mixup(self):
|
|
"""version bytes in addresses, xpubs, etc. differ between mainnet and testnet.
|
|
If the user tries to open a wallet for a different chain, try to show a reasonable error message.
|
|
"""
|
|
# we are on mainnet, and will try to open testnet wallets:
|
|
assert constants.net.TESTNET is False
|
|
|
|
# case 1: fresh wallet created on wrong network
|
|
with mock.patch("electrum.constants.net", constants.BitcoinTestnet):
|
|
path = self._restore_wallet_from_text("9dk", password=None)
|
|
with self.assertRaises(util.WalletFileException):
|
|
wallet = self.daemon.load_wallet(path, password=None, upgrade=True)
|
|
|
|
# case 2: existing older wallet (db v57) that gets populated with 'genesis_blockhash' in convert_version_71
|
|
path = self.get_wallet_file_path("client_4_5_2_9dk_with_ln")
|
|
with self.assertRaises(util.WalletFileException):
|
|
wallet = self.daemon.load_wallet(path, password=None, upgrade=True)
|
|
|
|
# case 3: existing older wallet (db v18) that gets populated with 'genesis_blockhash' in convert_version_71
|
|
# // this test does not work: convert_version_20 raises InvalidMasterKeyVersionBytes
|
|
# path = self.get_wallet_file_path("client_3_3_8_xpub_with_realistic_history")
|
|
# with self.assertRaises(util.WalletFileException):
|
|
# wallet = self.daemon.load_wallet(path, password=None, upgrade=True)
|