AaronFiore/electrum
1
0
Fork 0
Code Activity

hw DeviceMgr: mostly switch away from xpubs for device pairing

Browse Source 
- the DeviceMgr no longer uses xpubs to keep track of paired hw devices
- instead, introduce keystore.pairing_code(), based on soft_device_id
- xpubs are now only used in a single place when the actual pairing happens
- motivation is to allow pairing a single device with multiple generic
  output script descriptors, not just a single account-level xpub
- as a side-effect, we now allow pairing a device with multiple open
  windows simultaneously (if keystores have the same root fingerprint
  -- was already the case before if keystores had the same xpub)
This commit is contained in:
SomberNight
2022-12-19 08:19:19 +00:00
parent 8878059b2f
commit cea4238b81
5 changed files with 48 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
electrum/base_wizard.py
View File

@@ -622,7 +622,7 @@ class BaseWizard(Logger):
password = k.get_password_for_storage_encryption()
except UserCancelled:
devmgr = self.plugins.device_manager
devmgr.unpair_xpub(k.xpub)
devmgr.unpair_pairing_code(k.pairing_code())
raise ChooseHwDeviceAgain()
except BaseException as e:
self.logger.exception('')

6
electrum/keystore.py
View File

@@ -915,6 +915,12 @@ class Hardware_KeyStore(Xpub, KeyStore):
self.soft_device_id = client.get_soft_device_id()
self.is_requesting_to_be_rewritten_to_wallet_file = True
def pairing_code(self) -> Optional[str]:
"""Used by the DeviceMgr to keep track of paired hw devices."""
if not self.soft_device_id:
return None
return f"{self.plugin.name}/{self.soft_device_id}"
KeyStoreWithMPK = Union[KeyStore, MasterPublicKeyMixin] # intersection really...
AddressIndexGeneric = Union[Sequence[int], str] # can be hex pubkey str

69
electrum/plugin.py
View File

