Merge branch 'refactor-device'
Roman Zeyde
8 years ago
commit
c2158947c8
@ -1,148 +0,0 @@
|
||||
"""TREZOR-like interface for Ledger hardware wallet."""
|
||||
import binascii
|
||||
import struct
|
||||
|
||||
from trezorlib.types_pb2 import IdentityType # pylint: disable=import-error,unused-import
|
||||
from . import util
|
||||
|
||||
|
||||
class LedgerClientConnection(object):
|
||||
"""Mock for TREZOR-like connection object."""
|
||||
|
||||
def __init__(self, dongle):
|
||||
"""Create connection."""
|
||||
self.dongle = dongle
|
||||
|
||||
@staticmethod
|
||||
def expand_path(path):
|
||||
"""Convert BIP32 path into bytes."""
|
||||
return b''.join((struct.pack('>I', e) for e in path))
|
||||
|
||||
@staticmethod
|
||||
def convert_public_key(ecdsa_curve_name, result):
|
||||
"""Convert Ledger reply into PublicKey object."""
|
||||
from trezorlib.messages_pb2 import PublicKey # pylint: disable=import-error
|
||||
if ecdsa_curve_name == 'nist256p1':
|
||||
if (result[64] & 1) != 0:
|
||||
result = bytearray([0x03]) + result[1:33]
|
||||
else:
|
||||
result = bytearray([0x02]) + result[1:33]
|
||||
else:
|
||||
result = result[1:]
|
||||
keyX = bytearray(result[0:32])
|
||||
keyY = bytearray(result[32:][::-1])
|
||||
if (keyX[31] & 1) != 0:
|
||||
keyY[31] |= 0x80
|
||||
result = b'\x00' + bytes(keyY)
|
||||
publicKey = PublicKey()
|
||||
publicKey.node.public_key = bytes(result)
|
||||
return publicKey
|
||||
|
||||
# pylint: disable=unused-argument
|
||||
def get_public_node(self, n, ecdsa_curve_name='secp256k1', show_display=False):
|
||||
"""Get PublicKey object for specified BIP32 address and elliptic curve."""
|
||||
donglePath = LedgerClientConnection.expand_path(n)
|
||||
if ecdsa_curve_name == 'nist256p1':
|
||||
p2 = '01'
|
||||
else:
|
||||
p2 = '02'
|
||||
apdu = '800200' + p2
|
||||
apdu = binascii.unhexlify(apdu)
|
||||
apdu += bytearray([len(donglePath) + 1, len(donglePath) // 4])
|
||||
apdu += donglePath
|
||||
result = bytearray(self.dongle.exchange(bytes(apdu)))[1:]
|
||||
return LedgerClientConnection.convert_public_key(ecdsa_curve_name, result)
|
||||
|
||||
# pylint: disable=too-many-locals
|
||||
def sign_identity(self, identity, challenge_hidden, challenge_visual,
|
||||
ecdsa_curve_name='secp256k1'):
|
||||
"""Sign specified challenges using secret key derived from given identity."""
|
||||
from trezorlib.messages_pb2 import SignedIdentity # pylint: disable=import-error
|
||||
n = util.get_bip32_address(identity)
|
||||
donglePath = LedgerClientConnection.expand_path(n)
|
||||
if identity.proto == 'ssh':
|
||||
ins = '04'
|
||||
p1 = '00'
|
||||
else:
|
||||
ins = '08'
|
||||
p1 = '00'
|
||||
if ecdsa_curve_name == 'nist256p1':
|
||||
p2 = '81' if identity.proto == 'ssh' else '01'
|
||||
else:
|
||||
p2 = '82' if identity.proto == 'ssh' else '02'
|
||||
apdu = '80' + ins + p1 + p2
|
||||
apdu = binascii.unhexlify(apdu)
|
||||
apdu += bytearray([len(challenge_hidden) + len(donglePath) + 1])
|
||||
apdu += bytearray([len(donglePath) // 4]) + donglePath
|
||||
apdu += challenge_hidden
|
||||
result = bytearray(self.dongle.exchange(bytes(apdu)))
|
||||
if ecdsa_curve_name == 'nist256p1':
|
||||
offset = 3
|
||||
length = result[offset]
|
||||
r = result[offset+1:offset+1+length]
|
||||
if r[0] == 0:
|
||||
r = r[1:]
|
||||
offset = offset + 1 + length + 1
|
||||
length = result[offset]
|
||||
s = result[offset+1:offset+1+length]
|
||||
if s[0] == 0:
|
||||
s = s[1:]
|
||||
offset = offset + 1 + length
|
||||
signature = SignedIdentity()
|
||||
signature.signature = b'\x00' + bytes(r) + bytes(s)
|
||||
if identity.proto == 'ssh':
|
||||
keyData = result[offset:]
|
||||
pk = LedgerClientConnection.convert_public_key(ecdsa_curve_name, keyData)
|
||||
signature.public_key = pk.node.public_key
|
||||
return signature
|
||||
else:
|
||||
signature = SignedIdentity()
|
||||
signature.signature = b'\x00' + bytes(result[0:64])
|
||||
if identity.proto == 'ssh':
|
||||
keyData = result[64:]
|
||||
pk = LedgerClientConnection.convert_public_key(ecdsa_curve_name, keyData)
|
||||
signature.public_key = pk.node.public_key
|
||||
return signature
|
||||
|
||||
def get_ecdh_session_key(self, identity, peer_public_key, ecdsa_curve_name='secp256k1'):
|
||||
"""Create shared secret key for GPG decryption."""
|
||||
from trezorlib.messages_pb2 import ECDHSessionKey # pylint: disable=import-error
|
||||
n = util.get_bip32_address(identity, True)
|
||||
donglePath = LedgerClientConnection.expand_path(n)
|
||||
if ecdsa_curve_name == 'nist256p1':
|
||||
p2 = '01'
|
||||
else:
|
||||
p2 = '02'
|
||||
apdu = '800a00' + p2
|
||||
apdu = binascii.unhexlify(apdu)
|
||||
apdu += bytearray([len(peer_public_key) + len(donglePath) + 1])
|
||||
apdu += bytearray([len(donglePath) // 4]) + donglePath
|
||||
apdu += peer_public_key
|
||||
result = bytearray(self.dongle.exchange(bytes(apdu)))
|
||||
sessionKey = ECDHSessionKey()
|
||||
sessionKey.session_key = bytes(result)
|
||||
return sessionKey
|
||||
|
||||
def clear_session(self):
|
||||
"""Mock for TREZOR interface compatibility."""
|
||||
pass
|
||||
|
||||
def close(self):
|
||||
"""Close connection."""
|
||||
self.dongle.close()
|
||||
|
||||
# pylint: disable=unused-argument
|
||||
# pylint: disable=no-self-use
|
||||
def ping(self, msg, button_protection=False, pin_protection=False,
|
||||
passphrase_protection=False):
|
||||
"""Mock for TREZOR interface compatibility."""
|
||||
return msg
|
||||
|
||||
|
||||
class CallException(Exception):
|
||||
"""Ledger-related error (mainly for TREZOR compatibility)."""
|
||||
|
||||
def __init__(self, code, message):
|
||||
"""Create an error."""
|
||||
super(CallException, self).__init__()
|
||||
self.args = [code, message]
|
||||
@ -0,0 +1,28 @@
|
||||
"""Cryptographic hardware device management."""
|
||||
|
||||
import logging
|
||||
|
||||
from . import trezor
|
||||
from . import keepkey
|
||||
from . import ledger
|
||||
from . import interface
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
DEVICE_TYPES = [
|
||||
trezor.Trezor,
|
||||
keepkey.KeepKey,
|
||||
ledger.LedgerNanoS,
|
||||
]
|
||||
|
||||
|
||||
def detect(identity_str, curve_name):
|
||||
"""Detect the first available device and return it to the user."""
|
||||
for device_type in DEVICE_TYPES:
|
||||
try:
|
||||
with device_type(identity_str, curve_name) as d:
|
||||
return d
|
||||
except interface.NotFoundError as e:
|
||||
log.debug('device not found: %s', e)
|
||||
raise IOError('No device found: "{}" ({})'.format(identity_str,
|
||||
curve_name))
|
||||
@ -0,0 +1,125 @@
|
||||
"""Device abstraction layer."""
|
||||
|
||||
import hashlib
|
||||
import io
|
||||
import logging
|
||||
import re
|
||||
import struct
|
||||
|
||||
from .. import formats, util
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
_identity_regexp = re.compile(''.join([
|
||||
'^'
|
||||
r'(?:(?P<proto>.*)://)?',
|
||||
r'(?:(?P<user>.*)@)?',
|
||||
r'(?P<host>.*?)',
|
||||
r'(?::(?P<port>\w*))?',
|
||||
r'(?P<path>/.*)?',
|
||||
'$'
|
||||
]))
|
||||
|
||||
|
||||
def string_to_identity(identity_str):
|
||||
"""Parse string into Identity dictionary."""
|
||||
m = _identity_regexp.match(identity_str)
|
||||
result = m.groupdict()
|
||||
log.debug('parsed identity: %s', result)
|
||||
return {k: v for k, v in result.items() if v}
|
||||
|
||||
|
||||
def identity_to_string(identity_dict):
|
||||
"""Dump Identity dictionary into its string representation."""
|
||||
result = []
|
||||
if identity_dict.get('proto'):
|
||||
result.append(identity_dict['proto'] + '://')
|
||||
if identity_dict.get('user'):
|
||||
result.append(identity_dict['user'] + '@')
|
||||
result.append(identity_dict['host'])
|
||||
if identity_dict.get('port'):
|
||||
result.append(':' + identity_dict['port'])
|
||||
if identity_dict.get('path'):
|
||||
result.append(identity_dict['path'])
|
||||
log.debug('identity parts: %s', result)
|
||||
return ''.join(result)
|
||||
|
||||
|
||||
def get_bip32_address(identity_dict, ecdh=False):
|
||||
"""Compute BIP32 derivation address according to SLIP-0013/0017."""
|
||||
index = struct.pack('<L', identity_dict.get('index', 0))
|
||||
addr = index + identity_to_string(identity_dict).encode('ascii')
|
||||
log.debug('bip32 address string: %r', addr)
|
||||
digest = hashlib.sha256(addr).digest()
|
||||
s = io.BytesIO(bytearray(digest))
|
||||
|
||||
hardened = 0x80000000
|
||||
addr_0 = [13, 17][bool(ecdh)]
|
||||
address_n = [addr_0] + list(util.recv(s, '<LLLL'))
|
||||
return [(hardened | value) for value in address_n]
|
||||
|
||||
|
||||
class Error(Exception):
|
||||
"""Device-related error."""
|
||||
|
||||
|
||||
class NotFoundError(Error):
|
||||
"""Device could not be found."""
|
||||
|
||||
|
||||
class DeviceError(Error):
|
||||
""""Error during device operation."""
|
||||
|
||||
|
||||
class Device(object):
|
||||
"""Abstract cryptographic hardware device interface."""
|
||||
|
||||
def __init__(self, identity_str, curve_name):
|
||||
"""Configure for specific identity and elliptic curve usage."""
|
||||
self.identity_dict = string_to_identity(identity_str)
|
||||
self.curve_name = curve_name
|
||||
self.conn = None
|
||||
|
||||
def identity_str(self):
|
||||
"""Return identity serialized to string."""
|
||||
return identity_to_string(self.identity_dict)
|
||||
|
||||
def connect(self):
|
||||
"""Connect to device, otherwise raise NotFoundError."""
|
||||
raise NotImplementedError()
|
||||
|
||||
def __enter__(self):
|
||||
"""Allow usage as context manager."""
|
||||
self.conn = self.connect()
|
||||
return self
|
||||
|
||||
def __exit__(self, *args):
|
||||
"""Close and mark as disconnected."""
|
||||
try:
|
||||
self.conn.close()
|
||||
except Exception as e: # pylint: disable=broad-except
|
||||
log.exception('close failed: %s', e)
|
||||
self.conn = None
|
||||
|
||||
def pubkey(self, ecdh=False):
|
||||
"""Get public key (as bytes)."""
|
||||
raise NotImplementedError()
|
||||
|
||||
def sign(self, blob):
|
||||
"""Sign given blob and return the signature (as bytes)."""
|
||||
raise NotImplementedError()
|
||||
|
||||
def ecdh(self, pubkey):
|
||||
"""Get shared session key using Elliptic Curve Diffie-Hellman."""
|
||||
raise NotImplementedError()
|
||||
|
||||
def __str__(self):
|
||||
"""Human-readable representation."""
|
||||
return '{}'.format(self.__class__.__name__)
|
||||
|
||||
def get_curve_name(self, ecdh=False):
|
||||
"""Return correct curve name for device operations."""
|
||||
if ecdh:
|
||||
return formats.get_ecdh_curve_name(self.curve_name)
|
||||
else:
|
||||
return self.curve_name
|
||||
@ -0,0 +1,25 @@
|
||||
"""KeepKey-related code (see https://www.keepkey.com/)."""
|
||||
|
||||
from . import interface, trezor
|
||||
from .. import formats
|
||||
|
||||
|
||||
class KeepKey(trezor.Trezor):
|
||||
"""Connection to KeepKey device."""
|
||||
|
||||
from . import keepkey_defs as defs
|
||||
|
||||
required_version = '>=1.0.4'
|
||||
|
||||
def connect(self):
|
||||
"""No support for other than NIST256P elliptic curves."""
|
||||
if self.curve_name not in {formats.CURVE_NIST256}:
|
||||
fmt = 'KeepKey does not support {} curve'
|
||||
raise interface.NotFoundError(fmt.format(self.curve_name))
|
||||
|
||||
return trezor.Trezor.connect(self)
|
||||
|
||||
def ecdh(self, pubkey):
|
||||
"""No support for ECDH in KeepKey firmware."""
|
||||
msg = 'KeepKey does not support ECDH'
|
||||
raise interface.NotFoundError(msg)
|
||||
@ -0,0 +1,8 @@
|
||||
"""KeepKey-related definitions."""
|
||||
|
||||
# pylint: disable=unused-import
|
||||
from keepkeylib.client import KeepKeyClient as Client
|
||||
from keepkeylib.client import CallException
|
||||
from keepkeylib.transport_hid import HidTransport
|
||||
from keepkeylib.messages_pb2 import PassphraseAck
|
||||
from keepkeylib.types_pb2 import IdentityType
|
||||
@ -0,0 +1,114 @@
|
||||
"""Ledger-related code (see https://www.ledgerwallet.com/)."""
|
||||
|
||||
import binascii
|
||||
import logging
|
||||
import struct
|
||||
|
||||
from ledgerblue import comm
|
||||
|
||||
from . import interface
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _expand_path(path):
|
||||
"""Convert BIP32 path into bytes."""
|
||||
return b''.join((struct.pack('>I', e) for e in path))
|
||||
|
||||
|
||||
def _convert_public_key(ecdsa_curve_name, result):
|
||||
"""Convert Ledger reply into PublicKey object."""
|
||||
if ecdsa_curve_name == 'nist256p1':
|
||||
if (result[64] & 1) != 0:
|
||||
result = bytearray([0x03]) + result[1:33]
|
||||
else:
|
||||
result = bytearray([0x02]) + result[1:33]
|
||||
else:
|
||||
result = result[1:]
|
||||
keyX = bytearray(result[0:32])
|
||||
keyY = bytearray(result[32:][::-1])
|
||||
if (keyX[31] & 1) != 0:
|
||||
keyY[31] |= 0x80
|
||||
result = b'\x00' + bytes(keyY)
|
||||
return bytes(result)
|
||||
|
||||
|
||||
class LedgerNanoS(interface.Device):
|
||||
"""Connection to Ledger Nano S device."""
|
||||
|
||||
def connect(self):
|
||||
"""Enumerate and connect to the first USB HID interface."""
|
||||
try:
|
||||
return comm.getDongle()
|
||||
except comm.CommException as e:
|
||||
raise interface.NotFoundError(
|
||||
'{} not connected: "{}"'.format(self, e))
|
||||
|
||||
def pubkey(self, ecdh=False):
|
||||
"""Get PublicKey object for specified BIP32 address and elliptic curve."""
|
||||
curve_name = self.get_curve_name(ecdh)
|
||||
path = _expand_path(interface.get_bip32_address(self.identity_dict,
|
||||
ecdh=ecdh))
|
||||
if curve_name == 'nist256p1':
|
||||
p2 = '01'
|
||||
else:
|
||||
p2 = '02'
|
||||
apdu = '800200' + p2
|
||||
apdu = binascii.unhexlify(apdu)
|
||||
apdu += bytearray([len(path) + 1, len(path) // 4])
|
||||
apdu += path
|
||||
result = bytearray(self.conn.exchange(bytes(apdu)))[1:]
|
||||
return _convert_public_key(curve_name, result)
|
||||
|
||||
def sign(self, blob):
|
||||
"""Sign given blob and return the signature (as bytes)."""
|
||||
path = _expand_path(interface.get_bip32_address(self.identity_dict,
|
||||
ecdh=False))
|
||||
if self.identity_dict['proto'] == 'ssh':
|
||||
ins = '04'
|
||||
p1 = '00'
|
||||
else:
|
||||
ins = '08'
|
||||
p1 = '00'
|
||||
if self.curve_name == 'nist256p1':
|
||||
p2 = '81' if self.identity_dict['proto'] == 'ssh' else '01'
|
||||
else:
|
||||
p2 = '82' if self.identity_dict['proto'] == 'ssh' else '02'
|
||||
apdu = '80' + ins + p1 + p2
|
||||
apdu = binascii.unhexlify(apdu)
|
||||
apdu += bytearray([len(blob) + len(path) + 1])
|
||||
apdu += bytearray([len(path) // 4]) + path
|
||||
apdu += blob
|
||||
result = bytearray(self.conn.exchange(bytes(apdu)))
|
||||
if self.curve_name == 'nist256p1':
|
||||
offset = 3
|
||||
length = result[offset]
|
||||
r = result[offset+1:offset+1+length]
|
||||
if r[0] == 0:
|
||||
r = r[1:]
|
||||
offset = offset + 1 + length + 1
|
||||
length = result[offset]
|
||||
s = result[offset+1:offset+1+length]
|
||||
if s[0] == 0:
|
||||
s = s[1:]
|
||||
offset = offset + 1 + length
|
||||
return bytes(r) + bytes(s)
|
||||
else:
|
||||
return bytes(result[:64])
|
||||
|
||||
def ecdh(self, pubkey):
|
||||
"""Get shared session key using Elliptic Curve Diffie-Hellman."""
|
||||
path = _expand_path(interface.get_bip32_address(self.identity_dict,
|
||||
ecdh=True))
|
||||
if self.curve_name == 'nist256p1':
|
||||
p2 = '01'
|
||||
else:
|
||||
p2 = '02'
|
||||
apdu = '800a00' + p2
|
||||
apdu = binascii.unhexlify(apdu)
|
||||
apdu += bytearray([len(pubkey) + len(path) + 1])
|
||||
apdu += bytearray([len(path) // 4]) + path
|
||||
apdu += pubkey
|
||||
result = bytearray(self.conn.exchange(bytes(apdu)))
|
||||
assert result[0] == 0x04
|
||||
return bytes(result)
|
||||
@ -0,0 +1,108 @@
|
||||
"""TREZOR-related code (see http://bitcointrezor.com/)."""
|
||||
|
||||
import binascii
|
||||
import logging
|
||||
import semver
|
||||
|
||||
from . import interface
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Trezor(interface.Device):
|
||||
"""Connection to TREZOR device."""
|
||||
|
||||
from . import trezor_defs as defs
|
||||
|
||||
required_version = '>=1.4.0'
|
||||
|
||||
def connect(self):
|
||||
"""Enumerate and connect to the first USB HID interface."""
|
||||
def empty_passphrase_handler(_):
|
||||
return self.defs.PassphraseAck(passphrase='')
|
||||
|
||||
for d in self.defs.HidTransport.enumerate():
|
||||
log.debug('endpoint: %s', d)
|
||||
transport = self.defs.HidTransport(d)
|
||||
connection = self.defs.Client(transport)
|
||||
connection.callback_PassphraseRequest = empty_passphrase_handler
|
||||
f = connection.features
|
||||
log.debug('connected to %s %s', self, f.device_id)
|
||||
log.debug('label : %s', f.label)
|
||||
log.debug('vendor : %s', f.vendor)
|
||||
current_version = '{}.{}.{}'.format(f.major_version,
|
||||
f.minor_version,
|
||||
f.patch_version)
|
||||
log.debug('version : %s', current_version)
|
||||
log.debug('revision : %s', binascii.hexlify(f.revision))
|
||||
if not semver.match(current_version, self.required_version):
|
||||
fmt = ('Please upgrade your {} firmware to {} version'
|
||||
' (current: {})')
|
||||
raise ValueError(fmt.format(self, self.required_version,
|
||||
current_version))
|
||||
connection.ping(msg='', pin_protection=True) # unlock PIN
|
||||
return connection
|
||||
raise interface.NotFoundError('{} not connected'.format(self))
|
||||
|
||||
def close(self):
|
||||
"""Close connection."""
|
||||
self.conn.close()
|
||||
|
||||
def pubkey(self, ecdh=False):
|
||||
"""Return public key."""
|
||||
curve_name = self.get_curve_name(ecdh=ecdh)
|
||||
log.debug('"%s" getting public key (%s) from %s',
|
||||
interface.identity_to_string(self.identity_dict),
|
||||
curve_name, self)
|
||||
addr = interface.get_bip32_address(self.identity_dict, ecdh=ecdh)
|
||||
result = self.conn.get_public_node(n=addr,
|
||||
ecdsa_curve_name=curve_name)
|
||||
log.debug('result: %s', result)
|
||||
return result.node.public_key
|
||||
|
||||
def _identity_proto(self):
|
||||
result = self.defs.IdentityType()
|
||||
for name, value in self.identity_dict.items():
|
||||
setattr(result, name, value)
|
||||
return result
|
||||
|
||||
def sign(self, blob):
|
||||
"""Sign given blob and return the signature (as bytes)."""
|
||||
curve_name = self.get_curve_name(ecdh=False)
|
||||
log.debug('"%s" signing %r (%s) on %s',
|
||||
interface.identity_to_string(self.identity_dict), blob,
|
||||
curve_name, self)
|
||||
try:
|
||||
result = self.conn.sign_identity(
|
||||
identity=self._identity_proto(),
|
||||
challenge_hidden=blob,
|
||||
challenge_visual='',
|
||||
ecdsa_curve_name=curve_name)
|
||||
log.debug('result: %s', result)
|
||||
assert len(result.signature) == 65
|
||||
assert result.signature[:1] == b'\x00'
|
||||
return result.signature[1:]
|
||||
except self.defs.CallException as e:
|
||||
msg = '{} error: {}'.format(self, e)
|
||||
log.debug(msg, exc_info=True)
|
||||
raise interface.DeviceError(msg)
|
||||
|
||||
def ecdh(self, pubkey):
|
||||
"""Get shared session key using Elliptic Curve Diffie-Hellman."""
|
||||
curve_name = self.get_curve_name(ecdh=True)
|
||||
log.debug('"%s" shared session key (%s) for %r from %s',
|
||||
interface.identity_to_string(self.identity_dict),
|
||||
curve_name, pubkey, self)
|
||||
try:
|
||||
result = self.conn.get_ecdh_session_key(
|
||||
identity=self._identity_proto(),
|
||||
peer_public_key=pubkey,
|
||||
ecdsa_curve_name=curve_name)
|
||||
log.debug('result: %s', result)
|
||||
assert len(result.session_key) in {65, 33} # NIST256 or Curve25519
|
||||
assert result.session_key[:1] == b'\x04'
|
||||
return result.session_key
|
||||
except self.defs.CallException as e:
|
||||
msg = '{} error: {}'.format(self, e)
|
||||
log.debug(msg, exc_info=True)
|
||||
raise interface.DeviceError(msg)
|
||||
@ -0,0 +1,8 @@
|
||||
"""TREZOR-related definitions."""
|
||||
|
||||
# pylint: disable=unused-import
|
||||
from trezorlib.client import TrezorClient as Client
|
||||
from trezorlib.client import CallException
|
||||
from trezorlib.transport_hid import HidTransport
|
||||
from trezorlib.messages_pb2 import PassphraseAck
|
||||
from trezorlib.types_pb2 import IdentityType
|
||||
@ -1,124 +0,0 @@
|
||||
"""Thin wrapper around trezor/keepkey libraries."""
|
||||
from __future__ import absolute_import
|
||||
|
||||
import binascii
|
||||
import collections
|
||||
import logging
|
||||
|
||||
import semver
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
ClientWrapper = collections.namedtuple(
|
||||
'ClientWrapper',
|
||||
['connection', 'identity_type', 'device_name', 'call_exception'])
|
||||
|
||||
|
||||
# pylint: disable=too-many-arguments
|
||||
def _load_client(name, client_type, hid_transport,
|
||||
passphrase_ack, identity_type,
|
||||
required_version, call_exception):
|
||||
|
||||
def empty_passphrase_handler(_):
|
||||
return passphrase_ack(passphrase='')
|
||||
|
||||
for d in hid_transport.enumerate():
|
||||
connection = client_type(hid_transport(d))
|
||||
connection.callback_PassphraseRequest = empty_passphrase_handler
|
||||
f = connection.features
|
||||
log.debug('connected to %s %s', name, f.device_id)
|
||||
log.debug('label : %s', f.label)
|
||||
log.debug('vendor : %s', f.vendor)
|
||||
current_version = '{}.{}.{}'.format(f.major_version,
|
||||
f.minor_version,
|
||||
f.patch_version)
|
||||
log.debug('version : %s', current_version)
|
||||
log.debug('revision : %s', binascii.hexlify(f.revision))
|
||||
if not semver.match(current_version, required_version):
|
||||
fmt = 'Please upgrade your {} firmware to {} version (current: {})'
|
||||
raise ValueError(fmt.format(name,
|
||||
required_version,
|
||||
current_version))
|
||||
yield ClientWrapper(connection=connection,
|
||||
identity_type=identity_type,
|
||||
device_name=name,
|
||||
call_exception=call_exception)
|
||||
return
|
||||
|
||||
|
||||
def _load_trezor():
|
||||
try:
|
||||
from trezorlib.client import TrezorClient, CallException
|
||||
from trezorlib.transport_hid import HidTransport
|
||||
from trezorlib.messages_pb2 import PassphraseAck
|
||||
from trezorlib.types_pb2 import IdentityType
|
||||
return _load_client(name='Trezor',
|
||||
client_type=TrezorClient,
|
||||
hid_transport=HidTransport,
|
||||
passphrase_ack=PassphraseAck,
|
||||
identity_type=IdentityType,
|
||||
required_version='>=1.4.0',
|
||||
call_exception=CallException)
|
||||
except ImportError as e:
|
||||
log.warning('%s: install via "pip install trezor" '
|
||||
'if you need to support this device', e)
|
||||
|
||||
|
||||
def _load_keepkey():
|
||||
try:
|
||||
from keepkeylib.client import KeepKeyClient, CallException
|
||||
from keepkeylib.transport_hid import HidTransport
|
||||
from keepkeylib.messages_pb2 import PassphraseAck
|
||||
from keepkeylib.types_pb2 import IdentityType
|
||||
return _load_client(name='KeepKey',
|
||||
client_type=KeepKeyClient,
|
||||
hid_transport=HidTransport,
|
||||
passphrase_ack=PassphraseAck,
|
||||
identity_type=IdentityType,
|
||||
required_version='>=1.0.4',
|
||||
call_exception=CallException)
|
||||
except ImportError as e:
|
||||
log.warning('%s: install via "pip install keepkey" '
|
||||
'if you need to support this device', e)
|
||||
|
||||
|
||||
def _load_ledger():
|
||||
from ._ledger import LedgerClientConnection, CallException, IdentityType
|
||||
try:
|
||||
from ledgerblue.comm import getDongle, CommException
|
||||
except ImportError as e:
|
||||
log.warning('%s: install via "pip install ledgerblue" '
|
||||
'if you need to support this device', e)
|
||||
return
|
||||
try:
|
||||
dongle = getDongle()
|
||||
except CommException:
|
||||
return
|
||||
|
||||
yield ClientWrapper(connection=LedgerClientConnection(dongle),
|
||||
identity_type=IdentityType,
|
||||
device_name="ledger",
|
||||
call_exception=CallException)
|
||||
|
||||
|
||||
LOADERS = [
|
||||
_load_trezor,
|
||||
_load_keepkey,
|
||||
_load_ledger
|
||||
]
|
||||
|
||||
|
||||
def load(loaders=None):
|
||||
"""Load a single device, via specified loaders' list."""
|
||||
loaders = loaders if loaders is not None else LOADERS
|
||||
device_list = []
|
||||
for loader in loaders:
|
||||
device = loader()
|
||||
if device:
|
||||
device_list.extend(device)
|
||||
|
||||
if len(device_list) == 1:
|
||||
return device_list[0]
|
||||
|
||||
msg = '{:d} devices found'.format(len(device_list))
|
||||
raise IOError(msg)
|
||||
@ -1,97 +0,0 @@
|
||||
import mock
|
||||
import pytest
|
||||
|
||||
from .. import factory
|
||||
|
||||
|
||||
def test_load():
|
||||
|
||||
def single():
|
||||
return [0]
|
||||
|
||||
def nothing():
|
||||
return []
|
||||
|
||||
def double():
|
||||
return [1, 2]
|
||||
|
||||
assert factory.load(loaders=[single]) == 0
|
||||
assert factory.load(loaders=[single, nothing]) == 0
|
||||
assert factory.load(loaders=[nothing, single]) == 0
|
||||
|
||||
with pytest.raises(IOError):
|
||||
factory.load(loaders=[])
|
||||
|
||||
with pytest.raises(IOError):
|
||||
factory.load(loaders=[single, single])
|
||||
|
||||
with pytest.raises(IOError):
|
||||
factory.load(loaders=[double])
|
||||
|
||||
|
||||
def factory_load_client(**kwargs):
|
||||
# pylint: disable=protected-access
|
||||
return list(factory._load_client(**kwargs))
|
||||
|
||||
|
||||
def test_load_nothing():
|
||||
hid_transport = mock.Mock(spec_set=['enumerate'])
|
||||
hid_transport.enumerate.return_value = []
|
||||
result = factory_load_client(
|
||||
name=None,
|
||||
client_type=None,
|
||||
hid_transport=hid_transport,
|
||||
passphrase_ack=None,
|
||||
identity_type=None,
|
||||
required_version=None,
|
||||
call_exception=None)
|
||||
assert result == []
|
||||
|
||||
|
||||
def create_client_type(version):
|
||||
conn = mock.Mock(spec=[])
|
||||
conn.features = mock.Mock(spec=[])
|
||||
major, minor, patch = version.split('.')
|
||||
conn.features.device_id = 'DEVICE_ID'
|
||||
conn.features.label = 'LABEL'
|
||||
conn.features.vendor = 'VENDOR'
|
||||
conn.features.major_version = major
|
||||
conn.features.minor_version = minor
|
||||
conn.features.patch_version = patch
|
||||
conn.features.revision = b'\x12\x34\x56\x78'
|
||||
return mock.Mock(spec_set=[], return_value=conn)
|
||||
|
||||
|
||||
def test_load_single():
|
||||
hid_transport = mock.Mock(spec_set=['enumerate'])
|
||||
hid_transport.enumerate.return_value = [0]
|
||||
for version in ('1.3.4', '1.3.5', '1.4.0', '2.0.0'):
|
||||
passphrase_ack = mock.Mock(spec_set=[])
|
||||
client_type = create_client_type(version)
|
||||
client_wrapper, = factory_load_client(
|
||||
name='DEVICE_NAME',
|
||||
client_type=client_type,
|
||||
hid_transport=hid_transport,
|
||||
passphrase_ack=passphrase_ack,
|
||||
identity_type=None,
|
||||
required_version='>=1.3.4',
|
||||
call_exception=None)
|
||||
assert client_wrapper.connection is client_type.return_value
|
||||
assert client_wrapper.device_name == 'DEVICE_NAME'
|
||||
client_wrapper.connection.callback_PassphraseRequest('MESSAGE')
|
||||
assert passphrase_ack.mock_calls == [mock.call(passphrase='')]
|
||||
|
||||
|
||||
def test_load_old():
|
||||
hid_transport = mock.Mock(spec_set=['enumerate'])
|
||||
hid_transport.enumerate.return_value = [0]
|
||||
for version in ('1.3.3', '1.2.5', '1.1.0', '0.9.9'):
|
||||
with pytest.raises(ValueError):
|
||||
factory_load_client(
|
||||
name='DEVICE_NAME',
|
||||
client_type=create_client_type(version),
|
||||
hid_transport=hid_transport,
|
||||
passphrase_ack=None,
|
||||
identity_type=None,
|
||||
required_version='>=1.3.4',
|
||||
call_exception=None)
|
||||
Loading…
Reference in New Issue