arno/trezor-firmware
1
0
Fork 0
mirror of https://github.com/trezor/trezor-firmware.git synced 2024-12-27 00:28:10 +00:00
Code Issues Releases Wiki Activity
fae234e2d6
trezor-firmware/python/trezorctl

2034 lines
60 KiB
Python
Executable File
Raw Blame History

#!/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,
ontology,
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)
#
# Ontology functions
#
@cli.command(help="Get Ontology address for specified path.")
@click.option(
"-n", "--address", required=True, help="BIP-32 path, e.g. m/44'/888'/0'/0/0"
)
@click.option("-d", "--show-display", is_flag=True)
@click.pass_obj
def ontology_get_address(connect, address, show_display):
client = connect()
address_n = tools.parse_path(address)
return ontology.get_address(client, address_n, show_display)
@cli.command(help="Get Ontology public key for specified path.")
@click.option(
"-n", "--address", required=True, help="BIP-32 path, e.g. m/44'/888'/0'/0/0"
)
@click.option("-d", "--show-display", is_flag=True)
@click.pass_obj
def ontology_get_public_key(connect, address, show_display):
client = connect()
address_n = tools.parse_path(address)
result = ontology.get_public_key(client, address_n, show_display)
return result.public_key.hex()
@cli.command(help="Sign Ontology transfer.")
@click.option(
"-n",
"--address",
required=True,
help="BIP-32 path to signing key, e.g. m/44'/888'/0'/0/0",
)
@click.option(
"-tx",
"--transaction",
type=click.File("r"),
default="-",
help="Transaction in JSON format",
)
@click.option(
"-tr",
"--transfer",
type=click.File("r"),
default="-",
help="Transfer in JSON format",
)
@click.pass_obj
def ontology_sign_transfer(connect, address, transaction_f, transfer_f):
client = connect()
address_n = tools.parse_path(address)
transaction = protobuf.dict_to_proto(
proto.OntologyTransaction, json.load(transaction_f)
)
transfer = protobuf.dict_to_proto(proto.OntologyTransfer, json.load(transfer_f))
result = ontology.sign_transfer(client, address_n, transaction, transfer)
output = {"payload": result.payload.hex(), "signature": result.signature.hex()}
return output
@cli.command(help="Sign Ontology withdraw Ong.")
@click.option(
"-n",
"--address",
required=True,
help="BIP-32 path to signing key, e.g. m/44'/888'/0'/0/0",
)
@click.option(
"-tx",
"--transaction",
type=click.File("r"),
default="-",
help="Transaction in JSON format",
)
@click.option(
"-wi",
"--withdraw_ong",
type=click.File("r"),
default="-",
help="Withdrawal in JSON format",
)
@click.pass_obj
def ontology_sign_withdraw_ong(connect, address, transaction_f, withdraw_ong_f):
client = connect()
address_n = tools.parse_path(address)
transaction = protobuf.dict_to_proto(
proto.OntologyTransaction, json.load(transaction_f)
)
withdraw_ong = protobuf.dict_to_proto(
proto.OntologyWithdrawOng, json.load(withdraw_ong_f)
)
result = ontology.sign_withdrawal(client, address_n, transaction, withdraw_ong)
output = {"payload": result.payload.hex(), "signature": result.signature.hex()}
return output
@cli.command(help="Sign Ontology ONT ID Registration.")
@click.option(
"-n",
"--address",
required=True,
help="BIP-32 path to signing key, e.g. m/44'/888'/0'/0/0",
)
@click.option(
"-tx",
"--transaction",
type=click.File("r"),
default="-",
help="Transaction in JSON format",
)
@click.option(
"-re",
"--register",
type=click.File("r"),
default="-",
help="Register in JSON format",
)
@click.argument("transaction")
@click.argument("ont_id_register")
@click.pass_obj
def ontology_sign_ont_id_register(connect, address, transaction_f, ont_id_register_f):
client = connect()
address_n = tools.parse_path(address)
transaction = protobuf.dict_to_proto(
proto.OntologyTransaction, json.load(transaction_f)
)
ont_id_register = protobuf.dict_to_proto(
proto.OntologyOntIdRegister, json.load(ont_id_register_f)
)
result = ontology.sign_register(client, address_n, transaction, ont_id_register)
output = {"payload": result.payload.hex(), "signature": result.signature.hex()}
return output
@cli.command(help="Sign Ontology ONT ID Attributes adding.")
@click.option(
"-n",
"--address",
required=True,
help="BIP-32 path to signing key, e.g. m/44'/888'/0'/0/0",
)
@click.option(
"-tx",
"--transaction",
type=click.File("r"),
default="-",
help="Transaction in JSON format",
)
@click.option(
"-aa",
"--add_attr",
type=click.File("r"),
default="-",
help="Add attributes in JSON format",
)
@click.pass_obj
def ontology_sign_ont_id_add_attributes(
connect, address, transaction_f, ont_id_add_attributes_f
):
client = connect()
address_n = tools.parse_path(address)
transaction = protobuf.dict_to_proto(
proto.OntologyTransaction, json.load(transaction_f)
)
ont_id_add_attributes = protobuf.dict_to_proto(
proto.OntologyOntIdAddAttributes, json.load(ont_id_add_attributes_f)
)
result = ontology.sign_add_attr(
client, address_n, transaction, ont_id_add_attributes
)
output = {"payload": result.payload.hex(), "signature": result.signature.hex()}
return output
#
# Main
#
if __name__ == "__main__":
cli() # pylint: disable=E1120
Reference in New Issue View Git Blame Copy Permalink