@@ -401,8 +401,8 @@ class DeviceMgr(ThreadJob):
def __init__(self, config: SimpleConfig):
ThreadJob.__init__(self)
# An xpub->id_ map. Item only present if we have active pairing. Needs self.lock.
self.xpub_ids = {} # type: Dict[str, str]
# A pairing_code->id_ map. Item only present if we have active pairing. Needs self.lock.
self.pairing_code_to_id = {} # type: Dict[str, str]
# A client->id_ map. Needs self.lock.
self.clients = {} # type: Dict[HardwareClientBase, str]
# What we recognise. (vendor_id, product_id) -> Plugin
@@ -454,28 +454,28 @@ class DeviceMgr(ThreadJob):
self.clients[client] = device.id_
return client
def xpub_id(self, xpub):
def id_by_pairing_code(self, pairing_code):
with self.lock:
return self.xpub_ids.get(xpub)
return self.pairing_code_to_id.get(pairing_code)
def xpub_by_id(self, id_):
def pairing_code_by_id(self, id_):
with self.lock:
for xpub, xpub_id in self.xpub_ids.items():
if xpub_id == id_:
return xpub
for pairing_code, id2 in self.pairing_code_to_id.items():
if id2 == id_:
return pairing_code
return None
def unpair_xpub(self, xpub):
def unpair_pairing_code(self, pairing_code):
with self.lock:
if xpub not in self.xpub_ids:
if pairing_code not in self.pairing_code_to_id:
return
_id = self.xpub_ids.pop(xpub)
_id = self.pairing_code_to_id.pop(pairing_code)
self._close_client(_id)
def unpair_id(self, id_):
xpub = self.xpub_by_id(id_)
if xpub:
self.unpair_xpub(xpub)
pairing_code = self.pairing_code_by_id(id_)
if pairing_code:
self.unpair_pairing_code(pairing_code)
else:
self._close_client(id_)
@@ -486,10 +486,6 @@ class DeviceMgr(ThreadJob):
if client:
client.close()
def pair_xpub(self, xpub, id_):
with self.lock:
self.xpub_ids[xpub] = id_
def _client_by_id(self, id_) -> Optional['HardwareClientBase']:
with self.lock:
for client, client_id in self.clients.items():
@@ -517,10 +513,8 @@ class DeviceMgr(ThreadJob):
handler.update_status(False)
if devices is None:
devices = self.scan_devices()
xpub = keystore.xpub
derivation = keystore.get_derivation_prefix()
assert derivation is not None
client = self.client_by_xpub(plugin, xpub, handler, devices)
client = self.client_by_pairing_code(
plugin=plugin, pairing_code=keystore.pairing_code(), handler=handler, devices=devices)
if client is None and force_pair:
try:
info = self.select_device(plugin, handler, keystore, devices,
@@ -528,7 +522,7 @@ class DeviceMgr(ThreadJob):
except CannotAutoSelectDevice:
pass
else:
client = self.force_pair_xpub(plugin, handler, info, xpub, derivation)
client = self.force_pair_keystore(plugin=plugin, handler=handler, info=info, keystore=keystore)
if client:
handler.update_status(True)
# note: if select_device was called, we might also update label etc here:
@@ -536,9 +530,11 @@ class DeviceMgr(ThreadJob):
self.logger.info("end client for keystore")
return client
def client_by_xpub(self, plugin: 'HW_PluginBase', xpub, handler: 'HardwareHandlerBase',
devices: Sequence['Device']) -> Optional['HardwareClientBase']:
_id = self.xpub_id(xpub)
def client_by_pairing_code(
self, *, plugin: 'HW_PluginBase', pairing_code: str, handler: 'HardwareHandlerBase',
devices: Sequence['Device'],
) -> Optional['HardwareClientBase']:
_id = self.id_by_pairing_code(pairing_code)
client = self._client_by_id(_id)
if client:
if type(client.plugin) != type(plugin):
@@ -552,10 +548,17 @@ class DeviceMgr(ThreadJob):
if device.id_ == _id:
return self.create_client(device, handler, plugin)
def force_pair_xpub(self, plugin: 'HW_PluginBase', handler: 'HardwareHandlerBase',
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.
def force_pair_keystore(
self,
*,
plugin: 'HW_PluginBase',
handler: 'HardwareHandlerBase',
info: 'DeviceInfo',
keystore: 'Hardware_KeyStore',
) -> 'HardwareClientBase':
xpub = keystore.xpub
derivation = keystore.get_derivation_prefix()
assert derivation is not None
xtype = bip32.xpub_type(xpub)
client = self._client_by_id(info.device.id_)
if client and client.is_pairable() and type(client.plugin) == type(plugin):
@@ -568,7 +571,9 @@ class DeviceMgr(ThreadJob):
# Bad / cancelled PIN / passphrase
client_xpub = None
if client_xpub == xpub:
self.pair_xpub(xpub, info.device.id_)
keystore.opportunistically_fill_in_missing_info_from_device(client)
with self.lock:
self.pairing_code_to_id[keystore.pairing_code()] = info.device.id_
return client
# The user input has wrong PIN or passphrase, or cancelled input,
@@ -590,7 +595,7 @@ class DeviceMgr(ThreadJob):
raise HardwarePluginLibraryUnavailable(message)
if devices is None:
devices = self.scan_devices()
devices = [dev for dev in devices if not self.xpub_by_id(dev.id_)]
devices = [dev for dev in devices if not self.pairing_code_by_id(dev.id_)]
infos = []
for device in devices:
if not plugin.can_recognize_device(device):

6
electrum/plugins/hw_wallet/plugin.py
View File

@@ -86,7 +86,7 @@ class HW_PluginBase(BasePlugin):
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)
self.device_manager().unpair_pairing_code(keystore.pairing_code())
if keystore.thread:
keystore.thread.stop()
@@ -248,8 +248,8 @@ class HardwareClientBase:
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
root_fp = self.request_root_fingerprint_from_device()
return root_fp
def has_usable_connection_with_device(self) -> bool:
raise NotImplementedError()

2
electrum/plugins/hw_wallet/qt.py
View File

@@ -264,7 +264,7 @@ class QtPluginBase(object):
'''This dialog box should be usable even if the user has
forgotten their PIN or it is in bootloader mode.'''
assert window.gui_thread != threading.current_thread(), 'must not be called from GUI thread'
device_id = self.device_manager().xpub_id(keystore.xpub)
device_id = self.device_manager().id_by_pairing_code(keystore.pairing_code())
if not device_id:
try:
info = self.device_manager().select_device(self, keystore.handler, keystore)