tclientbase.py - electrum - Electrum Bitcoin wallet
HTML git clone https://git.parazyd.org/electrum
DIR Log
DIR Files
DIR Refs
DIR Submodules
---
tclientbase.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)))