mirror of
https://github.com/trezor/trezor-firmware.git
synced 2024-11-22 15:38:11 +00:00
829 lines
27 KiB
Python
Executable File
829 lines
27 KiB
Python
Executable File
#!/usr/bin/env python
|
|
|
|
# This file is part of the TREZOR project.
|
|
#
|
|
# Copyright (C) 2012-2017 Marek Palatinus <slush@satoshilabs.com>
|
|
# Copyright (C) 2012-2017 Pavol Rusnak <stick@satoshilabs.com>
|
|
# Copyright (C) 2016-2017 Jochen Hoenicke <hoenicke@gmail.com>
|
|
# Copyright (C) 2017 mruddy
|
|
#
|
|
# 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/>.
|
|
|
|
import base64
|
|
import binascii
|
|
import click
|
|
import functools
|
|
import json
|
|
import sys
|
|
|
|
from trezorlib.client import TrezorClient, TrezorClientVerbose, CallException
|
|
import trezorlib.types_pb2 as types
|
|
from trezorlib.coins import coins_txapi
|
|
|
|
|
|
def get_transport_class_by_name(name):
|
|
|
|
if name == 'usb':
|
|
from trezorlib.transport_hid import HidTransport
|
|
return HidTransport
|
|
|
|
if name == 'udp':
|
|
from trezorlib.transport_udp import UdpTransport
|
|
return UdpTransport
|
|
|
|
if name == 'pipe':
|
|
from trezorlib.transport_pipe import PipeTransport
|
|
return PipeTransport
|
|
|
|
if name == 'bridge':
|
|
from trezorlib.transport_bridge import BridgeTransport
|
|
return BridgeTransport
|
|
|
|
raise NotImplementedError('Unknown transport: "%s"' % name)
|
|
|
|
|
|
def get_transport(transport_name, path):
|
|
transport = get_transport_class_by_name(transport_name)
|
|
dev = transport.find_by_path(path)
|
|
return dev
|
|
|
|
|
|
@click.group()
|
|
@click.option('-t', '--transport', type=click.Choice(['usb', 'udp', 'pipe', 'bridge']), default='usb', help='Select transport used for communication.')
|
|
@click.option('-p', '--path', help='Select device by transport-specific path.')
|
|
@click.option('-v', '--verbose', is_flag=True, help='Show communication messages.')
|
|
@click.option('-j', '--json', 'is_json', is_flag=True, help='Print result as JSON object')
|
|
@click.pass_context
|
|
def cli(ctx, transport, path, verbose, is_json):
|
|
if ctx.invoked_subcommand == 'list':
|
|
ctx.obj = transport
|
|
else:
|
|
if verbose:
|
|
ctx.obj = lambda: TrezorClientVerbose(get_transport(transport, path))
|
|
else:
|
|
ctx.obj = lambda: TrezorClient(get_transport(transport, path))
|
|
|
|
|
|
@cli.resultcallback()
|
|
def print_result(res, transport, path, verbose, is_json):
|
|
if is_json:
|
|
from google.protobuf import json_format, message
|
|
if isinstance(res, message.Message):
|
|
click.echo(json_format.MessageToJson(res, preserving_proto_field_name=True))
|
|
else:
|
|
click.echo(json.dumps(res, sort_keys=True, indent=4))
|
|
else:
|
|
from google.protobuf import text_format, message
|
|
if isinstance(res, message.Message):
|
|
click.echo('%s {\n%s}' % (res.DESCRIPTOR.name, text_format.MessageToString(res, indent=4)))
|
|
elif isinstance(res, list):
|
|
for line in res:
|
|
click.echo(line)
|
|
elif isinstance(res, dict):
|
|
for k, v in res.items():
|
|
if isinstance(v, dict):
|
|
for kk, vv in v.items():
|
|
click.echo('%s.%s: %s' % (k, kk, vv))
|
|
else:
|
|
click.echo('%s: %s' % (k, v))
|
|
else:
|
|
click.echo(res)
|
|
|
|
|
|
#
|
|
# Common functions
|
|
#
|
|
|
|
|
|
@cli.command(name='list', help='List connected TREZOR devices.')
|
|
@click.pass_obj
|
|
def ls(transport_name):
|
|
transport_class = get_transport_class_by_name(transport_name)
|
|
devices = transport_class.enumerate()
|
|
return devices
|
|
|
|
|
|
@cli.command(help='Show version of trezorctl/trezorlib.')
|
|
def version():
|
|
from trezorlib import __version__ as VERSION
|
|
return VERSION
|
|
|
|
|
|
#
|
|
# Basic device functions
|
|
#
|
|
|
|
|
|
@cli.command(help='Send ping message.')
|
|
@click.argument('message')
|
|
@click.option('-b', '--button-protection', is_flag=True)
|
|
@click.option('-p', '--pin-protection', is_flag=True)
|
|
@click.option('-r', '--passphrase-protection', is_flag=True)
|
|
@click.pass_obj
|
|
def ping(connect, message, button_protection, pin_protection, passphrase_protection):
|
|
return connect().ping(message, button_protection=button_protection, pin_protection=pin_protection, passphrase_protection=passphrase_protection)
|
|
|
|
|
|
@cli.command(help='Clear session (remove cached PIN, passphrase, etc.).')
|
|
@click.pass_obj
|
|
def clear_session(connect):
|
|
return connect().clear_session()
|
|
|
|
|
|
@cli.command(help='Get example entropy.')
|
|
@click.argument('size', type=int)
|
|
@click.pass_obj
|
|
def get_entropy(connect, size):
|
|
return binascii.hexlify(connect().get_entropy(size))
|
|
|
|
|
|
@cli.command(help='Retrieve device features and settings.')
|
|
@click.pass_obj
|
|
def get_features(connect):
|
|
return connect().features
|
|
|
|
|
|
@cli.command(help='List all supported coin types by the device.')
|
|
@click.pass_obj
|
|
def list_coins(connect):
|
|
return [coin.coin_name for coin in connect().features.coins]
|
|
|
|
|
|
#
|
|
# Device management functions
|
|
#
|
|
|
|
|
|
@cli.command(help='Change new PIN or remove existing.')
|
|
@click.option('-r', '--remove', is_flag=True)
|
|
@click.pass_obj
|
|
def change_pin(connect, remove):
|
|
return connect().change_pin(remove)
|
|
|
|
|
|
@cli.command(help='Enable passphrase.')
|
|
@click.pass_obj
|
|
def enable_passphrase(connect):
|
|
return connect().apply_settings(use_passphrase=True)
|
|
|
|
|
|
@cli.command(help='Disable passphrase.')
|
|
@click.pass_obj
|
|
def disable_passphrase(connect):
|
|
return connect().apply_settings(use_passphrase=False)
|
|
|
|
|
|
@cli.command(help='Set new device label.')
|
|
@click.option('-l', '--label')
|
|
@click.pass_obj
|
|
def set_label(connect, label):
|
|
return connect().apply_settings(label=label)
|
|
|
|
|
|
@cli.command(help='Set device flags.')
|
|
@click.argument('flags')
|
|
@click.pass_obj
|
|
def set_flags(connect, flags):
|
|
flags = flags.lower()
|
|
if flags.startswith('0b'):
|
|
flags = int(flags, 2)
|
|
elif flags.startswith('0x'):
|
|
flags = int(flags, 16)
|
|
else:
|
|
flags = int(flags)
|
|
return connect().apply_flags(flags=flags)
|
|
|
|
|
|
@cli.command(help='Set new homescreen.')
|
|
@click.option('-f', '--filename', default=None)
|
|
@click.pass_obj
|
|
def set_homescreen(connect, filename):
|
|
if filename is not None:
|
|
from PIL import Image
|
|
im = Image.open(filename)
|
|
if im.size != (128, 64):
|
|
raise CallException(types.Failure_DataError, 'Wrong size of the image')
|
|
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)
|
|
else:
|
|
img = b'\x00'
|
|
return connect().apply_settings(homescreen=img)
|
|
|
|
|
|
@cli.command(help='Set U2F counter.')
|
|
@click.argument('counter', type=int)
|
|
@click.pass_obj
|
|
def set_u2f_counter(connect, counter):
|
|
return connect().set_u2f_counter(counter)
|
|
|
|
|
|
@cli.command(help='Reset device to factory defaults and remove all private data.')
|
|
@click.pass_obj
|
|
def wipe_device(connect):
|
|
return connect().wipe_device()
|
|
|
|
|
|
@cli.command(help='Load custom configuration to the device.')
|
|
@click.option('-m', '--mnemonic')
|
|
@click.option('-e', '--expand', is_flag=True)
|
|
@click.option('-x', '--xprv')
|
|
@click.option('-p', '--pin', default='')
|
|
@click.option('-r', '--passphrase-protection', is_flag=True)
|
|
@click.option('-l', '--label', default='')
|
|
@click.option('-i', '--ignore-checksum', is_flag=True)
|
|
@click.option('-s', '--slip0014', is_flag=True)
|
|
@click.pass_obj
|
|
def load_device(connect, mnemonic, expand, xprv, pin, passphrase_protection, label, ignore_checksum, slip0014):
|
|
if not mnemonic and not xprv and not slip0014:
|
|
raise CallException(types.Failure_DataError, 'Please provide mnemonic or xprv')
|
|
|
|
client = connect()
|
|
if mnemonic:
|
|
return client.load_device_by_mnemonic(
|
|
mnemonic,
|
|
pin,
|
|
passphrase_protection,
|
|
label,
|
|
'english',
|
|
ignore_checksum,
|
|
expand
|
|
)
|
|
if xprv:
|
|
return client.load_device_by_xprv(
|
|
xprv,
|
|
pin,
|
|
passphrase_protection,
|
|
label,
|
|
'english'
|
|
)
|
|
if slip0014:
|
|
return client.load_device_by_mnemonic(
|
|
' '.join(['all'] * 12),
|
|
pin,
|
|
passphrase_protection,
|
|
'SLIP-0014'
|
|
)
|
|
|
|
|
|
@cli.command(help='Start safe recovery workflow.')
|
|
@click.option('-w', '--words', type=click.Choice(['12', '18', '24']), default='24')
|
|
@click.option('-e', '--expand', is_flag=True)
|
|
@click.option('-p', '--pin-protection', is_flag=True)
|
|
@click.option('-r', '--passphrase-protection', is_flag=True)
|
|
@click.option('-l', '--label')
|
|
@click.option('-t', '--type', 'rec_type', type=click.Choice(['scrambled', 'matrix']), default='scrambled')
|
|
@click.option('-d', '--dry-run', is_flag=True)
|
|
@click.pass_obj
|
|
def recovery_device(connect, words, expand, pin_protection, passphrase_protection, label, rec_type, dry_run):
|
|
typemap = {
|
|
'scrambled': types.RecoveryDeviceType_ScrambledWords,
|
|
'matrix': types.RecoveryDeviceType_Matrix
|
|
}
|
|
return connect().recovery_device(
|
|
int(words),
|
|
passphrase_protection,
|
|
pin_protection,
|
|
label,
|
|
'english',
|
|
typemap[rec_type],
|
|
expand,
|
|
dry_run
|
|
)
|
|
|
|
|
|
@cli.command(help='Perform device setup and generate new seed.')
|
|
@click.option('-t', '--strength', type=click.Choice(['128', '192', '256']), default='256')
|
|
@click.option('-p', '--pin-protection', is_flag=True)
|
|
@click.option('-r', '--passphrase-protection', is_flag=True)
|
|
@click.option('-l', '--label')
|
|
@click.option('-u', '--u2f-counter', default=0)
|
|
@click.option('-s', '--skip-backup', is_flag=True)
|
|
@click.pass_obj
|
|
def reset_device(connect, strength, pin_protection, passphrase_protection, label, u2f_counter, skip_backup):
|
|
return connect().reset_device(
|
|
True,
|
|
int(strength),
|
|
passphrase_protection,
|
|
pin_protection,
|
|
label,
|
|
'english',
|
|
u2f_counter,
|
|
skip_backup
|
|
)
|
|
|
|
|
|
@cli.command(help='Perform device seed backup.')
|
|
@click.pass_obj
|
|
def backup_device(connect):
|
|
return connect().backup_device()
|
|
|
|
|
|
#
|
|
# Firmware update
|
|
#
|
|
|
|
|
|
@cli.command(help='Upload new firmware to device (must be in bootloader mode).')
|
|
@click.option('-f', '--filename')
|
|
@click.option('-u', '--url')
|
|
@click.option('-v', '--version')
|
|
@click.option('-s', '--skip-check', is_flag=True)
|
|
@click.pass_obj
|
|
def firmware_update(connect, filename, url, version, skip_check):
|
|
if filename:
|
|
fp = open(filename, 'rb').read()
|
|
elif url:
|
|
import requests
|
|
click.echo('Downloading from', url)
|
|
r = requests.get(url)
|
|
fp = r.content
|
|
else:
|
|
import requests
|
|
r = requests.get('https://wallet.trezor.io/data/firmware/releases.json')
|
|
releases = r.json()
|
|
|
|
def version_func(r):
|
|
return r['version']
|
|
|
|
def version_string(r):
|
|
return '.'.join(map(str, version_func(r)))
|
|
|
|
if version:
|
|
release = next((r for r in releases if version_string(r) == version))
|
|
else:
|
|
release = max(releases, key=version_func)
|
|
click.echo('Fetching version: %s' % version_string(release))
|
|
click.echo('Firmware fingerprint: %s' % release['fingerprint'])
|
|
url = 'https://wallet.trezor.io/' + release['url']
|
|
click.echo('Downloading from %s' % url)
|
|
r = requests.get(url)
|
|
fp = r.content
|
|
|
|
if not skip_check:
|
|
if fp[:8] == b'54525a52' or fp[:8] == b'54525a56':
|
|
fp = binascii.unhexlify(fp)
|
|
if fp[:4] != b'TRZR' and fp[:4] != b'TRZV':
|
|
raise CallException(types.Failure_FirmwareError, 'TREZOR firmware header expected')
|
|
|
|
click.echo('Please confirm action on device...')
|
|
|
|
from io import BytesIO
|
|
return connect().firmware_update(fp=BytesIO(fp))
|
|
|
|
|
|
@cli.command(help='Perform a self-test.')
|
|
@click.pass_obj
|
|
def self_test(connect):
|
|
return connect().self_test()
|
|
|
|
|
|
#
|
|
# Basic coin functions
|
|
#
|
|
|
|
|
|
@cli.command(help='Get address for specified path.')
|
|
@click.option('-c', '--coin', default='Bitcoin')
|
|
@click.option('-n', '--address', required=True, help="BIP-32 path, e.g. m/44'/0'/0'/0/0")
|
|
@click.option('-t', '--script-type', type=click.Choice(['address', 'segwit', 'p2shsegwit']), default='address')
|
|
@click.option('-d', '--show-display', is_flag=True)
|
|
@click.pass_obj
|
|
def get_address(connect, coin, address, script_type, show_display):
|
|
client = connect()
|
|
address_n = client.expand_path(address)
|
|
typemap = {
|
|
'address': types.SPENDADDRESS,
|
|
'segwit': types.SPENDWITNESS,
|
|
'p2shsegwit': types.SPENDP2SHWITNESS,
|
|
}
|
|
script_type = typemap[script_type]
|
|
return client.get_address(coin, address_n, show_display, script_type=script_type)
|
|
|
|
|
|
@cli.command(help='Get public node of given path.')
|
|
@click.option('-c', '--coin', default='Bitcoin')
|
|
@click.option('-n', '--address', required=True, help="BIP-32 path, e.g. m/44'/0'/0'")
|
|
@click.option('-e', '--curve')
|
|
@click.option('-d', '--show-display', is_flag=True)
|
|
@click.pass_obj
|
|
def get_public_node(connect, coin, address, curve, show_display):
|
|
client = connect()
|
|
address_n = client.expand_path(address)
|
|
result = client.get_public_node(address_n, ecdsa_curve_name=curve, show_display=show_display, coin_name=coin)
|
|
return {
|
|
'node': {
|
|
'depth': result.node.depth,
|
|
'fingerprint': "%08x" % result.node.fingerprint,
|
|
'child_num': result.node.child_num,
|
|
'chain_code': binascii.hexlify(result.node.chain_code),
|
|
'public_key': binascii.hexlify(result.node.public_key),
|
|
},
|
|
'xpub': result.xpub
|
|
}
|
|
|
|
|
|
#
|
|
# Signing options
|
|
#
|
|
|
|
@cli.command(help='Sign transaction.')
|
|
@click.option('-c', '--coin', default='Bitcoin')
|
|
# @click.option('-n', '--address', required=True, help="BIP-32 path, e.g. m/44'/0'/0'/0/0")
|
|
# @click.option('-t', '--script-type', type=click.Choice(['address', 'segwit', 'p2shsegwit']), default='address')
|
|
# @click.option('-o', '--output', required=True, help='Transaction output')
|
|
# @click.option('-f', '--fee', required=True, help='Transaction fee (sat/B)')
|
|
@click.pass_obj
|
|
def sign_tx(connect, coin):
|
|
client = connect()
|
|
if coin in coins_txapi:
|
|
txapi = coins_txapi[coin]
|
|
else:
|
|
click.echo('Coin "%s" is not recognized.' % coin, err=True)
|
|
click.echo('Supported coin types: %s' % ', '.join(coins_txapi.keys()), err=True)
|
|
sys.exit(1)
|
|
|
|
client.set_tx_api(txapi)
|
|
|
|
if sys.version_info.major < 3:
|
|
input = raw_input
|
|
|
|
inputs = []
|
|
while True:
|
|
click.echo()
|
|
prev = input('Input (prevhash:previndex, empty to move on): ').strip()
|
|
if prev == '':
|
|
break
|
|
prev_in_hash, prev_in_vout = prev.split(':')
|
|
addrn = input("Node path to sign with (e.g.- %s/0'/0/0): " % coin).strip()
|
|
inputs.append(types.TxInputType(
|
|
prev_hash=binascii.unhexlify(prev_in_hash),
|
|
prev_index=int(prev_in_vout, 10),
|
|
address_n=client.expand_path(addrn)
|
|
))
|
|
|
|
outputs = []
|
|
while True:
|
|
click.echo()
|
|
out_addr = input('Pay to address (empty to move on): ').strip()
|
|
if out_addr == '':
|
|
break
|
|
out_amount = input('Amount (in satoshis): ').strip()
|
|
outputs.append(types.TxOutputType(
|
|
amount=int(out_amount, 10),
|
|
script_type=types.PAYTOADDRESS,
|
|
address=out_addr
|
|
))
|
|
|
|
(signatures, serialized_tx) = client.sign_tx(coin, inputs, outputs)
|
|
|
|
client.close()
|
|
|
|
click.echo()
|
|
click.echo('Signed Transaction:')
|
|
click.echo(binascii.hexlify(serialized_tx))
|
|
click.echo()
|
|
click.echo('Use the following form to broadcast it to the network:')
|
|
click.echo(txapi.url.replace('/api/', '/tx/send'))
|
|
|
|
|
|
#
|
|
# Message functions
|
|
#
|
|
|
|
|
|
@cli.command(help='Sign message using address of given path.')
|
|
@click.option('-c', '--coin', default='Bitcoin')
|
|
@click.option('-n', '--address', required=True, help="BIP-32 path, e.g. m/44'/0'/0'/0/0")
|
|
@click.option('-t', '--script-type', type=click.Choice(['address', 'segwit', 'p2shsegwit']), default='address')
|
|
@click.argument('message')
|
|
@click.pass_obj
|
|
def sign_message(connect, coin, address, message, script_type):
|
|
client = connect()
|
|
address_n = client.expand_path(address)
|
|
typemap = {
|
|
'address': types.SPENDADDRESS,
|
|
'segwit': types.SPENDWITNESS,
|
|
'p2shsegwit': types.SPENDP2SHWITNESS,
|
|
}
|
|
script_type = typemap[script_type]
|
|
res = client.sign_message(coin, address_n, message, script_type)
|
|
return {
|
|
'message': message,
|
|
'address': res.address,
|
|
'signature': base64.b64encode(res.signature)
|
|
}
|
|
|
|
|
|
@cli.command(help='Verify message.')
|
|
@click.option('-c', '--coin', default='Bitcoin')
|
|
@click.argument('address')
|
|
@click.argument('signature')
|
|
@click.argument('message')
|
|
@click.pass_obj
|
|
def verify_message(connect, coin, address, signature, message):
|
|
signature = base64.b64decode(signature)
|
|
return connect().verify_message(coin, address, signature, message)
|
|
|
|
|
|
@cli.command(help='Sign message with Ethereum address.')
|
|
@click.option('-n', '--address', required=True, help="BIP-32 path, e.g. m/44'/60'/0'/0/0")
|
|
@click.argument('message')
|
|
@click.pass_obj
|
|
def ethereum_sign_message(connect, address, message):
|
|
client = connect()
|
|
address_n = client.expand_path(address)
|
|
ret = client.ethereum_sign_message(address_n, message)
|
|
output = {
|
|
'message': message,
|
|
'address': '0x%s' % binascii.hexlify(ret.address).decode(),
|
|
'signature': '0x%s' % binascii.hexlify(ret.signature).decode()
|
|
}
|
|
return output
|
|
|
|
|
|
def ethereum_decode_hex(value):
|
|
if value.startswith('0x') or value.startswith('0X'):
|
|
return value[2:].decode('hex')
|
|
else:
|
|
return value.decode('hex')
|
|
|
|
|
|
@cli.command(help='Verify message signed with Ethereum address.')
|
|
@click.argument('address')
|
|
@click.argument('signature')
|
|
@click.argument('message')
|
|
@click.pass_obj
|
|
def ethereum_verify_message(connect, address, signature, message):
|
|
address = ethereum_decode_hex(address)
|
|
signature = ethereum_decode_hex(signature)
|
|
return connect().ethereum_verify_message(address, signature, message)
|
|
|
|
|
|
@cli.command(help='Encrypt value by given key and path.')
|
|
@click.option('-n', '--address', required=True, help="BIP-32 path, e.g. m/10016'/0")
|
|
@click.argument('key')
|
|
@click.argument('value')
|
|
@click.pass_obj
|
|
def encrypt_keyvalue(connect, address, key, value):
|
|
client = connect()
|
|
address_n = client.expand_path(address)
|
|
res = client.encrypt_keyvalue(address_n, key, value)
|
|
return binascii.hexlify(res)
|
|
|
|
|
|
@cli.command(help='Decrypt value by given key and path.')
|
|
@click.option('-n', '--address', required=True, help="BIP-32 path, e.g. m/10016'/0")
|
|
@click.argument('key')
|
|
@click.argument('value')
|
|
@click.pass_obj
|
|
def decrypt_keyvalue(connect, address, key, value):
|
|
client = connect()
|
|
address_n = client.expand_path(address)
|
|
return client.decrypt_keyvalue(address_n, key, value.decode('hex'))
|
|
|
|
|
|
@cli.command(help='Encrypt message.')
|
|
@click.option('-c', '--coin', default='Bitcoin')
|
|
@click.option('-d', '--display-only', is_flag=True)
|
|
@click.option('-n', '--address', required=True, help="BIP-32 path, e.g. m/44'/0'/0'/0/0")
|
|
@click.argument('pubkey')
|
|
@click.argument('message')
|
|
@click.pass_obj
|
|
def encrypt_message(connect, coin, display_only, address, pubkey, message):
|
|
client = connect()
|
|
pubkey = binascii.unhexlify(pubkey)
|
|
address_n = client.expand_path(address)
|
|
res = client.encrypt_message(pubkey, message, display_only, coin, address_n)
|
|
return {
|
|
'nonce': binascii.hexlify(res.nonce),
|
|
'message': binascii.hexlify(res.message),
|
|
'hmac': binascii.hexlify(res.hmac),
|
|
'payload': base64.b64encode(res.nonce + res.message + res.hmac),
|
|
}
|
|
|
|
|
|
@cli.command(help='Decrypt message.')
|
|
@click.option('-n', '--address', required=True, help="BIP-32 path, e.g. m/44'/0'/0'/0/0")
|
|
@click.argument('payload')
|
|
@click.pass_obj
|
|
def decrypt_message(connect, address, payload):
|
|
client = connect()
|
|
address_n = client.expand_path(address)
|
|
payload = base64.b64decode(payload)
|
|
nonce, message, msg_hmac = payload[:33], payload[33:-8], payload[-8:]
|
|
return client.decrypt_message(address_n, nonce, message, msg_hmac)
|
|
|
|
|
|
#
|
|
# Ethereum functions
|
|
#
|
|
|
|
|
|
@cli.command(help='Get Ethereum address in hex encoding.')
|
|
@click.option('-n', '--address', required=True, help="BIP-32 path, e.g. m/44'/60'/0'/0/0")
|
|
@click.option('-d', '--show-display', is_flag=True)
|
|
@click.pass_obj
|
|
def ethereum_get_address(connect, address, show_display):
|
|
client = connect()
|
|
address_n = client.expand_path(address)
|
|
address = client.ethereum_get_address(address_n, show_display)
|
|
return '0x%s' % binascii.hexlify(address).decode()
|
|
|
|
|
|
@cli.command(help='Sign (and optionally publish) Ethereum transaction. Use TO as destination address or set TO to "" for contract creation.')
|
|
@click.option('-a', '--host', default='localhost:8545', help='RPC port of ethereum node for automatic gas/nonce estimation and publishing')
|
|
@click.option('-c', '--chain-id', type=int, help='EIP-155 chain id (replay protection)')
|
|
@click.option('-n', '--address', required=True, help="BIP-32 path to source address, e.g., m/44'/60'/0'/0/0")
|
|
@click.option('-v', '--value', default='0', help='Ether amount to transfer, e.g. "100 milliether"')
|
|
@click.option('-g', '--gas-limit', type=int, help='Gas limit - Required for offline signing')
|
|
@click.option('-t', '--gas-price', help='Gas price, e.g. "20 nanoether" - Required for offline signing')
|
|
@click.option('-i', '--nonce', type=int, help='Transaction counter - Required for offline signing')
|
|
@click.option('-d', '--data', default='', help='Data as hex string, e.g. 0x12345678')
|
|
@click.option('-p', '--publish', is_flag=True, help='Publish transaction via RPC')
|
|
@click.argument('to')
|
|
@click.pass_obj
|
|
def ethereum_sign_tx(connect, host, chain_id, address, value, gas_limit, gas_price, nonce, data, publish, to):
|
|
from ethjsonrpc import EthJsonRpc
|
|
import rlp
|
|
|
|
ether_units = {
|
|
'wei': 1,
|
|
'kwei': 1000,
|
|
'babbage': 1000,
|
|
'femtoether': 1000,
|
|
'mwei': 1000000,
|
|
'lovelace': 1000000,
|
|
'picoether': 1000000,
|
|
'gwei': 1000000000,
|
|
'shannon': 1000000000,
|
|
'nanoether': 1000000000,
|
|
'nano': 1000000000,
|
|
'szabo': 1000000000000,
|
|
'microether': 1000000000000,
|
|
'micro': 1000000000000,
|
|
'finney': 1000000000000000,
|
|
'milliether': 1000000000000000,
|
|
'milli': 1000000000000000,
|
|
'ether': 1000000000000000000,
|
|
'eth': 1000000000000000000,
|
|
}
|
|
|
|
if ' ' in value:
|
|
value, unit = value.split(' ', 1)
|
|
if unit.lower() not in ether_units:
|
|
raise CallException(types.Failure_DataError, 'Unrecognized ether unit %r' % unit)
|
|
value = int(value) * ether_units[unit.lower()]
|
|
else:
|
|
value = int(value)
|
|
|
|
if gas_price is not None:
|
|
if ' ' in gas_price:
|
|
gas_price, unit = gas_price.split(' ', 1)
|
|
if unit.lower() not in ether_units:
|
|
raise CallException(types.Failure_DataError, 'Unrecognized gas price unit %r' % unit)
|
|
gas_price = int(gas_price) * ether_units[unit.lower()]
|
|
else:
|
|
gas_price = int(gas_price)
|
|
|
|
if gas_limit is not None:
|
|
gas_limit = int(gas_limit)
|
|
|
|
to_address = ethereum_decode_hex(to)
|
|
|
|
client = connect()
|
|
address_n = client.expand_path(address)
|
|
address = '0x%s' % (binascii.hexlify(client.ethereum_get_address(address_n)),)
|
|
|
|
if gas_price is None or gas_limit is None or nonce is None:
|
|
host, port = host.split(':')
|
|
eth = EthJsonRpc(host, int(port))
|
|
|
|
if not data:
|
|
data = ''
|
|
data = ethereum_decode_hex(data)
|
|
|
|
if gas_price is None:
|
|
gas_price = eth.eth_gasPrice()
|
|
|
|
if gas_limit is None:
|
|
gas_limit = eth.eth_estimateGas(
|
|
to_address=to,
|
|
from_address=address,
|
|
value=('0x%x' % value),
|
|
data='0x' + data)
|
|
|
|
if nonce is None:
|
|
nonce = eth.eth_getTransactionCount(address)
|
|
|
|
sig = client.ethereum_sign_tx(
|
|
n=address_n,
|
|
nonce=nonce,
|
|
gas_price=gas_price,
|
|
gas_limit=gas_limit,
|
|
to=to_address,
|
|
value=value,
|
|
data=data,
|
|
chain_id=chain_id)
|
|
|
|
transaction = rlp.encode(
|
|
(nonce, gas_price, gas_limit, to_address, value, data) + sig)
|
|
tx_hex = '0x%s' % binascii.hexlify(transaction)
|
|
|
|
if publish:
|
|
tx_hash = eth.eth_sendRawTransaction(tx_hex)
|
|
return 'Transaction published with ID: %s' % tx_hash
|
|
else:
|
|
return 'Signed raw transaction: %s' % tx_hex
|
|
|
|
|
|
#
|
|
# NEM functions
|
|
#
|
|
|
|
|
|
@cli.command(help='Get NEM address for specified path.')
|
|
@click.option('-n', '--address', required=True, help="BIP-32 path, e.g. m/44'/0'/43'/0/0")
|
|
@click.option('-N', '--network', type=int, default=0x68)
|
|
@click.option('-d', '--show-display', is_flag=True)
|
|
@click.pass_obj
|
|
def nem_get_address(connect, address, network, show_display):
|
|
client = connect()
|
|
address_n = client.expand_path(address)
|
|
return client.nem_get_address(address_n, network, show_display)
|
|
|
|
|
|
@cli.command(help='Sign (and optionally broadcast) NEM transaction.')
|
|
@click.option('-n', '--address', help='BIP-32 path to signing key')
|
|
@click.option('-f', '--file', type=click.File('r'), default='-', help='Transaction in NIS (RequestPrepareAnnounce) format')
|
|
@click.option('-b', '--broadcast', help='NIS to announce transaction to')
|
|
@click.pass_obj
|
|
def nem_sign_tx(connect, address, file, broadcast):
|
|
client = connect()
|
|
address_n = client.expand_path(address)
|
|
transaction = client.nem_sign_tx(address_n, json.load(file))
|
|
|
|
payload = {
|
|
"data": binascii.hexlify(transaction.data).decode(),
|
|
"signature": binascii.hexlify(transaction.signature).decode()
|
|
}
|
|
|
|
if broadcast:
|
|
import requests
|
|
return requests.post("{}/transaction/announce".format(broadcast), json=payload).json()
|
|
else:
|
|
return payload
|
|
|
|
|
|
#
|
|
# CoSi functions
|
|
#
|
|
|
|
|
|
@cli.command(help='Ask device to commit to CoSi signing.')
|
|
@click.option('-n', '--address', required=True, help="BIP-32 path, e.g. m/44'/0'/0'/0/0")
|
|
@click.argument('data')
|
|
@click.pass_obj
|
|
def cosi_commit(connect, address, data):
|
|
client = connect()
|
|
address_n = client.expand_path(address)
|
|
return client.cosi_commit(address_n, binascii.unhexlify(data))
|
|
|
|
|
|
@cli.command(help='Ask device to sign using CoSi.')
|
|
@click.option('-n', '--address', required=True, help="BIP-32 path, e.g. m/44'/0'/0'/0/0")
|
|
@click.argument('data')
|
|
@click.argument('global_commitment')
|
|
@click.argument('global_pubkey')
|
|
@click.pass_obj
|
|
def cosi_sign(connect, address, data, global_commitment, global_pubkey):
|
|
client = connect()
|
|
address_n = client.expand_path(address)
|
|
return client.cosi_sign(address_n, binascii.unhexlify(data), binascii.unhexlify(global_commitment), binascii.unhexlify(global_pubkey))
|
|
|
|
|
|
#
|
|
# Main
|
|
#
|
|
|
|
|
|
if __name__ == '__main__':
|
|
cli()
|