hardware wallets: create base class for HW Clients. add some type hints

electrum/keystore.py

@@ -45,6 +45,7 @@ from .logging import Logger
 
 if TYPE_CHECKING:
     from .transaction import Transaction, PartialTransaction, PartialTxInput, PartialTxOutput
+    from .plugins.hw_wallet import HW_PluginBase
 
 
 class KeyStore(Logger):
@@ -624,12 +625,10 @@ class Old_KeyStore(Deterministic_KeyStore):
         self.pw_hash_version = PW_HASH_VERSION_LATEST
 
 
-
 class Hardware_KeyStore(KeyStore, Xpub):
-    # Derived classes must set:
-    #   - device
-    #   - DEVICE_IDS
-    #   - wallet_type
+    hw_type: str
+    device: str
+    plugin: 'HW_PluginBase'
 
     type = 'hardware'
 

electrum/plugin.py

@@ -28,7 +28,8 @@ import importlib.util
 import time
 import threading
 import sys
-from typing import NamedTuple, Any, Union, TYPE_CHECKING, Optional
+from typing import (NamedTuple, Any, Union, TYPE_CHECKING, Optional, Tuple,
+                    Dict, Iterable, List)
 
 from .i18n import _
 from .util import (profiler, DaemonThread, UserCancelled, ThreadJob, UserFacingException)
@@ -38,7 +39,7 @@ from .simple_config import SimpleConfig
 from .logging import get_logger, Logger
 
 if TYPE_CHECKING:
-    from .plugins.hw_wallet import HW_PluginBase
+    from .plugins.hw_wallet import HW_PluginBase, HardwareClientBase
     from .keystore import Hardware_KeyStore
 
 
@@ -234,7 +235,7 @@ def run_hook(name, *args):
 class BasePlugin(Logger):
 
     def __init__(self, parent, config, name):
-        self.parent = parent  # The plugins object
+        self.parent = parent  # type: Plugins  # The plugins object
         self.name = name
         self.config = config
         self.wallet = None
@@ -351,7 +352,7 @@ class DeviceMgr(ThreadJob):
         self.xpub_ids = {}
         # A list of clients.  The key is the client, the value is
         # a (path, id_) pair.
-        self.clients = {}
+        self.clients = {}  # type: Dict[HardwareClientBase, Tuple[Union[str, bytes], str]]
         # What we recognise.  Each entry is a (vendor_id, product_id)
         # pair.
         self.recognised_hardware = set()
@@ -382,7 +383,7 @@ class DeviceMgr(ThreadJob):
     def register_enumerate_func(self, func):
         self.enumerate_func.add(func)
 
-    def create_client(self, device, handler, plugin):
+    def create_client(self, device: 'Device', handler, plugin: 'HW_PluginBase') -> Optional['HardwareClientBase']:
         # Get from cache first
         client = self.client_lookup(device.id_)
         if client:
@@ -429,21 +430,22 @@ class DeviceMgr(ThreadJob):
         with self.lock:
             self.xpub_ids[xpub] = id_
 
-    def client_lookup(self, id_):
+    def client_lookup(self, id_) -> Optional['HardwareClientBase']:
         with self.lock:
             for client, (path, client_id) in self.clients.items():
                 if client_id == id_:
                     return client
         return None
 
-    def client_by_id(self, id_):
+    def client_by_id(self, id_) -> Optional['HardwareClientBase']:
         '''Returns a client for the device ID if one is registered.  If
         a device is wiped or in bootloader mode pairing is impossible;
         in such cases we communicate by device ID and not wallet.'''
         self.scan_devices()
         return self.client_lookup(id_)
 
-    def client_for_keystore(self, plugin, handler, keystore: 'Hardware_KeyStore', force_pair):
+    def client_for_keystore(self, plugin: 'HW_PluginBase', handler, keystore: 'Hardware_KeyStore',
+                            force_pair: bool) -> Optional['HardwareClientBase']:
         self.logger.info("getting client for keystore")
         if handler is None:
             raise Exception(_("Handler not found for") + ' ' + plugin.name + '\n' + _("A library is probably missing."))
