1
0
mirror of https://github.com/trezor/trezor-firmware.git synced 2024-12-28 00:58:09 +00:00
trezor-firmware/trezorlib/client.py

1002 lines
36 KiB
Python
Raw Normal View History

2016-11-25 21:53:55 +00:00
# This file is part of the TREZOR project.
#
# Copyright (C) 2012-2016 Marek Palatinus <slush@satoshilabs.com>
# Copyright (C) 2012-2016 Pavol Rusnak <stick@satoshilabs.com>
# Copyright (C) 2016 Jochen Hoenicke <hoenicke@gmail.com>
#
# This library is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this library. If not, see <http://www.gnu.org/licenses/>.
2016-09-27 20:49:51 +00:00
from __future__ import print_function, absolute_import
2016-05-20 20:27:20 +00:00
import os
import sys
2014-04-09 19:42:30 +00:00
import time
2014-01-06 00:54:53 +00:00
import binascii
2014-01-08 14:59:18 +00:00
import hashlib
import unicodedata
import json
import getpass
2014-02-21 00:48:11 +00:00
from mnemonic import Mnemonic
from . import tools
from . import mapping
from . import messages_pb2 as proto
from . import types_pb2 as types
from .debuglink import DebugLink
# try:
# from PIL import Image
# SCREENSHOT = True
# except:
# SCREENSHOT = False
SCREENSHOT = False
DEFAULT_CURVE = 'secp256k1'
# monkeypatching: text formatting of protobuf messages
tools.monkeypatch_google_protobuf_text_format()
def getch():
try:
import termios
except ImportError:
# Non-POSIX. Return msvcrt's (Windows') getch.
import msvcrt
return msvcrt.getch()
# POSIX system. Create and return a getch that manipulates the tty.
import sys, tty
def _getch():
fd = sys.stdin.fileno()
old_settings = termios.tcgetattr(fd)
try:
tty.setraw(fd)
ch = sys.stdin.read(1)
finally:
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
return ch
return _getch()
2014-02-13 15:46:21 +00:00
def get_buttonrequest_value(code):
# Converts integer code to its string representation of ButtonRequestType
return [ k for k, v in types.ButtonRequestType.items() if v == code][0]
2014-02-02 17:27:44 +00:00
def pprint(msg):
msg_class = msg.__class__.__name__
msg_size = msg.ByteSize()
"""
msg_ser = msg.SerializeToString()
msg_id = mapping.get_type(msg)
msg_json = json.dumps(protobuf_json.pb2json(msg))
"""
2014-02-21 18:13:14 +00:00
if isinstance(msg, proto.FirmwareUpload):
return "<%s> (%d bytes):\n" % (msg_class, msg_size)
2014-02-21 18:13:14 +00:00
else:
2016-05-20 11:55:43 +00:00
return "<%s> (%d bytes):\n%s" % (msg_class, msg_size, msg)
def log(msg):
2017-06-20 14:36:40 +00:00
sys.stderr.write(str(msg))
sys.stderr.write('\n')
sys.stderr.flush()
class CallException(Exception):
def __init__(self, code, message):
super(CallException, self).__init__()
self.args = [code, message]
class PinException(CallException):
pass
2014-02-13 15:46:21 +00:00
class field(object):
# Decorator extracts single value from
# protobuf object. If the field is not
# present, raises an exception.
def __init__(self, field):
self.field = field
def __call__(self, f):
def wrapped_f(*args, **kwargs):
ret = f(*args, **kwargs)
ret.HasField(self.field)
return getattr(ret, self.field)
return wrapped_f
class expect(object):
# Decorator checks if the method
# returned one of expected protobuf messages
# or raises an exception
def __init__(self, *expected):
self.expected = expected
2016-01-12 23:17:38 +00:00
2014-02-13 15:46:21 +00:00
def __call__(self, f):
def wrapped_f(*args, **kwargs):
ret = f(*args, **kwargs)
if not isinstance(ret, self.expected):
raise Exception("Got %s, expected %s" % (ret.__class__, self.expected))
return ret
return wrapped_f
def session(f):
# Decorator wraps a BaseClient method
# with session activation / deactivation
def wrapped_f(*args, **kwargs):
client = args[0]
try:
client.transport.session_begin()
return f(*args, **kwargs)
finally:
client.transport.session_end()
return wrapped_f
def normalize_nfc(txt):
2016-05-20 17:18:33 +00:00
if sys.version_info[0] < 3:
if isinstance(txt, unicode):
return unicodedata.normalize('NFC', txt)
2016-05-20 17:18:33 +00:00
if isinstance(txt, str):
return unicodedata.normalize('NFC', txt.decode('utf-8'))
else:
2016-05-20 17:18:33 +00:00
if isinstance(txt, bytes):
return unicodedata.normalize('NFC', txt.decode('utf-8'))
2016-05-20 17:18:33 +00:00
if isinstance(txt, str):
return unicodedata.normalize('NFC', txt)
2016-05-20 17:18:33 +00:00
raise Exception('unicode/str or bytes/str expected')
2014-02-13 15:46:21 +00:00
class BaseClient(object):
# Implements very basic layer of sending raw protobuf
# messages to device and getting its response back.
def __init__(self, transport, **kwargs):
2014-02-13 15:46:21 +00:00
self.transport = transport
super(BaseClient, self).__init__() # *args, **kwargs)
2016-02-10 15:46:58 +00:00
def cancel(self):
self.transport.write(proto.Cancel())
@session
def call_raw(self, msg):
self.transport.write(msg)
return self.transport.read_blocking()
@session
def call(self, msg):
resp = self.call_raw(msg)
handler_name = "callback_%s" % resp.__class__.__name__
handler = getattr(self, handler_name, None)
if handler != None:
msg = handler(resp)
if msg == None:
raise Exception("Callback %s must return protobuf message, not None" % handler)
resp = self.call(msg)
2014-02-13 15:46:21 +00:00
return resp
def callback_Failure(self, msg):
if msg.code in (types.Failure_PinInvalid,
types.Failure_PinCancelled, types.Failure_PinExpected):
raise PinException(msg.code, msg.message)
raise CallException(msg.code, msg.message)
def close(self):
self.transport.close()
class DebugWireMixin(object):
def call_raw(self, msg):
log("SENDING " + pprint(msg))
resp = super(DebugWireMixin, self).call_raw(msg)
log("RECEIVED " + pprint(resp))
return resp
2014-02-13 15:46:21 +00:00
class TextUIMixin(object):
# This class demonstrates easy test-based UI
# integration between the device and wallet.
# You can implement similar functionality
# by implementing your own GuiMixin with
# graphical widgets for every type of these callbacks.
def __init__(self, *args, **kwargs):
super(TextUIMixin, self).__init__(*args, **kwargs)
2014-02-13 15:46:21 +00:00
def callback_ButtonRequest(self, msg):
# log("Sending ButtonAck for %s " % get_buttonrequest_value(msg.code))
2014-02-13 15:46:21 +00:00
return proto.ButtonAck()
def callback_RecoveryMatrix(self, msg):
if self.recovery_matrix_first_pass:
self.recovery_matrix_first_pass = False
log("Use the numeric keypad to describe positions. For the word list use only left and right keys. The layout is:")
log(" 7 8 9 7 | 9")
log(" 4 5 6 4 | 6")
log(" 1 2 3 1 | 3")
while True:
character = getch()
if character in ('\x03', '\x04'):
return proto.Cancel()
if character in ('\x08', '\x7f'):
return proto.WordAck(word='\x08')
# ignore middle column if only 6 keys requested.
if (msg.type == types.WordRequestType_Matrix6 and
character in ('2', '5', '8')):
continue
if (ord(character) >= ord('1') and ord(character) <= ord('9')):
return proto.WordAck(word=character)
2014-02-13 15:46:21 +00:00
def callback_PinMatrixRequest(self, msg):
2014-03-28 15:26:48 +00:00
if msg.type == 1:
2015-02-28 13:06:23 +00:00
desc = 'current PIN'
2014-03-28 15:26:48 +00:00
elif msg.type == 2:
2014-04-02 18:06:54 +00:00
desc = 'new PIN'
2014-03-28 15:26:48 +00:00
elif msg.type == 3:
2014-04-02 18:06:54 +00:00
desc = 'new PIN again'
2014-03-28 15:26:48 +00:00
else:
2014-04-02 18:06:54 +00:00
desc = 'PIN'
2015-08-21 13:16:27 +00:00
log("Use the numeric keypad to describe number positions. The layout is:")
log(" 7 8 9")
log(" 4 5 6")
log(" 1 2 3")
log("Please enter %s: " % desc)
pin = getpass.getpass('')
2014-02-13 15:46:21 +00:00
return proto.PinMatrixAck(pin=pin)
def callback_PassphraseRequest(self, msg):
log("Passphrase required: ")
passphrase = getpass.getpass('')
log("Confirm your Passphrase: ")
2014-11-23 12:28:09 +00:00
if passphrase == getpass.getpass(''):
2016-05-20 17:18:33 +00:00
passphrase = normalize_nfc(passphrase)
return proto.PassphraseAck(passphrase=passphrase)
else:
log("Passphrase did not match! ")
exit()
2014-02-13 15:46:21 +00:00
def callback_WordRequest(self, msg):
if msg.type in (types.WordRequestType_Matrix9,
types.WordRequestType_Matrix6):
return self.callback_RecoveryMatrix(msg)
log("Enter one word of mnemonic: ")
2016-05-20 11:36:17 +00:00
try:
word = raw_input()
except NameError:
word = input() # Python 3
if self.expand:
word = self.mnemonic_wordlist.expand_word(word)
2014-02-13 15:46:21 +00:00
return proto.WordAck(word=word)
2014-02-13 15:46:21 +00:00
class DebugLinkMixin(object):
# This class implements automatic responses
# and other functionality for unit tests
# for various callbacks, created in order
# to automatically pass unit tests.
#
# This mixing should be used only for purposes
# of unit testing, because it will fail to work
# without special DebugLink interface provided
# by the device.
def __init__(self, *args, **kwargs):
super(DebugLinkMixin, self).__init__(*args, **kwargs)
self.debug = None
self.in_with_statement = 0
2014-12-10 14:26:18 +00:00
self.button_wait = 0
self.screenshot_id = 0
2014-02-13 15:46:21 +00:00
# Always press Yes and provide correct pin
self.setup_debuglink(True, True)
2016-01-12 23:17:38 +00:00
# Do not expect any specific response from device
self.expected_responses = None
# Use blank passphrase
self.set_passphrase('')
2014-02-13 15:46:21 +00:00
def close(self):
super(DebugLinkMixin, self).close()
if self.debug:
self.debug.close()
def set_debuglink(self, debug_transport):
self.debug = DebugLink(debug_transport)
2014-12-10 14:26:18 +00:00
def set_buttonwait(self, secs):
self.button_wait = secs
def __enter__(self):
# For usage in with/expected_responses
self.in_with_statement += 1
return self
2014-02-21 20:00:56 +00:00
def __exit__(self, _type, value, traceback):
self.in_with_statement -= 1
2014-02-21 20:00:56 +00:00
if _type != None:
# Another exception raised
return False
# return isinstance(value, TypeError)
# Evaluate missed responses in 'with' statement
if self.expected_responses != None and len(self.expected_responses):
raise Exception("Some of expected responses didn't come from device: %s" % \
[ pprint(x) for x in self.expected_responses ])
# Cleanup
self.expected_responses = None
return False
def set_expected_responses(self, expected):
if not self.in_with_statement:
raise Exception("Must be called inside 'with' statement")
self.expected_responses = expected
2014-02-13 15:46:21 +00:00
def setup_debuglink(self, button, pin_correct):
self.button = button # True -> YES button, False -> NO button
self.pin_correct = pin_correct
def set_passphrase(self, passphrase):
2016-05-20 17:18:33 +00:00
self.passphrase = normalize_nfc(passphrase)
2014-02-21 20:00:56 +00:00
def set_mnemonic(self, mnemonic):
self.mnemonic = normalize_nfc(mnemonic).split(' ')
def call_raw(self, msg):
if SCREENSHOT and self.debug:
layout = self.debug.read_layout()
im = Image.new("RGB", (128, 64))
pix = im.load()
for x in range(128):
for y in range(64):
rx, ry = 127 - x, 63 - y
if (ord(layout[rx + (ry / 8) * 128]) & (1 << (ry % 8))) > 0:
pix[x, y] = (255, 255, 255)
im.save('scr%05d.png' % self.screenshot_id)
self.screenshot_id += 1
resp = super(DebugLinkMixin, self).call_raw(msg)
self._check_request(resp)
return resp
2016-01-12 23:17:38 +00:00
def _check_request(self, msg):
if self.expected_responses != None:
2014-02-13 15:46:21 +00:00
try:
expected = self.expected_responses.pop(0)
2014-02-13 15:46:21 +00:00
except IndexError:
raise CallException(types.Failure_UnexpectedMessage,
"Got %s, but no message has been expected" % pprint(msg))
2014-02-13 15:46:21 +00:00
if msg.__class__ != expected.__class__:
raise CallException(types.Failure_UnexpectedMessage,
"Expected %s, got %s" % (pprint(expected), pprint(msg)))
fields = expected.ListFields() # only filled (including extensions)
for field, value in fields:
if not msg.HasField(field.name) or getattr(msg, field.name) != value:
raise CallException(types.Failure_UnexpectedMessage,
"Expected %s, got %s" % (pprint(expected), pprint(msg)))
2016-01-12 23:17:38 +00:00
def callback_ButtonRequest(self, msg):
log("ButtonRequest code: " + get_buttonrequest_value(msg.code))
2014-02-13 15:46:21 +00:00
2014-06-13 17:24:53 +00:00
log("Pressing button " + str(self.button))
2014-12-10 14:26:18 +00:00
if self.button_wait:
log("Waiting %d seconds " % self.button_wait)
time.sleep(self.button_wait)
2014-02-13 15:46:21 +00:00
self.debug.press_button(self.button)
return proto.ButtonAck()
def callback_PinMatrixRequest(self, msg):
if self.pin_correct:
pin = self.debug.read_pin_encoded()
else:
2014-02-13 15:46:21 +00:00
pin = '444222'
return proto.PinMatrixAck(pin=pin)
def callback_PassphraseRequest(self, msg):
log("Provided passphrase: '%s'" % self.passphrase)
return proto.PassphraseAck(passphrase=self.passphrase)
2014-02-13 15:46:21 +00:00
def callback_WordRequest(self, msg):
(word, pos) = self.debug.read_recovery_word()
2014-02-21 20:00:56 +00:00
if word != '':
return proto.WordAck(word=word)
if pos != 0:
return proto.WordAck(word=self.mnemonic[pos - 1])
raise Exception("Unexpected call")
2014-02-13 15:46:21 +00:00
class ProtocolMixin(object):
PRIME_DERIVATION_FLAG = 0x80000000
2017-01-17 13:13:02 +00:00
VENDORS = ('bitcointrezor.com', 'trezor.io')
2014-02-13 15:46:21 +00:00
def __init__(self, *args, **kwargs):
super(ProtocolMixin, self).__init__(*args, **kwargs)
self.init_device()
2014-03-28 18:47:53 +00:00
self.tx_api = None
2014-02-13 15:46:21 +00:00
2014-03-28 18:47:53 +00:00
def set_tx_api(self, tx_api):
self.tx_api = tx_api
2014-02-13 15:46:21 +00:00
def init_device(self):
self.features = expect(proto.Features)(self.call)(proto.Initialize())
2015-01-28 04:31:30 +00:00
if str(self.features.vendor) not in self.VENDORS:
raise Exception("Unsupported device")
def _get_local_entropy(self):
return os.urandom(32)
def _convert_prime(self, n):
# Convert minus signs to uint32 with flag
2014-02-13 15:46:21 +00:00
return [ int(abs(x) | self.PRIME_DERIVATION_FLAG) if x < 0 else x for x in n ]
2016-02-10 14:53:14 +00:00
@staticmethod
def expand_path(n):
# Convert string of bip32 path to list of uint32 integers with prime flags
# 0/-1/1' -> [0, 0x80000001, 0x80000001]
2014-01-09 23:11:03 +00:00
if not n:
return []
n = n.split('/')
# m/a/b/c => a/b/c
if n[0] == 'm':
n = n[1:]
# coin_name/a/b/c => 44'/SLIP44_constant'/a/b/c
2017-03-04 15:20:38 +00:00
coins = {
"Bitcoin": 0,
"Testnet": 1,
"Namecoin": 7,
"Litecoin": 2,
"Dogecoin": 3,
"Dash": 5,
"Ether": 60,
"EtherClassic": 61,
"Zcash": 133,
2017-04-21 17:14:04 +00:00
"Decred": 42
2017-03-04 15:20:38 +00:00
}
if n[0] in coins:
n = ["44'", "%d'" % coins[n[0]] ] + n[1:]
path = []
for x in n:
prime = False
2014-01-09 23:11:03 +00:00
if x.endswith("'"):
x = x.replace('\'', '')
prime = True
2014-01-09 23:11:03 +00:00
if x.startswith('-'):
prime = True
2014-01-09 23:11:03 +00:00
x = abs(int(x))
if prime:
2016-02-10 14:53:14 +00:00
x |= ProtocolMixin.PRIME_DERIVATION_FLAG
2014-01-09 23:11:03 +00:00
path.append(x)
return path
2014-02-13 15:46:21 +00:00
@expect(proto.PublicKey)
def get_public_node(self, n, ecdsa_curve_name=DEFAULT_CURVE, show_display=False, coin_name=None):
n = self._convert_prime(n)
if not ecdsa_curve_name:
ecdsa_curve_name=DEFAULT_CURVE
return self.call(proto.GetPublicKey(address_n=n, ecdsa_curve_name=ecdsa_curve_name, show_display=show_display, coin_name=coin_name))
2014-02-13 15:46:21 +00:00
@field('address')
@expect(proto.Address)
2017-01-02 20:25:07 +00:00
def get_address(self, coin_name, n, show_display=False, multisig=None, script_type=types.SPENDADDRESS):
n = self._convert_prime(n)
if multisig:
2017-01-02 20:25:07 +00:00
return self.call(proto.GetAddress(address_n=n, coin_name=coin_name, show_display=show_display, multisig=multisig, script_type=script_type))
else:
2017-01-02 20:25:07 +00:00
return self.call(proto.GetAddress(address_n=n, coin_name=coin_name, show_display=show_display, script_type=script_type))
@field('address')
@expect(proto.EthereumAddress)
def ethereum_get_address(self, n, show_display=False, multisig=None):
n = self._convert_prime(n)
return self.call(proto.EthereumGetAddress(address_n=n, show_display=show_display))
@session
def ethereum_sign_tx(self, n, nonce, gas_price, gas_limit, to, value, data=None, chain_id=None):
def int_to_big_endian(value):
import rlp.utils
if value == 0:
return b''
return rlp.utils.int_to_big_endian(value)
2016-05-27 06:13:23 +00:00
n = self._convert_prime(n)
msg = proto.EthereumSignTx(
address_n=n,
nonce=int_to_big_endian(nonce),
gas_price=int_to_big_endian(gas_price),
gas_limit=int_to_big_endian(gas_limit),
value=int_to_big_endian(value))
2016-08-10 16:13:05 +00:00
if to:
msg.to = to
if data:
msg.data_length = len(data)
data, chunk = data[1024:], data[:1024]
msg.data_initial_chunk = chunk
2016-08-10 16:13:05 +00:00
if chain_id:
msg.chain_id = chain_id
response = self.call(msg)
2016-05-27 06:13:23 +00:00
while response.HasField('data_length'):
data_length = response.data_length
data, chunk = data[data_length:], data[:data_length]
response = self.call(proto.EthereumTxAck(data_chunk=chunk))
2016-05-27 06:13:23 +00:00
return response.signature_v, response.signature_r, response.signature_s
2016-08-10 16:13:05 +00:00
2016-05-27 06:13:23 +00:00
2014-02-13 15:46:21 +00:00
@field('entropy')
@expect(proto.Entropy)
def get_entropy(self, size):
2014-02-13 15:46:21 +00:00
return self.call(proto.GetEntropy(size=size))
2014-02-13 15:46:21 +00:00
@field('message')
@expect(proto.Success)
def ping(self, msg, button_protection=False, pin_protection=False, passphrase_protection=False):
2014-02-03 23:31:44 +00:00
msg = proto.Ping(message=msg,
2014-02-04 00:02:16 +00:00
button_protection=button_protection,
pin_protection=pin_protection,
passphrase_protection=passphrase_protection)
2014-02-13 15:46:21 +00:00
return self.call(msg)
2013-10-08 18:33:39 +00:00
def get_device_id(self):
return self.features.device_id
2014-02-13 15:46:21 +00:00
@field('message')
@expect(proto.Success)
2015-02-04 19:53:22 +00:00
def apply_settings(self, label=None, language=None, use_passphrase=None, homescreen=None):
settings = proto.ApplySettings()
2014-01-09 23:11:03 +00:00
if label != None:
settings.label = label
if language:
settings.language = language
2014-12-13 18:07:30 +00:00
if use_passphrase != None:
settings.use_passphrase = use_passphrase
2015-02-04 19:53:22 +00:00
if homescreen != None:
settings.homescreen = homescreen
2014-02-13 15:46:21 +00:00
out = self.call(settings)
self.init_device() # Reload Features
return out
@field('message')
@expect(proto.Success)
def clear_session(self):
return self.call(proto.ClearSession())
2014-02-13 15:46:21 +00:00
@field('message')
@expect(proto.Success)
2014-01-31 18:48:19 +00:00
def change_pin(self, remove=False):
ret = self.call(proto.ChangePin(remove=remove))
self.init_device() # Re-read features
return ret
2014-02-13 15:46:21 +00:00
@expect(proto.MessageSignature)
def sign_message(self, coin_name, n, message):
2014-02-13 15:46:21 +00:00
n = self._convert_prime(n)
2016-05-20 17:18:33 +00:00
# Convert message to UTF8 NFC (seems to be a bitcoin-qt standard)
message = normalize_nfc(message).encode("utf-8")
return self.call(proto.SignMessage(coin_name=coin_name, address_n=n, message=message))
2014-02-13 15:46:21 +00:00
2015-02-20 17:50:53 +00:00
@expect(proto.SignedIdentity)
def sign_identity(self, identity, challenge_hidden, challenge_visual, ecdsa_curve_name=DEFAULT_CURVE):
return self.call(proto.SignIdentity(identity=identity, challenge_hidden=challenge_hidden, challenge_visual=challenge_visual, ecdsa_curve_name=ecdsa_curve_name))
2015-02-20 17:50:53 +00:00
2016-06-12 12:39:04 +00:00
@expect(proto.ECDHSessionKey)
def get_ecdh_session_key(self, identity, peer_public_key, ecdsa_curve_name=DEFAULT_CURVE):
return self.call(proto.GetECDHSessionKey(identity=identity, peer_public_key=peer_public_key, ecdsa_curve_name=ecdsa_curve_name))
2016-06-12 21:49:52 +00:00
@field('message')
@expect(proto.Success)
def set_u2f_counter(self, u2f_counter):
ret = self.call(proto.SetU2FCounter(u2f_counter = u2f_counter))
return ret
2016-10-10 09:01:40 +00:00
def verify_message(self, coin_name, address, signature, message):
2016-05-20 17:18:33 +00:00
# Convert message to UTF8 NFC (seems to be a bitcoin-qt standard)
message = normalize_nfc(message).encode("utf-8")
2014-02-13 15:46:21 +00:00
try:
2016-10-10 09:01:40 +00:00
resp = self.call(proto.VerifyMessage(address=address, signature=signature, message=message, coin_name=coin_name))
2014-02-13 15:46:21 +00:00
except CallException as e:
resp = e
if isinstance(resp, proto.Success):
return True
return False
2014-11-03 18:42:22 +00:00
@expect(proto.EncryptedMessage)
2014-10-29 23:48:06 +00:00
def encrypt_message(self, pubkey, message, display_only, coin_name, n):
if coin_name and n:
n = self._convert_prime(n)
return self.call(proto.EncryptMessage(pubkey=pubkey, message=message, display_only=display_only, coin_name=coin_name, address_n=n))
else:
return self.call(proto.EncryptMessage(pubkey=pubkey, message=message, display_only=display_only))
2014-06-12 15:02:46 +00:00
2014-11-03 18:42:22 +00:00
@expect(proto.DecryptedMessage)
def decrypt_message(self, n, nonce, message, msg_hmac):
2014-06-12 15:02:46 +00:00
n = self._convert_prime(n)
2014-11-03 18:42:22 +00:00
return self.call(proto.DecryptMessage(address_n=n, nonce=nonce, message=message, hmac=msg_hmac))
2014-06-12 15:02:46 +00:00
2014-11-03 18:42:22 +00:00
@field('value')
@expect(proto.CipheredKeyValue)
def encrypt_keyvalue(self, n, key, value, ask_on_encrypt=True, ask_on_decrypt=True, iv=b''):
n = self._convert_prime(n)
return self.call(proto.CipherKeyValue(address_n=n,
key=key,
value=value,
encrypt=True,
ask_on_encrypt=ask_on_encrypt,
ask_on_decrypt=ask_on_decrypt,
iv=iv))
2014-11-03 18:42:22 +00:00
@field('value')
@expect(proto.CipheredKeyValue)
def decrypt_keyvalue(self, n, key, value, ask_on_encrypt=True, ask_on_decrypt=True, iv=b''):
n = self._convert_prime(n)
return self.call(proto.CipherKeyValue(address_n=n,
key=key,
value=value,
encrypt=False,
ask_on_encrypt=ask_on_encrypt,
ask_on_decrypt=ask_on_decrypt,
iv=iv))
2014-02-13 15:46:21 +00:00
@field('tx_size')
@expect(proto.TxSize)
def estimate_tx_size(self, coin_name, inputs, outputs):
msg = proto.EstimateTxSize()
msg.coin_name = coin_name
msg.inputs_count = len(inputs)
msg.outputs_count = len(outputs)
return self.call(msg)
def _prepare_simple_sign_tx(self, coin_name, inputs, outputs):
msg = proto.SimpleSignTx()
msg.coin_name = coin_name
msg.inputs.extend(inputs)
msg.outputs.extend(outputs)
known_hashes = []
for inp in inputs:
if inp.prev_hash in known_hashes:
continue
tx = msg.transactions.add()
2015-06-03 12:53:53 +00:00
if self.tx_api:
2017-06-18 22:00:26 +00:00
tx.CopyFrom(self.tx_api.get_tx(binascii.hexlify(inp.prev_hash).decode('utf-8')))
2015-06-03 12:53:53 +00:00
else:
raise Exception('TX_API not defined')
2014-02-13 15:46:21 +00:00
known_hashes.append(inp.prev_hash)
return msg
def simple_sign_tx(self, coin_name, inputs, outputs):
msg = self._prepare_simple_sign_tx(coin_name, inputs, outputs)
return self.call(msg).serialized.serialized_tx
2014-04-09 19:42:30 +00:00
def _prepare_sign_tx(self, coin_name, inputs, outputs):
tx = types.TransactionType()
tx.inputs.extend(inputs)
2014-04-17 06:30:34 +00:00
tx.outputs.extend(outputs)
2014-04-09 19:42:30 +00:00
txes = {}
txes[b''] = tx
2014-04-09 19:42:30 +00:00
known_hashes = []
for inp in inputs:
if inp.prev_hash in known_hashes:
continue
2015-06-03 12:53:53 +00:00
if self.tx_api:
2017-06-18 22:00:26 +00:00
txes[inp.prev_hash] = self.tx_api.get_tx(binascii.hexlify(inp.prev_hash).decode('utf-8'))
2015-06-03 12:53:53 +00:00
else:
raise Exception('TX_API not defined')
2014-04-09 19:42:30 +00:00
known_hashes.append(inp.prev_hash)
return txes
2014-02-13 15:46:21 +00:00
@session
def sign_tx(self, coin_name, inputs, outputs, version=None, lock_time=None, debug_processor=None):
2014-04-09 19:42:30 +00:00
start = time.time()
txes = self._prepare_sign_tx(coin_name, inputs, outputs)
# Prepare and send initial message
tx = proto.SignTx()
tx.inputs_count = len(inputs)
tx.outputs_count = len(outputs)
tx.coin_name = coin_name
if version is not None:
tx.version = version
if lock_time is not None:
tx.lock_time = lock_time
res = self.call(tx)
# Prepare structure for signatures
signatures = [None] * len(inputs)
serialized_tx = b''
counter = 0
while True:
counter += 1
if isinstance(res, proto.Failure):
raise CallException("Signing failed")
if not isinstance(res, proto.TxRequest):
raise CallException("Unexpected message")
# If there's some part of signed transaction, let's add it
if res.HasField('serialized') and res.serialized.HasField('serialized_tx'):
log("RECEIVED PART OF SERIALIZED TX (%d BYTES)" % len(res.serialized.serialized_tx))
serialized_tx += res.serialized.serialized_tx
if res.HasField('serialized') and res.serialized.HasField('signature_index'):
if signatures[res.serialized.signature_index] != None:
raise Exception("Signature for index %d already filled" % res.serialized.signature_index)
signatures[res.serialized.signature_index] = res.serialized.signature
if res.request_type == types.TXFINISHED:
# Device didn't ask for more information, finish workflow
break
# Device asked for one more information, let's process it.
current_tx = txes[res.details.tx_hash]
if res.request_type == types.TXMETA:
msg = types.TransactionType()
msg.version = current_tx.version
msg.lock_time = current_tx.lock_time
msg.inputs_cnt = len(current_tx.inputs)
if res.details.tx_hash:
msg.outputs_cnt = len(current_tx.bin_outputs)
else:
msg.outputs_cnt = len(current_tx.outputs)
2016-10-21 13:24:30 +00:00
msg.extra_data_len = len(current_tx.extra_data)
res = self.call(proto.TxAck(tx=msg))
continue
2014-04-09 19:42:30 +00:00
elif res.request_type == types.TXINPUT:
msg = types.TransactionType()
msg.inputs.extend([current_tx.inputs[res.details.request_index], ])
res = self.call(proto.TxAck(tx=msg))
continue
elif res.request_type == types.TXOUTPUT:
msg = types.TransactionType()
if res.details.tx_hash:
msg.bin_outputs.extend([current_tx.bin_outputs[res.details.request_index], ])
else:
msg.outputs.extend([current_tx.outputs[res.details.request_index], ])
if debug_processor != None:
# If debug_processor function is provided,
# pass thru it the request and prepared response.
# This is useful for unit tests, see test_msg_signtx
msg = debug_processor(res, msg)
res = self.call(proto.TxAck(tx=msg))
continue
2014-04-09 19:42:30 +00:00
2016-10-21 13:24:30 +00:00
elif res.request_type == types.TXEXTRADATA:
o, l = res.details.extra_data_offset, res.details.extra_data_len
msg = types.TransactionType()
msg.extra_data = current_tx.extra_data[o:o + l]
res = self.call(proto.TxAck(tx=msg))
continue
if None in signatures:
raise Exception("Some signatures are missing!")
log("SIGNED IN %.03f SECONDS, CALLED %d MESSAGES, %d BYTES" % \
(time.time() - start, counter, len(serialized_tx)))
2014-04-09 19:42:30 +00:00
return (signatures, serialized_tx)
2014-02-13 15:46:21 +00:00
@field('message')
@expect(proto.Success)
def wipe_device(self):
ret = self.call(proto.WipeDevice())
self.init_device()
return ret
@field('message')
@expect(proto.Success)
def recovery_device(self, word_count, passphrase_protection, pin_protection, label, language, type=types.RecoveryDeviceType_ScrambledWords, expand=False):
2014-02-13 15:46:21 +00:00
if self.features.initialized:
raise Exception("Device is initialized already. Call wipe_device() and try again.")
if word_count not in (12, 18, 24):
raise Exception("Invalid word count. Use 12/18/24")
self.recovery_matrix_first_pass = True
self.expand = expand
if self.expand:
# optimization to load the wordlist once, instead of for each recovery word
self.mnemonic_wordlist = Mnemonic('english')
2014-02-13 15:46:21 +00:00
res = self.call(proto.RecoveryDevice(word_count=int(word_count),
passphrase_protection=bool(passphrase_protection),
pin_protection=bool(pin_protection),
label=label,
language=language,
enforce_wordlist=True,
type=type))
2014-02-13 15:46:21 +00:00
self.init_device()
return res
@field('message')
@expect(proto.Success)
@session
2014-02-13 15:46:21 +00:00
def reset_device(self, display_random, strength, passphrase_protection, pin_protection, label, language):
if self.features.initialized:
raise Exception("Device is initialized already. Call wipe_device() and try again.")
# Begin with device reset workflow
msg = proto.ResetDevice(display_random=display_random,
strength=strength,
language=language,
passphrase_protection=bool(passphrase_protection),
pin_protection=bool(pin_protection),
label=label)
resp = self.call(msg)
if not isinstance(resp, proto.EntropyRequest):
raise Exception("Invalid response, expected EntropyRequest")
external_entropy = self._get_local_entropy()
log("Computer generated entropy: " + binascii.hexlify(external_entropy).decode('ascii'))
2014-02-21 20:00:56 +00:00
ret = self.call(proto.EntropyAck(entropy=external_entropy))
self.init_device()
return ret
2014-02-13 15:46:21 +00:00
@field('message')
@expect(proto.Success)
def load_device_by_mnemonic(self, mnemonic, pin, passphrase_protection, label, language, skip_checksum=False, expand=False):
2014-02-21 00:48:11 +00:00
# Convert mnemonic to UTF8 NKFD
mnemonic = Mnemonic.normalize_string(mnemonic)
# Convert mnemonic to ASCII stream
2016-05-20 17:18:33 +00:00
mnemonic = normalize_nfc(mnemonic)
2014-02-21 00:48:11 +00:00
m = Mnemonic('english')
if expand:
mnemonic = m.expand(mnemonic)
if not skip_checksum and not m.check(mnemonic):
raise Exception("Invalid mnemonic checksum")
2014-02-13 15:46:21 +00:00
if self.features.initialized:
raise Exception("Device is initialized already. Call wipe_device() and try again.")
resp = self.call(proto.LoadDevice(mnemonic=mnemonic, pin=pin,
passphrase_protection=passphrase_protection,
language=language,
2014-02-21 00:48:11 +00:00
label=label,
skip_checksum=skip_checksum))
2014-02-13 15:46:21 +00:00
self.init_device()
return resp
@field('message')
@expect(proto.Success)
2014-03-07 20:44:06 +00:00
def load_device_by_xprv(self, xprv, pin, passphrase_protection, label, language):
2014-02-13 15:46:21 +00:00
if self.features.initialized:
raise Exception("Device is initialized already. Call wipe_device() and try again.")
if xprv[0:4] not in ('xprv', 'tprv'):
raise Exception("Unknown type of xprv")
if len(xprv) < 100 and len(xprv) > 112:
raise Exception("Invalid length of xprv")
node = types.HDNodeType()
2016-05-20 17:18:33 +00:00
data = binascii.hexlify(tools.b58decode(xprv, None))
2014-02-13 15:46:21 +00:00
if data[90:92] != b'00':
2014-02-13 15:46:21 +00:00
raise Exception("Contain invalid private key")
checksum = binascii.hexlify(hashlib.sha256(hashlib.sha256(binascii.unhexlify(data[:156])).digest()).digest()[:4])
2014-02-13 15:46:21 +00:00
if checksum != data[156:]:
raise Exception("Checksum doesn't match")
# version 0488ade4
# depth 00
# fingerprint 00000000
# child_num 00000000
# chaincode 873dff81c02f525623fd1fe5167eac3a55a049de3d314bb42ee227ffed37d508
# privkey 00e8f32e723decf4051aefac8e2c93c9c5b214313817cdb01a1494b917c8436b35
# checksum e77e9d71
node.depth = int(data[8:10], 16)
node.fingerprint = int(data[10:18], 16)
node.child_num = int(data[18:26], 16)
node.chain_code = binascii.unhexlify(data[26:90])
node.private_key = binascii.unhexlify(data[92:156]) # skip 0x00 indicating privkey
2014-02-13 15:46:21 +00:00
resp = self.call(proto.LoadDevice(node=node,
pin=pin,
passphrase_protection=passphrase_protection,
2014-03-07 20:44:06 +00:00
language=language,
2014-02-13 15:46:21 +00:00
label=label))
self.init_device()
return resp
@session
2014-02-13 15:46:21 +00:00
def firmware_update(self, fp):
if self.features.bootloader_mode == False:
raise Exception("Device must be in bootloader mode")
2017-06-20 14:32:54 +00:00
data = fp.read()
resp = self.call(proto.FirmwareErase(length=len(data)))
2014-02-13 15:46:21 +00:00
if isinstance(resp, proto.Failure) and resp.code == types.Failure_FirmwareError:
return False
# TREZORv1 method
2014-02-13 15:46:21 +00:00
if isinstance(resp, proto.Success):
fingerprint = hashlib.sha256(data[256:]).hexdigest()
log("Firmware fingerprint: " + fingerprint)
resp = self.call(proto.FirmwareUpload(payload=data))
if isinstance(resp, proto.Success):
return True
elif isinstance(resp, proto.Failure) and resp.code == types.Failure_FirmwareError:
return False
raise Exception("Unexpected result %s" % resp)
# TREZORv2 method
if isinstance(resp, proto.FirmwareRequest):
import pyblake2
while True:
payload = data[resp.offset:resp.offset + resp.length]
digest = pyblake2.blake2s(payload).digest()
resp = self.call(proto.FirmwareUpload(payload=payload, hash=digest))
if isinstance(resp, proto.FirmwareRequest):
continue
elif isinstance(resp, proto.Success):
return True
elif isinstance(resp, proto.Failure) and resp.code == types.Failure_FirmwareError:
return False
raise Exception("Unexpected result %s" % resp)
raise Exception("Unexpected message %s" % resp)
2014-02-13 15:46:21 +00:00
class TrezorClient(ProtocolMixin, TextUIMixin, BaseClient):
2014-02-13 15:46:21 +00:00
pass
class TrezorClientDebug(ProtocolMixin, TextUIMixin, DebugWireMixin, BaseClient):
pass
class TrezorDebugClient(ProtocolMixin, DebugLinkMixin, DebugWireMixin, BaseClient):
2014-02-13 15:46:21 +00:00
pass