Merge #124: Add a switch for interactivity and make setup and restore interactive only commands

4453555 Add interactivity to Trezor setup and restore (Andrew Chow)
18857d9 Add interactive option and move setup and restore to interactive only (Andrew Chow)

Pull request description:

  Currently using setup and restore on the Trezor and Keepkey (which are basically the only devices that support setup and restore) are broken. They require user interaction. Instead of removing them entirely, move those commands behind a switch, `--interactive`.

Tree-SHA512: 4c76a94d7dc3b876f57eb417f22b14cb0a4ecbb97dc79ba675dc49670369e5682edbaf846946e246a028532f92c32f3e0f67b66d000d341368937c856dea5347

hwilib/cli.py

@@ -9,7 +9,8 @@ from .errors import (
     DEVICE_CONN_ERROR,
     NO_PASSWORD,
     UNKNWON_DEVICE_TYPE,
-    UNKNOWN_ERROR
+    UNKNOWN_ERROR,
+    UNAVAILABLE_ACTION
 )
 from . import __version__
 
@@ -38,10 +39,14 @@ def getkeypool_handler(args, client):
     return getkeypool(client, path=args.path, start=args.start, end=args.end, internal=args.internal, keypool=args.keypool, account=args.account, sh_wpkh=args.sh_wpkh, wpkh=args.wpkh)
 
 def restore_device_handler(args, client):
-    return restore_device(client, label=args.label)
+    if args.interactive:
+        return restore_device(client, label=args.label)
+    return {'error': 'restore requires interactive mode', 'code': UNAVAILABLE_ACTION}
 
 def setup_device_handler(args, client):
-    return setup_device(client, label=args.label, backup_passphrase=args.backup_passphrase)
+    if args.interactive:
+        return setup_device(client, label=args.label, backup_passphrase=args.backup_passphrase)
+    return {'error': 'setup requires interactive mode', 'code': UNAVAILABLE_ACTION}
 
 def signmessage_handler(args, client):
     return signmessage(client, message=args.message, path=args.path)
@@ -69,6 +74,7 @@ def process_commands(cli_args):
     parser.add_argument('--fingerprint', '-f', help='Specify the device to connect to using the first 4 bytes of the hash160 of the master public key. It will connect to the first device that matches this fingerprint.')
     parser.add_argument('--version', action='version', version='%(prog)s {}'.format(__version__))
     parser.add_argument('--stdin', help='Enter commands and arguments via stdin', action='store_true')
+    parser.add_argument('--interactive', '-i', help='Use some commands interactively. Currently required for all device configuration commands', action='store_true')
 
     subparsers = parser.add_subparsers(description='Commands', dest='command')
     # work-around to make subparser required
@@ -112,7 +118,7 @@ def process_commands(cli_args):
     displayaddr_parser.add_argument('--wpkh', action='store_true', help='Display the bech32 version of the address associated with this key path')
     displayaddr_parser.set_defaults(func=displayaddress_handler)
 
-    setupdev_parser = subparsers.add_parser('setup', help='Setup a device. Passphrase protection uses the password given by -p')
+    setupdev_parser = subparsers.add_parser('setup', help='Setup a device. Passphrase protection uses the password given by -p. Requires interactive mode')
     setupdev_parser.add_argument('--label', '-l', help='The name to give to the device', default='')
     setupdev_parser.add_argument('--backup_passphrase', '-b', help='The passphrase to use for the backup, if applicable', default='')
     setupdev_parser.set_defaults(func=setup_device_handler)
@@ -120,7 +126,7 @@ def process_commands(cli_args):
     wipedev_parser = subparsers.add_parser('wipe', help='Wipe a device')
     wipedev_parser.set_defaults(func=wipe_device_handler)
 
-    restore_parser = subparsers.add_parser('restore', help='Initiate the device restoring process')
+    restore_parser = subparsers.add_parser('restore', help='Initiate the device restoring process. Requires interactive mode')
     restore_parser.add_argument('--label', '-l', help='The name to give to the device', default='')
     restore_parser.set_defaults(func=restore_device_handler)
 