@@ -455,7 +457,7 @@ class DeviceMgr(ThreadJob):
         client = self.client_by_xpub(plugin, xpub, handler, devices)
         if client is None and force_pair:
             info = self.select_device(plugin, handler, keystore, devices)
-            client = self.force_pair_xpub(plugin, handler, info, xpub, derivation, devices)
+            client = self.force_pair_xpub(plugin, handler, info, xpub, derivation)
         if client:
             handler.update_status(True)
         if client:
@@ -463,7 +465,8 @@ class DeviceMgr(ThreadJob):
         self.logger.info("end client for keystore")
         return client
 
-    def client_by_xpub(self, plugin, xpub, handler, devices):
+    def client_by_xpub(self, plugin: 'HW_PluginBase', xpub, handler,
+                       devices: Iterable['Device']) -> Optional['HardwareClientBase']:
         _id = self.xpub_id(xpub)
         client = self.client_lookup(_id)
         if client:
@@ -476,8 +479,8 @@ class DeviceMgr(ThreadJob):
             if device.id_ == _id:
                 return self.create_client(device, handler, plugin)
 
-
-    def force_pair_xpub(self, plugin, handler, info, xpub, derivation, devices):
+    def force_pair_xpub(self, plugin: 'HW_PluginBase', handler,
+                        info: 'DeviceInfo', xpub, derivation) -> Optional['HardwareClientBase']:
         # The wallet has not been previously paired, so let the user
         # choose an unpaired device and compare its first address.
         xtype = bip32.xpub_type(xpub)
@@ -533,7 +536,8 @@ class DeviceMgr(ThreadJob):
 
         return infos
 
-    def select_device(self, plugin, handler, keystore, devices=None):
+    def select_device(self, plugin: 'HW_PluginBase', handler,
+                      keystore: 'Hardware_KeyStore', devices=None) -> 'DeviceInfo':
         '''Ask the user to select a device to use if there is more than one,
         and return the DeviceInfo for the device.'''
         while True:
@@ -569,7 +573,7 @@ class DeviceMgr(ThreadJob):
             handler.win.wallet.save_keystore()
         return info
 
-    def _scan_devices_with_hid(self):
+    def _scan_devices_with_hid(self) -> List['Device']:
         try:
             import hid
         except ImportError:
@@ -597,7 +601,7 @@ class DeviceMgr(ThreadJob):
                                       transport_ui_string='hid'))
         return devices
 
-    def scan_devices(self):
+    def scan_devices(self) -> List['Device']:
         self.logger.info("scanning devices...")
 
         # First see what's connected that we know about

electrum/plugins/coldcard/coldcard.py

@@ -18,7 +18,7 @@ from electrum.util import bfh, bh2u, versiontuple, UserFacingException
 from electrum.base_wizard import ScriptTypeNotSupported
 from electrum.logging import get_logger
 
-from ..hw_wallet import HW_PluginBase
+from ..hw_wallet import HW_PluginBase, HardwareClientBase
 from ..hw_wallet.plugin import LibraryFoundButUnusable, only_hook_if_libraries_available
 
 if TYPE_CHECKING:
@@ -60,9 +60,8 @@ except ImportError:
 
 CKCC_SIMULATED_PID = CKCC_PID ^ 0x55aa
 
-class CKCCClient:
-    # Challenge: I haven't found anywhere that defines a base class for this 'client',
-    # nor an API (interface) to be met. Winging it. Gets called from lib/plugins.py mostly?
+
+class CKCCClient(HardwareClientBase):
 
     def __init__(self, plugin, handler, dev_path, is_simulator=False):
         self.device = plugin.device
@@ -463,11 +462,9 @@ class Coldcard_KeyStore(Hardware_KeyStore):
             self.handler.show_error(exc)
 
 
-
 class ColdcardPlugin(HW_PluginBase):
     keystore_class = Coldcard_KeyStore
     minimum_library = (0, 7, 7)
