mirror of
https://github.com/trezor/trezor-firmware.git
synced 2024-12-27 00:28:10 +00:00
cc5c8ae7e0
This reverts commit402e72e36e. This reverts commit21326d3a41. This reverts commit18739987ee.
1842 lines
55 KiB
Python
Executable File
1842 lines
55 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
# 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 json
|
|
import os
|
|
import re
|
|
import sys
|
|
from decimal import Decimal
|
|
|
|
import click
|
|
import requests
|
|
|
|
from trezorlib import (
|
|
btc,
|
|
cardano,
|
|
coins,
|
|
cosi,
|
|
debuglink,
|
|
device,
|
|
eos,
|
|
ethereum,
|
|
exceptions,
|
|
firmware,
|
|
lisk,
|
|
log,
|
|
messages as proto,
|
|
misc,
|
|
monero,
|
|
nem,
|
|
protobuf,
|
|
ripple,
|
|
stellar,
|
|
tezos,
|
|
tools,
|
|
ui,
|
|
)
|
|
from trezorlib.client import TrezorClient
|
|
from trezorlib.transport import enumerate_devices, get_transport
|
|
|
|
try:
|
|
import rlp
|
|
import web3
|
|
|
|
ETHEREUM_SIGN_TX = True
|
|
except Exception:
|
|
ETHEREUM_SIGN_TX = False
|
|
|
|
|
|
class ChoiceType(click.Choice):
|
|
def __init__(self, typemap):
|
|
super(ChoiceType, self).__init__(typemap.keys())
|
|
self.typemap = typemap
|
|
|
|
def convert(self, value, param, ctx):
|
|
value = super(ChoiceType, self).convert(value, param, ctx)
|
|
return self.typemap[value]
|
|
|
|
|
|
CHOICE_PASSPHRASE_SOURCE_TYPE = ChoiceType(
|
|
{
|
|
"ask": proto.PassphraseSourceType.ASK,
|
|
"device": proto.PassphraseSourceType.DEVICE,
|
|
"host": proto.PassphraseSourceType.HOST,
|
|
}
|
|
)
|
|
|
|
|
|
CHOICE_DISPLAY_ROTATION_TYPE = ChoiceType(
|
|
{"north": 0, "east": 90, "south": 180, "west": 270}
|
|
)
|
|
|
|
|
|
CHOICE_RECOVERY_DEVICE_TYPE = ChoiceType(
|
|
{
|
|
"scrambled": proto.RecoveryDeviceType.ScrambledWords,
|
|
"matrix": proto.RecoveryDeviceType.Matrix,
|
|
}
|
|
)
|
|
|
|
CHOICE_INPUT_SCRIPT_TYPE = ChoiceType(
|
|
{
|
|
"address": proto.InputScriptType.SPENDADDRESS,
|
|
"segwit": proto.InputScriptType.SPENDWITNESS,
|
|
"p2shsegwit": proto.InputScriptType.SPENDP2SHWITNESS,
|
|
}
|
|
)
|
|
|
|
CHOICE_OUTPUT_SCRIPT_TYPE = ChoiceType(
|
|
{
|
|
"address": proto.OutputScriptType.PAYTOADDRESS,
|
|
"segwit": proto.OutputScriptType.PAYTOWITNESS,
|
|
"p2shsegwit": proto.OutputScriptType.PAYTOP2SHWITNESS,
|
|
}
|
|
)
|
|
|
|
|
|
class UnderscoreAgnosticGroup(click.Group):
|
|
"""Command group that normalizes dashes and underscores.
|
|
|
|
Click 7.0 silently switched all underscore_commands to dash-commands.
|
|
This implementation of `click.Group` responds to underscore_commands by invoking
|
|
the respective dash-command.
|
|
"""
|
|
|
|
def get_command(self, ctx, cmd_name):
|
|
cmd = super().get_command(ctx, cmd_name)
|
|
if cmd is None:
|
|
cmd = super().get_command(ctx, cmd_name.replace("_", "-"))
|
|
return cmd
|
|
|
|
|
|
def enable_logging():
|
|
log.enable_debug_output()
|
|
log.OMITTED_MESSAGES.add(proto.Features)
|
|
|
|
|
|
@click.command(cls=UnderscoreAgnosticGroup, context_settings={"max_content_width": 400})
|
|
@click.option(
|
|
"-p",
|
|
"--path",
|
|
help="Select device by specific path.",
|
|
default=os.environ.get("TREZOR_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, path, verbose, is_json):
|
|
if verbose:
|
|
enable_logging()
|
|
|
|
def get_device():
|
|
try:
|
|
device = get_transport(path, prefix_search=False)
|
|
except Exception:
|
|
try:
|
|
device = get_transport(path, prefix_search=True)
|
|
except Exception:
|
|
click.echo("Failed to find a TREZOR device.")
|
|
if path is not None:
|
|
click.echo("Using path: {}".format(path))
|
|
sys.exit(1)
|
|
return TrezorClient(transport=device, ui=ui.ClickUI())
|
|
|
|
ctx.obj = get_device
|
|
|
|
|
|
@cli.resultcallback()
|
|
def print_result(res, path, verbose, is_json):
|
|
if is_json:
|
|
if isinstance(res, protobuf.MessageType):
|
|
click.echo(json.dumps({res.__class__.__name__: res.__dict__}))
|
|
else:
|
|
click.echo(json.dumps(res, sort_keys=True, indent=4))
|
|
else:
|
|
if 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))
|
|
elif isinstance(res, protobuf.MessageType):
|
|
click.echo(protobuf.format_message(res))
|
|
else:
|
|
click.echo(res)
|
|
|
|
|
|
#
|
|
# Common functions
|
|
#
|
|
|
|
|
|
@cli.command(name="list", help="List connected TREZOR devices.")
|
|
def ls():
|
|
return enumerate_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 misc.get_entropy(connect(), size).hex()
|
|
|
|
|
|
@cli.command(help="Retrieve device features and settings.")
|
|
@click.pass_obj
|
|
def get_features(connect):
|
|
return connect().features
|
|
|
|
|
|
#
|
|
# 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 device.change_pin(connect(), remove)
|
|
|
|
|
|
@cli.command(help="Enable passphrase.")
|
|
@click.pass_obj
|
|
def enable_passphrase(connect):
|
|
return device.apply_settings(connect(), use_passphrase=True)
|
|
|
|
|
|
@cli.command(help="Disable passphrase.")
|
|
@click.pass_obj
|
|
def disable_passphrase(connect):
|
|
return device.apply_settings(connect(), use_passphrase=False)
|
|
|
|
|
|
@cli.command(help="Set new device label.")
|
|
@click.option("-l", "--label")
|
|
@click.pass_obj
|
|
def set_label(connect, label):
|
|
return device.apply_settings(connect(), label=label)
|
|
|
|
|
|
@cli.command()
|
|
@click.argument("source", type=CHOICE_PASSPHRASE_SOURCE_TYPE)
|
|
@click.pass_obj
|
|
def set_passphrase_source(connect, source):
|
|
"""Set passphrase source.
|
|
|
|
Configure how to enter passphrase on Trezor Model T. The options are:
|
|
|
|
\b
|
|
ask - always ask where to enter passphrase
|
|
device - always enter passphrase on device
|
|
host - always enter passphrase on host
|
|
"""
|
|
return device.apply_settings(connect(), passphrase_source=source)
|
|
|
|
|
|
@cli.command()
|
|
@click.argument("rotation", type=CHOICE_DISPLAY_ROTATION_TYPE)
|
|
@click.pass_obj
|
|
def set_display_rotation(connect, rotation):
|
|
"""Set display rotation.
|
|
|
|
Configure display rotation for Trezor Model T. The options are
|
|
north, east, south or west.
|
|
"""
|
|
return device.apply_settings(connect(), display_rotation=rotation)
|
|
|
|
|
|
@cli.command(help="Set auto-lock delay (in seconds).")
|
|
@click.argument("delay", type=str)
|
|
@click.pass_obj
|
|
def set_auto_lock_delay(connect, delay):
|
|
value, unit = delay[:-1], delay[-1:]
|
|
units = {"s": 1, "m": 60, "h": 3600}
|
|
if unit in units:
|
|
seconds = float(value) * units[unit]
|
|
else:
|
|
seconds = float(delay) # assume seconds if no unit is specified
|
|
return device.apply_settings(connect(), auto_lock_delay_ms=int(seconds * 1000))
|
|
|
|
|
|
@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 device.apply_flags(connect(), 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 None:
|
|
img = b"\x00"
|
|
elif filename.endswith(".toif"):
|
|
img = open(filename, "rb").read()
|
|
if img[:8] != b"TOIf\x90\x00\x90\x00":
|
|
raise tools.CallException(
|
|
proto.FailureType.DataError,
|
|
"File is not a TOIF file with size of 144x144",
|
|
)
|
|
else:
|
|
from PIL import Image
|
|
|
|
im = Image.open(filename)
|
|
if im.size != (128, 64):
|
|
raise tools.CallException(
|
|
proto.FailureType.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)
|
|
return device.apply_settings(connect(), homescreen=img)
|
|
|
|
|
|
@cli.command(help="Set U2F counter.")
|
|
@click.argument("counter", type=int)
|
|
@click.pass_obj
|
|
def set_u2f_counter(connect, counter):
|
|
return device.set_u2f_counter(connect(), counter)
|
|
|
|
|
|
@cli.command(help="Reset device to factory defaults and remove all private data.")
|
|
@click.option(
|
|
"-b",
|
|
"--bootloader",
|
|
help="Wipe device in bootloader mode. This also erases the firmware.",
|
|
is_flag=True,
|
|
)
|
|
@click.pass_obj
|
|
def wipe_device(connect, bootloader):
|
|
client = connect()
|
|
if bootloader:
|
|
if not client.features.bootloader_mode:
|
|
click.echo("Please switch your device to bootloader mode.")
|
|
sys.exit(1)
|
|
else:
|
|
click.echo("Wiping user data and firmware!")
|
|
else:
|
|
if client.features.bootloader_mode:
|
|
click.echo(
|
|
"Your device is in bootloader mode. This operation would also erase firmware."
|
|
)
|
|
click.echo(
|
|
'Specify "--bootloader" if that is what you want, or disconnect and reconnect device in normal mode.'
|
|
)
|
|
click.echo("Aborting.")
|
|
sys.exit(1)
|
|
else:
|
|
click.echo("Wiping user data!")
|
|
|
|
try:
|
|
return device.wipe(connect())
|
|
except tools.CallException as e:
|
|
click.echo("Action failed: {} {}".format(*e.args))
|
|
sys.exit(3)
|
|
|
|
|
|
@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 tools.CallException(
|
|
proto.FailureType.DataError, "Please provide mnemonic or xprv"
|
|
)
|
|
|
|
client = connect()
|
|
if mnemonic:
|
|
return debuglink.load_device_by_mnemonic(
|
|
client,
|
|
mnemonic,
|
|
pin,
|
|
passphrase_protection,
|
|
label,
|
|
"english",
|
|
ignore_checksum,
|
|
expand,
|
|
)
|
|
if xprv:
|
|
return debuglink.load_device_by_xprv(
|
|
client, xprv, pin, passphrase_protection, label, "english"
|
|
)
|
|
if slip0014:
|
|
return debuglink.load_device_by_mnemonic(
|
|
client, " ".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=CHOICE_RECOVERY_DEVICE_TYPE, 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,
|
|
):
|
|
if rec_type == proto.RecoveryDeviceType.ScrambledWords:
|
|
input_callback = ui.mnemonic_words(expand)
|
|
else:
|
|
input_callback = ui.matrix_words
|
|
click.echo(ui.RECOVERY_MATRIX_DESCRIPTION)
|
|
|
|
return device.recover(
|
|
connect(),
|
|
word_count=int(words),
|
|
passphrase_protection=passphrase_protection,
|
|
pin_protection=pin_protection,
|
|
label=label,
|
|
language="english",
|
|
input_callback=input_callback,
|
|
type=rec_type,
|
|
dry_run=dry_run,
|
|
)
|
|
|
|
|
|
@cli.command(help="Perform device setup and generate new seed.")
|
|
@click.option("-e", "--show-entropy", is_flag=True)
|
|
@click.option("-t", "--strength", type=click.Choice(["128", "192", "256"]))
|
|
@click.option("-r", "--passphrase-protection", is_flag=True)
|
|
@click.option("-p", "--pin-protection", is_flag=True)
|
|
@click.option("-l", "--label")
|
|
@click.option("-u", "--u2f-counter", default=0)
|
|
@click.option("-s", "--skip-backup", is_flag=True)
|
|
@click.option("-n", "--no-backup", is_flag=True)
|
|
@click.pass_obj
|
|
def reset_device(
|
|
connect,
|
|
show_entropy,
|
|
strength,
|
|
passphrase_protection,
|
|
pin_protection,
|
|
label,
|
|
u2f_counter,
|
|
skip_backup,
|
|
no_backup,
|
|
):
|
|
if strength:
|
|
strength = int(strength)
|
|
return device.reset(
|
|
connect(),
|
|
display_random=show_entropy,
|
|
strength=strength,
|
|
passphrase_protection=passphrase_protection,
|
|
pin_protection=pin_protection,
|
|
label=label,
|
|
language="english",
|
|
u2f_counter=u2f_counter,
|
|
skip_backup=skip_backup,
|
|
no_backup=no_backup,
|
|
)
|
|
|
|
|
|
@cli.command(help="Perform device seed backup.")
|
|
@click.pass_obj
|
|
def backup_device(connect):
|
|
return device.backup(connect())
|
|
|
|
|
|
#
|
|
# Firmware update
|
|
#
|
|
|
|
|
|
ALLOWED_FIRMWARE_FORMATS = {
|
|
1: (firmware.FirmwareFormat.TREZOR_ONE, firmware.FirmwareFormat.TREZOR_ONE_V2),
|
|
2: (firmware.FirmwareFormat.TREZOR_T,),
|
|
}
|
|
|
|
|
|
def _print_version(version):
|
|
vstr = "Firmware version {major}.{minor}.{patch} build {build}".format(**version)
|
|
click.echo(vstr)
|
|
|
|
|
|
def validate_firmware(version, fw, expected_fingerprint=None):
|
|
if version == firmware.FirmwareFormat.TREZOR_ONE:
|
|
if fw.embedded_onev2:
|
|
click.echo("Trezor One firmware with embedded v2 image (1.8.0 or later)")
|
|
_print_version(fw.embedded_onev2.firmware_header.version)
|
|
else:
|
|
click.echo("Trezor One firmware image.")
|
|
elif version == firmware.FirmwareFormat.TREZOR_ONE_V2:
|
|
click.echo("Trezor One v2 firmware (1.8.0 or later)")
|
|
_print_version(fw.firmware_header.version)
|
|
elif version == firmware.FirmwareFormat.TREZOR_T:
|
|
click.echo("Trezor T firmware image.")
|
|
vendor = fw.vendor_header.vendor_string
|
|
vendor_version = "{major}.{minor}".format(**fw.vendor_header.version)
|
|
click.echo("Vendor header from {}, version {}".format(vendor, vendor_version))
|
|
_print_version(fw.firmware_header.version)
|
|
|
|
try:
|
|
firmware.validate(version, fw, allow_unsigned=False)
|
|
click.echo("Signatures are valid.")
|
|
except firmware.Unsigned:
|
|
if not click.confirm("No signatures found. Continue?", default=False):
|
|
sys.exit(1)
|
|
try:
|
|
firmware.validate(version, fw, allow_unsigned=True)
|
|
click.echo("Unsigned firmware looking OK.")
|
|
except firmware.FirmwareIntegrityError as e:
|
|
click.echo(e)
|
|
click.echo("Firmware validation failed, aborting.")
|
|
sys.exit(4)
|
|
except firmware.FirmwareIntegrityError as e:
|
|
click.echo(e)
|
|
click.echo("Firmware validation failed, aborting.")
|
|
sys.exit(4)
|
|
|
|
fingerprint = firmware.digest(version, fw).hex()
|
|
click.echo("Firmware fingerprint: {}".format(fingerprint))
|
|
if expected_fingerprint and fingerprint != expected_fingerprint:
|
|
click.echo("Expected fingerprint: {}".format(expected_fingerprint))
|
|
click.echo("Fingerprints do not match, aborting.")
|
|
sys.exit(5)
|
|
|
|
|
|
def find_best_firmware_version(bootloader_version, requested_version=None):
|
|
url = "https://beta-wallet.trezor.io/data/firmware/{}/releases.json"
|
|
releases = requests.get(url.format(bootloader_version[0])).json()
|
|
if not releases:
|
|
raise click.ClickException("Failed to get list of releases")
|
|
|
|
releases.sort(key=lambda r: r["version"], reverse=True)
|
|
|
|
def version_str(version):
|
|
return ".".join(map(str, version))
|
|
|
|
want_version = requested_version
|
|
|
|
if want_version is None:
|
|
want_version = releases[0]["version"]
|
|
click.echo("Best available version: {}".format(version_str(want_version)))
|
|
|
|
confirm_different_version = False
|
|
while True:
|
|
want_version_str = version_str(want_version)
|
|
try:
|
|
release = next(r for r in releases if r["version"] == want_version)
|
|
except StopIteration:
|
|
click.echo("Version {} not found.".format(want_version_str))
|
|
sys.exit(1)
|
|
|
|
if (
|
|
"min_bootloader_version" in release
|
|
and release["min_bootloader_version"] > bootloader_version
|
|
):
|
|
need_version_str = version_str(release["min_firmware_version"])
|
|
click.echo(
|
|
"Version {} is required before upgrading to {}.".format(
|
|
need_version_str, want_version_str
|
|
)
|
|
)
|
|
want_version = release["min_firmware_version"]
|
|
confirm_different_version = True
|
|
else:
|
|
break
|
|
|
|
if confirm_different_version:
|
|
installing_different = "Installing version {} instead.".format(want_version_str)
|
|
if requested_version is None:
|
|
click.echo(installing_different)
|
|
else:
|
|
ok = click.confirm(installing_different + " Continue?", default=True)
|
|
if not ok:
|
|
sys.exit(1)
|
|
|
|
url = "https://beta-wallet.trezor.io/" + release["url"]
|
|
if url.endswith(".hex"):
|
|
url = url[:-4]
|
|
|
|
return url, release["fingerprint"]
|
|
|
|
|
|
@cli.command()
|
|
# fmt: off
|
|
@click.option("-f", "--filename")
|
|
@click.option("-u", "--url")
|
|
@click.option("-v", "--version")
|
|
@click.option("-s", "--skip-check", is_flag=True, help="Do not validate firmware integrity")
|
|
@click.option("-n", "--dry-run", is_flag=True, help="Perform all steps but do not actually upload the firmware")
|
|
@click.option("--raw", is_flag=True, help="Push raw data to Trezor")
|
|
@click.option("--fingerprint", help="Expected firmware fingerprint in hex")
|
|
@click.option("--skip-vendor-header", help="Skip vendor header validation on Trezor T")
|
|
# fmt: on
|
|
@click.pass_obj
|
|
def firmware_update(
|
|
connect,
|
|
filename,
|
|
url,
|
|
version,
|
|
skip_check,
|
|
fingerprint,
|
|
skip_vendor_header,
|
|
raw,
|
|
dry_run,
|
|
):
|
|
"""Upload new firmware to device.
|
|
|
|
Device must be in bootloader mode.
|
|
|
|
You can specify a filename or URL from which the firmware can be downloaded.
|
|
You can also explicitly specify a firmware version that you want.
|
|
Otherwise, trezorctl will attempt to find latest available version
|
|
from wallet.trezor.io.
|
|
|
|
If you provide a fingerprint via the --fingerprint option, it will be checked
|
|
against downloaded firmware fingerprint. Otherwise fingerprint is checked
|
|
against wallet.trezor.io information, if available.
|
|
|
|
If you are customizing Model T bootloader and providing your own vendor header,
|
|
you can use --skip-vendor-header to ignore vendor header signatures.
|
|
"""
|
|
if sum(bool(x) for x in (filename, url, version)) > 1:
|
|
click.echo("You can use only one of: filename, url, version.")
|
|
sys.exit(1)
|
|
|
|
client = connect()
|
|
if not dry_run and not client.features.bootloader_mode:
|
|
click.echo("Please switch your device to bootloader mode.")
|
|
sys.exit(1)
|
|
|
|
f = client.features
|
|
bootloader_onev2 = f.major_version == 1 and f.minor_version >= 8
|
|
|
|
if filename:
|
|
data = open(filename, "rb").read()
|
|
else:
|
|
if not url:
|
|
bootloader_version = [f.major_version, f.minor_version, f.patch_version]
|
|
version_list = [int(x) for x in version.split(".")] if version else None
|
|
url, fp = find_best_firmware_version(bootloader_version, version_list)
|
|
if not fingerprint:
|
|
fingerprint = fp
|
|
|
|
click.echo("Downloading from {}".format(url))
|
|
r = requests.get(url)
|
|
data = r.content
|
|
|
|
if not raw and not skip_check:
|
|
try:
|
|
version, fw = firmware.parse(data)
|
|
except Exception as e:
|
|
click.echo(e)
|
|
sys.exit(2)
|
|
|
|
validate_firmware(version, fw, fingerprint)
|
|
if (
|
|
bootloader_onev2
|
|
and version == firmware.FirmwareFormat.TREZOR_ONE
|
|
and not fw.embedded_onev2
|
|
):
|
|
click.echo("Firmware is too old for your device. Aborting.")
|
|
sys.exit(3)
|
|
elif not bootloader_onev2 and version == firmware.FirmwareFormat.TREZOR_ONE_V2:
|
|
click.echo("You need to upgrade to bootloader 1.8.0 first.")
|
|
sys.exit(3)
|
|
|
|
if f.major_version not in ALLOWED_FIRMWARE_FORMATS:
|
|
click.echo("trezorctl doesn't know your device version. Aborting.")
|
|
sys.exit(3)
|
|
elif version not in ALLOWED_FIRMWARE_FORMATS[f.major_version]:
|
|
click.echo("Firmware does not match your device, aborting.")
|
|
sys.exit(3)
|
|
|
|
if not raw:
|
|
# special handling for embedded-OneV2 format:
|
|
# for bootloader < 1.8, keep the embedding
|
|
# for bootloader 1.8.0 and up, strip the old OneV1 header
|
|
if bootloader_onev2 and data[:4] == b"TRZR" and data[256 : 256 + 4] == b"TRZF":
|
|
click.echo("Extracting embedded firmware image (fingerprint may change).")
|
|
data = data[256:]
|
|
|
|
if dry_run:
|
|
click.echo("Dry run. Not uploading firmware to device.")
|
|
else:
|
|
try:
|
|
if f.major_version == 1 and f.firmware_present:
|
|
# Trezor One does not send ButtonRequest
|
|
click.echo("Please confirm action on your Trezor device")
|
|
return firmware.update(client, data)
|
|
except exceptions.Cancelled:
|
|
click.echo("Update aborted on device.")
|
|
except exceptions.TrezorException as e:
|
|
click.echo("Update failed: {}".format(e))
|
|
sys.exit(3)
|
|
|
|
|
|
@cli.command(help="Perform a self-test.")
|
|
@click.pass_obj
|
|
def self_test(connect):
|
|
return debuglink.self_test(connect())
|
|
|
|
|
|
@cli.command()
|
|
def usb_reset():
|
|
"""Perform USB reset on a stuck device.
|
|
|
|
This can fix LIBUSB_ERROR_PIPE and similar errors when connecting to a device
|
|
in a messed state.
|
|
"""
|
|
from trezorlib.transport.webusb import WebUsbTransport
|
|
|
|
WebUsbTransport.enumerate(usb_reset=True)
|
|
|
|
|
|
#
|
|
# 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=CHOICE_INPUT_SCRIPT_TYPE, 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 = tools.parse_path(address)
|
|
return btc.get_address(
|
|
client, 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("-t", "--script-type", type=CHOICE_INPUT_SCRIPT_TYPE, default="address")
|
|
@click.option("-d", "--show-display", is_flag=True)
|
|
@click.pass_obj
|
|
def get_public_node(connect, coin, address, curve, script_type, show_display):
|
|
client = connect()
|
|
address_n = tools.parse_path(address)
|
|
result = btc.get_public_node(
|
|
client,
|
|
address_n,
|
|
ecdsa_curve_name=curve,
|
|
show_display=show_display,
|
|
coin_name=coin,
|
|
script_type=script_type,
|
|
)
|
|
return {
|
|
"node": {
|
|
"depth": result.node.depth,
|
|
"fingerprint": "%08x" % result.node.fingerprint,
|
|
"child_num": result.node.child_num,
|
|
"chain_code": result.node.chain_code.hex(),
|
|
"public_key": result.node.public_key.hex(),
|
|
},
|
|
"xpub": result.xpub,
|
|
}
|
|
|
|
|
|
#
|
|
# Signing options
|
|
#
|
|
|
|
|
|
@cli.command(help="Sign transaction.")
|
|
@click.option("-c", "--coin", default="Bitcoin")
|
|
@click.argument("json_file", type=click.File(), required=False)
|
|
@click.pass_obj
|
|
def sign_tx(connect, coin, json_file):
|
|
client = connect()
|
|
|
|
# XXX this is the future code of this function
|
|
if json_file is not None:
|
|
data = json.load(json_file)
|
|
coin = data["coin_name"]
|
|
details = protobuf.dict_to_proto(proto.SignTx, data["details"])
|
|
inputs = [protobuf.dict_to_proto(proto.TxInputType, i) for i in data["inputs"]]
|
|
outputs = [
|
|
protobuf.dict_to_proto(proto.TxOutputType, output)
|
|
for output in data["outputs"]
|
|
]
|
|
prev_txes = {
|
|
bytes.fromhex(txid): protobuf.dict_to_proto(proto.TransactionType, tx)
|
|
for txid, tx in data["prev_txes"].items()
|
|
}
|
|
|
|
_, serialized_tx = btc.sign_tx(
|
|
client, coin, inputs, outputs, details, prev_txes
|
|
)
|
|
|
|
client.close()
|
|
|
|
click.echo()
|
|
click.echo("Signed Transaction:")
|
|
click.echo(serialized_tx.hex())
|
|
return
|
|
|
|
# XXX ALL THE REST is legacy code and will be dropped
|
|
click.echo("Warning: interactive sign-tx mode is deprecated.", err=True)
|
|
click.echo(
|
|
"Instead, you should format your transaction data as JSON and "
|
|
"supply the file as an argument to sign-tx"
|
|
)
|
|
if coin in coins.tx_api:
|
|
coin_data = coins.by_name[coin]
|
|
txapi = coins.tx_api[coin]
|
|
else:
|
|
click.echo('Coin "%s" is not recognized.' % coin, err=True)
|
|
click.echo(
|
|
"Supported coin types: %s" % ", ".join(coins.tx_api.keys()), err=True
|
|
)
|
|
sys.exit(1)
|
|
|
|
def default_script_type(address_n):
|
|
script_type = "address"
|
|
|
|
if address_n is None:
|
|
pass
|
|
elif address_n[0] == tools.H_(49):
|
|
script_type = "p2shsegwit"
|
|
|
|
return script_type
|
|
|
|
def outpoint(s):
|
|
txid, vout = s.split(":")
|
|
return bytes.fromhex(txid), int(vout)
|
|
|
|
inputs = []
|
|
txes = {}
|
|
while True:
|
|
click.echo()
|
|
prev = click.prompt(
|
|
"Previous output to spend (txid:vout)", type=outpoint, default=""
|
|
)
|
|
if not prev:
|
|
break
|
|
prev_hash, prev_index = prev
|
|
address_n = click.prompt("BIP-32 path to derive the key", type=tools.parse_path)
|
|
try:
|
|
tx = txapi[prev_hash]
|
|
txes[prev_hash] = tx
|
|
amount = tx.bin_outputs[prev_index].amount
|
|
click.echo("Prefilling input amount: {}".format(amount))
|
|
except Exception as e:
|
|
print(e)
|
|
click.echo("Failed to fetch transation. This might bite you later.")
|
|
amount = click.prompt("Input amount (satoshis)", type=int, default=0)
|
|
sequence = click.prompt(
|
|
"Sequence Number to use (RBF opt-in enabled by default)",
|
|
type=int,
|
|
default=0xFFFFFFFD,
|
|
)
|
|
script_type = click.prompt(
|
|
"Input type",
|
|
type=CHOICE_INPUT_SCRIPT_TYPE,
|
|
default=default_script_type(address_n),
|
|
)
|
|
script_type = (
|
|
script_type
|
|
if isinstance(script_type, int)
|
|
else CHOICE_INPUT_SCRIPT_TYPE.typemap[script_type]
|
|
)
|
|
|
|
new_input = proto.TxInputType(
|
|
address_n=address_n,
|
|
prev_hash=prev_hash,
|
|
prev_index=prev_index,
|
|
amount=amount,
|
|
script_type=script_type,
|
|
sequence=sequence,
|
|
)
|
|
if coin_data["bip115"]:
|
|
prev_output = txapi.get_tx(prev_hash.hex()).bin_outputs[prev_index]
|
|
new_input.prev_block_hash_bip115 = prev_output.block_hash
|
|
new_input.prev_block_height_bip115 = prev_output.block_height
|
|
|
|
inputs.append(new_input)
|
|
|
|
if coin_data["bip115"]:
|
|
current_block_height = txapi.current_height()
|
|
# Zencash recommendation for the better protection
|
|
block_height = current_block_height - 300
|
|
block_hash = txapi.get_block_hash(block_height)
|
|
# Blockhash passed in reverse order
|
|
block_hash = block_hash[::-1]
|
|
else:
|
|
block_height = None
|
|
block_hash = None
|
|
|
|
outputs = []
|
|
while True:
|
|
click.echo()
|
|
address = click.prompt("Output address (for non-change output)", default="")
|
|
if address:
|
|
address_n = None
|
|
else:
|
|
address = None
|
|
address_n = click.prompt(
|
|
"BIP-32 path (for change output)", type=tools.parse_path, default=""
|
|
)
|
|
if not address_n:
|
|
break
|
|
amount = click.prompt("Amount to spend (satoshis)", type=int)
|
|
script_type = click.prompt(
|
|
"Output type",
|
|
type=CHOICE_OUTPUT_SCRIPT_TYPE,
|
|
default=default_script_type(address_n),
|
|
)
|
|
script_type = (
|
|
script_type
|
|
if isinstance(script_type, int)
|
|
else CHOICE_OUTPUT_SCRIPT_TYPE.typemap[script_type]
|
|
)
|
|
outputs.append(
|
|
proto.TxOutputType(
|
|
address_n=address_n,
|
|
address=address,
|
|
amount=amount,
|
|
script_type=script_type,
|
|
block_hash_bip115=block_hash,
|
|
block_height_bip115=block_height,
|
|
)
|
|
)
|
|
|
|
signtx = proto.SignTx()
|
|
signtx.version = click.prompt("Transaction version", type=int, default=2)
|
|
signtx.lock_time = click.prompt("Transaction locktime", type=int, default=0)
|
|
if coin == "Capricoin":
|
|
signtx.timestamp = click.prompt("Transaction timestamp", type=int)
|
|
|
|
_, serialized_tx = btc.sign_tx(
|
|
client, coin, inputs, outputs, details=signtx, prev_txes=txes
|
|
)
|
|
|
|
client.close()
|
|
|
|
click.echo()
|
|
click.echo("Signed Transaction:")
|
|
click.echo(serialized_tx.hex())
|
|
click.echo()
|
|
click.echo("Use the following form to broadcast it to the network:")
|
|
click.echo(txapi.pushtx_url)
|
|
|
|
|
|
#
|
|
# 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 = tools.parse_path(address)
|
|
typemap = {
|
|
"address": proto.InputScriptType.SPENDADDRESS,
|
|
"segwit": proto.InputScriptType.SPENDWITNESS,
|
|
"p2shsegwit": proto.InputScriptType.SPENDP2SHWITNESS,
|
|
}
|
|
script_type = typemap[script_type]
|
|
res = btc.sign_message(client, 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 btc.verify_message(connect(), 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 = tools.parse_path(address)
|
|
ret = ethereum.sign_message(client, address_n, message)
|
|
output = {
|
|
"message": message,
|
|
"address": ret.address,
|
|
"signature": "0x%s" % ret.signature.hex(),
|
|
}
|
|
return output
|
|
|
|
|
|
def ethereum_decode_hex(value):
|
|
if value.startswith("0x") or value.startswith("0X"):
|
|
return bytes.fromhex(value[2:])
|
|
else:
|
|
return bytes.fromhex(value)
|
|
|
|
|
|
@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):
|
|
signature = ethereum_decode_hex(signature)
|
|
return ethereum.verify_message(connect(), 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 = tools.parse_path(address)
|
|
res = misc.encrypt_keyvalue(client, address_n, key, value.encode())
|
|
return res.hex()
|
|
|
|
|
|
@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 = tools.parse_path(address)
|
|
return misc.decrypt_keyvalue(client, address_n, key, bytes.fromhex(value))
|
|
|
|
|
|
# @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 = bytes.fromhex(pubkey)
|
|
# address_n = tools.parse_path(address)
|
|
# res = client.encrypt_message(pubkey, message, display_only, coin, address_n)
|
|
# return {
|
|
# 'nonce': res.nonce.hex(),
|
|
# 'message': res.message.hex(),
|
|
# 'hmac': res.hmac.hex(),
|
|
# '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 = tools.parse_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 = tools.parse_path(address)
|
|
return ethereum.get_address(client, address_n, show_display)
|
|
|
|
|
|
@cli.command(help="Get Ethereum public node of given path.")
|
|
@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_public_node(connect, address, show_display):
|
|
client = connect()
|
|
address_n = tools.parse_path(address)
|
|
result = ethereum.get_public_node(client, address_n, show_display=show_display)
|
|
return {
|
|
"node": {
|
|
"depth": result.node.depth,
|
|
"fingerprint": "%08x" % result.node.fingerprint,
|
|
"child_num": result.node.child_num,
|
|
"chain_code": result.node.chain_code.hex(),
|
|
"public_key": result.node.public_key.hex(),
|
|
},
|
|
"xpub": result.xpub,
|
|
}
|
|
|
|
|
|
# fmt: off
|
|
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,
|
|
}
|
|
# fmt: on
|
|
|
|
|
|
def ethereum_amount_to_int(ctx, param, value):
|
|
if value is None:
|
|
return None
|
|
if value.isdigit():
|
|
return int(value)
|
|
try:
|
|
number, unit = re.match(r"^(\d+(?:.\d+)?)([a-z]+)", value).groups()
|
|
scale = ETHER_UNITS[unit]
|
|
decoded_number = Decimal(number)
|
|
return int(decoded_number * scale)
|
|
|
|
except Exception:
|
|
import traceback
|
|
|
|
traceback.print_exc()
|
|
raise click.BadParameter("Amount not understood")
|
|
|
|
|
|
def ethereum_list_units(ctx, param, value):
|
|
if not value or ctx.resilient_parsing:
|
|
return
|
|
maxlen = max(len(k) for k in ETHER_UNITS.keys()) + 1
|
|
for unit, scale in ETHER_UNITS.items():
|
|
click.echo("{:{maxlen}}: {}".format(unit, scale, maxlen=maxlen))
|
|
ctx.exit()
|
|
|
|
|
|
def ethereum_erc20_contract(w3, token_address, to_address, amount):
|
|
min_abi = [
|
|
{
|
|
"name": "transfer",
|
|
"type": "function",
|
|
"constant": False,
|
|
"inputs": [
|
|
{"name": "_to", "type": "address"},
|
|
{"name": "_value", "type": "uint256"},
|
|
],
|
|
"outputs": [{"name": "", "type": "bool"}],
|
|
}
|
|
]
|
|
contract = w3.eth.contract(address=token_address, abi=min_abi)
|
|
return contract.encodeABI("transfer", [to_address, amount])
|
|
|
|
|
|
@cli.command()
|
|
@click.option(
|
|
"-c", "--chain-id", type=int, default=1, 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(
|
|
"-g", "--gas-limit", type=int, help="Gas limit (required for offline signing)"
|
|
)
|
|
@click.option(
|
|
"-t",
|
|
"--gas-price",
|
|
help="Gas price (required for offline signing)",
|
|
callback=ethereum_amount_to_int,
|
|
)
|
|
@click.option(
|
|
"-i", "--nonce", type=int, help="Transaction counter (required for offline signing)"
|
|
)
|
|
@click.option("-d", "--data", help="Data as hex string, e.g. 0x12345678")
|
|
@click.option("-p", "--publish", is_flag=True, help="Publish transaction via RPC")
|
|
@click.option("-x", "--tx-type", type=int, help="TX type (used only for Wanchain)")
|
|
@click.option("-t", "--token", help="ERC20 token address")
|
|
@click.option(
|
|
"--list-units",
|
|
is_flag=True,
|
|
help="List known currency units and exit.",
|
|
is_eager=True,
|
|
callback=ethereum_list_units,
|
|
expose_value=False,
|
|
)
|
|
@click.argument("to_address")
|
|
@click.argument("amount", callback=ethereum_amount_to_int)
|
|
@click.pass_obj
|
|
def ethereum_sign_tx(
|
|
connect,
|
|
chain_id,
|
|
address,
|
|
amount,
|
|
gas_limit,
|
|
gas_price,
|
|
nonce,
|
|
data,
|
|
publish,
|
|
to_address,
|
|
tx_type,
|
|
token,
|
|
):
|
|
"""Sign (and optionally publish) Ethereum transaction.
|
|
|
|
Use TO_ADDRESS as destination address, or set to "" for contract creation.
|
|
|
|
Specify a contract address with the --token option to send an ERC20 token.
|
|
|
|
You can specify AMOUNT and gas price either as a number of wei,
|
|
or you can use a unit suffix.
|
|
|
|
Use the --list-units option to show all known currency units.
|
|
ERC20 token amounts are specified in eth/wei, custom units are not supported.
|
|
|
|
If any of gas price, gas limit and nonce is not specified, this command will
|
|
try to connect to an ethereum node and auto-fill these values. You can configure
|
|
the connection with WEB3_PROVIDER_URI environment variable.
|
|
"""
|
|
if not ETHEREUM_SIGN_TX:
|
|
click.echo("Ethereum requirements not installed.")
|
|
click.echo("Please run:")
|
|
click.echo()
|
|
click.echo(" pip install web3 rlp")
|
|
sys.exit(1)
|
|
|
|
w3 = web3.Web3()
|
|
if (
|
|
gas_price is None or gas_limit is None or nonce is None or publish
|
|
) and not w3.isConnected():
|
|
click.echo("Failed to connect to Ethereum node.")
|
|
click.echo(
|
|
"If you want to sign offline, make sure you provide --gas-price, "
|
|
"--gas-limit and --nonce arguments"
|
|
)
|
|
sys.exit(1)
|
|
|
|
if data is not None and token is not None:
|
|
click.echo("Can't send tokens and custom data at the same time")
|
|
sys.exit(1)
|
|
|
|
client = connect()
|
|
address_n = tools.parse_path(address)
|
|
from_address = ethereum.get_address(client, address_n)
|
|
|
|
if token:
|
|
data = ethereum_erc20_contract(w3, token, to_address, amount)
|
|
to_address = token
|
|
amount = 0
|
|
|
|
if data:
|
|
data = ethereum_decode_hex(data)
|
|
else:
|
|
data = b""
|
|
|
|
if gas_price is None:
|
|
gas_price = w3.eth.gasPrice
|
|
|
|
if gas_limit is None:
|
|
gas_limit = w3.eth.estimateGas(
|
|
{
|
|
"to": to_address,
|
|
"from": from_address,
|
|
"value": amount,
|
|
"data": "0x%s" % data.hex(),
|
|
}
|
|
)
|
|
|
|
if nonce is None:
|
|
nonce = w3.eth.getTransactionCount(from_address)
|
|
|
|
sig = ethereum.sign_tx(
|
|
client,
|
|
n=address_n,
|
|
tx_type=tx_type,
|
|
nonce=nonce,
|
|
gas_price=gas_price,
|
|
gas_limit=gas_limit,
|
|
to=to_address,
|
|
value=amount,
|
|
data=data,
|
|
chain_id=chain_id,
|
|
)
|
|
|
|
to = ethereum_decode_hex(to_address)
|
|
if tx_type is None:
|
|
transaction = rlp.encode((nonce, gas_price, gas_limit, to, amount, data) + sig)
|
|
else:
|
|
transaction = rlp.encode(
|
|
(tx_type, nonce, gas_price, gas_limit, to, amount, data) + sig
|
|
)
|
|
tx_hex = "0x%s" % transaction.hex()
|
|
|
|
if publish:
|
|
tx_hash = w3.eth.sendRawTransaction(tx_hex).hex()
|
|
return "Transaction published with ID: %s" % tx_hash
|
|
else:
|
|
return "Signed raw transaction:\n%s" % tx_hex
|
|
|
|
|
|
#
|
|
# EOS functions
|
|
#
|
|
|
|
|
|
@cli.command(help="Get Eos public key in base58 encoding.")
|
|
@click.option(
|
|
"-n", "--address", required=True, help="BIP-32 path, e.g. m/44'/194'/0'/0/0"
|
|
)
|
|
@click.option("-d", "--show-display", is_flag=True)
|
|
@click.pass_obj
|
|
def eos_get_public_key(connect, address, show_display):
|
|
client = connect()
|
|
address_n = tools.parse_path(address)
|
|
res = eos.get_public_key(client, address_n, show_display)
|
|
return "WIF: {}\nRaw: {}".format(res.wif_public_key, res.raw_public_key.hex())
|
|
|
|
|
|
@cli.command(help="Init sign (and optionally publish) EOS transaction. ")
|
|
@click.option(
|
|
"-n",
|
|
"--address",
|
|
required=True,
|
|
help="BIP-32 path to source address, e.g., m/44'/194'/0'/0/0",
|
|
)
|
|
@click.option(
|
|
"-f",
|
|
"--file",
|
|
type=click.File("r"),
|
|
required=True,
|
|
help="Transaction in JSON format",
|
|
)
|
|
@click.pass_obj
|
|
def eos_sign_transaction(connect, address, file):
|
|
client = connect()
|
|
|
|
tx_json = json.load(file)
|
|
|
|
address_n = tools.parse_path(address)
|
|
return eos.sign_tx(client, address_n, tx_json["transaction"], tx_json["chain_id"])
|
|
|
|
|
|
#
|
|
# ADA functions
|
|
#
|
|
|
|
|
|
@cli.command(help="Sign Cardano transaction.")
|
|
@click.option(
|
|
"-f",
|
|
"--file",
|
|
type=click.File("r"),
|
|
required=True,
|
|
help="Transaction in JSON format",
|
|
)
|
|
@click.option("-N", "--network", type=int, default=1)
|
|
@click.pass_obj
|
|
def cardano_sign_tx(connect, file, network):
|
|
client = connect()
|
|
|
|
transaction = json.load(file)
|
|
|
|
inputs = [cardano.create_input(input) for input in transaction["inputs"]]
|
|
outputs = [cardano.create_output(output) for output in transaction["outputs"]]
|
|
transactions = transaction["transactions"]
|
|
|
|
signed_transaction = cardano.sign_tx(client, inputs, outputs, transactions, network)
|
|
|
|
return {
|
|
"tx_hash": signed_transaction.tx_hash.hex(),
|
|
"tx_body": signed_transaction.tx_body.hex(),
|
|
}
|
|
|
|
|
|
@cli.command(help="Get Cardano address.")
|
|
@click.option(
|
|
"-n", "--address", required=True, help="BIP-32 path to key, e.g. m/44'/1815'/0'/0/0"
|
|
)
|
|
@click.option("-d", "--show-display", is_flag=True)
|
|
@click.pass_obj
|
|
def cardano_get_address(connect, address, show_display):
|
|
client = connect()
|
|
address_n = tools.parse_path(address)
|
|
|
|
return cardano.get_address(client, address_n, show_display)
|
|
|
|
|
|
@cli.command(help="Get Cardano public key.")
|
|
@click.option(
|
|
"-n", "--address", required=True, help="BIP-32 path to key, e.g. m/44'/1815'/0'/0/0"
|
|
)
|
|
@click.pass_obj
|
|
def cardano_get_public_key(connect, address):
|
|
client = connect()
|
|
address_n = tools.parse_path(address)
|
|
|
|
return cardano.get_public_key(client, address_n)
|
|
|
|
|
|
#
|
|
# 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 = tools.parse_path(address)
|
|
return nem.get_address(client, 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 = tools.parse_path(address)
|
|
transaction = nem.sign_tx(client, address_n, json.load(file))
|
|
|
|
payload = {"data": transaction.data.hex(), "signature": transaction.signature.hex()}
|
|
|
|
if broadcast:
|
|
return requests.post(
|
|
"{}/transaction/announce".format(broadcast), json=payload
|
|
).json()
|
|
else:
|
|
return payload
|
|
|
|
|
|
#
|
|
# Lisk functions
|
|
#
|
|
|
|
|
|
@cli.command(help="Get Lisk address for specified path.")
|
|
@click.option(
|
|
"-n", "--address", required=True, help="BIP-32 path, e.g. m/44'/134'/0'/0'"
|
|
)
|
|
@click.option("-d", "--show-display", is_flag=True)
|
|
@click.pass_obj
|
|
def lisk_get_address(connect, address, show_display):
|
|
client = connect()
|
|
address_n = tools.parse_path(address)
|
|
return lisk.get_address(client, address_n, show_display)
|
|
|
|
|
|
@cli.command(help="Get Lisk public key for specified path.")
|
|
@click.option(
|
|
"-n", "--address", required=True, help="BIP-32 path, e.g. m/44'/134'/0'/0'"
|
|
)
|
|
@click.option("-d", "--show-display", is_flag=True)
|
|
@click.pass_obj
|
|
def lisk_get_public_key(connect, address, show_display):
|
|
client = connect()
|
|
address_n = tools.parse_path(address)
|
|
res = lisk.get_public_key(client, address_n, show_display)
|
|
output = {"public_key": res.public_key.hex()}
|
|
return output
|
|
|
|
|
|
@cli.command(help="Sign Lisk transaction.")
|
|
@click.option(
|
|
"-n",
|
|
"--address",
|
|
required=True,
|
|
help="BIP-32 path to signing key, e.g. m/44'/134'/0'/0'",
|
|
)
|
|
@click.option(
|
|
"-f", "--file", type=click.File("r"), default="-", help="Transaction in JSON format"
|
|
)
|
|
# @click.option('-b', '--broadcast', help='Broadcast Lisk transaction')
|
|
@click.pass_obj
|
|
def lisk_sign_tx(connect, address, file):
|
|
client = connect()
|
|
address_n = tools.parse_path(address)
|
|
transaction = lisk.sign_tx(client, address_n, json.load(file))
|
|
|
|
payload = {"signature": transaction.signature.hex()}
|
|
|
|
return payload
|
|
|
|
|
|
@cli.command(help="Sign message with Lisk address.")
|
|
@click.option(
|
|
"-n", "--address", required=True, help="BIP-32 path, e.g. m/44'/134'/0'/0'"
|
|
)
|
|
@click.argument("message")
|
|
@click.pass_obj
|
|
def lisk_sign_message(connect, address, message):
|
|
client = connect()
|
|
address_n = client.expand_path(address)
|
|
res = lisk.sign_message(client, address_n, message)
|
|
output = {
|
|
"message": message,
|
|
"public_key": res.public_key.hex(),
|
|
"signature": res.signature.hex(),
|
|
}
|
|
return output
|
|
|
|
|
|
@cli.command(help="Verify message signed with Lisk address.")
|
|
@click.argument("pubkey")
|
|
@click.argument("signature")
|
|
@click.argument("message")
|
|
@click.pass_obj
|
|
def lisk_verify_message(connect, pubkey, signature, message):
|
|
signature = bytes.fromhex(signature)
|
|
pubkey = bytes.fromhex(pubkey)
|
|
return lisk.verify_message(connect(), pubkey, signature, message)
|
|
|
|
|
|
#
|
|
# Monero functions
|
|
#
|
|
|
|
|
|
@cli.command(help="Get Monero address for specified path.")
|
|
@click.option("-n", "--address", required=True, help="BIP-32 path, e.g. m/44'/128'/0'")
|
|
@click.option("-d", "--show-display", is_flag=True)
|
|
@click.option(
|
|
"-t", "--network-type", type=click.Choice(["0", "1", "2", "3"]), default="0"
|
|
)
|
|
@click.pass_obj
|
|
def monero_get_address(connect, address, show_display, network_type):
|
|
client = connect()
|
|
address_n = tools.parse_path(address)
|
|
network_type = int(network_type)
|
|
return monero.get_address(client, address_n, show_display, network_type)
|
|
|
|
|
|
@cli.command(help="Get Monero watch key for specified path.")
|
|
@click.option("-n", "--address", required=True, help="BIP-32 path, e.g. m/44'/128'/0'")
|
|
@click.option(
|
|
"-t", "--network-type", type=click.Choice(["0", "1", "2", "3"]), default="0"
|
|
)
|
|
@click.pass_obj
|
|
def monero_get_watch_key(connect, address, network_type):
|
|
client = connect()
|
|
address_n = tools.parse_path(address)
|
|
network_type = int(network_type)
|
|
res = monero.get_watch_key(client, address_n, network_type)
|
|
output = {"address": res.address.decode(), "watch_key": res.watch_key.hex()}
|
|
return output
|
|
|
|
|
|
#
|
|
# 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 = tools.parse_path(address)
|
|
return cosi.commit(client, address_n, bytes.fromhex(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 = tools.parse_path(address)
|
|
return cosi.sign(
|
|
client,
|
|
address_n,
|
|
bytes.fromhex(data),
|
|
bytes.fromhex(global_commitment),
|
|
bytes.fromhex(global_pubkey),
|
|
)
|
|
|
|
|
|
#
|
|
# Stellar functions
|
|
#
|
|
@cli.command(help="Get Stellar public address")
|
|
@click.option(
|
|
"-n",
|
|
"--address",
|
|
required=False,
|
|
help="BIP32 path. Always use hardened paths and the m/44'/148'/ prefix",
|
|
default=stellar.DEFAULT_BIP32_PATH,
|
|
)
|
|
@click.option("-d", "--show-display", is_flag=True)
|
|
@click.pass_obj
|
|
def stellar_get_address(connect, address, show_display):
|
|
client = connect()
|
|
address_n = tools.parse_path(address)
|
|
return stellar.get_address(client, address_n, show_display)
|
|
|
|
|
|
@cli.command(help="Sign a base64-encoded transaction envelope")
|
|
@click.option(
|
|
"-n",
|
|
"--address",
|
|
required=False,
|
|
help="BIP32 path. Always use hardened paths and the m/44'/148'/ prefix",
|
|
default=stellar.DEFAULT_BIP32_PATH,
|
|
)
|
|
@click.option(
|
|
"-n",
|
|
"--network-passphrase",
|
|
default=stellar.DEFAULT_NETWORK_PASSPHRASE,
|
|
required=False,
|
|
help="Network passphrase (blank for public network). Testnet is: 'Test SDF Network ; September 2015'",
|
|
)
|
|
@click.argument("b64envelope")
|
|
@click.pass_obj
|
|
def stellar_sign_transaction(connect, b64envelope, address, network_passphrase):
|
|
client = connect()
|
|
address_n = tools.parse_path(address)
|
|
tx, operations = stellar.parse_transaction_bytes(base64.b64decode(b64envelope))
|
|
resp = stellar.sign_tx(client, tx, operations, address_n, network_passphrase)
|
|
|
|
return base64.b64encode(resp.signature)
|
|
|
|
|
|
#
|
|
# Ripple functions
|
|
#
|
|
@cli.command(help="Get Ripple address")
|
|
@click.option(
|
|
"-n", "--address", required=True, help="BIP-32 path to key, e.g. m/44'/144'/0'/0/0"
|
|
)
|
|
@click.option("-d", "--show-display", is_flag=True)
|
|
@click.pass_obj
|
|
def ripple_get_address(connect, address, show_display):
|
|
client = connect()
|
|
address_n = tools.parse_path(address)
|
|
return ripple.get_address(client, address_n, show_display)
|
|
|
|
|
|
@cli.command(help="Sign Ripple transaction")
|
|
@click.option(
|
|
"-n", "--address", required=True, help="BIP-32 path to key, e.g. m/44'/144'/0'/0/0"
|
|
)
|
|
@click.option(
|
|
"-f", "--file", type=click.File("r"), default="-", help="Transaction in JSON format"
|
|
)
|
|
@click.pass_obj
|
|
def ripple_sign_tx(connect, address, file):
|
|
client = connect()
|
|
address_n = tools.parse_path(address)
|
|
msg = ripple.create_sign_tx_msg(json.load(file))
|
|
|
|
result = ripple.sign_tx(client, address_n, msg)
|
|
click.echo("Signature:")
|
|
click.echo(result.signature.hex())
|
|
click.echo()
|
|
click.echo("Serialized tx including the signature:")
|
|
click.echo(result.serialized_tx.hex())
|
|
|
|
|
|
#
|
|
# Tezos functions
|
|
#
|
|
@cli.command(help="Get Tezos address for specified path.")
|
|
@click.option("-n", "--address", required=True, help="BIP-32 path, e.g. m/44'/1729'/0'")
|
|
@click.option("-d", "--show-display", is_flag=True)
|
|
@click.pass_obj
|
|
def tezos_get_address(connect, address, show_display):
|
|
client = connect()
|
|
address_n = tools.parse_path(address)
|
|
return tezos.get_address(client, address_n, show_display)
|
|
|
|
|
|
@cli.command(help="Get Tezos public key.")
|
|
@click.option("-n", "--address", required=True, help="BIP-32 path, e.g. m/44'/1729'/0'")
|
|
@click.option("-d", "--show-display", is_flag=True)
|
|
@click.pass_obj
|
|
def tezos_get_public_key(connect, address, show_display):
|
|
client = connect()
|
|
address_n = tools.parse_path(address)
|
|
return tezos.get_public_key(client, address_n, show_display)
|
|
|
|
|
|
@cli.command(help="Sign Tezos transaction.")
|
|
@click.option("-n", "--address", required=True, help="BIP-32 path, e.g. m/44'/1729'/0'")
|
|
@click.option(
|
|
"-f",
|
|
"--file",
|
|
type=click.File("r"),
|
|
default="-",
|
|
help="Transaction in JSON format (byte fields should be hexlified)",
|
|
)
|
|
@click.pass_obj
|
|
def tezos_sign_tx(connect, address, file):
|
|
client = connect()
|
|
address_n = tools.parse_path(address)
|
|
msg = protobuf.dict_to_proto(proto.TezosSignTx, json.load(file))
|
|
return tezos.sign_tx(client, address_n, msg)
|
|
|
|
|
|
#
|
|
# Main
|
|
#
|
|
|
|
|
|
if __name__ == "__main__":
|
|
cli() # pylint: disable=E1120
|