diff --git a/electrum/plugins/hw_wallet/plugin.py b/electrum/plugins/hw_wallet/plugin.py index 251ee4244..b4e6edd90 100644 --- a/electrum/plugins/hw_wallet/plugin.py +++ b/electrum/plugins/hw_wallet/plugin.py @@ -24,25 +24,34 @@ # CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. -from typing import TYPE_CHECKING, Dict, List, Union, Tuple, Sequence, Optional, Type +from typing import TYPE_CHECKING, Dict, List, Union, Tuple, Sequence, Optional, Type, Iterable, Any +from functools import partial -from electrum.plugin import BasePlugin, hook, Device, DeviceMgr +from electrum.plugin import (BasePlugin, hook, Device, DeviceMgr, DeviceInfo, + assert_runs_in_hwd_thread, runs_in_hwd_thread) from electrum.i18n import _ 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: + import threading from electrum.wallet import Abstract_Wallet - from electrum.keystore import Hardware_KeyStore + from electrum.base_wizard import BaseWizard class HW_PluginBase(BasePlugin): keystore_class: Type['Hardware_KeyStore'] libraries_available: bool + # define supported library versions: minimum_library <= x < maximum_library minimum_library = (0, ) + maximum_library = (float('inf'), ) + + DEVICE_IDS: Iterable[Any] def __init__(self, parent, config, name): BasePlugin.__init__(self, parent, config, name) @@ -56,21 +65,58 @@ class HW_PluginBase(BasePlugin): def device_manager(self) -> 'DeviceMgr': return self.parent.device_manager + def create_device_from_hid_enumeration(self, d: dict, *, product_key) -> Optional['Device']: + # Older versions of hid don't provide interface_number + interface_number = d.get('interface_number', -1) + usage_page = d['usage_page'] + id_ = d['serial_number'] + if len(id_) == 0: + id_ = str(d['path']) + id_ += str(interface_number) + str(usage_page) + device = Device(path=d['path'], + interface_number=interface_number, + id_=id_, + product_key=product_key, + usage_page=usage_page, + transport_ui_string='hid') + return device + @hook def close_wallet(self, wallet: 'Abstract_Wallet'): for keystore in wallet.get_keystores(): if isinstance(keystore, self.keystore_class): self.device_manager().unpair_xpub(keystore.xpub) + if keystore.thread: + keystore.thread.stop() - def setup_device(self, device_info, wizard, purpose): + def scan_and_create_client_for_device(self, *, device_id: str, wizard: 'BaseWizard') -> 'HardwareClientBase': + devmgr = self.device_manager() + client = wizard.run_task_without_blocking_gui( + task=partial(devmgr.client_by_id, device_id)) + if client is None: + raise UserFacingException(_('Failed to create a client for this device.') + '\n' + + _('Make sure it is in the correct state.')) + client.handler = self.create_handler(wizard) + return client + + def setup_device(self, device_info: DeviceInfo, wizard: 'BaseWizard', purpose) -> 'HardwareClientBase': """Called when creating a new wallet or when using the device to decrypt an existing wallet. Select the device to use. If the device is uninitialized, go through the initialization process. + + Runs in GUI thread. """ raise NotImplementedError() - def get_client(self, keystore: 'Hardware_KeyStore', force_pair: bool = True): - raise NotImplementedError() + def get_client(self, keystore: 'Hardware_KeyStore', force_pair: bool = True, *, + devices: Sequence['Device'] = None, + allow_user_interaction: bool = True) -> Optional['HardwareClientBase']: + devmgr = self.device_manager() + handler = keystore.handler + client = devmgr.client_for_keystore(self, handler, keystore, force_pair, + devices=devices, + allow_user_interaction=allow_user_interaction) + return client def show_address(self, wallet: 'Abstract_Wallet', address, keystore: 'Hardware_KeyStore' = None): pass # implemented in child classes @@ -108,17 +154,16 @@ class HW_PluginBase(BasePlugin): # if no exception so far, we might still raise LibraryFoundButUnusable if (library_version == 'unknown' or versiontuple(library_version) < self.minimum_library - or hasattr(self, "maximum_library") and versiontuple(library_version) >= self.maximum_library): + or versiontuple(library_version) >= self.maximum_library): raise LibraryFoundButUnusable(library_version=library_version) except ImportError: return False except LibraryFoundButUnusable as e: library_version = e.library_version - max_version_str = version_str(self.maximum_library) if hasattr(self, "maximum_library") else "inf" self.libraries_available_message = ( _("Library version for '{}' is incompatible.").format(self.name) + '\nInstalled: {}, Needed: {} <= x < {}' - .format(library_version, version_str(self.minimum_library), max_version_str)) + .format(library_version, version_str(self.minimum_library), version_str(self.maximum_library))) self.logger.warning(self.libraries_available_message) return False @@ -138,15 +183,35 @@ class HW_PluginBase(BasePlugin): def is_outdated_fw_ignored(self) -> bool: return self._ignore_outdated_fw - def create_client(self, device: 'Device', handler) -> Optional['HardwareClientBase']: + def create_client(self, device: 'Device', + handler: Optional['HardwareHandlerBase']) -> Optional['HardwareClientBase']: raise NotImplementedError() - def get_xpub(self, device_id, derivation: str, xtype, wizard) -> str: + def get_xpub(self, device_id: str, derivation: str, xtype, wizard: 'BaseWizard') -> str: raise NotImplementedError() + def create_handler(self, window) -> 'HardwareHandlerBase': + # note: in Qt GUI, 'window' is either an ElectrumWindow or an InstallWizard + raise NotImplementedError() + + def can_recognize_device(self, device: Device) -> bool: + """Whether the plugin thinks it can handle the given device. + Used for filtering all connected hardware devices to only those by this vendor. + """ + return device.product_key in self.DEVICE_IDS + class HardwareClientBase: + handler = None # type: Optional['HardwareHandlerBase'] + + def __init__(self, *, plugin: 'HW_PluginBase'): + assert_runs_in_hwd_thread() + self.plugin = plugin + + def device_manager(self) -> 'DeviceMgr': + return self.plugin.device_manager() + def is_pairable(self) -> bool: raise NotImplementedError() @@ -167,7 +232,19 @@ class HardwareClientBase: and they are also used as a fallback to distinguish devices programmatically. So ideally, different devices would have different labels. """ - raise NotImplementedError() + # When returning a constant here (i.e. not implementing the method in the way + # it is supposed to work), make sure the return value is in electrum.plugin.PLACEHOLDER_HW_CLIENT_LABELS + return " " + + def get_soft_device_id(self) -> Optional[str]: + """An id-like string that is used to distinguish devices programmatically. + This is a long term id for the device, that does not change between reconnects. + This method should not prompt the user, i.e. no user interaction, as it is used + during USB device enumeration (called for each unpaired device). + Stored in the wallet file. + """ + # This functionality is optional. If not implemented just return None: + return None def has_usable_connection_with_device(self) -> bool: raise NotImplementedError() @@ -175,6 +252,7 @@ class HardwareClientBase: def get_xpub(self, bip32_path: str, xtype) -> str: raise NotImplementedError() + @runs_in_hwd_thread def request_root_fingerprint_from_device(self) -> str: # digitalbitbox (at least) does not reveal xpubs corresponding to unhardened paths # so ask for a direct child, and read out fingerprint from that: @@ -182,6 +260,70 @@ class HardwareClientBase: root_fingerprint = BIP32Node.from_xkey(child_of_root_xpub).fingerprint.hex().lower() return root_fingerprint + @runs_in_hwd_thread + 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 device_model_name(self) -> Optional[str]: + """Return the name of the model of this device, which might be displayed in the UI. + E.g. for Trezor, "Trezor One" or "Trezor T". + """ + return None + + def manipulate_keystore_dict_during_wizard_setup(self, d: dict) -> None: + """Called during wallet creation in the wizard, before the keystore + is constructed for the first time. 'd' is the dict that will be + passed to the keystore constructor. + """ + pass + + +class HardwareHandlerBase: + """An interface between the GUI and the device handling logic for handling I/O.""" + win = None + device: str + + def get_wallet(self) -> Optional['Abstract_Wallet']: + if self.win is not None: + if hasattr(self.win, 'wallet'): + return self.win.wallet + + def get_gui_thread(self) -> Optional['threading.Thread']: + if self.win is not None: + if hasattr(self.win, 'gui_thread'): + return self.win.gui_thread + + def update_status(self, paired: bool) -> None: + pass + + def query_choice(self, msg: str, labels: Sequence[str]) -> Optional[int]: + raise NotImplementedError() + + def yes_no_question(self, msg: str) -> bool: + raise NotImplementedError() + + def show_message(self, msg: str, on_cancel=None) -> None: + raise NotImplementedError() + + def show_error(self, msg: str, blocking: bool = False) -> None: + raise NotImplementedError() + + def finished(self) -> None: + pass + + def get_word(self, msg: str) -> str: + raise NotImplementedError() + + def get_passphrase(self, msg: str, confirm: bool) -> Optional[str]: + raise NotImplementedError() + + def get_pin(self, msg: str, *, show_strength: bool = True) -> str: + raise NotImplementedError() + def is_any_tx_output_on_change_branch(tx: PartialTransaction) -> bool: return any([txout.is_change for txout in tx.outputs()])