From f8ba66058308afc73109d4b59b587a38733e1d3a Mon Sep 17 00:00:00 2001 From: SomberNight Date: Fri, 28 Feb 2020 19:47:56 +0100 Subject: [PATCH] clean-up hw-wallet "get_password_for_storage_encryption"-related code --- electrum/base_wizard.py | 15 ++++++--------- electrum/keystore.py | 6 +----- electrum/plugin.py | 6 +++--- electrum/plugins/hw_wallet/plugin.py | 12 ++++++++++-- run_electrum | 27 ++++++++++++++++----------- 5 files changed, 36 insertions(+), 30 deletions(-) diff --git a/electrum/base_wizard.py b/electrum/base_wizard.py index 7f1f5e1a5..8cd7d4a11 100644 --- a/electrum/base_wizard.py +++ b/electrum/base_wizard.py @@ -37,8 +37,7 @@ from .bip32 import is_bip32_derivation, xpub_type, normalize_bip32_derivation, B from .keystore import bip44_derivation, purpose48_derivation, Hardware_KeyStore, KeyStore from .wallet import (Imported_Wallet, Standard_Wallet, Multisig_Wallet, wallet_types, Wallet, Abstract_Wallet) -from .storage import (WalletStorage, StorageEncryptionVersion, - get_derivation_used_for_hw_device_encryption) +from .storage import WalletStorage, StorageEncryptionVersion from .wallet_db import WalletDB from .i18n import _ from .util import UserCancelled, InvalidPassword, WalletFileException @@ -334,7 +333,9 @@ class BaseWizard(Logger): run_next=lambda *args: self.on_device(*args, purpose=purpose, storage=storage)) def on_device(self, name, device_info, *, purpose, storage=None): - self.plugin = self.plugins.get_plugin(name) # type: HW_PluginBase + self.plugin = self.plugins.get_plugin(name) + assert isinstance(self.plugin, HW_PluginBase) + devmgr = self.plugins.device_manager try: self.plugin.setup_device(device_info, self, purpose) except OSError as e: @@ -342,7 +343,6 @@ class BaseWizard(Logger): + '\n' + str(e) + '\n' + _('To try to fix this, we will now re-pair with your device.') + '\n' + _('Please try again.')) - devmgr = self.plugins.device_manager devmgr.unpair_id(device_info.device.id_) self.choose_hw_device(purpose, storage=storage) return @@ -350,7 +350,6 @@ class BaseWizard(Logger): if self.question(e.text_ignore_old_fw_and_continue(), title=_("Outdated device firmware")): self.plugin.set_ignore_outdated_fw() # will need to re-pair - devmgr = self.plugins.device_manager devmgr.unpair_id(device_info.device.id_) self.choose_hw_device(purpose, storage=storage) return @@ -368,14 +367,12 @@ class BaseWizard(Logger): self.run('on_hw_derivation', name, device_info, derivation, script_type) self.derivation_and_script_type_dialog(f) elif purpose == HWD_SETUP_DECRYPT_WALLET: - derivation = get_derivation_used_for_hw_device_encryption() - xpub = self.plugin.get_xpub(device_info.device.id_, derivation, 'standard', self) - password = keystore.Xpub.get_pubkey_from_xpub(xpub, ()).hex() + client = devmgr.client_by_id(device_info.device.id_) + password = client.get_password_for_storage_encryption() try: storage.decrypt(password) except InvalidPassword: # try to clear session so that user can type another passphrase - devmgr = self.plugins.device_manager client = devmgr.client_by_id(device_info.device.id_) if hasattr(client, 'clear_session'): # FIXME not all hw wallet plugins have this client.clear_session() diff --git a/electrum/keystore.py b/electrum/keystore.py index 713f89898..0ecb12fc8 100644 --- a/electrum/keystore.py +++ b/electrum/keystore.py @@ -765,12 +765,8 @@ class Hardware_KeyStore(Xpub, KeyStore): return False def get_password_for_storage_encryption(self) -> str: - from .storage import get_derivation_used_for_hw_device_encryption client = self.plugin.get_client(self) - derivation = get_derivation_used_for_hw_device_encryption() - xpub = client.get_xpub(derivation, "standard") - password = self.get_pubkey_from_xpub(xpub, ()).hex() - return password + return client.get_password_for_storage_encryption() def has_usable_connection_with_device(self) -> bool: if not hasattr(self, 'plugin'): diff --git a/electrum/plugin.py b/electrum/plugin.py index f8524e4a3..0bcc5d20e 100644 --- a/electrum/plugin.py +++ b/electrum/plugin.py @@ -60,7 +60,7 @@ class Plugins(DaemonThread): self.pkgpath = os.path.dirname(plugins.__file__) self.config = config self.hw_wallets = {} - self.plugins = {} + self.plugins = {} # type: Dict[str, BasePlugin] self.gui_name = gui_name self.descriptions = {} self.device_manager = DeviceMgr(config) @@ -198,8 +198,8 @@ class Plugins(DaemonThread): self.logger.info(f"registering hardware {name}: {details}") register_keystore(details[1], dynamic_constructor) - def get_plugin(self, name): - if not name in self.plugins: + def get_plugin(self, name: str) -> 'BasePlugin': + if name not in self.plugins: self.load_plugin(name) return self.plugins[name] diff --git a/electrum/plugins/hw_wallet/plugin.py b/electrum/plugins/hw_wallet/plugin.py index 251ee4244..97201f40e 100644 --- a/electrum/plugins/hw_wallet/plugin.py +++ b/electrum/plugins/hw_wallet/plugin.py @@ -32,10 +32,11 @@ from electrum.bitcoin import is_address, opcodes from electrum.util import bfh, versiontuple, UserFacingException from electrum.transaction import TxOutput, Transaction, PartialTransaction, PartialTxInput, PartialTxOutput from electrum.bip32 import BIP32Node +from electrum.storage import get_derivation_used_for_hw_device_encryption +from electrum.keystore import Xpub, Hardware_KeyStore if TYPE_CHECKING: from electrum.wallet import Abstract_Wallet - from electrum.keystore import Hardware_KeyStore class HW_PluginBase(BasePlugin): @@ -69,7 +70,7 @@ class HW_PluginBase(BasePlugin): """ raise NotImplementedError() - def get_client(self, keystore: 'Hardware_KeyStore', force_pair: bool = True): + def get_client(self, keystore: 'Hardware_KeyStore', force_pair: bool = True) -> Optional['HardwareClientBase']: raise NotImplementedError() def show_address(self, wallet: 'Abstract_Wallet', address, keystore: 'Hardware_KeyStore' = None): @@ -182,6 +183,13 @@ class HardwareClientBase: root_fingerprint = BIP32Node.from_xkey(child_of_root_xpub).fingerprint.hex().lower() return root_fingerprint + def get_password_for_storage_encryption(self) -> str: + # note: using a different password based on hw device type is highly undesirable! see #5993 + derivation = get_derivation_used_for_hw_device_encryption() + xpub = self.get_xpub(derivation, "standard") + password = Xpub.get_pubkey_from_xpub(xpub, ()).hex() + return password + def is_any_tx_output_on_change_branch(tx: PartialTransaction) -> bool: return any([txout.is_change for txout in tx.outputs()]) diff --git a/run_electrum b/run_electrum index 83fb8707d..54236da85 100755 --- a/run_electrum +++ b/run_electrum @@ -25,8 +25,7 @@ # SOFTWARE. import os import sys -import warnings -import asyncio + MIN_PYTHON_VERSION = "3.6.1" # FIXME duplicated from setup.py _min_python_version_tuple = tuple(map(int, (MIN_PYTHON_VERSION.split(".")))) @@ -36,6 +35,11 @@ if sys.version_info[:3] < _min_python_version_tuple: sys.exit("Error: Electrum requires Python version >= %s..." % MIN_PYTHON_VERSION) +import warnings +import asyncio +from typing import TYPE_CHECKING + + script_dir = os.path.dirname(os.path.realpath(__file__)) is_bundle = getattr(sys, 'frozen', False) is_local = not is_bundle and os.path.exists(os.path.join(script_dir, "electrum.desktop")) @@ -83,7 +87,7 @@ from electrum import constants from electrum import SimpleConfig from electrum.wallet_db import WalletDB from electrum.wallet import Wallet -from electrum.storage import WalletStorage, get_derivation_used_for_hw_device_encryption +from electrum.storage import WalletStorage from electrum.util import print_msg, print_stderr, json_encode, json_decode, UserCancelled from electrum.util import InvalidPassword from electrum.commands import get_parser, known_commands, Commands, config_variables @@ -91,6 +95,9 @@ from electrum import daemon from electrum import keystore from electrum.util import create_and_start_event_loop +if TYPE_CHECKING: + from electrum.plugin import Plugins + _logger = get_logger(__name__) @@ -166,7 +173,7 @@ def init_cmdline(config_options, wallet_path, server): config_options['new_password'] = new_password -def get_connected_hw_devices(plugins): +def get_connected_hw_devices(plugins: 'Plugins'): supported_plugins = plugins.get_hardware_support() # scan devices devices = [] @@ -186,7 +193,7 @@ def get_connected_hw_devices(plugins): return devices -def get_password_for_hw_device_encrypted_storage(plugins) -> str: +def get_password_for_hw_device_encrypted_storage(plugins: 'Plugins') -> str: devices = get_connected_hw_devices(plugins) if len(devices) == 0: print_msg("Error: No connected hw device found. Cannot decrypt this wallet.") @@ -196,17 +203,15 @@ def get_password_for_hw_device_encrypted_storage(plugins) -> str: "The first one will be used to decrypt the wallet.") # FIXME we use the "first" device, in case of multiple ones name, device_info = devices[0] - plugin = plugins.get_plugin(name) - derivation = get_derivation_used_for_hw_device_encryption() + devmgr = plugins.device_manager try: - xpub = plugin.get_xpub(device_info.device.id_, derivation, 'standard', plugin.handler) + client = devmgr.client_by_id(device_info.device.id_) + return client.get_password_for_storage_encryption() except UserCancelled: sys.exit(0) - password = keystore.Xpub.get_pubkey_from_xpub(xpub, ()).hex() - return password -async def run_offline_command(config, config_options, plugins): +async def run_offline_command(config, config_options, plugins: 'Plugins'): cmdname = config.get('cmd') cmd = known_commands[cmdname] password = config_options.get('password')