zebra.lucky
/
electrum
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Merge pull request #2821 from benma/dbb_py3 

fix digital bitbox plugin, port to Python 3
master
ThomasV 8 years ago committed by GitHub
parent
936b92fa00 3143179907
commit
1063847369
1 changed files with 75 additions and 74 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 83
      plugins/digitalbitbox/digitalbitbox.py

83
plugins/digitalbitbox/digitalbitbox.py
Unescape View File

@ -15,6 +15,7 @@ try:
import hid import hid
import json import json
import math import math
import binascii
import struct import struct
import hashlib import hashlib
from ecdsa.ecdsa import generator_secp256k1 from ecdsa.ecdsa import generator_secp256k1
@ -30,6 +31,9 @@ except ImportError as e:
# USB HID interface # USB HID interface
# #
def to_hexstr(s):
return binascii.hexlify(s).decode('ascii')
class DigitalBitbox_Client(): class DigitalBitbox_Client():
def __init__(self, hidDevice): def __init__(self, hidDevice):
@ -72,14 +76,14 @@ class DigitalBitbox_Client():
def get_xpub(self, bip32_path): def get_xpub(self, bip32_path):
if self.check_device_dialog(): if self.check_device_dialog():
msg = '{"xpub":"' + bip32_path + '"}' msg = b'{"xpub": "%s"}' % bip32_path.encode('utf8')
reply = self.hid_send_encrypt(msg) reply = self.hid_send_encrypt(msg)
return reply['xpub'] return reply['xpub']
return None return None
def dbb_has_password(self): def dbb_has_password(self):
reply = self.hid_send_plain('{"ping":""}') reply = self.hid_send_plain(b'{"ping":""}')
if 'ping' not in reply: if 'ping' not in reply:
raise Exception('Device communication error. Please unplug and replug your Digital Bitbox.') raise Exception('Device communication error. Please unplug and replug your Digital Bitbox.')
if reply['ping'] == 'password': if reply['ping'] == 'password':
@ -89,7 +93,7 @@ class DigitalBitbox_Client():
def stretch_key(self, key): def stretch_key(self, key):
import pbkdf2, hmac import pbkdf2, hmac
return pbkdf2.PBKDF2(key, 'Digital Bitbox', iterations = 20480, macmodule = hmac, digestmodule = hashlib.sha512).read(64).encode('hex') return binascii.hexlify(pbkdf2.PBKDF2(key, b'Digital Bitbox', iterations = 20480, macmodule = hmac, digestmodule = hashlib.sha512).read(64))
def backup_password_dialog(self): def backup_password_dialog(self):
@ -103,7 +107,7 @@ class DigitalBitbox_Client():
elif len(password) > 64: elif len(password) > 64:
msg = _("Password must have less than 64 characters.\r\n\r\nEnter password:") msg = _("Password must have less than 64 characters.\r\n\r\nEnter password:")
else: else:
return str(password) return password.encode('utf8')
def password_dialog(self, msg): def password_dialog(self, msg):
@ -116,7 +120,7 @@ class DigitalBitbox_Client():
elif len(password) > 64: elif len(password) > 64:
msg = _("Password must have less than 64 characters.\r\n\r\nEnter password:") msg = _("Password must have less than 64 characters.\r\n\r\nEnter password:")
else: else:
self.password = str(password) self.password = password.encode('utf8')
return True return True
@ -130,7 +134,7 @@ class DigitalBitbox_Client():
"You cannot access your coins or a backup without the password.\r\n" \ "You cannot access your coins or a backup without the password.\r\n" \
"A backup is saved automatically when generating a new wallet.") "A backup is saved automatically when generating a new wallet.")
if self.password_dialog(msg): if self.password_dialog(msg):
reply = self.hid_send_plain('{"password":"' + self.password + '"}') reply = self.hid_send_plain(b'{"password":"' + self.password + '"}')
else: else:
return False return False
@ -139,7 +143,7 @@ class DigitalBitbox_Client():
while self.password is None: while self.password is None:
if not self.password_dialog(msg): if not self.password_dialog(msg):
return False return False
reply = self.hid_send_encrypt('{"led":"blink"}') reply = self.hid_send_encrypt(b'{"led":"blink"}')
if 'error' in reply: if 'error' in reply:
self.password = None self.password = None
if reply['error']['code'] == 109: if reply['error']['code'] == 109:
@ -156,7 +160,7 @@ class DigitalBitbox_Client():
if not self.setupRunning: if not self.setupRunning:
self.isInitialized = True # Wallet exists. Electrum code later checks if the device matches the wallet self.isInitialized = True # Wallet exists. Electrum code later checks if the device matches the wallet
elif not self.isInitialized: elif not self.isInitialized:
reply = self.hid_send_encrypt('{"device":"info"}') reply = self.hid_send_encrypt(b'{"device":"info"}')
if reply['device']['id'] != "": if reply['device']['id'] != "":
self.recover_or_erase_dialog() # Already seeded self.recover_or_erase_dialog() # Already seeded
else: else:
@ -206,8 +210,8 @@ class DigitalBitbox_Client():
def dbb_generate_wallet(self): def dbb_generate_wallet(self):
key = self.stretch_key(self.password) key = self.stretch_key(self.password)
filename = "Electrum-" + time.strftime("%Y-%m-%d-%H-%M-%S") + ".pdf" filename = b"Electrum-" + time.strftime("%Y-%m-%d-%H-%M-%S") + ".pdf"
msg = '{"seed":{"source": "create", "key": "%s", "filename": "%s", "entropy": "%s"}}' % (key, filename, 'Digital Bitbox Electrum Plugin') msg = b'{"seed":{"source": "create", "key": "%s", "filename": "%s", "entropy": "%s"}}' % (key, filename, b'Digital Bitbox Electrum Plugin')
reply = self.hid_send_encrypt(msg) reply = self.hid_send_encrypt(msg)
if 'error' in reply: if 'error' in reply:
raise Exception(reply['error']['message']) raise Exception(reply['error']['message'])
@ -217,7 +221,7 @@ class DigitalBitbox_Client():
self.handler.show_message(_("Are you sure you want to erase the Digital Bitbox?\r\n\r\n" \ self.handler.show_message(_("Are you sure you want to erase the Digital Bitbox?\r\n\r\n" \
"To continue, touch the Digital Bitbox's light for 3 seconds.\r\n\r\n" \ "To continue, touch the Digital Bitbox's light for 3 seconds.\r\n\r\n" \
"To cancel, briefly touch the light or wait for the timeout.")) "To cancel, briefly touch the light or wait for the timeout."))
hid_reply = self.hid_send_encrypt('{"reset":"__ERASE__"}') hid_reply = self.hid_send_encrypt(b'{"reset":"__ERASE__"}')
self.handler.clear_dialog() self.handler.clear_dialog()
if 'error' in hid_reply: if 'error' in hid_reply:
raise Exception(hid_reply['error']['message']) raise Exception(hid_reply['error']['message'])
@ -227,7 +231,7 @@ class DigitalBitbox_Client():
def dbb_load_backup(self, show_msg=True): def dbb_load_backup(self, show_msg=True):
backups = self.hid_send_encrypt('{"backup":"list"}') backups = self.hid_send_encrypt(b'{"backup":"list"}')
if 'error' in backups: if 'error' in backups:
raise Exception(backups['error']['message']) raise Exception(backups['error']['message'])
try: try:
@ -242,7 +246,7 @@ class DigitalBitbox_Client():
self.handler.show_message(_("Loading backup...\r\n\r\n" \ self.handler.show_message(_("Loading backup...\r\n\r\n" \
"To continue, touch the Digital Bitbox's light for 3 seconds.\r\n\r\n" \ "To continue, touch the Digital Bitbox's light for 3 seconds.\r\n\r\n" \
"To cancel, briefly touch the light or wait for the timeout.")) "To cancel, briefly touch the light or wait for the timeout."))
msg = '{"seed":{"source": "backup", "key": "%s", "filename": "%s"}}' % (key, backups['backup'][f]) msg = b'{"seed":{"source": "backup", "key": "%s", "filename": "%s"}}' % (key, backups['backup'][f].encode('utf8'))
hid_reply = self.hid_send_encrypt(msg) hid_reply = self.hid_send_encrypt(msg)
self.handler.clear_dialog() self.handler.clear_dialog()
if 'error' in hid_reply: if 'error' in hid_reply:
@ -253,7 +257,6 @@ class DigitalBitbox_Client():
def hid_send_frame(self, data): def hid_send_frame(self, data):
HWW_CID = 0xFF000000 HWW_CID = 0xFF000000
HWW_CMD = 0x80 + 0x40 + 0x01 HWW_CMD = 0x80 + 0x40 + 0x01
data = bytearray(data)
data_len = len(data) data_len = len(data)
seq = 0; seq = 0;
idx = 0; idx = 0;
@ -262,18 +265,18 @@ class DigitalBitbox_Client():
if idx == 0: if idx == 0:
# INIT frame # INIT frame
write = data[idx : idx + min(data_len, self.usbReportSize - 7)] write = data[idx : idx + min(data_len, self.usbReportSize - 7)]
self.dbb_hid.write('\0' + struct.pack(">IBH", HWW_CID, HWW_CMD, data_len & 0xFFFF) + write + '\xEE' * (self.usbReportSize - 7 - len(write))) self.dbb_hid.write(b'\0' + struct.pack(">IBH", HWW_CID, HWW_CMD, data_len & 0xFFFF) + write + b'\xEE' * (self.usbReportSize - 7 - len(write)))
else: else:
# CONT frame # CONT frame
write = data[idx : idx + min(data_len, self.usbReportSize - 5)] write = data[idx : idx + min(data_len, self.usbReportSize - 5)]
self.dbb_hid.write('\0' + struct.pack(">IB", HWW_CID, seq) + write + '\xEE' * (self.usbReportSize - 5 - len(write))) self.dbb_hid.write(b'\0' + struct.pack(">IB", HWW_CID, seq) + write + b'\xEE' * (self.usbReportSize - 5 - len(write)))
seq += 1 seq += 1
idx += len(write) idx += len(write)
def hid_read_frame(self): def hid_read_frame(self):
# INIT response # INIT response
read = self.dbb_hid.read(self.usbReportSize) read = bytearray(self.dbb_hid.read(self.usbReportSize))
cid = ((read[0] * 256 + read[1]) * 256 + read[2]) * 256 + read[3] cid = ((read[0] * 256 + read[1]) * 256 + read[2]) * 256 + read[3]
cmd = read[4] cmd = read[4]
data_len = read[5] * 256 + read[6] data_len = read[5] * 256 + read[6]
@ -281,7 +284,7 @@ class DigitalBitbox_Client():
idx = len(read) - 7; idx = len(read) - 7;
while idx < data_len: while idx < data_len:
# CONT response # CONT response
read = self.dbb_hid.read(self.usbReportSize) read = bytearray(self.dbb_hid.read(self.usbReportSize))
data += read[5:] data += read[5:]
idx += len(read) - 5 idx += len(read) - 5
return data return data
@ -293,15 +296,15 @@ class DigitalBitbox_Client():
serial_number = self.dbb_hid.get_serial_number_string() serial_number = self.dbb_hid.get_serial_number_string()
if "v2.0." in serial_number or "v1." in serial_number: if "v2.0." in serial_number or "v1." in serial_number:
hidBufSize = 4096 hidBufSize = 4096
self.dbb_hid.write('\0' + bytearray(msg) + '\0' * (hidBufSize - len(msg))) self.dbb_hid.write('\0' + msg + '\0' * (hidBufSize - len(msg)))
r = [] r = bytearray()
while len(r) < hidBufSize: while len(r) < hidBufSize:
r = r + self.dbb_hid.read(hidBufSize) r += bytearray(self.dbb_hid.read(hidBufSize))
else: else:
self.hid_send_frame(msg) self.hid_send_frame(msg)
r = self.hid_read_frame() r = self.hid_read_frame()
r = str(bytearray(r)).rstrip(' \t\r\n\0') r = r.rstrip(b' \t\r\n\0')
r = r.replace("\0", '') r = r.replace(b"\0", b'')
reply = json.loads(r) reply = json.loads(r)
except Exception as e: except Exception as e:
print_error('Exception caught ' + str(e)) print_error('Exception caught ' + str(e))
@ -357,13 +360,15 @@ class DigitalBitbox_KeyStore(Hardware_KeyStore):
def sign_message(self, sequence, message, password): def sign_message(self, sequence, message, password):
sig = None sig = None
try: try:
message = message.encode('utf8')
inputPath = self.get_derivation() + "/%d/%d" % sequence inputPath = self.get_derivation() + "/%d/%d" % sequence
inputHash = Hash(msg_magic(message)).encode('hex') msg_hash = Hash(msg_magic(message))
inputHash = to_hexstr(msg_hash)
hasharray = [] hasharray = []
hasharray.append({'hash': inputHash, 'keypath': inputPath}) hasharray.append({'hash': inputHash, 'keypath': inputPath})
hasharray = json.dumps(hasharray) hasharray = json.dumps(hasharray)
msg = '{"sign":{"meta":"sign message", "data":%s}}' % (hasharray) msg = b'{"sign":{"meta":"sign message", "data":%s}}' % hasharray.encode('utf8')
dbb_client = self.plugin.get_client(self) dbb_client = self.plugin.get_client(self)
@ -385,9 +390,8 @@ class DigitalBitbox_KeyStore(Hardware_KeyStore):
if 'recid' in reply['sign'][0]: if 'recid' in reply['sign'][0]:
# firmware > v2.1.1 # firmware > v2.1.1
sig = chr(27 + int(reply['sign'][0]['recid'], 16) + 4) + reply['sign'][0]['sig'].decode('hex') sig = bytes([27 + int(reply['sign'][0]['recid'], 16) + 4]) + binascii.unhexlify(reply['sign'][0]['sig'])
h = Hash(msg_magic(message)) pk, compressed = pubkey_from_signature(sig, msg_hash)
pk, compressed = pubkey_from_signature(sig, h)
pk = point_to_ser(pk.pubkey.point, compressed) pk = point_to_ser(pk.pubkey.point, compressed)
addr = public_key_to_p2pkh(pk) addr = public_key_to_p2pkh(pk)
if verify_message(addr, sig, message) is False: if verify_message(addr, sig, message) is False:
@ -395,9 +399,9 @@ class DigitalBitbox_KeyStore(Hardware_KeyStore):
elif 'pubkey' in reply['sign'][0]: elif 'pubkey' in reply['sign'][0]:
# firmware <= v2.1.1 # firmware <= v2.1.1
for i in range(4): for i in range(4):
sig = chr(27 + i + 4) + reply['sign'][0]['sig'].decode('hex') sig = bytes([27 + i + 4]) + binascii.unhexlify(reply['sign'][0]['sig'])
try: try:
addr = public_key_to_p2pkh(reply['sign'][0]['pubkey'].decode('hex')) addr = public_key_to_p2pkh(binascii.unhexlify(reply['sign'][0]['pubkey']))
if verify_message(addr, sig, message): if verify_message(addr, sig, message):
break break
except Exception: except Exception:
@ -434,8 +438,8 @@ class DigitalBitbox_KeyStore(Hardware_KeyStore):
if x_pubkey in derivations: if x_pubkey in derivations:
index = derivations.get(x_pubkey) index = derivations.get(x_pubkey)
inputPath = "%s/%d/%d" % (self.get_derivation(), index[0], index[1]) inputPath = "%s/%d/%d" % (self.get_derivation(), index[0], index[1])
inputHash = Hash(tx.serialize_preimage(i).decode('hex')) inputHash = Hash(binascii.unhexlify(tx.serialize_preimage(i)))
hasharray_i = {'hash': inputHash.encode('hex'), 'keypath': inputPath} hasharray_i = {'hash': to_hexstr(inputHash), 'keypath': inputPath}
hasharray.append(hasharray_i) hasharray.append(hasharray_i)
inputhasharray.append(inputHash) inputhasharray.append(inputHash)
break break
@ -466,8 +470,8 @@ class DigitalBitbox_KeyStore(Hardware_KeyStore):
for step in range(int(steps)): for step in range(int(steps)):
hashes = hasharray[step * self.maxInputs : (step + 1) * self.maxInputs] hashes = hasharray[step * self.maxInputs : (step + 1) * self.maxInputs]
msg = '{"sign": {"meta":"%s", "data":%s, "checkpub":%s} }' % \ msg = ('{"sign": {"meta":"%s", "data":%s, "checkpub":%s} }' % \
(Hash(tx.serialize()).encode('hex'), json.dumps(hashes), json.dumps(pubkeyarray)) (to_hexstr(Hash(tx.serialize())), json.dumps(hashes), json.dumps(pubkeyarray))).encode('utf8')
dbb_client = self.plugin.get_client(self) dbb_client = self.plugin.get_client(self)
@ -475,7 +479,6 @@ class DigitalBitbox_KeyStore(Hardware_KeyStore):
raise Exception("Could not sign transaction.") raise Exception("Could not sign transaction.")
reply = dbb_client.hid_send_encrypt(msg) reply = dbb_client.hid_send_encrypt(msg)
if 'error' in reply: if 'error' in reply:
raise Exception(reply['error']['message']) raise Exception(reply['error']['message'])
@ -509,7 +512,7 @@ class DigitalBitbox_KeyStore(Hardware_KeyStore):
for i, txin in enumerate(tx.inputs()): for i, txin in enumerate(tx.inputs()):
num = txin['num_sig'] num = txin['num_sig']
for pubkey in txin['pubkeys']: for pubkey in txin['pubkeys']:
signatures = filter(None, txin['signatures']) signatures = list(filter(None, txin['signatures']))
if len(signatures) == num: if len(signatures) == num:
break # txin is complete break # txin is complete
ii = txin['pubkeys'].index(pubkey) ii = txin['pubkeys'].index(pubkey)
@ -517,10 +520,10 @@ class DigitalBitbox_KeyStore(Hardware_KeyStore):
if 'recid' in signed: if 'recid' in signed:
# firmware > v2.1.1 # firmware > v2.1.1
recid = int(signed['recid'], 16) recid = int(signed['recid'], 16)
s = signed['sig'].decode('hex') s = binascii.unhexlify(signed['sig'])
h = inputhasharray[i] h = inputhasharray[i]
pk = MyVerifyingKey.from_signature(s, recid, h, curve = SECP256k1) pk = MyVerifyingKey.from_signature(s, recid, h, curve = SECP256k1)
pk = point_to_ser(pk.pubkey.point, True).encode('hex') pk = to_hexstr(point_to_ser(pk.pubkey.point, True))
elif 'pubkey' in signed: elif 'pubkey' in signed:
# firmware <= v2.1.1 # firmware <= v2.1.1
pk = signed['pubkey'] pk = signed['pubkey']
@ -529,9 +532,8 @@ class DigitalBitbox_KeyStore(Hardware_KeyStore):
sig_r = int(signed['sig'][:64], 16) sig_r = int(signed['sig'][:64], 16)
sig_s = int(signed['sig'][64:], 16) sig_s = int(signed['sig'][64:], 16)
sig = sigencode_der(sig_r, sig_s, generator_secp256k1.order()) sig = sigencode_der(sig_r, sig_s, generator_secp256k1.order())
txin['signatures'][ii] = sig.encode('hex') + '01' txin['signatures'][ii] = to_hexstr(sig) + '01'
tx._inputs[i] = txin tx._inputs[i] = txin
except BaseException as e: except BaseException as e:
self.give_error(e, True) self.give_error(e, True)
else: else:
@ -597,4 +599,3 @@ class DigitalBitboxPlugin(HW_PluginBase):
if client is not None: if client is not None:
client.check_device_dialog() client.check_device_dialog()
return client return client

Write Preview
Loading…
Cancel
Save