hwilib/devices/trezor.py

@@ -3,16 +3,17 @@
 from ..hwwclient import HardwareWalletClient
 from ..errors import ActionCanceledError, BadArgumentError, DeviceAlreadyInitError, DeviceAlreadyUnlockedError, DeviceConnectionError, UnavailableActionError, DeviceNotReadyError
 from .trezorlib.client import TrezorClient as Trezor
-from .trezorlib.debuglink import TrezorClientDebugLink
+from .trezorlib.debuglink import TrezorClientDebugLink, DebugUI
 from .trezorlib.exceptions import Cancelled
 from .trezorlib.transport import enumerate_devices, get_transport
-from .trezorlib.ui import PassphraseUI, mnemonic_words, PIN_MATRIX_DESCRIPTION
+from .trezorlib.ui import echo, PassphraseUI, mnemonic_words, PIN_CURRENT, PIN_NEW, PIN_CONFIRM, PIN_MATRIX_DESCRIPTION, prompt
 from .trezorlib import protobuf, tools, btc, device
 from .trezorlib import messages as proto
 from ..base58 import get_xpub_fingerprint, decode, to_address, xpub_main_2_test, get_xpub_fingerprint_hex
 from ..serializations import ser_uint256, uint256_from_str
 from .. import bech32
 from usb1 import USBErrorNoDevice
+from types import MethodType
 
 import base64
 import binascii
@@ -69,15 +70,36 @@ def trezor_exception(f):
             raise DeviceConnectionError('Device disconnected')
     return func
 
+def interactive_get_pin(self, code=None):
+    if code == PIN_CURRENT:
+        desc = "current PIN"
+    elif code == PIN_NEW:
+        desc = "new PIN"
+    elif code == PIN_CONFIRM:
+        desc = "new PIN again"
+    else:
+        desc = "PIN"
+
+    echo(PIN_MATRIX_DESCRIPTION)
+
+    while True:
+        pin = prompt("Please enter {}".format(desc), hide_input=True)
+        if not pin.isdigit():
+            echo("Non-numerical PIN provided, please try again")
+        else:
+            return pin
+
 # This class extends the HardwareWalletClient for Trezor specific things
 class TrezorClient(HardwareWalletClient):
 
     def __init__(self, path, password=''):
         super(TrezorClient, self).__init__(path, password)
+        self.simulator = False
         if path.startswith('udp'):
             logging.debug('Simulator found, using DebugLink')
             transport = get_transport(path)
             self.client = TrezorClientDebugLink(transport=transport)
+            self.simulator = True
         else:
             self.client = Trezor(transport=get_transport(path), ui=PassphraseUI(password))
 
@@ -314,6 +336,10 @@ class TrezorClient(HardwareWalletClient):
     @trezor_exception
     def setup_device(self, label='', passphrase=''):
         self.client.init_device()
+        if not self.simulator:
+            # Use interactive_get_pin
+            self.client.ui.get_pin = MethodType(interactive_get_pin, self.client.ui)
+
         if self.client.features.initialized:
             raise DeviceAlreadyInitError('Device is already initialized. Use wipe first and try again')
         passphrase_enabled = False
@@ -332,8 +358,13 @@ class TrezorClient(HardwareWalletClient):
     # Restore device from mnemonic or xprv
     @trezor_exception
     def restore_device(self, label=''):
+        self.client.init_device()
+        if not self.simulator:
+            # Use interactive_get_pin
+            self.client.ui.get_pin = MethodType(interactive_get_pin, self.client.ui)
+
         passphrase_enabled = False
-        device.recover(self.client, label=label, input_callback=mnemonic_words, passphrase_protection=bool(self.password))
+        device.recover(self.client, label=label, input_callback=mnemonic_words(), passphrase_protection=bool(self.password))
         return {'success': True}
 
     # Begin backup process

hwilib/devices/trezorlib/ui.py

