clientbase.py (11974B)
1 import time 2 from struct import pack 3 4 from electrum import ecc 5 from electrum.i18n import _ 6 from electrum.util import UserCancelled, UserFacingException 7 from electrum.keystore import bip39_normalize_passphrase 8 from electrum.bip32 import BIP32Node, convert_bip32_path_to_list_of_uint32 as parse_path 9 from electrum.logging import Logger 10 from electrum.plugin import runs_in_hwd_thread 11 from electrum.plugins.hw_wallet.plugin import OutdatedHwFirmwareException, HardwareClientBase 12 13 from trezorlib.client import TrezorClient, PASSPHRASE_ON_DEVICE 14 from trezorlib.exceptions import TrezorFailure, Cancelled, OutdatedFirmwareError 15 from trezorlib.messages import WordRequestType, FailureType, RecoveryDeviceType, ButtonRequestType 16 import trezorlib.btc 17 import trezorlib.device 18 19 MESSAGES = { 20 ButtonRequestType.ConfirmOutput: 21 _("Confirm the transaction output on your {} device"), 22 ButtonRequestType.ResetDevice: 23 _("Complete the initialization process on your {} device"), 24 ButtonRequestType.ConfirmWord: 25 _("Write down the seed word shown on your {}"), 26 ButtonRequestType.WipeDevice: 27 _("Confirm on your {} that you want to wipe it clean"), 28 ButtonRequestType.ProtectCall: 29 _("Confirm on your {} device the message to sign"), 30 ButtonRequestType.SignTx: 31 _("Confirm the total amount spent and the transaction fee on your {} device"), 32 ButtonRequestType.Address: 33 _("Confirm wallet address on your {} device"), 34 ButtonRequestType._Deprecated_ButtonRequest_PassphraseType: 35 _("Choose on your {} device where to enter your passphrase"), 36 ButtonRequestType.PassphraseEntry: 37 _("Please enter your passphrase on the {} device"), 38 'default': _("Check your {} device to continue"), 39 } 40 41 42 class TrezorClientBase(HardwareClientBase, Logger): 43 def __init__(self, transport, handler, plugin): 44 HardwareClientBase.__init__(self, plugin=plugin) 45 if plugin.is_outdated_fw_ignored(): 46 TrezorClient.is_outdated = lambda *args, **kwargs: False 47 self.client = TrezorClient(transport, ui=self) 48 self.device = plugin.device 49 self.handler = handler 50 Logger.__init__(self) 51 52 self.msg = None 53 self.creating_wallet = False 54 55 self.in_flow = False 56 57 self.used() 58 59 def run_flow(self, message=None, creating_wallet=False): 60 if self.in_flow: 61 raise RuntimeError("Overlapping call to run_flow") 62 63 self.in_flow = True 64 self.msg = message 65 self.creating_wallet = creating_wallet 66 self.prevent_timeouts() 67 return self 68 69 def end_flow(self): 70 self.in_flow = False 71 self.msg = None 72 self.creating_wallet = False 73 self.handler.finished() 74 self.used() 75 76 def __enter__(self): 77 return self 78 79 def __exit__(self, exc_type, e, traceback): 80 self.end_flow() 81 if e is not None: 82 if isinstance(e, Cancelled): 83 raise UserCancelled() from e 84 elif isinstance(e, TrezorFailure): 85 raise RuntimeError(str(e)) from e 86 elif isinstance(e, OutdatedFirmwareError): 87 raise OutdatedHwFirmwareException(e) from e 88 else: 89 return False 90 return True 91 92 @property 93 def features(self): 94 return self.client.features 95 96 def __str__(self): 97 return "%s/%s" % (self.label(), self.features.device_id) 98 99 def label(self): 100 return self.features.label 101 102 def get_soft_device_id(self): 103 return self.features.device_id 104 105 def is_initialized(self): 106 return self.features.initialized 107 108 def is_pairable(self): 109 return not self.features.bootloader_mode 110 111 @runs_in_hwd_thread 112 def has_usable_connection_with_device(self): 113 if self.in_flow: 114 return True 115 116 try: 117 self.client.init_device() 118 except BaseException: 119 return False 120 return True 121 122 def used(self): 123 self.last_operation = time.time() 124 125 def prevent_timeouts(self): 126 self.last_operation = float('inf') 127 128 @runs_in_hwd_thread 129 def timeout(self, cutoff): 130 '''Time out the client if the last operation was before cutoff.''' 131 if self.last_operation < cutoff: 132 self.logger.info("timed out") 133 self.clear_session() 134 135 def i4b(self, x): 136 return pack('>I', x) 137 138 @runs_in_hwd_thread 139 def get_xpub(self, bip32_path, xtype, creating=False): 140 address_n = parse_path(bip32_path) 141 with self.run_flow(creating_wallet=creating): 142 node = trezorlib.btc.get_public_node(self.client, address_n).node 143 return BIP32Node(xtype=xtype, 144 eckey=ecc.ECPubkey(node.public_key), 145 chaincode=node.chain_code, 146 depth=node.depth, 147 fingerprint=self.i4b(node.fingerprint), 148 child_number=self.i4b(node.child_num)).to_xpub() 149 150 @runs_in_hwd_thread 151 def toggle_passphrase(self): 152 if self.features.passphrase_protection: 153 msg = _("Confirm on your {} device to disable passphrases") 154 else: 155 msg = _("Confirm on your {} device to enable passphrases") 156 enabled = not self.features.passphrase_protection 157 with self.run_flow(msg): 158 trezorlib.device.apply_settings(self.client, use_passphrase=enabled) 159 160 @runs_in_hwd_thread 161 def change_label(self, label): 162 with self.run_flow(_("Confirm the new label on your {} device")): 163 trezorlib.device.apply_settings(self.client, label=label) 164 165 @runs_in_hwd_thread 166 def change_homescreen(self, homescreen): 167 with self.run_flow(_("Confirm on your {} device to change your home screen")): 168 trezorlib.device.apply_settings(self.client, homescreen=homescreen) 169 170 @runs_in_hwd_thread 171 def set_pin(self, remove): 172 if remove: 173 msg = _("Confirm on your {} device to disable PIN protection") 174 elif self.features.pin_protection: 175 msg = _("Confirm on your {} device to change your PIN") 176 else: 177 msg = _("Confirm on your {} device to set a PIN") 178 with self.run_flow(msg): 179 trezorlib.device.change_pin(self.client, remove) 180 181 @runs_in_hwd_thread 182 def clear_session(self): 183 '''Clear the session to force pin (and passphrase if enabled) 184 re-entry. Does not leak exceptions.''' 185 self.logger.info(f"clear session: {self}") 186 self.prevent_timeouts() 187 try: 188 self.client.clear_session() 189 except BaseException as e: 190 # If the device was removed it has the same effect... 191 self.logger.info(f"clear_session: ignoring error {e}") 192 193 @runs_in_hwd_thread 194 def close(self): 195 '''Called when Our wallet was closed or the device removed.''' 196 self.logger.info("closing client") 197 self.clear_session() 198 199 @runs_in_hwd_thread 200 def is_uptodate(self): 201 if self.client.is_outdated(): 202 return False 203 return self.client.version >= self.plugin.minimum_firmware 204 205 def get_trezor_model(self): 206 """Returns '1' for Trezor One, 'T' for Trezor T.""" 207 return self.features.model 208 209 def device_model_name(self): 210 model = self.get_trezor_model() 211 if model == '1': 212 return "Trezor One" 213 elif model == 'T': 214 return "Trezor T" 215 return None 216 217 @runs_in_hwd_thread 218 def show_address(self, address_str, script_type, multisig=None): 219 coin_name = self.plugin.get_coin_name() 220 address_n = parse_path(address_str) 221 with self.run_flow(): 222 return trezorlib.btc.get_address( 223 self.client, 224 coin_name, 225 address_n, 226 show_display=True, 227 script_type=script_type, 228 multisig=multisig) 229 230 @runs_in_hwd_thread 231 def sign_message(self, address_str, message): 232 coin_name = self.plugin.get_coin_name() 233 address_n = parse_path(address_str) 234 with self.run_flow(): 235 return trezorlib.btc.sign_message( 236 self.client, 237 coin_name, 238 address_n, 239 message) 240 241 @runs_in_hwd_thread 242 def recover_device(self, recovery_type, *args, **kwargs): 243 input_callback = self.mnemonic_callback(recovery_type) 244 with self.run_flow(): 245 return trezorlib.device.recover( 246 self.client, 247 *args, 248 input_callback=input_callback, 249 type=recovery_type, 250 **kwargs) 251 252 # ========= Unmodified trezorlib methods ========= 253 254 @runs_in_hwd_thread 255 def sign_tx(self, *args, **kwargs): 256 with self.run_flow(): 257 return trezorlib.btc.sign_tx(self.client, *args, **kwargs) 258 259 @runs_in_hwd_thread 260 def reset_device(self, *args, **kwargs): 261 with self.run_flow(): 262 return trezorlib.device.reset(self.client, *args, **kwargs) 263 264 @runs_in_hwd_thread 265 def wipe_device(self, *args, **kwargs): 266 with self.run_flow(): 267 return trezorlib.device.wipe(self.client, *args, **kwargs) 268 269 # ========= UI methods ========== 270 271 def button_request(self, code): 272 message = self.msg or MESSAGES.get(code) or MESSAGES['default'] 273 self.handler.show_message(message.format(self.device), self.client.cancel) 274 275 def get_pin(self, code=None): 276 show_strength = True 277 if code == 2: 278 msg = _("Enter a new PIN for your {}:") 279 elif code == 3: 280 msg = (_("Re-enter the new PIN for your {}.\n\n" 281 "NOTE: the positions of the numbers have changed!")) 282 else: 283 msg = _("Enter your current {} PIN:") 284 show_strength = False 285 pin = self.handler.get_pin(msg.format(self.device), show_strength=show_strength) 286 if not pin: 287 raise Cancelled 288 if len(pin) > 9: 289 self.handler.show_error(_('The PIN cannot be longer than 9 characters.')) 290 raise Cancelled 291 return pin 292 293 def get_passphrase(self, available_on_device): 294 if self.creating_wallet: 295 msg = _("Enter a passphrase to generate this wallet. Each time " 296 "you use this wallet your {} will prompt you for the " 297 "passphrase. If you forget the passphrase you cannot " 298 "access the bitcoins in the wallet.").format(self.device) 299 else: 300 msg = _("Enter the passphrase to unlock this wallet:") 301 302 self.handler.passphrase_on_device = available_on_device 303 passphrase = self.handler.get_passphrase(msg, self.creating_wallet) 304 if passphrase is PASSPHRASE_ON_DEVICE: 305 return passphrase 306 if passphrase is None: 307 raise Cancelled 308 passphrase = bip39_normalize_passphrase(passphrase) 309 length = len(passphrase) 310 if length > 50: 311 self.handler.show_error(_("Too long passphrase ({} > 50 chars).").format(length)) 312 raise Cancelled 313 return passphrase 314 315 def _matrix_char(self, matrix_type): 316 num = 9 if matrix_type == WordRequestType.Matrix9 else 6 317 char = self.handler.get_matrix(num) 318 if char == 'x': 319 raise Cancelled 320 return char 321 322 def mnemonic_callback(self, recovery_type): 323 if recovery_type is None: 324 return None 325 326 if recovery_type == RecoveryDeviceType.Matrix: 327 return self._matrix_char 328 329 step = 0 330 def word_callback(_ignored): 331 nonlocal step 332 step += 1 333 msg = _("Step {}/24. Enter seed word as explained on your {}:").format(step, self.device) 334 word = self.handler.get_word(msg) 335 if not word: 336 raise Cancelled 337 return word 338 return word_callback