Handle errors to have cli output always be JSON

Handle HWWErrors generically when the command is invoked in
process_commands. The library functions that used to handle the
errors will instead raise the exceptions.

All exceptions possible are handled, with unknown ones resulting
in an UNKNOWN_ERROR code.
This commit is contained in:
Andrew Chow 2019-01-30 21:33:42 -05:00
parent 4c3a6df1b7
commit 0a809b2754
3 changed files with 38 additions and 68 deletions

View File

@ -8,7 +8,8 @@ from .errors import (
NO_DEVICE_PATH,
DEVICE_CONN_ERROR,
NO_PASSWORD,
UNKNWON_DEVICE_TYPE
UNKNWON_DEVICE_TYPE,
UNKNOWN_ERROR
)
import argparse
@ -157,10 +158,8 @@ def process_commands(args):
elif args.device_type and args.device_path:
try:
client = get_client(device_type, device_path, password)
except NoPasswordError as e:
return {'error':str(e),'code':NO_PASSWORD}
except UnknownDeviceError as e:
return {'error':str(e),'code':UNKNWON_DEVICE_TYPE}
except HWWError as e:
return {'error': e.get_msg(), 'code': e.get_code()}
except Exception as e:
return {'error':str(e),'code':DEVICE_CONN_ERROR}
else:
@ -169,10 +168,26 @@ def process_commands(args):
client.is_testnet = args.testnet
# Do the commands
result = args.func(args, client)
try:
result = args.func(args, client)
except HWWError as e:
result = {'error': e.get_msg(), 'code': e.get_code()}
except Exception as e:
if args.debug:
import traceback
traceback.print_exc()
result = {'error': str(e), 'code': UNKNOWN_ERROR}
# Close the device
client.close()
try:
client.close()
except HWWError as e:
result = {'error': e.get_msg(), 'code': e.get_code()}
except Exception as e:
if args.debug:
import traceback
traceback.print_exc()
result = {'error': str(e), 'code': UNKNOWN_ERROR}
return result

View File

@ -66,37 +66,19 @@ def find_device(device_path, password='', device_type=None, fingerprint=None):
return None
def getmasterxpub(client):
try:
return client.get_master_xpub()
except NotImplementedError as e:
return {'error': str(e), 'code': NOT_IMPLEMENTED}
return client.get_master_xpub()
def signtx(client, psbt):
# Deserialize the transaction
try:
tx = PSBT()
tx.deserialize(psbt)
return client.sign_tx(tx)
except NotImplementedError as e:
return {'error': str(e), 'code': NOT_IMPLEMENTED}
except IOError as e:
return {'error':'You must provide a PSBT','code':INVALID_TX}
except Exception as e:
return {'error': str(e), 'code': BAD_ARGUMENT}
tx = PSBT()
tx.deserialize(psbt)
return client.sign_tx(tx)
def getxpub(client, path):
try:
return client.get_pubkey_at_path(path)
except NotImplementedError as e:
return {'error': str(e), 'code': NOT_IMPLEMENTED}
return client.get_pubkey_at_path(path)
def signmessage(client, message, path):
try:
return client.sign_message(message, path)
except NotImplementedError as e:
return {'error': str(e), 'code': NOT_IMPLEMENTED}
except ValueError as e:
return {'error': str(e), 'code': BAD_ARGUMENT}
return client.sign_message(message, path)
def getkeypool(client, path, start, end, internal=False, keypool=False, account=0, sh_wpkh=False, wpkh=True):
if sh_wpkh == True and wpkh == True:
@ -178,49 +160,19 @@ def displayaddress(client, path, sh_wpkh=False, wpkh=False):
return client.display_address(path, sh_wpkh, wpkh)
def setup_device(client, label='', backup_passphrase=''):
try:
return client.setup_device(label, backup_passphrase)
except UnavailableActionError as e:
return {'error': str(e), 'code': UNAVAILABLE_ACTION}
except DeviceAlreadyInitError as e:
return {'error': str(e), 'code': DEVICE_ALREADY_INIT}
except ValueError as e:
return {'error': str(e), 'code': BAD_ARGUMENT}
return client.setup_device(label, backup_passphrase)
def wipe_device(client):
try:
return client.wipe_device()
except UnavailableActionError as e:
return {'error': str(e), 'code': UNAVAILABLE_ACTION}
return client.wipe_device()
def restore_device(client, label):
try:
return client.restore_device(label)
except UnavailableActionError as e:
return {'error': str(e), 'code': UNAVAILABLE_ACTION}
except DeviceAlreadyInitError as e:
return {'error': str(e), 'code': DEVICE_ALREADY_INIT}
except ValueError as e:
return {'error': str(e), 'code': BAD_ARGUMENT}
return client.restore_device(label)
def backup_device(client, label='', backup_passphrase=''):
try:
return client.backup_device(label, backup_passphrase)
except UnavailableActionError as e:
return {'error': str(e), 'code': UNAVAILABLE_ACTION}
except ValueError as e:
return {'error': str(e), 'code': BAD_ARGUMENT}
return client.backup_device(label, backup_passphrase)
def prompt_pin(client):
try:
return client.prompt_pin()
except DeviceAlreadyUnlockedError as e:
return {'error': str(e), 'code': DEVICE_ALREADY_UNLOCKED}
return client.prompt_pin()
def send_pin(client, pin):
try:
return client.send_pin(pin)
except DeviceAlreadyUnlockedError as e:
return {'error': str(e), 'code': DEVICE_ALREADY_UNLOCKED}
except ValueError as e:
return {'error': str(e), 'code': BAD_ARGUMENT}
return client.send_pin(pin)

View File

@ -95,7 +95,10 @@ class TrezorClient(HardwareWalletClient):
# Retrieves the public key at the specified BIP 32 derivation path
def get_pubkey_at_path(self, path):
self._check_unlocked()
expanded_path = tools.parse_path(path)
try:
expanded_path = tools.parse_path(path)
except ValueError as e:
raise BadArgumentError(str(e))
output = btc.get_public_node(self.client, expanded_path)
if self.is_testnet:
return {'xpub':xpub_main_2_test(output.xpub)}