-    client = None
 
     DEVICE_IDS = [
         (COINKITE_VID, CKCC_PID),

electrum/plugins/digitalbitbox/digitalbitbox.py

@@ -27,12 +27,13 @@ from electrum import constants
 from electrum.transaction import Transaction, PartialTransaction, PartialTxInput
 from electrum.i18n import _
 from electrum.keystore import Hardware_KeyStore
-from ..hw_wallet import HW_PluginBase
 from electrum.util import to_string, UserCancelled, UserFacingException, bfh
 from electrum.base_wizard import ScriptTypeNotSupported, HWD_SETUP_NEW_WALLET
 from electrum.network import Network
 from electrum.logging import get_logger
 
+from ..hw_wallet import HW_PluginBase, HardwareClientBase
+
 
 _logger = get_logger(__name__)
 
@@ -63,7 +64,7 @@ MIN_MAJOR_VERSION = 5
 ENCRYPTION_PRIVKEY_KEY = 'encryptionprivkey'
 CHANNEL_ID_KEY = 'comserverchannelid'
 
-class DigitalBitbox_Client():
+class DigitalBitbox_Client(HardwareClientBase):
 
     def __init__(self, plugin, hidDevice):
         self.plugin = plugin

electrum/plugins/hw_wallet/__init__.py

@@ -1,2 +1,2 @@
-from .plugin import HW_PluginBase
+from .plugin import HW_PluginBase, HardwareClientBase
 from .cmdline import CmdLineHandler

electrum/plugins/hw_wallet/plugin.py

@@ -24,9 +24,9 @@
 # CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 # SOFTWARE.
 
-from typing import TYPE_CHECKING, Dict, List, Union, Tuple, Sequence
+from typing import TYPE_CHECKING, Dict, List, Union, Tuple, Sequence, Optional, Type
 
-from electrum.plugin import BasePlugin, hook
+from electrum.plugin import BasePlugin, hook, Device, DeviceMgr
 from electrum.i18n import _
 from electrum.bitcoin import is_address, TYPE_SCRIPT, opcodes
 from electrum.util import bfh, versiontuple, UserFacingException
@@ -39,11 +39,7 @@ if TYPE_CHECKING:
 
 
 class HW_PluginBase(BasePlugin):
-    # Derived classes provide:
-    #
-    #  class-static variables: client_class, firmware_URL, handler_class,
-    #     libraries_available, libraries_URL, minimum_firmware,
-    #     wallet_class, ckd_public, types, HidTransport
+    keystore_class: Type['Hardware_KeyStore']
 
     minimum_library = (0, )
 
@@ -56,11 +52,11 @@ class HW_PluginBase(BasePlugin):
     def is_enabled(self):
         return True
 
-    def device_manager(self):
+    def device_manager(self) -> 'DeviceMgr':
         return self.parent.device_manager
 
     @hook
-    def close_wallet(self, wallet):
+    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)
@@ -141,6 +137,38 @@ 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']:
+        raise NotImplementedError()
+
+    def get_xpub(self, device_id, derivation: str, xtype, wizard) -> str:
+        raise NotImplementedError()
+
+
+class HardwareClientBase:
+
+    def is_pairable(self) -> bool:
+        raise NotImplementedError()
+
+    def close(self):
+        raise NotImplementedError()
+
+    def timeout(self, cutoff) -> None:
+        pass
+
+    def is_initialized(self) -> bool:
+        """True if initialized, False if wiped."""
+        raise NotImplementedError()
+
+    def label(self) -> str:
+        """The name given by the user to the device."""
+        raise NotImplementedError()
+
+    def has_usable_connection_with_device(self) -> bool:
+        raise NotImplementedError()
+
+    def get_xpub(self, bip32_path: str, xtype) -> str:
+        raise NotImplementedError()
+
 
 def is_any_tx_output_on_change_branch(tx: PartialTransaction) -> bool:
     return any([txout.is_change for txout in tx.outputs()])

electrum/plugins/keepkey/clientbase.py

@@ -7,6 +7,7 @@ from electrum.util import UserCancelled
 from electrum.keystore import bip39_normalize_passphrase
 from electrum.bip32 import BIP32Node, convert_bip32_path_to_list_of_uint32
 from electrum.logging import Logger
+from electrum.plugins.hw_wallet.plugin import HardwareClientBase
 
 
 class GuiMixin(object):