@@ -50,8 +50,12 @@ PIN_CONFIRM = PinMatrixRequestType.NewSecond
 def echo(msg):
     print(msg, file=sys.stderr)
 
-def prompt(msg):
-    return input(msg)
+def prompt(msg, hide_input=False):
+    if hide_input:
+        import getpass
+        return getpass.getpass(msg + ' :\n')
+    else:
+        return input(msg + ':\n')
 
 class PassphraseUI:
     def __init__(self, passphrase):

test/test_coldcard.py

@@ -30,7 +30,7 @@ def coldcard_test_suite(simulator, rpc, userpass, interface):
     # Coldcard specific management command tests
     class TestColdcardManCommands(DeviceTestCase):
         def test_setup(self):
-            result = self.do_command(self.dev_args + ['setup'])
+            result = self.do_command(self.dev_args + ['-i', 'setup'])
             self.assertIn('error', result)
             self.assertIn('code', result)
             self.assertEqual(result['error'], 'The Coldcard does not support software setup')
@@ -44,7 +44,7 @@ def coldcard_test_suite(simulator, rpc, userpass, interface):
             self.assertEqual(result['code'], -9)
 
         def test_restore(self):
-            result = self.do_command(self.dev_args + ['restore'])
+            result = self.do_command(self.dev_args + ['-i', 'restore'])
             self.assertIn('error', result)
             self.assertIn('code', result)
             self.assertEqual(result['error'], 'The Coldcard does not support restoring via software')

test/test_digitalbitbox.py

@@ -44,7 +44,7 @@ def digitalbitbox_test_suite(rpc, userpass, simulator, interface):
     # DigitalBitbox specific management command tests
     class TestDBBManCommands(DeviceTestCase):
         def test_restore(self):
-            result = self.do_command(self.dev_args + ['restore'])
+            result = self.do_command(self.dev_args + ['-i', 'restore'])
             self.assertIn('error', result)
             self.assertIn('code', result)
             self.assertEqual(result['error'], 'The Digital Bitbox does not support restoring via software')
@@ -72,7 +72,7 @@ def digitalbitbox_test_suite(rpc, userpass, simulator, interface):
 
         def test_setup_wipe(self):
             # Device is init, setup should fail
-            result = self.do_command(self.dev_args + ['setup', '--label', 'setup_test', '--backup_passphrase', 'testpass'])
+            result = self.do_command(self.dev_args + ['-i', 'setup', '--label', 'setup_test', '--backup_passphrase', 'testpass'])
             self.assertEquals(result['code'], -10)
             self.assertEquals(result['error'], 'Device is already initialized. Use wipe first and try again')
 
@@ -81,15 +81,15 @@ def digitalbitbox_test_suite(rpc, userpass, simulator, interface):
             self.assertTrue(result['success'])
 
             # Check arguments
-            result = self.do_command(self.dev_args + ['setup', '--label', 'setup_test'])
+            result = self.do_command(self.dev_args + ['-i', 'setup', '--label', 'setup_test'])
             self.assertEquals(result['code'], -7)
             self.assertEquals(result['error'], 'The label and backup passphrase for a new Digital Bitbox wallet must be specified and cannot be empty')
-            result = self.do_command(self.dev_args + ['setup', '--backup_passphrase', 'testpass'])
+            result = self.do_command(self.dev_args + ['-i', 'setup', '--backup_passphrase', 'testpass'])
             self.assertEquals(result['code'], -7)
             self.assertEquals(result['error'], 'The label and backup passphrase for a new Digital Bitbox wallet must be specified and cannot be empty')
 
             # Setup
-            result = self.do_command(self.dev_args + ['setup', '--label', 'setup_test', '--backup_passphrase', 'testpass'])
+            result = self.do_command(self.dev_args + ['-i', 'setup', '--label', 'setup_test', '--backup_passphrase', 'testpass'])
             self.assertTrue(result['success'])
 
             # Reset back to original
