tiagotrs
6 years ago
53 changed files with 1881 additions and 309 deletions
@ -0,0 +1,8 @@ |
|||
from electrum.i18n import _ |
|||
|
|||
fullname = 'Safe-T mini Wallet' |
|||
description = _('Provides support for Safe-T mini hardware wallet') |
|||
requires = [('safetlib','github.com/archos-safe-t/python-safet')] |
|||
registers_keystore = ('hardware', 'safe_t', _("Safe-T mini wallet")) |
|||
available_for = ['qt', 'cmdline'] |
|||
|
|||
@ -0,0 +1,11 @@ |
|||
from safetlib.client import proto, BaseClient, ProtocolMixin |
|||
from .clientbase import SafeTClientBase |
|||
|
|||
class SafeTClient(SafeTClientBase, ProtocolMixin, BaseClient): |
|||
def __init__(self, transport, handler, plugin): |
|||
BaseClient.__init__(self, transport=transport) |
|||
ProtocolMixin.__init__(self, transport=transport) |
|||
SafeTClientBase.__init__(self, handler, plugin, proto) |
|||
|
|||
|
|||
SafeTClientBase.wrap_methods(SafeTClient) |
|||
@ -0,0 +1,252 @@ |
|||
import time |
|||
from struct import pack |
|||
|
|||
from electrum.i18n import _ |
|||
from electrum.util import PrintError, UserCancelled |
|||
from electrum.keystore import bip39_normalize_passphrase |
|||
from electrum.bitcoin import serialize_xpub |
|||
|
|||
|
|||
class GuiMixin(object): |
|||
# Requires: self.proto, self.device |
|||
|
|||
# ref: https://github.com/trezor/trezor-common/blob/44dfb07cfaafffada4b2ce0d15ba1d90d17cf35e/protob/types.proto#L89 |
|||
messages = { |
|||
3: _("Confirm the transaction output on your {} device"), |
|||
4: _("Confirm internal entropy on your {} device to begin"), |
|||
5: _("Write down the seed word shown on your {}"), |
|||
6: _("Confirm on your {} that you want to wipe it clean"), |
|||
7: _("Confirm on your {} device the message to sign"), |
|||
8: _("Confirm the total amount spent and the transaction fee on your " |
|||
"{} device"), |
|||
10: _("Confirm wallet address on your {} device"), |
|||
14: _("Choose on your {} device where to enter your passphrase"), |
|||
'default': _("Check your {} device to continue"), |
|||
} |
|||
|
|||
def callback_Failure(self, msg): |
|||
# BaseClient's unfortunate call() implementation forces us to |
|||
# raise exceptions on failure in order to unwind the stack. |
|||
# However, making the user acknowledge they cancelled |
|||
# gets old very quickly, so we suppress those. The NotInitialized |
|||
# one is misnamed and indicates a passphrase request was cancelled. |
|||
if msg.code in (self.types.FailureType.PinCancelled, |
|||
self.types.FailureType.ActionCancelled, |
|||
self.types.FailureType.NotInitialized): |
|||
raise UserCancelled() |
|||
raise RuntimeError(msg.message) |
|||
|
|||
def callback_ButtonRequest(self, msg): |
|||
message = self.msg |
|||
if not message: |
|||
message = self.messages.get(msg.code, self.messages['default']) |
|||
self.handler.show_message(message.format(self.device), self.cancel) |
|||
return self.proto.ButtonAck() |
|||
|
|||
def callback_PinMatrixRequest(self, msg): |
|||
if msg.type == 2: |
|||
msg = _("Enter a new PIN for your {}:") |
|||
elif msg.type == 3: |
|||
msg = (_("Re-enter the new PIN for your {}.\n\n" |
|||
"NOTE: the positions of the numbers have changed!")) |
|||
else: |
|||
msg = _("Enter your current {} PIN:") |
|||
pin = self.handler.get_pin(msg.format(self.device)) |
|||
if len(pin) > 9: |
|||
self.handler.show_error(_('The PIN cannot be longer than 9 characters.')) |
|||
pin = '' # to cancel below |
|||
if not pin: |
|||
return self.proto.Cancel() |
|||
return self.proto.PinMatrixAck(pin=pin) |
|||
|
|||
def callback_PassphraseRequest(self, req): |
|||
if req and hasattr(req, 'on_device') and req.on_device is True: |
|||
return self.proto.PassphraseAck() |
|||
|
|||
if self.creating_wallet: |
|||
msg = _("Enter a passphrase to generate this wallet. Each time " |
|||
"you use this wallet your {} will prompt you for the " |
|||
"passphrase. If you forget the passphrase you cannot " |
|||
"access the bitcoins in the wallet.").format(self.device) |
|||
else: |
|||
msg = _("Enter the passphrase to unlock this wallet:") |
|||
passphrase = self.handler.get_passphrase(msg, self.creating_wallet) |
|||
if passphrase is None: |
|||
return self.proto.Cancel() |
|||
passphrase = bip39_normalize_passphrase(passphrase) |
|||
|
|||
ack = self.proto.PassphraseAck(passphrase=passphrase) |
|||
length = len(ack.passphrase) |
|||
if length > 50: |
|||
self.handler.show_error(_("Too long passphrase ({} > 50 chars).").format(length)) |
|||
return self.proto.Cancel() |
|||
return ack |
|||
|
|||
def callback_PassphraseStateRequest(self, msg): |
|||
return self.proto.PassphraseStateAck() |
|||
|
|||
def callback_WordRequest(self, msg): |
|||
self.step += 1 |
|||
msg = _("Step {}/24. Enter seed word as explained on " |
|||
"your {}:").format(self.step, self.device) |
|||
word = self.handler.get_word(msg) |
|||
# Unfortunately the device can't handle self.proto.Cancel() |
|||
return self.proto.WordAck(word=word) |
|||
|
|||
|
|||
class SafeTClientBase(GuiMixin, PrintError): |
|||
|
|||
def __init__(self, handler, plugin, proto): |
|||
assert hasattr(self, 'tx_api') # ProtocolMixin already constructed? |
|||
self.proto = proto |
|||
self.device = plugin.device |
|||
self.handler = handler |
|||
self.tx_api = plugin |
|||
self.types = plugin.types |
|||
self.msg = None |
|||
self.creating_wallet = False |
|||
self.used() |
|||
|
|||
def __str__(self): |
|||
return "%s/%s" % (self.label(), self.features.device_id) |
|||
|
|||
def label(self): |
|||
'''The name given by the user to the device.''' |
|||
return self.features.label |
|||
|
|||
def is_initialized(self): |
|||
'''True if initialized, False if wiped.''' |
|||
return self.features.initialized |
|||
|
|||
def is_pairable(self): |
|||
return not self.features.bootloader_mode |
|||
|
|||
def has_usable_connection_with_device(self): |
|||
try: |
|||
res = self.ping("electrum pinging device") |
|||
assert res == "electrum pinging device" |
|||
except BaseException: |
|||
return False |
|||
return True |
|||
|
|||
def used(self): |
|||
self.last_operation = time.time() |
|||
|
|||
def prevent_timeouts(self): |
|||
self.last_operation = float('inf') |
|||
|
|||
def timeout(self, cutoff): |
|||
'''Time out the client if the last operation was before cutoff.''' |
|||
if self.last_operation < cutoff: |
|||
self.print_error("timed out") |
|||
self.clear_session() |
|||
|
|||
@staticmethod |
|||
def expand_path(n): |
|||
'''Convert bip32 path to list of uint32 integers with prime flags |
|||
0/-1/1' -> [0, 0x80000001, 0x80000001]''' |
|||
# This code is similar to code in safetlib where it unfortunately |
|||
# is not declared as a staticmethod. Our n has an extra element. |
|||
PRIME_DERIVATION_FLAG = 0x80000000 |
|||
path = [] |
|||
for x in n.split('/')[1:]: |
|||
prime = 0 |
|||
if x.endswith("'"): |
|||
x = x.replace('\'', '') |
|||
prime = PRIME_DERIVATION_FLAG |
|||
if x.startswith('-'): |
|||
prime = PRIME_DERIVATION_FLAG |
|||
path.append(abs(int(x)) | prime) |
|||
return path |
|||
|
|||
def cancel(self): |
|||
'''Provided here as in keepkeylib but not safetlib.''' |
|||
self.transport.write(self.proto.Cancel()) |
|||
|
|||
def i4b(self, x): |
|||
return pack('>I', x) |
|||
|
|||
def get_xpub(self, bip32_path, xtype): |
|||
address_n = self.expand_path(bip32_path) |
|||
creating = False |
|||
node = self.get_public_node(address_n, creating).node |
|||
return serialize_xpub(xtype, node.chain_code, node.public_key, node.depth, self.i4b(node.fingerprint), self.i4b(node.child_num)) |
|||
|
|||
def toggle_passphrase(self): |
|||
if self.features.passphrase_protection: |
|||
self.msg = _("Confirm on your {} device to disable passphrases") |
|||
else: |
|||
self.msg = _("Confirm on your {} device to enable passphrases") |
|||
enabled = not self.features.passphrase_protection |
|||
self.apply_settings(use_passphrase=enabled) |
|||
|
|||
def change_label(self, label): |
|||
self.msg = _("Confirm the new label on your {} device") |
|||
self.apply_settings(label=label) |
|||
|
|||
def change_homescreen(self, homescreen): |
|||
self.msg = _("Confirm on your {} device to change your home screen") |
|||
self.apply_settings(homescreen=homescreen) |
|||
|
|||
def set_pin(self, remove): |
|||
if remove: |
|||
self.msg = _("Confirm on your {} device to disable PIN protection") |
|||
elif self.features.pin_protection: |
|||
self.msg = _("Confirm on your {} device to change your PIN") |
|||
else: |
|||
self.msg = _("Confirm on your {} device to set a PIN") |
|||
self.change_pin(remove) |
|||
|
|||
def clear_session(self): |
|||
'''Clear the session to force pin (and passphrase if enabled) |
|||
re-entry. Does not leak exceptions.''' |
|||
self.print_error("clear session:", self) |
|||
self.prevent_timeouts() |
|||
try: |
|||
super(SafeTClientBase, self).clear_session() |
|||
except BaseException as e: |
|||
# If the device was removed it has the same effect... |
|||
self.print_error("clear_session: ignoring error", str(e)) |
|||
|
|||
def get_public_node(self, address_n, creating): |
|||
self.creating_wallet = creating |
|||
return super(SafeTClientBase, self).get_public_node(address_n) |
|||
|
|||
def close(self): |
|||
'''Called when Our wallet was closed or the device removed.''' |
|||
self.print_error("closing client") |
|||
self.clear_session() |
|||
# Release the device |
|||
self.transport.close() |
|||
|
|||
def firmware_version(self): |
|||
f = self.features |
|||
return (f.major_version, f.minor_version, f.patch_version) |
|||
|
|||
def atleast_version(self, major, minor=0, patch=0): |
|||
return self.firmware_version() >= (major, minor, patch) |
|||
|
|||
@staticmethod |
|||
def wrapper(func): |
|||
'''Wrap methods to clear any message box they opened.''' |
|||
|
|||
def wrapped(self, *args, **kwargs): |
|||
try: |
|||
self.prevent_timeouts() |
|||
return func(self, *args, **kwargs) |
|||
finally: |
|||
self.used() |
|||
self.handler.finished() |
|||
self.creating_wallet = False |
|||
self.msg = None |
|||
|
|||
return wrapped |
|||
|
|||
@staticmethod |
|||
def wrap_methods(cls): |
|||
for method in ['apply_settings', 'change_pin', |
|||
'get_address', 'get_public_node', |
|||
'load_device_by_mnemonic', 'load_device_by_xprv', |
|||
'recovery_device', 'reset_device', 'sign_message', |
|||
'sign_tx', 'wipe_device']: |
|||
setattr(cls, method, cls.wrapper(getattr(cls, method))) |
|||
@ -0,0 +1,14 @@ |
|||
from electrum.plugin import hook |
|||
from .safe_t import SafeTPlugin |
|||
from ..hw_wallet import CmdLineHandler |
|||
|
|||
class Plugin(SafeTPlugin): |
|||
handler = CmdLineHandler() |
|||
@hook |
|||
def init_keystore(self, keystore): |
|||
if not isinstance(keystore, self.keystore_class): |
|||
return |
|||
keystore.handler = self.handler |
|||
|
|||
def create_handler(self, window): |
|||
return self.handler |
|||
@ -0,0 +1,492 @@ |
|||
from functools import partial |
|||
import threading |
|||
|
|||
from PyQt5.Qt import Qt |
|||
from PyQt5.Qt import QGridLayout, QInputDialog, QPushButton |
|||
from PyQt5.Qt import QVBoxLayout, QLabel |
|||
|
|||
from electrum.gui.qt.util import * |
|||
from electrum.i18n import _ |
|||
from electrum.plugin import hook, DeviceMgr |
|||
from electrum.util import PrintError, UserCancelled, bh2u |
|||
from electrum.wallet import Wallet, Standard_Wallet |
|||
|
|||
from ..hw_wallet.qt import QtHandlerBase, QtPluginBase |
|||
from .safe_t import SafeTPlugin, TIM_NEW, TIM_RECOVER, TIM_MNEMONIC |
|||
|
|||
|
|||
PASSPHRASE_HELP_SHORT =_( |
|||
"Passphrases allow you to access new wallets, each " |
|||
"hidden behind a particular case-sensitive passphrase.") |
|||
PASSPHRASE_HELP = PASSPHRASE_HELP_SHORT + " " + _( |
|||
"You need to create a separate Electrum wallet for each passphrase " |
|||
"you use as they each generate different addresses. Changing " |
|||
"your passphrase does not lose other wallets, each is still " |
|||
"accessible behind its own passphrase.") |
|||
RECOMMEND_PIN = _( |
|||
"You should enable PIN protection. Your PIN is the only protection " |
|||
"for your bitcoins if your device is lost or stolen.") |
|||
PASSPHRASE_NOT_PIN = _( |
|||
"If you forget a passphrase you will be unable to access any " |
|||
"bitcoins in the wallet behind it. A passphrase is not a PIN. " |
|||
"Only change this if you are sure you understand it.") |
|||
|
|||
|
|||
class QtHandler(QtHandlerBase): |
|||
|
|||
pin_signal = pyqtSignal(object) |
|||
|
|||
def __init__(self, win, pin_matrix_widget_class, device): |
|||
super(QtHandler, self).__init__(win, device) |
|||
self.pin_signal.connect(self.pin_dialog) |
|||
self.pin_matrix_widget_class = pin_matrix_widget_class |
|||
|
|||
def get_pin(self, msg): |
|||
self.done.clear() |
|||
self.pin_signal.emit(msg) |
|||
self.done.wait() |
|||
return self.response |
|||
|
|||
def pin_dialog(self, msg): |
|||
# Needed e.g. when resetting a device |
|||
self.clear_dialog() |
|||
dialog = WindowModalDialog(self.top_level_window(), _("Enter PIN")) |
|||
matrix = self.pin_matrix_widget_class() |
|||
vbox = QVBoxLayout() |
|||
vbox.addWidget(QLabel(msg)) |
|||
vbox.addWidget(matrix) |
|||
vbox.addLayout(Buttons(CancelButton(dialog), OkButton(dialog))) |
|||
dialog.setLayout(vbox) |
|||
dialog.exec_() |
|||
self.response = str(matrix.get_value()) |
|||
self.done.set() |
|||
|
|||
|
|||
class QtPlugin(QtPluginBase): |
|||
# Derived classes must provide the following class-static variables: |
|||
# icon_file |
|||
# pin_matrix_widget_class |
|||
|
|||
def create_handler(self, window): |
|||
return QtHandler(window, self.pin_matrix_widget_class(), self.device) |
|||
|
|||
@hook |
|||
def receive_menu(self, menu, addrs, wallet): |
|||
if len(addrs) != 1: |
|||
return |
|||
for keystore in wallet.get_keystores(): |
|||
if type(keystore) == self.keystore_class: |
|||
def show_address(): |
|||
keystore.thread.add(partial(self.show_address, wallet, addrs[0], keystore)) |
|||
menu.addAction(_("Show on {}").format(self.device), show_address) |
|||
break |
|||
|
|||
def show_settings_dialog(self, window, keystore): |
|||
device_id = self.choose_device(window, keystore) |
|||
if device_id: |
|||
SettingsDialog(window, self, keystore, device_id).exec_() |
|||
|
|||
def request_safe_t_init_settings(self, wizard, method, device): |
|||
vbox = QVBoxLayout() |
|||
next_enabled = True |
|||
label = QLabel(_("Enter a label to name your device:")) |
|||
name = QLineEdit() |
|||
hl = QHBoxLayout() |
|||
hl.addWidget(label) |
|||
hl.addWidget(name) |
|||
hl.addStretch(1) |
|||
vbox.addLayout(hl) |
|||
|
|||
def clean_text(widget): |
|||
text = widget.toPlainText().strip() |
|||
return ' '.join(text.split()) |
|||
|
|||
if method in [TIM_NEW, TIM_RECOVER]: |
|||
gb = QGroupBox() |
|||
hbox1 = QHBoxLayout() |
|||
gb.setLayout(hbox1) |
|||
vbox.addWidget(gb) |
|||
gb.setTitle(_("Select your seed length:")) |
|||
bg = QButtonGroup() |
|||
for i, count in enumerate([12, 18, 24]): |
|||
rb = QRadioButton(gb) |
|||
rb.setText(_("%d words") % count) |
|||
bg.addButton(rb) |
|||
bg.setId(rb, i) |
|||
hbox1.addWidget(rb) |
|||
rb.setChecked(True) |
|||
cb_pin = QCheckBox(_('Enable PIN protection')) |
|||
cb_pin.setChecked(True) |
|||
else: |
|||
text = QTextEdit() |
|||
text.setMaximumHeight(60) |
|||
if method == TIM_MNEMONIC: |
|||
msg = _("Enter your BIP39 mnemonic:") |
|||
else: |
|||
msg = _("Enter the master private key beginning with xprv:") |
|||
def set_enabled(): |
|||
from electrum.keystore import is_xprv |
|||
wizard.next_button.setEnabled(is_xprv(clean_text(text))) |
|||
text.textChanged.connect(set_enabled) |
|||
next_enabled = False |
|||
|
|||
vbox.addWidget(QLabel(msg)) |
|||
vbox.addWidget(text) |
|||
pin = QLineEdit() |
|||
pin.setValidator(QRegExpValidator(QRegExp('[1-9]{0,9}'))) |
|||
pin.setMaximumWidth(100) |
|||
hbox_pin = QHBoxLayout() |
|||
hbox_pin.addWidget(QLabel(_("Enter your PIN (digits 1-9):"))) |
|||
hbox_pin.addWidget(pin) |
|||
hbox_pin.addStretch(1) |
|||
|
|||
if method in [TIM_NEW, TIM_RECOVER]: |
|||
vbox.addWidget(WWLabel(RECOMMEND_PIN)) |
|||
vbox.addWidget(cb_pin) |
|||
else: |
|||
vbox.addLayout(hbox_pin) |
|||
|
|||
passphrase_msg = WWLabel(PASSPHRASE_HELP_SHORT) |
|||
passphrase_warning = WWLabel(PASSPHRASE_NOT_PIN) |
|||
passphrase_warning.setStyleSheet("color: red") |
|||
cb_phrase = QCheckBox(_('Enable passphrases')) |
|||
cb_phrase.setChecked(False) |
|||
vbox.addWidget(passphrase_msg) |
|||
vbox.addWidget(passphrase_warning) |
|||
vbox.addWidget(cb_phrase) |
|||
|
|||
wizard.exec_layout(vbox, next_enabled=next_enabled) |
|||
|
|||
if method in [TIM_NEW, TIM_RECOVER]: |
|||
item = bg.checkedId() |
|||
pin = cb_pin.isChecked() |
|||
else: |
|||
item = ' '.join(str(clean_text(text)).split()) |
|||
pin = str(pin.text()) |
|||
|
|||
return (item, name.text(), pin, cb_phrase.isChecked()) |
|||
|
|||
|
|||
class Plugin(SafeTPlugin, QtPlugin): |
|||
icon_unpaired = ":icons/safe-t_unpaired.png" |
|||
icon_paired = ":icons/safe-t.png" |
|||
|
|||
@classmethod |
|||
def pin_matrix_widget_class(self): |
|||
from safetlib.qt.pinmatrix import PinMatrixWidget |
|||
return PinMatrixWidget |
|||
|
|||
|
|||
class SettingsDialog(WindowModalDialog): |
|||
'''This dialog doesn't require a device be paired with a wallet. |
|||
We want users to be able to wipe a device even if they've forgotten |
|||
their PIN.''' |
|||
|
|||
def __init__(self, window, plugin, keystore, device_id): |
|||
title = _("{} Settings").format(plugin.device) |
|||
super(SettingsDialog, self).__init__(window, title) |
|||
self.setMaximumWidth(540) |
|||
|
|||
devmgr = plugin.device_manager() |
|||
config = devmgr.config |
|||
handler = keystore.handler |
|||
thread = keystore.thread |
|||
hs_rows, hs_cols = (64, 128) |
|||
|
|||
def invoke_client(method, *args, **kw_args): |
|||
unpair_after = kw_args.pop('unpair_after', False) |
|||
|
|||
def task(): |
|||
client = devmgr.client_by_id(device_id) |
|||
if not client: |
|||
raise RuntimeError("Device not connected") |
|||
if method: |
|||
getattr(client, method)(*args, **kw_args) |
|||
if unpair_after: |
|||
devmgr.unpair_id(device_id) |
|||
return client.features |
|||
|
|||
thread.add(task, on_success=update) |
|||
|
|||
def update(features): |
|||
self.features = features |
|||
set_label_enabled() |
|||
if features.bootloader_hash: |
|||
bl_hash = bh2u(features.bootloader_hash) |
|||
bl_hash = "\n".join([bl_hash[:32], bl_hash[32:]]) |
|||
else: |
|||
bl_hash = "N/A" |
|||
noyes = [_("No"), _("Yes")] |
|||
endis = [_("Enable Passphrases"), _("Disable Passphrases")] |
|||
disen = [_("Disabled"), _("Enabled")] |
|||
setchange = [_("Set a PIN"), _("Change PIN")] |
|||
|
|||
version = "%d.%d.%d" % (features.major_version, |
|||
features.minor_version, |
|||
features.patch_version) |
|||
|
|||
device_label.setText(features.label) |
|||
pin_set_label.setText(noyes[features.pin_protection]) |
|||
passphrases_label.setText(disen[features.passphrase_protection]) |
|||
bl_hash_label.setText(bl_hash) |
|||
label_edit.setText(features.label) |
|||
device_id_label.setText(features.device_id) |
|||
initialized_label.setText(noyes[features.initialized]) |
|||
version_label.setText(version) |
|||
clear_pin_button.setVisible(features.pin_protection) |
|||
clear_pin_warning.setVisible(features.pin_protection) |
|||
pin_button.setText(setchange[features.pin_protection]) |
|||
pin_msg.setVisible(not features.pin_protection) |
|||
passphrase_button.setText(endis[features.passphrase_protection]) |
|||
language_label.setText(features.language) |
|||
|
|||
def set_label_enabled(): |
|||
label_apply.setEnabled(label_edit.text() != self.features.label) |
|||
|
|||
def rename(): |
|||
invoke_client('change_label', label_edit.text()) |
|||
|
|||
def toggle_passphrase(): |
|||
title = _("Confirm Toggle Passphrase Protection") |
|||
currently_enabled = self.features.passphrase_protection |
|||
if currently_enabled: |
|||
msg = _("After disabling passphrases, you can only pair this " |
|||
"Electrum wallet if it had an empty passphrase. " |
|||
"If its passphrase was not empty, you will need to " |
|||
"create a new wallet with the install wizard. You " |
|||
"can use this wallet again at any time by re-enabling " |
|||
"passphrases and entering its passphrase.") |
|||
else: |
|||
msg = _("Your current Electrum wallet can only be used with " |
|||
"an empty passphrase. You must create a separate " |
|||
"wallet with the install wizard for other passphrases " |
|||
"as each one generates a new set of addresses.") |
|||
msg += "\n\n" + _("Are you sure you want to proceed?") |
|||
if not self.question(msg, title=title): |
|||
return |
|||
invoke_client('toggle_passphrase', unpair_after=currently_enabled) |
|||
|
|||
def change_homescreen(): |
|||
dialog = QFileDialog(self, _("Choose Homescreen")) |
|||
filename, __ = dialog.getOpenFileName() |
|||
if not filename: |
|||
return # user cancelled |
|||
|
|||
if filename.endswith('.toif'): |
|||
img = open(filename, 'rb').read() |
|||
if img[:8] != b'TOIf\x90\x00\x90\x00': |
|||
handler.show_error('File is not a TOIF file with size of 144x144') |
|||
return |
|||
else: |
|||
from PIL import Image # FIXME |
|||
im = Image.open(filename) |
|||
if im.size != (128, 64): |
|||
handler.show_error('Image must be 128 x 64 pixels') |
|||
return |
|||
im = im.convert('1') |
|||
pix = im.load() |
|||
img = bytearray(1024) |
|||
for j in range(64): |
|||
for i in range(128): |
|||
if pix[i, j]: |
|||
o = (i + j * 128) |
|||
img[o // 8] |= (1 << (7 - o % 8)) |
|||
img = bytes(img) |
|||
invoke_client('change_homescreen', img) |
|||
|
|||
def clear_homescreen(): |
|||
invoke_client('change_homescreen', b'\x00') |
|||
|
|||
def set_pin(): |
|||
invoke_client('set_pin', remove=False) |
|||
|
|||
def clear_pin(): |
|||
invoke_client('set_pin', remove=True) |
|||
|
|||
def wipe_device(): |
|||
wallet = window.wallet |
|||
if wallet and sum(wallet.get_balance()): |
|||
title = _("Confirm Device Wipe") |
|||
msg = _("Are you SURE you want to wipe the device?\n" |
|||
"Your wallet still has bitcoins in it!") |
|||
if not self.question(msg, title=title, |
|||
icon=QMessageBox.Critical): |
|||
return |
|||
invoke_client('wipe_device', unpair_after=True) |
|||
|
|||
def slider_moved(): |
|||
mins = timeout_slider.sliderPosition() |
|||
timeout_minutes.setText(_("%2d minutes") % mins) |
|||
|
|||
def slider_released(): |
|||
config.set_session_timeout(timeout_slider.sliderPosition() * 60) |
|||
|
|||
# Information tab |
|||
info_tab = QWidget() |
|||
info_layout = QVBoxLayout(info_tab) |
|||
info_glayout = QGridLayout() |
|||
info_glayout.setColumnStretch(2, 1) |
|||
device_label = QLabel() |
|||
pin_set_label = QLabel() |
|||
passphrases_label = QLabel() |
|||
version_label = QLabel() |
|||
device_id_label = QLabel() |
|||
bl_hash_label = QLabel() |
|||
bl_hash_label.setWordWrap(True) |
|||
language_label = QLabel() |
|||
initialized_label = QLabel() |
|||
rows = [ |
|||
(_("Device Label"), device_label), |
|||
(_("PIN set"), pin_set_label), |
|||
(_("Passphrases"), passphrases_label), |
|||
(_("Firmware Version"), version_label), |
|||
(_("Device ID"), device_id_label), |
|||
(_("Bootloader Hash"), bl_hash_label), |
|||
(_("Language"), language_label), |
|||
(_("Initialized"), initialized_label), |
|||
] |
|||
for row_num, (label, widget) in enumerate(rows): |
|||
info_glayout.addWidget(QLabel(label), row_num, 0) |
|||
info_glayout.addWidget(widget, row_num, 1) |
|||
info_layout.addLayout(info_glayout) |
|||
|
|||
# Settings tab |
|||
settings_tab = QWidget() |
|||
settings_layout = QVBoxLayout(settings_tab) |
|||
settings_glayout = QGridLayout() |
|||
|
|||
# Settings tab - Label |
|||
label_msg = QLabel(_("Name this {}. If you have multiple devices " |
|||
"their labels help distinguish them.") |
|||
.format(plugin.device)) |
|||
label_msg.setWordWrap(True) |
|||
label_label = QLabel(_("Device Label")) |
|||
label_edit = QLineEdit() |
|||
label_edit.setMinimumWidth(150) |
|||
label_edit.setMaxLength(plugin.MAX_LABEL_LEN) |
|||
label_apply = QPushButton(_("Apply")) |
|||
label_apply.clicked.connect(rename) |
|||
label_edit.textChanged.connect(set_label_enabled) |
|||
settings_glayout.addWidget(label_label, 0, 0) |
|||
settings_glayout.addWidget(label_edit, 0, 1, 1, 2) |
|||
settings_glayout.addWidget(label_apply, 0, 3) |
|||
settings_glayout.addWidget(label_msg, 1, 1, 1, -1) |
|||
|
|||
# Settings tab - PIN |
|||
pin_label = QLabel(_("PIN Protection")) |
|||
pin_button = QPushButton() |
|||
pin_button.clicked.connect(set_pin) |
|||
settings_glayout.addWidget(pin_label, 2, 0) |
|||
settings_glayout.addWidget(pin_button, 2, 1) |
|||
pin_msg = QLabel(_("PIN protection is strongly recommended. " |
|||
"A PIN is your only protection against someone " |
|||
"stealing your bitcoins if they obtain physical " |
|||
"access to your {}.").format(plugin.device)) |
|||
pin_msg.setWordWrap(True) |
|||
pin_msg.setStyleSheet("color: red") |
|||
settings_glayout.addWidget(pin_msg, 3, 1, 1, -1) |
|||
|
|||
# Settings tab - Homescreen |
|||
homescreen_label = QLabel(_("Homescreen")) |
|||
homescreen_change_button = QPushButton(_("Change...")) |
|||
homescreen_clear_button = QPushButton(_("Reset")) |
|||
homescreen_change_button.clicked.connect(change_homescreen) |
|||
try: |
|||
import PIL |
|||
except ImportError: |
|||
homescreen_change_button.setDisabled(True) |
|||
homescreen_change_button.setToolTip( |
|||
_("Required package 'PIL' is not available - Please install it.") |
|||
) |
|||
homescreen_clear_button.clicked.connect(clear_homescreen) |
|||
homescreen_msg = QLabel(_("You can set the homescreen on your " |
|||
"device to personalize it. You must " |
|||
"choose a {} x {} monochrome black and " |
|||
"white image.").format(hs_rows, hs_cols)) |
|||
homescreen_msg.setWordWrap(True) |
|||
settings_glayout.addWidget(homescreen_label, 4, 0) |
|||
settings_glayout.addWidget(homescreen_change_button, 4, 1) |
|||
settings_glayout.addWidget(homescreen_clear_button, 4, 2) |
|||
settings_glayout.addWidget(homescreen_msg, 5, 1, 1, -1) |
|||
|
|||
# Settings tab - Session Timeout |
|||
timeout_label = QLabel(_("Session Timeout")) |
|||
timeout_minutes = QLabel() |
|||
timeout_slider = QSlider(Qt.Horizontal) |
|||
timeout_slider.setRange(1, 60) |
|||
timeout_slider.setSingleStep(1) |
|||
timeout_slider.setTickInterval(5) |
|||
timeout_slider.setTickPosition(QSlider.TicksBelow) |
|||
timeout_slider.setTracking(True) |
|||
timeout_msg = QLabel( |
|||
_("Clear the session after the specified period " |
|||
"of inactivity. Once a session has timed out, " |
|||
"your PIN and passphrase (if enabled) must be " |
|||
"re-entered to use the device.")) |
|||
timeout_msg.setWordWrap(True) |
|||
timeout_slider.setSliderPosition(config.get_session_timeout() // 60) |
|||
slider_moved() |
|||
timeout_slider.valueChanged.connect(slider_moved) |
|||
timeout_slider.sliderReleased.connect(slider_released) |
|||
settings_glayout.addWidget(timeout_label, 6, 0) |
|||
settings_glayout.addWidget(timeout_slider, 6, 1, 1, 3) |
|||
settings_glayout.addWidget(timeout_minutes, 6, 4) |
|||
settings_glayout.addWidget(timeout_msg, 7, 1, 1, -1) |
|||
settings_layout.addLayout(settings_glayout) |
|||
settings_layout.addStretch(1) |
|||
|
|||
# Advanced tab |
|||
advanced_tab = QWidget() |
|||
advanced_layout = QVBoxLayout(advanced_tab) |
|||
advanced_glayout = QGridLayout() |
|||
|
|||
# Advanced tab - clear PIN |
|||
clear_pin_button = QPushButton(_("Disable PIN")) |
|||
clear_pin_button.clicked.connect(clear_pin) |
|||
clear_pin_warning = QLabel( |
|||
_("If you disable your PIN, anyone with physical access to your " |
|||
"{} device can spend your bitcoins.").format(plugin.device)) |
|||
clear_pin_warning.setWordWrap(True) |
|||
clear_pin_warning.setStyleSheet("color: red") |
|||
advanced_glayout.addWidget(clear_pin_button, 0, 2) |
|||
advanced_glayout.addWidget(clear_pin_warning, 1, 0, 1, 5) |
|||
|
|||
# Advanced tab - toggle passphrase protection |
|||
passphrase_button = QPushButton() |
|||
passphrase_button.clicked.connect(toggle_passphrase) |
|||
passphrase_msg = WWLabel(PASSPHRASE_HELP) |
|||
passphrase_warning = WWLabel(PASSPHRASE_NOT_PIN) |
|||
passphrase_warning.setStyleSheet("color: red") |
|||
advanced_glayout.addWidget(passphrase_button, 3, 2) |
|||
advanced_glayout.addWidget(passphrase_msg, 4, 0, 1, 5) |
|||
advanced_glayout.addWidget(passphrase_warning, 5, 0, 1, 5) |
|||
|
|||
# Advanced tab - wipe device |
|||
wipe_device_button = QPushButton(_("Wipe Device")) |
|||
wipe_device_button.clicked.connect(wipe_device) |
|||
wipe_device_msg = QLabel( |
|||
_("Wipe the device, removing all data from it. The firmware " |
|||
"is left unchanged.")) |
|||
wipe_device_msg.setWordWrap(True) |
|||
wipe_device_warning = QLabel( |
|||
_("Only wipe a device if you have the recovery seed written down " |
|||
"and the device wallet(s) are empty, otherwise the bitcoins " |
|||
"will be lost forever.")) |
|||
wipe_device_warning.setWordWrap(True) |
|||
wipe_device_warning.setStyleSheet("color: red") |
|||
advanced_glayout.addWidget(wipe_device_button, 6, 2) |
|||
advanced_glayout.addWidget(wipe_device_msg, 7, 0, 1, 5) |
|||
advanced_glayout.addWidget(wipe_device_warning, 8, 0, 1, 5) |
|||
advanced_layout.addLayout(advanced_glayout) |
|||
advanced_layout.addStretch(1) |
|||
|
|||
tabs = QTabWidget(self) |
|||
tabs.addTab(info_tab, _("Information")) |
|||
tabs.addTab(settings_tab, _("Settings")) |
|||
tabs.addTab(advanced_tab, _("Advanced")) |
|||
dialog_vbox = QVBoxLayout(self) |
|||
dialog_vbox.addWidget(tabs) |
|||
dialog_vbox.addLayout(Buttons(CloseButton(self))) |
|||
|
|||
# Update information |
|||
invoke_client(None) |
|||
@ -0,0 +1,509 @@ |
|||
from binascii import hexlify, unhexlify |
|||
import traceback |
|||
import sys |
|||
|
|||
from electrum.util import bfh, bh2u, versiontuple, UserCancelled |
|||
from electrum.bitcoin import (b58_address_to_hash160, xpub_from_pubkey, deserialize_xpub, |
|||
TYPE_ADDRESS, TYPE_SCRIPT, is_address) |
|||
from electrum import constants |
|||
from electrum.i18n import _ |
|||
from electrum.plugin import BasePlugin, Device |
|||
from electrum.transaction import deserialize, Transaction |
|||
from electrum.keystore import Hardware_KeyStore, is_xpubkey, parse_xpubkey, xtype_from_derivation |
|||
from electrum.base_wizard import ScriptTypeNotSupported |
|||
|
|||
from ..hw_wallet import HW_PluginBase |
|||
from ..hw_wallet.plugin import is_any_tx_output_on_change_branch, trezor_validate_op_return_output_and_get_data |
|||
|
|||
|
|||
# Safe-T mini initialization methods |
|||
TIM_NEW, TIM_RECOVER, TIM_MNEMONIC, TIM_PRIVKEY = range(0, 4) |
|||
|
|||
# script "generation" |
|||
SCRIPT_GEN_LEGACY, SCRIPT_GEN_P2SH_SEGWIT, SCRIPT_GEN_NATIVE_SEGWIT = range(0, 3) |
|||
|
|||
|
|||
class SafeTKeyStore(Hardware_KeyStore): |
|||
hw_type = 'safe_t' |
|||
device = 'Safe-T mini' |
|||
|
|||
def get_derivation(self): |
|||
return self.derivation |
|||
|
|||
def get_script_gen(self): |
|||
xtype = xtype_from_derivation(self.derivation) |
|||
if xtype in ('p2wpkh', 'p2wsh'): |
|||
return SCRIPT_GEN_NATIVE_SEGWIT |
|||
elif xtype in ('p2wpkh-p2sh', 'p2wsh-p2sh'): |
|||
return SCRIPT_GEN_P2SH_SEGWIT |
|||
else: |
|||
return SCRIPT_GEN_LEGACY |
|||
|
|||
def get_client(self, force_pair=True): |
|||
return self.plugin.get_client(self, force_pair) |
|||
|
|||
def decrypt_message(self, sequence, message, password): |
|||
raise RuntimeError(_('Encryption and decryption are not implemented by {}').format(self.device)) |
|||
|
|||
def sign_message(self, sequence, message, password): |
|||
client = self.get_client() |
|||
address_path = self.get_derivation() + "/%d/%d"%sequence |
|||
address_n = client.expand_path(address_path) |
|||
msg_sig = client.sign_message(self.plugin.get_coin_name(), address_n, message) |
|||
return msg_sig.signature |
|||
|
|||
def sign_transaction(self, tx, password): |
|||
if tx.is_complete(): |
|||
return |
|||
# previous transactions used as inputs |
|||
prev_tx = {} |
|||
# path of the xpubs that are involved |
|||
xpub_path = {} |
|||
for txin in tx.inputs(): |
|||
pubkeys, x_pubkeys = tx.get_sorted_pubkeys(txin) |
|||
tx_hash = txin['prevout_hash'] |
|||
if txin.get('prev_tx') is None and not Transaction.is_segwit_input(txin): |
|||
raise Exception(_('Offline signing with {} is not supported for legacy inputs.').format(self.device)) |
|||
prev_tx[tx_hash] = txin['prev_tx'] |
|||
for x_pubkey in x_pubkeys: |
|||
if not is_xpubkey(x_pubkey): |
|||
continue |
|||
xpub, s = parse_xpubkey(x_pubkey) |
|||
if xpub == self.get_master_public_key(): |
|||
xpub_path[xpub] = self.get_derivation() |
|||
|
|||
self.plugin.sign_transaction(self, tx, prev_tx, xpub_path) |
|||
|
|||
|
|||
class SafeTPlugin(HW_PluginBase): |
|||
# Derived classes provide: |
|||
# |
|||
# class-static variables: client_class, firmware_URL, handler_class, |
|||
# libraries_available, libraries_URL, minimum_firmware, |
|||
# wallet_class, types |
|||
|
|||
firmware_URL = 'https://safe-t.io' |
|||
libraries_URL = 'https://github.com/archos-safe-t/python-safet' |
|||
minimum_firmware = (1, 0, 5) |
|||
keystore_class = SafeTKeyStore |
|||
minimum_library = (0, 1, 0) |
|||
SUPPORTED_XTYPES = ('standard', 'p2wpkh-p2sh', 'p2wpkh', 'p2wsh-p2sh', 'p2wsh') |
|||
|
|||
MAX_LABEL_LEN = 32 |
|||
|
|||
def __init__(self, parent, config, name): |
|||
HW_PluginBase.__init__(self, parent, config, name) |
|||
|
|||
try: |
|||
# Minimal test if python-safet is installed |
|||
import safetlib |
|||
try: |
|||
library_version = safetlib.__version__ |
|||
except AttributeError: |
|||
# python-safet only introduced __version__ in 0.1.0 |
|||
library_version = 'unknown' |
|||
if library_version == 'unknown' or \ |
|||
versiontuple(library_version) < self.minimum_library: |
|||
self.libraries_available_message = ( |
|||
_("Library version for '{}' is too old.").format(name) |
|||
+ '\nInstalled: {}, Needed: {}' |
|||
.format(library_version, self.minimum_library)) |
|||
self.print_stderr(self.libraries_available_message) |
|||
raise ImportError() |
|||
self.libraries_available = True |
|||
except ImportError: |
|||
self.libraries_available = False |
|||
return |
|||
|
|||
from . import client |
|||
from . import transport |
|||
import safetlib.messages |
|||
self.client_class = client.SafeTClient |
|||
self.types = safetlib.messages |
|||
self.DEVICE_IDS = ('Safe-T mini',) |
|||
|
|||
self.transport_handler = transport.SafeTTransport() |
|||
self.device_manager().register_enumerate_func(self.enumerate) |
|||
|
|||
def enumerate(self): |
|||
devices = self.transport_handler.enumerate_devices() |
|||
return [Device(d.get_path(), -1, d.get_path(), 'Safe-T mini', 0) for d in devices] |
|||
|
|||
def create_client(self, device, handler): |
|||
try: |
|||
self.print_error("connecting to device at", device.path) |
|||
transport = self.transport_handler.get_transport(device.path) |
|||
except BaseException as e: |
|||
self.print_error("cannot connect at", device.path, str(e)) |
|||
return None |
|||
|
|||
if not transport: |
|||
self.print_error("cannot connect at", device.path) |
|||
return |
|||
|
|||
self.print_error("connected to device at", device.path) |
|||
client = self.client_class(transport, handler, self) |
|||
|
|||
# Try a ping for device sanity |
|||
try: |
|||
client.ping('t') |
|||
except BaseException as e: |
|||
self.print_error("ping failed", str(e)) |
|||
return None |
|||
|
|||
if not client.atleast_version(*self.minimum_firmware): |
|||
msg = (_('Outdated {} firmware for device labelled {}. Please ' |
|||
'download the updated firmware from {}') |
|||
.format(self.device, client.label(), self.firmware_URL)) |
|||
self.print_error(msg) |
|||
if handler: |
|||
handler.show_error(msg) |
|||
else: |
|||
raise Exception(msg) |
|||
return None |
|||
|
|||
return client |
|||
|
|||
def get_client(self, keystore, force_pair=True): |
|||
devmgr = self.device_manager() |
|||
handler = keystore.handler |
|||
with devmgr.hid_lock: |
|||
client = devmgr.client_for_keystore(self, handler, keystore, force_pair) |
|||
# returns the client for a given keystore. can use xpub |
|||
if client: |
|||
client.used() |
|||
return client |
|||
|
|||
def get_coin_name(self): |
|||
return "Testnet" if constants.net.TESTNET else "Bitcoin" |
|||
|
|||
def initialize_device(self, device_id, wizard, handler): |
|||
# Initialization method |
|||
msg = _("Choose how you want to initialize your {}.\n\n" |
|||
"The first two methods are secure as no secret information " |
|||
"is entered into your computer.\n\n" |
|||
"For the last two methods you input secrets on your keyboard " |
|||
"and upload them to your {}, and so you should " |
|||
"only do those on a computer you know to be trustworthy " |
|||
"and free of malware." |
|||
).format(self.device, self.device) |
|||
choices = [ |
|||
# Must be short as QT doesn't word-wrap radio button text |
|||
(TIM_NEW, _("Let the device generate a completely new seed randomly")), |
|||
(TIM_RECOVER, _("Recover from a seed you have previously written down")), |
|||
(TIM_MNEMONIC, _("Upload a BIP39 mnemonic to generate the seed")), |
|||
(TIM_PRIVKEY, _("Upload a master private key")) |
|||
] |
|||
def f(method): |
|||
import threading |
|||
settings = self.request_safe_t_init_settings(wizard, method, self.device) |
|||
t = threading.Thread(target=self._initialize_device_safe, args=(settings, method, device_id, wizard, handler)) |
|||
t.setDaemon(True) |
|||
t.start() |
|||
exit_code = wizard.loop.exec_() |
|||
if exit_code != 0: |
|||
# this method (initialize_device) was called with the expectation |
|||
# of leaving the device in an initialized state when finishing. |
|||
# signal that this is not the case: |
|||
raise UserCancelled() |
|||
wizard.choice_dialog(title=_('Initialize Device'), message=msg, choices=choices, run_next=f) |
|||
|
|||
def _initialize_device_safe(self, settings, method, device_id, wizard, handler): |
|||
exit_code = 0 |
|||
try: |
|||
self._initialize_device(settings, method, device_id, wizard, handler) |
|||
except UserCancelled: |
|||
exit_code = 1 |
|||
except BaseException as e: |
|||
traceback.print_exc(file=sys.stderr) |
|||
handler.show_error(str(e)) |
|||
exit_code = 1 |
|||
finally: |
|||
wizard.loop.exit(exit_code) |
|||
|
|||
def _initialize_device(self, settings, method, device_id, wizard, handler): |
|||
item, label, pin_protection, passphrase_protection = settings |
|||
|
|||
if method == TIM_RECOVER: |
|||
handler.show_error(_( |
|||
"You will be asked to enter 24 words regardless of your " |
|||
"seed's actual length. If you enter a word incorrectly or " |
|||
"misspell it, you cannot change it or go back - you will need " |
|||
"to start again from the beginning.\n\nSo please enter " |
|||
"the words carefully!"), |
|||
blocking=True) |
|||
|
|||
language = 'english' |
|||
devmgr = self.device_manager() |
|||
client = devmgr.client_by_id(device_id) |
|||
|
|||
if method == TIM_NEW: |
|||
strength = 64 * (item + 2) # 128, 192 or 256 |
|||
u2f_counter = 0 |
|||
skip_backup = False |
|||
client.reset_device(True, strength, passphrase_protection, |
|||
pin_protection, label, language, |
|||
u2f_counter, skip_backup) |
|||
elif method == TIM_RECOVER: |
|||
word_count = 6 * (item + 2) # 12, 18 or 24 |
|||
client.step = 0 |
|||
client.recovery_device(word_count, passphrase_protection, |
|||
pin_protection, label, language) |
|||
elif method == TIM_MNEMONIC: |
|||
pin = pin_protection # It's the pin, not a boolean |
|||
client.load_device_by_mnemonic(str(item), pin, |
|||
passphrase_protection, |
|||
label, language) |
|||
else: |
|||
pin = pin_protection # It's the pin, not a boolean |
|||
client.load_device_by_xprv(item, pin, passphrase_protection, |
|||
label, language) |
|||
|
|||
def _make_node_path(self, xpub, address_n): |
|||
_, depth, fingerprint, child_num, chain_code, key = deserialize_xpub(xpub) |
|||
node = self.types.HDNodeType( |
|||
depth=depth, |
|||
fingerprint=int.from_bytes(fingerprint, 'big'), |
|||
child_num=int.from_bytes(child_num, 'big'), |
|||
chain_code=chain_code, |
|||
public_key=key, |
|||
) |
|||
return self.types.HDNodePathType(node=node, address_n=address_n) |
|||
|
|||
def setup_device(self, device_info, wizard, purpose): |
|||
devmgr = self.device_manager() |
|||
device_id = device_info.device.id_ |
|||
client = devmgr.client_by_id(device_id) |
|||
if client is None: |
|||
raise Exception(_('Failed to create a client for this device.') + '\n' + |
|||
_('Make sure it is in the correct state.')) |
|||
# fixme: we should use: client.handler = wizard |
|||
client.handler = self.create_handler(wizard) |
|||
if not device_info.initialized: |
|||
self.initialize_device(device_id, wizard, client.handler) |
|||
client.get_xpub('m', 'standard') |
|||
client.used() |
|||
|
|||
def get_xpub(self, device_id, derivation, xtype, wizard): |
|||
if xtype not in self.SUPPORTED_XTYPES: |
|||
raise ScriptTypeNotSupported(_('This type of script is not supported with {}.').format(self.device)) |
|||
devmgr = self.device_manager() |
|||
client = devmgr.client_by_id(device_id) |
|||
client.handler = wizard |
|||
xpub = client.get_xpub(derivation, xtype) |
|||
client.used() |
|||
return xpub |
|||
|
|||
def get_safet_input_script_type(self, script_gen, is_multisig): |
|||
if script_gen == SCRIPT_GEN_NATIVE_SEGWIT: |
|||
return self.types.InputScriptType.SPENDWITNESS |
|||
elif script_gen == SCRIPT_GEN_P2SH_SEGWIT: |
|||
return self.types.InputScriptType.SPENDP2SHWITNESS |
|||
else: |
|||
if is_multisig: |
|||
return self.types.InputScriptType.SPENDMULTISIG |
|||
else: |
|||
return self.types.InputScriptType.SPENDADDRESS |
|||
|
|||
def sign_transaction(self, keystore, tx, prev_tx, xpub_path): |
|||
self.prev_tx = prev_tx |
|||
self.xpub_path = xpub_path |
|||
client = self.get_client(keystore) |
|||
inputs = self.tx_inputs(tx, True, keystore.get_script_gen()) |
|||
outputs = self.tx_outputs(keystore.get_derivation(), tx, keystore.get_script_gen()) |
|||
signatures = client.sign_tx(self.get_coin_name(), inputs, outputs, lock_time=tx.locktime)[0] |
|||
signatures = [(bh2u(x) + '01') for x in signatures] |
|||
tx.update_signatures(signatures) |
|||
|
|||
def show_address(self, wallet, address, keystore=None): |
|||
if keystore is None: |
|||
keystore = wallet.get_keystore() |
|||
if not self.show_address_helper(wallet, address, keystore): |
|||
return |
|||
client = self.get_client(keystore) |
|||
if not client.atleast_version(1, 0): |
|||
keystore.handler.show_error(_("Your device firmware is too old")) |
|||
return |
|||
change, index = wallet.get_address_index(address) |
|||
derivation = keystore.derivation |
|||
address_path = "%s/%d/%d"%(derivation, change, index) |
|||
address_n = client.expand_path(address_path) |
|||
xpubs = wallet.get_master_public_keys() |
|||
if len(xpubs) == 1: |
|||
script_gen = keystore.get_script_gen() |
|||
script_type = self.get_safet_input_script_type(script_gen, is_multisig=False) |
|||
client.get_address(self.get_coin_name(), address_n, True, script_type=script_type) |
|||
else: |
|||
def f(xpub): |
|||
return self._make_node_path(xpub, [change, index]) |
|||
pubkeys = wallet.get_public_keys(address) |
|||
# sort xpubs using the order of pubkeys |
|||
sorted_pubkeys, sorted_xpubs = zip(*sorted(zip(pubkeys, xpubs))) |
|||
pubkeys = list(map(f, sorted_xpubs)) |
|||
multisig = self.types.MultisigRedeemScriptType( |
|||
pubkeys=pubkeys, |
|||
signatures=[b''] * wallet.n, |
|||
m=wallet.m, |
|||
) |
|||
script_gen = keystore.get_script_gen() |
|||
script_type = self.get_safet_input_script_type(script_gen, is_multisig=True) |
|||
client.get_address(self.get_coin_name(), address_n, True, multisig=multisig, script_type=script_type) |
|||
|
|||
def tx_inputs(self, tx, for_sig=False, script_gen=SCRIPT_GEN_LEGACY): |
|||
inputs = [] |
|||
for txin in tx.inputs(): |
|||
txinputtype = self.types.TxInputType() |
|||
if txin['type'] == 'coinbase': |
|||
prev_hash = b"\x00"*32 |
|||
prev_index = 0xffffffff # signed int -1 |
|||
else: |
|||
if for_sig: |
|||
x_pubkeys = txin['x_pubkeys'] |
|||
if len(x_pubkeys) == 1: |
|||
x_pubkey = x_pubkeys[0] |
|||
xpub, s = parse_xpubkey(x_pubkey) |
|||
xpub_n = self.client_class.expand_path(self.xpub_path[xpub]) |
|||
txinputtype._extend_address_n(xpub_n + s) |
|||
txinputtype.script_type = self.get_safet_input_script_type(script_gen, is_multisig=False) |
|||
else: |
|||
def f(x_pubkey): |
|||
if is_xpubkey(x_pubkey): |
|||
xpub, s = parse_xpubkey(x_pubkey) |
|||
else: |
|||
xpub = xpub_from_pubkey(0, bfh(x_pubkey)) |
|||
s = [] |
|||
return self._make_node_path(xpub, s) |
|||
pubkeys = list(map(f, x_pubkeys)) |
|||
multisig = self.types.MultisigRedeemScriptType( |
|||
pubkeys=pubkeys, |
|||
signatures=list(map(lambda x: bfh(x)[:-1] if x else b'', txin.get('signatures'))), |
|||
m=txin.get('num_sig'), |
|||
) |
|||
script_type = self.get_safet_input_script_type(script_gen, is_multisig=True) |
|||
txinputtype = self.types.TxInputType( |
|||
script_type=script_type, |
|||
multisig=multisig |
|||
) |
|||
# find which key is mine |
|||
for x_pubkey in x_pubkeys: |
|||
if is_xpubkey(x_pubkey): |
|||
xpub, s = parse_xpubkey(x_pubkey) |
|||
if xpub in self.xpub_path: |
|||
xpub_n = self.client_class.expand_path(self.xpub_path[xpub]) |
|||
txinputtype._extend_address_n(xpub_n + s) |
|||
break |
|||
|
|||
prev_hash = unhexlify(txin['prevout_hash']) |
|||
prev_index = txin['prevout_n'] |
|||
|
|||
if 'value' in txin: |
|||
txinputtype.amount = txin['value'] |
|||
txinputtype.prev_hash = prev_hash |
|||
txinputtype.prev_index = prev_index |
|||
|
|||
if txin.get('scriptSig') is not None: |
|||
script_sig = bfh(txin['scriptSig']) |
|||
txinputtype.script_sig = script_sig |
|||
|
|||
txinputtype.sequence = txin.get('sequence', 0xffffffff - 1) |
|||
|
|||
inputs.append(txinputtype) |
|||
|
|||
return inputs |
|||
|
|||
def tx_outputs(self, derivation, tx, script_gen=SCRIPT_GEN_LEGACY): |
|||
|
|||
def create_output_by_derivation(info): |
|||
index, xpubs, m = info |
|||
if len(xpubs) == 1: |
|||
if script_gen == SCRIPT_GEN_NATIVE_SEGWIT: |
|||
script_type = self.types.OutputScriptType.PAYTOWITNESS |
|||
elif script_gen == SCRIPT_GEN_P2SH_SEGWIT: |
|||
script_type = self.types.OutputScriptType.PAYTOP2SHWITNESS |
|||
else: |
|||
script_type = self.types.OutputScriptType.PAYTOADDRESS |
|||
address_n = self.client_class.expand_path(derivation + "/%d/%d" % index) |
|||
txoutputtype = self.types.TxOutputType( |
|||
amount=amount, |
|||
script_type=script_type, |
|||
address_n=address_n, |
|||
) |
|||
else: |
|||
if script_gen == SCRIPT_GEN_NATIVE_SEGWIT: |
|||
script_type = self.types.OutputScriptType.PAYTOWITNESS |
|||
elif script_gen == SCRIPT_GEN_P2SH_SEGWIT: |
|||
script_type = self.types.OutputScriptType.PAYTOP2SHWITNESS |
|||
else: |
|||
script_type = self.types.OutputScriptType.PAYTOMULTISIG |
|||
address_n = self.client_class.expand_path("/%d/%d" % index) |
|||
pubkeys = [self._make_node_path(xpub, address_n) for xpub in xpubs] |
|||
multisig = self.types.MultisigRedeemScriptType( |
|||
pubkeys=pubkeys, |
|||
signatures=[b''] * len(pubkeys), |
|||
m=m) |
|||
txoutputtype = self.types.TxOutputType( |
|||
multisig=multisig, |
|||
amount=amount, |
|||
address_n=self.client_class.expand_path(derivation + "/%d/%d" % index), |
|||
script_type=script_type) |
|||
return txoutputtype |
|||
|
|||
def create_output_by_address(): |
|||
txoutputtype = self.types.TxOutputType() |
|||
txoutputtype.amount = amount |
|||
if _type == TYPE_SCRIPT: |
|||
txoutputtype.script_type = self.types.OutputScriptType.PAYTOOPRETURN |
|||
txoutputtype.op_return_data = trezor_validate_op_return_output_and_get_data(o) |
|||
elif _type == TYPE_ADDRESS: |
|||
txoutputtype.script_type = self.types.OutputScriptType.PAYTOADDRESS |
|||
txoutputtype.address = address |
|||
return txoutputtype |
|||
|
|||
outputs = [] |
|||
has_change = False |
|||
any_output_on_change_branch = is_any_tx_output_on_change_branch(tx) |
|||
|
|||
for o in tx.outputs(): |
|||
_type, address, amount = o.type, o.address, o.value |
|||
use_create_by_derivation = False |
|||
|
|||
info = tx.output_info.get(address) |
|||
if info is not None and not has_change: |
|||
index, xpubs, m = info |
|||
on_change_branch = index[0] == 1 |
|||
# prioritise hiding outputs on the 'change' branch from user |
|||
# because no more than one change address allowed |
|||
# note: ^ restriction can be removed once we require fw |
|||
# that has https://github.com/trezor/trezor-mcu/pull/306 |
|||
if on_change_branch == any_output_on_change_branch: |
|||
use_create_by_derivation = True |
|||
has_change = True |
|||
|
|||
if use_create_by_derivation: |
|||
txoutputtype = create_output_by_derivation(info) |
|||
else: |
|||
txoutputtype = create_output_by_address() |
|||
outputs.append(txoutputtype) |
|||
|
|||
return outputs |
|||
|
|||
def electrum_tx_to_txtype(self, tx): |
|||
t = self.types.TransactionType() |
|||
if tx is None: |
|||
# probably for segwit input and we don't need this prev txn |
|||
return t |
|||
d = deserialize(tx.raw) |
|||
t.version = d['version'] |
|||
t.lock_time = d['lockTime'] |
|||
inputs = self.tx_inputs(tx) |
|||
t._extend_inputs(inputs) |
|||
for vout in d['outputs']: |
|||
o = t._add_bin_outputs() |
|||
o.amount = vout['value'] |
|||
o.script_pubkey = bfh(vout['scriptPubKey']) |
|||
return t |
|||
|
|||
# This function is called from the TREZOR libraries (via tx_api) |
|||
def get_tx(self, tx_hash): |
|||
tx = self.prev_tx[tx_hash] |
|||
return self.electrum_tx_to_txtype(tx) |
|||
@ -0,0 +1,95 @@ |
|||
from electrum.util import PrintError |
|||
|
|||
|
|||
class SafeTTransport(PrintError): |
|||
|
|||
@staticmethod |
|||
def all_transports(): |
|||
"""Reimplemented safetlib.transport.all_transports so that we can |
|||
enable/disable specific transports. |
|||
""" |
|||
try: |
|||
# only to detect safetlib version |
|||
from safetlib.transport import all_transports |
|||
except ImportError: |
|||
# old safetlib. compat for safetlib < 0.9.2 |
|||
transports = [] |
|||
#try: |
|||
# from safetlib.transport_bridge import BridgeTransport |
|||
# transports.append(BridgeTransport) |
|||
#except BaseException: |
|||
# pass |
|||
try: |
|||
from safetlib.transport_hid import HidTransport |
|||
transports.append(HidTransport) |
|||
except BaseException: |
|||
pass |
|||
try: |
|||
from safetlib.transport_udp import UdpTransport |
|||
transports.append(UdpTransport) |
|||
except BaseException: |
|||
pass |
|||
try: |
|||
from safetlib.transport_webusb import WebUsbTransport |
|||
transports.append(WebUsbTransport) |
|||
except BaseException: |
|||
pass |
|||
else: |
|||
# new safetlib. |
|||
transports = [] |
|||
#try: |
|||
# from safetlib.transport.bridge import BridgeTransport |
|||
# transports.append(BridgeTransport) |
|||
#except BaseException: |
|||
# pass |
|||
try: |
|||
from safetlib.transport.hid import HidTransport |
|||
transports.append(HidTransport) |
|||
except BaseException: |
|||
pass |
|||
try: |
|||
from safetlib.transport.udp import UdpTransport |
|||
transports.append(UdpTransport) |
|||
except BaseException: |
|||
pass |
|||
try: |
|||
from safetlib.transport.webusb import WebUsbTransport |
|||
transports.append(WebUsbTransport) |
|||
except BaseException: |
|||
pass |
|||
return transports |
|||
return transports |
|||
|
|||
def enumerate_devices(self): |
|||
"""Just like safetlib.transport.enumerate_devices, |
|||
but with exception catching, so that transports can fail separately. |
|||
""" |
|||
devices = [] |
|||
for transport in self.all_transports(): |
|||
try: |
|||
new_devices = transport.enumerate() |
|||
except BaseException as e: |
|||
self.print_error('enumerate failed for {}. error {}' |
|||
.format(transport.__name__, str(e))) |
|||
else: |
|||
devices.extend(new_devices) |
|||
return devices |
|||
|
|||
def get_transport(self, path=None): |
|||
"""Reimplemented safetlib.transport.get_transport, |
|||
(1) for old safetlib |
|||
(2) to be able to disable specific transports |
|||
(3) to call our own enumerate_devices that catches exceptions |
|||
""" |
|||
if path is None: |
|||
try: |
|||
return self.enumerate_devices()[0] |
|||
except IndexError: |
|||
raise Exception("No Safe-T mini found") from None |
|||
|
|||
def match_prefix(a, b): |
|||
return a.startswith(b) or b.startswith(a) |
|||
transports = [t for t in self.all_transports() if match_prefix(path, t.PATH_PREFIX)] |
|||
if transports: |
|||
return transports[0].find_by_path(path) |
|||
raise Exception("Unknown path prefix '%s'" % path) |
|||
|
After Width: | Height: | Size: 3.8 KiB |
|
After Width: | Height: | Size: 3.6 KiB |
Loading…
Reference in new issue