clientbase.py (9904B)
1 import time 2 from struct import pack 3 from typing import Optional 4 5 from electrum import ecc 6 from electrum.i18n import _ 7 from electrum.util import UserCancelled 8 from electrum.keystore import bip39_normalize_passphrase 9 from electrum.bip32 import BIP32Node, convert_bip32_path_to_list_of_uint32 10 from electrum.logging import Logger 11 from electrum.plugin import runs_in_hwd_thread 12 from electrum.plugins.hw_wallet.plugin import HardwareClientBase, HardwareHandlerBase 13 14 15 class GuiMixin(object): 16 # Requires: self.proto, self.device 17 handler: Optional[HardwareHandlerBase] 18 19 messages = { 20 3: _("Confirm the transaction output on your {} device"), 21 4: _("Confirm internal entropy on your {} device to begin"), 22 5: _("Write down the seed word shown on your {}"), 23 6: _("Confirm on your {} that you want to wipe it clean"), 24 7: _("Confirm on your {} device the message to sign"), 25 8: _("Confirm the total amount spent and the transaction fee on your " 26 "{} device"), 27 10: _("Confirm wallet address on your {} device"), 28 'default': _("Check your {} device to continue"), 29 } 30 31 def callback_Failure(self, msg): 32 # BaseClient's unfortunate call() implementation forces us to 33 # raise exceptions on failure in order to unwind the stack. 34 # However, making the user acknowledge they cancelled 35 # gets old very quickly, so we suppress those. The NotInitialized 36 # one is misnamed and indicates a passphrase request was cancelled. 37 if msg.code in (self.types.Failure_PinCancelled, 38 self.types.Failure_ActionCancelled, 39 self.types.Failure_NotInitialized): 40 raise UserCancelled() 41 raise RuntimeError(msg.message) 42 43 def callback_ButtonRequest(self, msg): 44 message = self.msg 45 if not message: 46 message = self.messages.get(msg.code, self.messages['default']) 47 self.handler.show_message(message.format(self.device), self.cancel) 48 return self.proto.ButtonAck() 49 50 def callback_PinMatrixRequest(self, msg): 51 show_strength = True 52 if msg.type == 2: 53 msg = _("Enter a new PIN for your {}:") 54 elif msg.type == 3: 55 msg = (_("Re-enter the new PIN for your {}.\n\n" 56 "NOTE: the positions of the numbers have changed!")) 57 else: 58 msg = _("Enter your current {} PIN:") 59 show_strength = False 60 pin = self.handler.get_pin(msg.format(self.device), show_strength=show_strength) 61 if len(pin) > 9: 62 self.handler.show_error(_('The PIN cannot be longer than 9 characters.')) 63 pin = '' # to cancel below 64 if not pin: 65 return self.proto.Cancel() 66 return self.proto.PinMatrixAck(pin=pin) 67 68 def callback_PassphraseRequest(self, req): 69 if self.creating_wallet: 70 msg = _("Enter a passphrase to generate this wallet. Each time " 71 "you use this wallet your {} will prompt you for the " 72 "passphrase. If you forget the passphrase you cannot " 73 "access the bitcoins in the wallet.").format(self.device) 74 else: 75 msg = _("Enter the passphrase to unlock this wallet:") 76 passphrase = self.handler.get_passphrase(msg, self.creating_wallet) 77 if passphrase is None: 78 return self.proto.Cancel() 79 passphrase = bip39_normalize_passphrase(passphrase) 80 81 ack = self.proto.PassphraseAck(passphrase=passphrase) 82 length = len(ack.passphrase) 83 if length > 50: 84 self.handler.show_error(_("Too long passphrase ({} > 50 chars).").format(length)) 85 return self.proto.Cancel() 86 return ack 87 88 def callback_WordRequest(self, msg): 89 self.step += 1 90 msg = _("Step {}/24. Enter seed word as explained on " 91 "your {}:").format(self.step, self.device) 92 word = self.handler.get_word(msg) 93 # Unfortunately the device can't handle self.proto.Cancel() 94 return self.proto.WordAck(word=word) 95 96 def callback_CharacterRequest(self, msg): 97 char_info = self.handler.get_char(msg) 98 if not char_info: 99 return self.proto.Cancel() 100 return self.proto.CharacterAck(**char_info) 101 102 103 class KeepKeyClientBase(HardwareClientBase, GuiMixin, Logger): 104 105 def __init__(self, handler, plugin, proto): 106 assert hasattr(self, 'tx_api') # ProtocolMixin already constructed? 107 HardwareClientBase.__init__(self, plugin=plugin) 108 self.proto = proto 109 self.device = plugin.device 110 self.handler = handler 111 self.tx_api = plugin 112 self.types = plugin.types 113 self.msg = None 114 self.creating_wallet = False 115 Logger.__init__(self) 116 self.used() 117 118 def __str__(self): 119 return "%s/%s" % (self.label(), self.features.device_id) 120 121 def label(self): 122 return self.features.label 123 124 def get_soft_device_id(self): 125 return self.features.device_id 126 127 def is_initialized(self): 128 return self.features.initialized 129 130 def is_pairable(self): 131 return not self.features.bootloader_mode 132 133 @runs_in_hwd_thread 134 def has_usable_connection_with_device(self): 135 try: 136 res = self.ping("electrum pinging device") 137 assert res == "electrum pinging device" 138 except BaseException: 139 return False 140 return True 141 142 def used(self): 143 self.last_operation = time.time() 144 145 def prevent_timeouts(self): 146 self.last_operation = float('inf') 147 148 @runs_in_hwd_thread 149 def timeout(self, cutoff): 150 '''Time out the client if the last operation was before cutoff.''' 151 if self.last_operation < cutoff: 152 self.logger.info("timed out") 153 self.clear_session() 154 155 @staticmethod 156 def expand_path(n): 157 return convert_bip32_path_to_list_of_uint32(n) 158 159 @runs_in_hwd_thread 160 def cancel(self): 161 '''Provided here as in keepkeylib but not trezorlib.''' 162 self.transport.write(self.proto.Cancel()) 163 164 def i4b(self, x): 165 return pack('>I', x) 166 167 @runs_in_hwd_thread 168 def get_xpub(self, bip32_path, xtype): 169 address_n = self.expand_path(bip32_path) 170 creating = False 171 node = self.get_public_node(address_n, creating).node 172 return BIP32Node(xtype=xtype, 173 eckey=ecc.ECPubkey(node.public_key), 174 chaincode=node.chain_code, 175 depth=node.depth, 176 fingerprint=self.i4b(node.fingerprint), 177 child_number=self.i4b(node.child_num)).to_xpub() 178 179 @runs_in_hwd_thread 180 def toggle_passphrase(self): 181 if self.features.passphrase_protection: 182 self.msg = _("Confirm on your {} device to disable passphrases") 183 else: 184 self.msg = _("Confirm on your {} device to enable passphrases") 185 enabled = not self.features.passphrase_protection 186 self.apply_settings(use_passphrase=enabled) 187 188 @runs_in_hwd_thread 189 def change_label(self, label): 190 self.msg = _("Confirm the new label on your {} device") 191 self.apply_settings(label=label) 192 193 @runs_in_hwd_thread 194 def change_homescreen(self, homescreen): 195 self.msg = _("Confirm on your {} device to change your home screen") 196 self.apply_settings(homescreen=homescreen) 197 198 @runs_in_hwd_thread 199 def set_pin(self, remove): 200 if remove: 201 self.msg = _("Confirm on your {} device to disable PIN protection") 202 elif self.features.pin_protection: 203 self.msg = _("Confirm on your {} device to change your PIN") 204 else: 205 self.msg = _("Confirm on your {} device to set a PIN") 206 self.change_pin(remove) 207 208 @runs_in_hwd_thread 209 def clear_session(self): 210 '''Clear the session to force pin (and passphrase if enabled) 211 re-entry. Does not leak exceptions.''' 212 self.logger.info(f"clear session: {self}") 213 self.prevent_timeouts() 214 try: 215 super(KeepKeyClientBase, self).clear_session() 216 except BaseException as e: 217 # If the device was removed it has the same effect... 218 self.logger.info(f"clear_session: ignoring error {e}") 219 220 @runs_in_hwd_thread 221 def get_public_node(self, address_n, creating): 222 self.creating_wallet = creating 223 return super(KeepKeyClientBase, self).get_public_node(address_n) 224 225 @runs_in_hwd_thread 226 def close(self): 227 '''Called when Our wallet was closed or the device removed.''' 228 self.logger.info("closing client") 229 self.clear_session() 230 # Release the device 231 self.transport.close() 232 233 def firmware_version(self): 234 f = self.features 235 return (f.major_version, f.minor_version, f.patch_version) 236 237 def atleast_version(self, major, minor=0, patch=0): 238 return self.firmware_version() >= (major, minor, patch) 239 240 @staticmethod 241 def wrapper(func): 242 '''Wrap methods to clear any message box they opened.''' 243 244 def wrapped(self, *args, **kwargs): 245 try: 246 self.prevent_timeouts() 247 return func(self, *args, **kwargs) 248 finally: 249 self.used() 250 self.handler.finished() 251 self.creating_wallet = False 252 self.msg = None 253 254 return wrapped 255 256 @staticmethod 257 def wrap_methods(cls): 258 for method in ['apply_settings', 'change_pin', 259 'get_address', 'get_public_node', 260 'load_device_by_mnemonic', 'load_device_by_xprv', 261 'recovery_device', 'reset_device', 'sign_message', 262 'sign_tx', 'wipe_device']: 263 setattr(cls, method, cls.wrapper(getattr(cls, method)))