@@ -94,7 +95,7 @@ class GuiMixin(object):
         return self.proto.CharacterAck(**char_info)
 
 
-class KeepKeyClientBase(GuiMixin, Logger):
+class KeepKeyClientBase(HardwareClientBase, GuiMixin, Logger):
 
     def __init__(self, handler, plugin, proto):
         assert hasattr(self, 'tx_api')  # ProtocolMixin already constructed?
@@ -112,11 +113,9 @@ class KeepKeyClientBase(GuiMixin, Logger):
         return "%s/%s" % (self.label(), self.features.device_id)
 
     def label(self):
-        '''The name given by the user to the device.'''
         return self.features.label
 
     def is_initialized(self):
-        '''True if initialized, False if wiped.'''
         return self.features.initialized
 
     def is_pairable(self):

electrum/plugins/ledger/ledger.py

@@ -16,7 +16,7 @@ from electrum.util import bfh, bh2u, versiontuple, UserFacingException
 from electrum.base_wizard import ScriptTypeNotSupported
 from electrum.logging import get_logger
 
-from ..hw_wallet import HW_PluginBase
+from ..hw_wallet import HW_PluginBase, HardwareClientBase
 from ..hw_wallet.plugin import is_any_tx_output_on_change_branch
 
 
@@ -60,7 +60,7 @@ def test_pin_unlocked(func):
     return catch_exception
 
 
-class Ledger_Client():
+class Ledger_Client(HardwareClientBase):
     def __init__(self, hidDevice):
         self.dongleObject = btchip(hidDevice)
         self.preflightDone = False

electrum/plugins/safe_t/clientbase.py

@@ -7,6 +7,7 @@ from electrum.util import UserCancelled
 from electrum.keystore import bip39_normalize_passphrase
 from electrum.bip32 import BIP32Node, convert_bip32_path_to_list_of_uint32
 from electrum.logging import Logger
+from electrum.plugins.hw_wallet.plugin import HardwareClientBase
 
 
 class GuiMixin(object):
@@ -96,7 +97,7 @@ class GuiMixin(object):
         return self.proto.WordAck(word=word)
 
 
-class SafeTClientBase(GuiMixin, Logger):
+class SafeTClientBase(HardwareClientBase, GuiMixin, Logger):
 
     def __init__(self, handler, plugin, proto):
         assert hasattr(self, 'tx_api')  # ProtocolMixin already constructed?
@@ -114,11 +115,9 @@ class SafeTClientBase(GuiMixin, Logger):
         return "%s/%s" % (self.label(), self.features.device_id)
 
     def label(self):
-        '''The name given by the user to the device.'''
         return self.features.label
 
     def is_initialized(self):
-        '''True if initialized, False if wiped.'''
         return self.features.initialized
 
     def is_pairable(self):

electrum/plugins/trezor/clientbase.py

@@ -7,7 +7,7 @@ from electrum.util import UserCancelled, UserFacingException
 from electrum.keystore import bip39_normalize_passphrase
 from electrum.bip32 import BIP32Node, convert_bip32_path_to_list_of_uint32 as parse_path
 from electrum.logging import Logger
-from electrum.plugins.hw_wallet.plugin import OutdatedHwFirmwareException
+from electrum.plugins.hw_wallet.plugin import OutdatedHwFirmwareException, HardwareClientBase
 
 from trezorlib.client import TrezorClient
 from trezorlib.exceptions import TrezorFailure, Cancelled, OutdatedFirmwareError
@@ -28,7 +28,7 @@ MESSAGES = {
 }
 
 
-class TrezorClientBase(Logger):
+class TrezorClientBase(HardwareClientBase, Logger):
     def __init__(self, transport, handler, plugin):
         if plugin.is_outdated_fw_ignored():
             TrezorClient.is_outdated = lambda *args, **kwargs: False
@@ -86,11 +86,9 @@ class TrezorClientBase(Logger):
         return "%s/%s" % (self.label(), self.features.device_id)
 
     def label(self):
-        '''The name given by the user to the device.'''
         return self.features.label
 
     def is_initialized(self):
-        '''True if initialized, False if wiped.'''
         return self.features.initialized
 
     def is_pairable(self):