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, NO_DEVICE_PATH,
DEVICE_CONN_ERROR, DEVICE_CONN_ERROR,
NO_PASSWORD, NO_PASSWORD,
UNKNWON_DEVICE_TYPE UNKNWON_DEVICE_TYPE,
UNKNOWN_ERROR
) )
import argparse import argparse
@ -157,10 +158,8 @@ def process_commands(args):
elif args.device_type and args.device_path: elif args.device_type and args.device_path:
try: try:
client = get_client(device_type, device_path, password) client = get_client(device_type, device_path, password)
except NoPasswordError as e: except HWWError as e:
return {'error':str(e),'code':NO_PASSWORD} return {'error': e.get_msg(), 'code': e.get_code()}
except UnknownDeviceError as e:
return {'error':str(e),'code':UNKNWON_DEVICE_TYPE}
except Exception as e: except Exception as e:
return {'error':str(e),'code':DEVICE_CONN_ERROR} return {'error':str(e),'code':DEVICE_CONN_ERROR}
else: else:
@ -169,10 +168,26 @@ def process_commands(args):
client.is_testnet = args.testnet client.is_testnet = args.testnet
# Do the commands # 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 # 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 return result

View File

@ -66,37 +66,19 @@ def find_device(device_path, password='', device_type=None, fingerprint=None):
return None return None
def getmasterxpub(client): def getmasterxpub(client):
try: return client.get_master_xpub()
return client.get_master_xpub()
except NotImplementedError as e:
return {'error': str(e), 'code': NOT_IMPLEMENTED}
def signtx(client, psbt): def signtx(client, psbt):
# Deserialize the transaction # Deserialize the transaction
try: tx = PSBT()
tx = PSBT() tx.deserialize(psbt)
tx.deserialize(psbt) return client.sign_tx(tx)
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}
def getxpub(client, path): def getxpub(client, path):
try: return client.get_pubkey_at_path(path)
return client.get_pubkey_at_path(path)
except NotImplementedError as e:
return {'error': str(e), 'code': NOT_IMPLEMENTED}
def signmessage(client, message, path): def signmessage(client, message, path):
try: return client.sign_message(message, path)
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}
def getkeypool(client, path, start, end, internal=False, keypool=False, account=0, sh_wpkh=False, wpkh=True): def getkeypool(client, path, start, end, internal=False, keypool=False, account=0, sh_wpkh=False, wpkh=True):
if sh_wpkh == True and 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) return client.display_address(path, sh_wpkh, wpkh)
def setup_device(client, label='', backup_passphrase=''): def setup_device(client, label='', backup_passphrase=''):
try: return client.setup_device(label, backup_passphrase)
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}
def wipe_device(client): def wipe_device(client):
try: return client.wipe_device()
return client.wipe_device()
except UnavailableActionError as e:
return {'error': str(e), 'code': UNAVAILABLE_ACTION}
def restore_device(client, label): def restore_device(client, label):
try: return client.restore_device(label)
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}
def backup_device(client, label='', backup_passphrase=''): def backup_device(client, label='', backup_passphrase=''):
try: return client.backup_device(label, backup_passphrase)
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}
def prompt_pin(client): def prompt_pin(client):
try: return client.prompt_pin()
return client.prompt_pin()
except DeviceAlreadyUnlockedError as e:
return {'error': str(e), 'code': DEVICE_ALREADY_UNLOCKED}
def send_pin(client, pin): def send_pin(client, pin):
try: return client.send_pin(pin)
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}

View File

@ -95,7 +95,10 @@ class TrezorClient(HardwareWalletClient):
# Retrieves the public key at the specified BIP 32 derivation path # Retrieves the public key at the specified BIP 32 derivation path
def get_pubkey_at_path(self, path): def get_pubkey_at_path(self, path):
self._check_unlocked() 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) output = btc.get_public_node(self.client, expanded_path)
if self.is_testnet: if self.is_testnet:
return {'xpub':xpub_main_2_test(output.xpub)} return {'xpub':xpub_main_2_test(output.xpub)}