tplugin.py - electrum - Electrum Bitcoin wallet
HTML git clone https://git.parazyd.org/electrum
DIR Log
DIR Files
DIR Refs
DIR Submodules
---
tplugin.py (16379B)
---
1 #!/usr/bin/env python2
2 # -*- mode: python -*-
3 #
4 # Electrum - lightweight Bitcoin client
5 # Copyright (C) 2016 The Electrum developers
6 #
7 # Permission is hereby granted, free of charge, to any person
8 # obtaining a copy of this software and associated documentation files
9 # (the "Software"), to deal in the Software without restriction,
10 # including without limitation the rights to use, copy, modify, merge,
11 # publish, distribute, sublicense, and/or sell copies of the Software,
12 # and to permit persons to whom the Software is furnished to do so,
13 # subject to the following conditions:
14 #
15 # The above copyright notice and this permission notice shall be
16 # included in all copies or substantial portions of the Software.
17 #
18 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
19 # EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
20 # MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
21 # NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
22 # BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
23 # ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
24 # CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
25 # SOFTWARE.
26
27 from typing import TYPE_CHECKING, Dict, List, Union, Tuple, Sequence, Optional, Type, Iterable, Any
28 from functools import partial
29
30 from electrum.plugin import (BasePlugin, hook, Device, DeviceMgr, DeviceInfo,
31 assert_runs_in_hwd_thread, runs_in_hwd_thread)
32 from electrum.i18n import _
33 from electrum.bitcoin import is_address, opcodes
34 from electrum.util import bfh, versiontuple, UserFacingException
35 from electrum.transaction import TxOutput, Transaction, PartialTransaction, PartialTxInput, PartialTxOutput
36 from electrum.bip32 import BIP32Node
37 from electrum.storage import get_derivation_used_for_hw_device_encryption
38 from electrum.keystore import Xpub, Hardware_KeyStore
39
40 if TYPE_CHECKING:
41 import threading
42 from electrum.wallet import Abstract_Wallet
43 from electrum.base_wizard import BaseWizard
44
45
46 class HW_PluginBase(BasePlugin):
47 keystore_class: Type['Hardware_KeyStore']
48 libraries_available: bool
49
50 # define supported library versions: minimum_library <= x < maximum_library
51 minimum_library = (0, )
52 maximum_library = (float('inf'), )
53
54 DEVICE_IDS: Iterable[Any]
55
56 def __init__(self, parent, config, name):
57 BasePlugin.__init__(self, parent, config, name)
58 self.device = self.keystore_class.device
59 self.keystore_class.plugin = self
60 self._ignore_outdated_fw = False
61
62 def is_enabled(self):
63 return True
64
65 def device_manager(self) -> 'DeviceMgr':
66 return self.parent.device_manager
67
68 def create_device_from_hid_enumeration(self, d: dict, *, product_key) -> Optional['Device']:
69 # Older versions of hid don't provide interface_number
70 interface_number = d.get('interface_number', -1)
71 usage_page = d['usage_page']
72 id_ = d['serial_number']
73 if len(id_) == 0:
74 id_ = str(d['path'])
75 id_ += str(interface_number) + str(usage_page)
76 device = Device(path=d['path'],
77 interface_number=interface_number,
78 id_=id_,
79 product_key=product_key,
80 usage_page=usage_page,
81 transport_ui_string='hid')
82 return device
83
84 @hook
85 def close_wallet(self, wallet: 'Abstract_Wallet'):
86 for keystore in wallet.get_keystores():
87 if isinstance(keystore, self.keystore_class):
88 self.device_manager().unpair_xpub(keystore.xpub)
89 if keystore.thread:
90 keystore.thread.stop()
91
92 def scan_and_create_client_for_device(self, *, device_id: str, wizard: 'BaseWizard') -> 'HardwareClientBase':
93 devmgr = self.device_manager()
94 client = wizard.run_task_without_blocking_gui(
95 task=partial(devmgr.client_by_id, device_id))
96 if client is None:
97 raise UserFacingException(_('Failed to create a client for this device.') + '\n' +
98 _('Make sure it is in the correct state.'))
99 client.handler = self.create_handler(wizard)
100 return client
101
102 def setup_device(self, device_info: DeviceInfo, wizard: 'BaseWizard', purpose) -> 'HardwareClientBase':
103 """Called when creating a new wallet or when using the device to decrypt
104 an existing wallet. Select the device to use. If the device is
105 uninitialized, go through the initialization process.
106
107 Runs in GUI thread.
108 """
109 raise NotImplementedError()
110
111 def get_client(self, keystore: 'Hardware_KeyStore', force_pair: bool = True, *,
112 devices: Sequence['Device'] = None,
113 allow_user_interaction: bool = True) -> Optional['HardwareClientBase']:
114 devmgr = self.device_manager()
115 handler = keystore.handler
116 client = devmgr.client_for_keystore(self, handler, keystore, force_pair,
117 devices=devices,
118 allow_user_interaction=allow_user_interaction)
119 return client
120
121 def show_address(self, wallet: 'Abstract_Wallet', address, keystore: 'Hardware_KeyStore' = None):
122 pass # implemented in child classes
123
124 def show_address_helper(self, wallet, address, keystore=None):
125 if keystore is None:
126 keystore = wallet.get_keystore()
127 if not is_address(address):
128 keystore.handler.show_error(_('Invalid Bitcoin Address'))
129 return False
130 if not wallet.is_mine(address):
131 keystore.handler.show_error(_('Address not in wallet.'))
132 return False
133 if type(keystore) != self.keystore_class:
134 return False
135 return True
136
137 def get_library_version(self) -> str:
138 """Returns the version of the 3rd party python library
139 for the hw wallet. For example '0.9.0'
140
141 Returns 'unknown' if library is found but cannot determine version.
142 Raises 'ImportError' if library is not found.
143 Raises 'LibraryFoundButUnusable' if found but there was some problem (includes version num).
144 """
145 raise NotImplementedError()
146
147 def check_libraries_available(self) -> bool:
148 def version_str(t):
149 return ".".join(str(i) for i in t)
150
151 try:
152 # this might raise ImportError or LibraryFoundButUnusable
153 library_version = self.get_library_version()
154 # if no exception so far, we might still raise LibraryFoundButUnusable
155 if (library_version == 'unknown'
156 or versiontuple(library_version) < self.minimum_library
157 or versiontuple(library_version) >= self.maximum_library):
158 raise LibraryFoundButUnusable(library_version=library_version)
159 except ImportError:
160 return False
161 except LibraryFoundButUnusable as e:
162 library_version = e.library_version
163 self.libraries_available_message = (
164 _("Library version for '{}' is incompatible.").format(self.name)
165 + '\nInstalled: {}, Needed: {} <= x < {}'
166 .format(library_version, version_str(self.minimum_library), version_str(self.maximum_library)))
167 self.logger.warning(self.libraries_available_message)
168 return False
169
170 return True
171
172 def get_library_not_available_message(self) -> str:
173 if hasattr(self, 'libraries_available_message'):
174 message = self.libraries_available_message
175 else:
176 message = _("Missing libraries for {}.").format(self.name)
177 message += '\n' + _("Make sure you install it with python3")
178 return message
179
180 def set_ignore_outdated_fw(self):
181 self._ignore_outdated_fw = True
182
183 def is_outdated_fw_ignored(self) -> bool:
184 return self._ignore_outdated_fw
185
186 def create_client(self, device: 'Device',
187 handler: Optional['HardwareHandlerBase']) -> Optional['HardwareClientBase']:
188 raise NotImplementedError()
189
190 def get_xpub(self, device_id: str, derivation: str, xtype, wizard: 'BaseWizard') -> str:
191 raise NotImplementedError()
192
193 def create_handler(self, window) -> 'HardwareHandlerBase':
194 # note: in Qt GUI, 'window' is either an ElectrumWindow or an InstallWizard
195 raise NotImplementedError()
196
197 def can_recognize_device(self, device: Device) -> bool:
198 """Whether the plugin thinks it can handle the given device.
199 Used for filtering all connected hardware devices to only those by this vendor.
200 """
201 return device.product_key in self.DEVICE_IDS
202
203
204 class HardwareClientBase:
205
206 handler = None # type: Optional['HardwareHandlerBase']
207
208 def __init__(self, *, plugin: 'HW_PluginBase'):
209 assert_runs_in_hwd_thread()
210 self.plugin = plugin
211
212 def device_manager(self) -> 'DeviceMgr':
213 return self.plugin.device_manager()
214
215 def is_pairable(self) -> bool:
216 raise NotImplementedError()
217
218 def close(self):
219 raise NotImplementedError()
220
221 def timeout(self, cutoff) -> None:
222 pass
223
224 def is_initialized(self) -> bool:
225 """True if initialized, False if wiped."""
226 raise NotImplementedError()
227
228 def label(self) -> Optional[str]:
229 """The name given by the user to the device.
230
231 Note: labels are shown to the user to help distinguish their devices,
232 and they are also used as a fallback to distinguish devices programmatically.
233 So ideally, different devices would have different labels.
234 """
235 # When returning a constant here (i.e. not implementing the method in the way
236 # it is supposed to work), make sure the return value is in electrum.plugin.PLACEHOLDER_HW_CLIENT_LABELS
237 return " "
238
239 def get_soft_device_id(self) -> Optional[str]:
240 """An id-like string that is used to distinguish devices programmatically.
241 This is a long term id for the device, that does not change between reconnects.
242 This method should not prompt the user, i.e. no user interaction, as it is used
243 during USB device enumeration (called for each unpaired device).
244 Stored in the wallet file.
245 """
246 # This functionality is optional. If not implemented just return None:
247 return None
248
249 def has_usable_connection_with_device(self) -> bool:
250 raise NotImplementedError()
251
252 def get_xpub(self, bip32_path: str, xtype) -> str:
253 raise NotImplementedError()
254
255 @runs_in_hwd_thread
256 def request_root_fingerprint_from_device(self) -> str:
257 # digitalbitbox (at least) does not reveal xpubs corresponding to unhardened paths
258 # so ask for a direct child, and read out fingerprint from that:
259 child_of_root_xpub = self.get_xpub("m/0'", xtype='standard')
260 root_fingerprint = BIP32Node.from_xkey(child_of_root_xpub).fingerprint.hex().lower()
261 return root_fingerprint
262
263 @runs_in_hwd_thread
264 def get_password_for_storage_encryption(self) -> str:
265 # note: using a different password based on hw device type is highly undesirable! see #5993
266 derivation = get_derivation_used_for_hw_device_encryption()
267 xpub = self.get_xpub(derivation, "standard")
268 password = Xpub.get_pubkey_from_xpub(xpub, ()).hex()
269 return password
270
271 def device_model_name(self) -> Optional[str]:
272 """Return the name of the model of this device, which might be displayed in the UI.
273 E.g. for Trezor, "Trezor One" or "Trezor T".
274 """
275 return None
276
277 def manipulate_keystore_dict_during_wizard_setup(self, d: dict) -> None:
278 """Called during wallet creation in the wizard, before the keystore
279 is constructed for the first time. 'd' is the dict that will be
280 passed to the keystore constructor.
281 """
282 pass
283
284
285 class HardwareHandlerBase:
286 """An interface between the GUI and the device handling logic for handling I/O."""
287 win = None
288 device: str
289
290 def get_wallet(self) -> Optional['Abstract_Wallet']:
291 if self.win is not None:
292 if hasattr(self.win, 'wallet'):
293 return self.win.wallet
294
295 def get_gui_thread(self) -> Optional['threading.Thread']:
296 if self.win is not None:
297 if hasattr(self.win, 'gui_thread'):
298 return self.win.gui_thread
299
300 def update_status(self, paired: bool) -> None:
301 pass
302
303 def query_choice(self, msg: str, labels: Sequence[str]) -> Optional[int]:
304 raise NotImplementedError()
305
306 def yes_no_question(self, msg: str) -> bool:
307 raise NotImplementedError()
308
309 def show_message(self, msg: str, on_cancel=None) -> None:
310 raise NotImplementedError()
311
312 def show_error(self, msg: str, blocking: bool = False) -> None:
313 raise NotImplementedError()
314
315 def finished(self) -> None:
316 pass
317
318 def get_word(self, msg: str) -> str:
319 raise NotImplementedError()
320
321 def get_passphrase(self, msg: str, confirm: bool) -> Optional[str]:
322 raise NotImplementedError()
323
324 def get_pin(self, msg: str, *, show_strength: bool = True) -> str:
325 raise NotImplementedError()
326
327
328 def is_any_tx_output_on_change_branch(tx: PartialTransaction) -> bool:
329 return any([txout.is_change for txout in tx.outputs()])
330
331
332 def trezor_validate_op_return_output_and_get_data(output: TxOutput) -> bytes:
333 validate_op_return_output(output)
334 script = output.scriptpubkey
335 if not (script[0] == opcodes.OP_RETURN and
336 script[1] == len(script) - 2 and script[1] <= 75):
337 raise UserFacingException(_("Only OP_RETURN scripts, with one constant push, are supported."))
338 return script[2:]
339
340
341 def validate_op_return_output(output: TxOutput, *, max_size: int = None) -> None:
342 script = output.scriptpubkey
343 if script[0] != opcodes.OP_RETURN:
344 raise UserFacingException(_("Only OP_RETURN scripts are supported."))
345 if max_size is not None and len(script) > max_size:
346 raise UserFacingException(_("OP_RETURN payload too large." + "\n"
347 + f"(scriptpubkey size {len(script)} > {max_size})"))
348 if output.value != 0:
349 raise UserFacingException(_("Amount for OP_RETURN output must be zero."))
350
351
352 def get_xpubs_and_der_suffixes_from_txinout(tx: PartialTransaction,
353 txinout: Union[PartialTxInput, PartialTxOutput]) \
354 -> List[Tuple[str, List[int]]]:
355 xfp_to_xpub_map = {xfp: bip32node for bip32node, (xfp, path)
356 in tx.xpubs.items()} # type: Dict[bytes, BIP32Node]
357 xfps = [txinout.bip32_paths[pubkey][0] for pubkey in txinout.pubkeys]
358 try:
359 xpubs = [xfp_to_xpub_map[xfp] for xfp in xfps]
360 except KeyError as e:
361 raise Exception(f"Partial transaction is missing global xpub for "
362 f"fingerprint ({str(e)}) in input/output") from e
363 xpubs_and_deriv_suffixes = []
364 for bip32node, pubkey in zip(xpubs, txinout.pubkeys):
365 xfp, path = txinout.bip32_paths[pubkey]
366 der_suffix = list(path)[bip32node.depth:]
367 xpubs_and_deriv_suffixes.append((bip32node.to_xpub(), der_suffix))
368 return xpubs_and_deriv_suffixes
369
370
371 def only_hook_if_libraries_available(func):
372 # note: this decorator must wrap @hook, not the other way around,
373 # as 'hook' uses the name of the function it wraps
374 def wrapper(self: 'HW_PluginBase', *args, **kwargs):
375 if not self.libraries_available: return None
376 return func(self, *args, **kwargs)
377 return wrapper
378
379
380 class LibraryFoundButUnusable(Exception):
381 def __init__(self, library_version='unknown'):
382 self.library_version = library_version
383
384
385 class OutdatedHwFirmwareException(UserFacingException):
386
387 def text_ignore_old_fw_and_continue(self) -> str:
388 suffix = (_("The firmware of your hardware device is too old. "
389 "If possible, you should upgrade it. "
390 "You can ignore this error and try to continue, however things are likely to break.") + "\n\n" +
391 _("Ignore and continue?"))
392 if str(self):
393 return str(self) + "\n\n" + suffix
394 else:
395 return suffix