@@ -99,7 +99,7 @@ def digitalbitbox_test_suite(rpc, userpass, simulator, interface):
             send_encrypt(json.dumps({"seed":{"source":"backup","filename":"test_backup.pdf","key":"key"}}), '0000', dev)
 
             # Make sure device is init, setup should fail
-            result = self.do_command(self.dev_args + ['setup', '--label', 'setup_test', '--backup_passphrase', 'testpass'])
+            result = self.do_command(self.dev_args + ['-i', 'setup', '--label', 'setup_test', '--backup_passphrase', 'testpass'])
             self.assertEquals(result['code'], -10)
             self.assertEquals(result['error'], 'Device is already initialized. Use wipe first and try again')
 
@@ -117,7 +117,7 @@ def digitalbitbox_test_suite(rpc, userpass, simulator, interface):
             self.assertTrue(result['success'])
 
             # Setup
-            result = self.do_command(self.dev_args + ['setup', '--label', 'backup_test', '--backup_passphrase', 'testpass'])
+            result = self.do_command(self.dev_args + ['-i', 'setup', '--label', 'backup_test', '--backup_passphrase', 'testpass'])
             self.assertTrue(result['success'])
 
             # make the backup

test/test_keepkey.py

@@ -141,7 +141,7 @@ class TestKeepkeyManCommands(KeepkeyTestCase):
 
     def test_setup_wipe(self):
         # Device is init, setup should fail
-        result = self.do_command(self.dev_args + ['setup'])
+        result = self.do_command(self.dev_args + ['-i', 'setup'])
         self.assertEquals(result['code'], -10)
         self.assertEquals(result['error'], 'Device is already initialized. Use wipe first and try again')
 
@@ -157,7 +157,7 @@ class TestKeepkeyManCommands(KeepkeyTestCase):
         self.assertTrue(result['success'])
 
         # Make sure device is init, setup should fail
-        result = self.do_command(self.dev_args + ['setup'])
+        result = self.do_command(self.dev_args + ['-i', 'setup'])
         self.assertEquals(result['code'], -10)
         self.assertEquals(result['error'], 'Device is already initialized. Use wipe first and try again')
 

test/test_ledger.py

@@ -44,7 +44,7 @@ def ledger_test_suite(rpc, userpass, interface):
             self.assertEqual(result['code'], -9)
 
         def test_setup(self):
-            result = self.do_command(self.dev_args + ['setup'])
+            result = self.do_command(self.dev_args + ['-i', 'setup'])
             self.assertIn('error', result)
             self.assertIn('code', result)
             self.assertEqual(result['error'], 'The Ledger Nano S does not support software setup')
@@ -58,7 +58,7 @@ def ledger_test_suite(rpc, userpass, interface):
             self.assertEqual(result['code'], -9)
 
         def test_restore(self):
-            result = self.do_command(self.dev_args + ['restore'])
+            result = self.do_command(self.dev_args + ['-i', 'restore'])
             self.assertIn('error', result)
             self.assertIn('code', result)
             self.assertEqual(result['error'], 'The Ledger Nano S does not support restoring via software')

test/test_trezor.py

@@ -141,7 +141,7 @@ class TestTrezorManCommands(TrezorTestCase):
 
     def test_setup_wipe(self):
         # Device is init, setup should fail
-        result = self.do_command(self.dev_args + ['setup'])
+        result = self.do_command(self.dev_args + ['-i', 'setup'])
         self.assertEquals(result['code'], -10)
         self.assertEquals(result['error'], 'Device is already initialized. Use wipe first and try again')
 
@@ -157,7 +157,7 @@ class TestTrezorManCommands(TrezorTestCase):
         self.assertTrue(result['success'])
 
         # Make sure device is init, setup should fail
-        result = self.do_command(self.dev_args + ['setup'])
+        result = self.do_command(self.dev_args + ['-i', 'setup'])
         self.assertEquals(result['code'], -10)
         self.assertEquals(result['error'], 'Device is already initialized. Use wipe first and try again')