WIP: descriptor request using "getkeys"

hwilib/cli.py

@@ -1,7 +1,7 @@
 #! /usr/bin/env python3
 
 from .commands import backup_device, displayaddress, enumerate, find_device, \
-    get_client, getmasterxpub, getxpub, getkeypool, restore_device, setup_device, \
+    get_client, getmasterxpub, getxpub, getkeypool, getkeys, restore_device, setup_device, \
     signmessage, signtx, wipe_device, NO_DEVICE_PATH, DEVICE_CONN_ERROR, NO_PASSWORD, \
     UNKNWON_DEVICE_TYPE
 
@@ -15,7 +15,7 @@ def backup_device_handler(args, client):
     return backup_device(client, label=args.label, backup_passphrase=args.backup_passphrase)
 
 def displayaddress_handler(args, client):
-    return displayaddress(client, path=args.path, sh_wpkh=args.sh_wpkh, wpkh=args.wpkh)
+    return displayaddress(client, desc=args.desc, path=args.path, sh_wpkh=args.sh_wpkh, wpkh=args.wpkh)
 
 def enumerate_handler(args):
     return enumerate(password=args.password)
@@ -29,6 +29,9 @@ def getxpub_handler(args, client):
 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 getkeys_handler(args, client):
+    return getkeys(client, desc=args.desc)
+
 def restore_device_handler(args, client):
     return restore_device(client, label=args.label)
 
@@ -88,8 +91,14 @@ def process_commands(args):
     getkeypool_parser.add_argument('end', type=int, help='The index to end at.')
     getkeypool_parser.set_defaults(func=getkeypool_handler)
 
+    getkeys_parser = subparsers.add_parser('getkeys', help='Get JSON array of descriptors')
+    getkeys_parser.add_argument('desc', help='Descriptor request for which to return a descriptor with keys, e.g. wpkh([00000000/84h/1h/0h]/0/*)')
+    getkeys_parser.set_defaults(func=getkeys_handler)
+
     displayaddr_parser = subparsers.add_parser('displayaddress', help='Display an address')
-    displayaddr_parser.add_argument('path', help='The BIP 32 derivation path of the key embedded in the address')
+    group = displayaddr_parser.add_mutually_exclusive_group()
+    group.add_argument('--desc', help='Descriptor request for which to return a descriptor with keys, e.g. wpkh([00000000/84h/1h/0h]/0/*). Alternatively, use --path with --sh_wpkh / --wpkh')
+    group.add_argument('--path', help='The BIP 32 derivation path of the key embedded in the address, default follows BIP43 convention, e.g. m/84h/0h/0h/1/*')
     displayaddr_parser.add_argument('--sh_wpkh', action='store_true', help='Display the p2sh-nested segwit address associated with this key path')
     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)

hwilib/commands.py

@@ -9,6 +9,7 @@
 from .base58 import xpub_to_address, xpub_to_pub_hex, get_xpub_fingerprint_as_id, get_xpub_fingerprint_hex
 from os.path import dirname, basename, isfile
 from .hwwclient import NoPasswordError, UnavailableActionError, DeviceAlreadyInitError
+from .descriptor import Descriptor
 
 # Error codes
 NO_DEVICE_PATH = -1
@@ -182,10 +183,32 @@ def getkeypool(client, path, start, end, internal=False, keypool=False, account=
     import_data.append(this_import)
     return import_data
 
-def displayaddress(client, path, sh_wpkh=False, wpkh=False):
-    if sh_wpkh == True and wpkh == True:
-        return {'error':'Both `--wpkh` and `--sh_wpkh` can not be selected at the same time.','code':BAD_ARGUMENT}
-    return client.display_address(path, sh_wpkh, wpkh)
+def displayaddress(client, path=None, desc=None, sh_wpkh=False, wpkh=False):
+    if path is not None:
+        if sh_wpkh == True and wpkh == True:
+            return {'error':'Both `--wpkh` and `--sh_wpkh` can not be selected at the same time.','code':BAD_ARGUMENT}
+        client.display_address(path, sh_wpkh, wpkh)
+    elif desc is not None:
+        descriptor = Descriptor.parse(desc, client.is_testnet)
+        if descriptor is None:
+            return {'error':'Unable to parse descriptor: ' + desc,'code':BAD_ARGUMENT}
+        if descriptor.m_path is None:
+            return {'error':'Descriptor missing origin info: ' + desc,'code':BAD_ARGUMENT}
+        client.display_address(descriptor.m_path, descriptor.sh_wpkh, descriptor.wpkh)
+
+def getkeys(client, desc=None):
+    descriptor = Descriptor.parse(desc, client.is_testnet)
+
+    if descriptor is None:
+        return {'error':'Unable to parse descriptor: ' + desc,'code':BAD_ARGUMENT}
+    if descriptor.m_path is None:
+        return {'error':'Descriptor missing origin info: ' + desc,'code':BAD_ARGUMENT}
+
+    # Get the key at the base
+    # TODO: if the last path component is hardened, obtain a list of keys
+    descriptor.base_key = client.get_pubkey_at_path(descriptor.m_path_base)['xpub']
+
+    return [descriptor.serialize()]
 
 def setup_device(client, label='', backup_passphrase=''):
     try:

hwilib/descriptor.py

@@ -0,0 +1,89 @@
+import re
+
+class Descriptor:
+    def __init__(self, origin_fingerprint, origin_path, base_key, path_suffix, testnet, sh_wpkh, wpkh, is_request):
+        self.origin_fingerprint = origin_fingerprint
+        self.origin_path = origin_path
+        self.path_suffix = path_suffix
+        self.base_key = base_key
+        self.testnet = testnet
+        self.sh_wpkh = sh_wpkh
+        self.wpkh = wpkh
+        self.m_path = None
+        self.is_request = is_request
+
+        if origin_path:
+            self.m_path = "m" + origin_path + (path_suffix or "")
+
+            if is_request:
+                self.m_path_base = "m" + origin_path
+
+    @classmethod
+    def parse(cls, desc, testnet = False):
+        sh_wpkh = None
+        wpkh = None
+        origin_fingerprint = None
+        origin_path = None
+        base_key_and_path_match = None
+        base_key = None
+        is_request = False
+
+        if desc.startswith("sh(wpkh("):
+            sh_wpkh = True
+        elif desc.startswith("wpkh("):
+            wpkh = True
+
+        origin_match = re.search(r"\[(.*)\]", desc)
+        if origin_match:
+            origin = origin_match.group(1)
+            match = re.search(r"^([0-9a-fA-F]{8})(\/.*)", origin)
+            if  match:
+              origin_fingerprint = match.group(1)
+              origin_path = match.group(2)
+              # Replace h with '
+              origin_path = origin_path.replace('h', '\'')
+
+            base_key_and_path_match = re.search(r"\[.*\](\w+)([\/\)][\d'\/\*]*)", desc)
+        else:
+            base_key_and_path_match = re.search(r"\((\w+)([\/\)][\d'\/\*]*)", desc)
+
+        if base_key_and_path_match:
+            base_key = base_key_and_path_match.group(1)
+            path_suffix = base_key_and_path_match.group(2)
+            if path_suffix == ")":
+                path_suffix = None
+        else:
+            if origin_match == None:
+                return None
+            else:
+                # Check if this is a descriptor request, which does not contain
+                # a key, must contain an origin and the last path element must be * or *'
+                request_match = re.search(r"\[.*\]([\/\)][\d'\/\*]*\*[']?)", desc)
+                if request_match is None:
+                    return None
+
+                path_suffix = request_match.group(1)
+                is_request = True
+
+        return cls(origin_fingerprint, origin_path, base_key, path_suffix, testnet, sh_wpkh, wpkh, is_request)
+
+
+    def serialize(self):
+        descriptor_open = 'pkh('
+        descriptor_close = ')'
+        origin = ''
+        path_suffix = ''
+
+        if self.wpkh == True:
+            descriptor_open = 'wpkh('
+        elif self.sh_wpkh == True:
+            descriptor_open = 'sh(wpkh('
+            descriptor_close = '))'
+
+        if self.origin_fingerprint and self.origin_path:
+            origin = '[' + self.origin_fingerprint + self.origin_path + ']'
+
+        if self.path_suffix:
+            path_suffix = self.path_suffix
+
+        return descriptor_open + origin + self.base_key + path_suffix + descriptor_close

test/run_tests.py

@@ -7,6 +7,7 @@
 
 from test_bech32 import TestSegwitAddress
 from test_coldcard import coldcard_test_suite
+from test_descriptor import TestDescriptor
 from test_device import start_bitcoind
 from test_psbt import TestPSBT
 from test_trezor import trezor_test_suite
@@ -35,6 +36,7 @@
 
 # Run tests
 suite = unittest.TestSuite()
+suite.addTests(unittest.defaultTestLoader.loadTestsFromTestCase(TestDescriptor))
 suite.addTests(unittest.defaultTestLoader.loadTestsFromTestCase(TestSegwitAddress))
 suite.addTests(unittest.defaultTestLoader.loadTestsFromTestCase(TestPSBT))
 

test/test_descriptor.py

@@ -0,0 +1,90 @@
+#! /usr/bin/env python3
+
+from hwilib.descriptor import Descriptor
+import unittest
+
+class TestDescriptor(unittest.TestCase):
+    def test_parse_descriptor_with_origin(self):
+        desc = Descriptor.parse("wpkh([00000001/84'/1'/0']tpubD6NzVbkrYhZ4WaWSyoBvQwbpLkojyoTZPRsgXELWz3Popb3qkjcJyJUGLnL4qHHoQvao8ESaAstxYSnhyswJ76uZPStJRJCTKvosUCJZL5B/0/0)", True)
+        self.assertIsNotNone(desc)
+        self.assertEqual(desc.wpkh, True)
+        self.assertEqual(desc.sh_wpkh, None)
+        self.assertEqual(desc.origin_fingerprint, "00000001")
+        self.assertEqual(desc.origin_path, "/84'/1'/0'")
+        self.assertEqual(desc.base_key, "tpubD6NzVbkrYhZ4WaWSyoBvQwbpLkojyoTZPRsgXELWz3Popb3qkjcJyJUGLnL4qHHoQvao8ESaAstxYSnhyswJ76uZPStJRJCTKvosUCJZL5B")
+        self.assertEqual(desc.path_suffix, "/0/0")
+        self.assertEqual(desc.testnet, True)
+        self.assertEqual(desc.m_path, "m/84'/1'/0'/0/0")
+
+    def test_parse_descriptor_without_origin(self):
+        desc = Descriptor.parse("wpkh(tpubD6NzVbkrYhZ4WaWSyoBvQwbpLkojyoTZPRsgXELWz3Popb3qkjcJyJUGLnL4qHHoQvao8ESaAstxYSnhyswJ76uZPStJRJCTKvosUCJZL5B/0/0)", True)
+        self.assertIsNotNone(desc)
+        self.assertEqual(desc.wpkh, True)
+        self.assertEqual(desc.sh_wpkh, None)
+        self.assertEqual(desc.origin_fingerprint, None)
+        self.assertEqual(desc.origin_path, None)
+        self.assertEqual(desc.base_key, "tpubD6NzVbkrYhZ4WaWSyoBvQwbpLkojyoTZPRsgXELWz3Popb3qkjcJyJUGLnL4qHHoQvao8ESaAstxYSnhyswJ76uZPStJRJCTKvosUCJZL5B")
+        self.assertEqual(desc.path_suffix, "/0/0")
+        self.assertEqual(desc.testnet, True)
+        self.assertEqual(desc.m_path, None)
+
+    def test_parse_descriptor_with_key_at_end_with_origin(self):
+        desc = Descriptor.parse("wpkh([00000001/84'/1'/0'/0/0]0297dc3f4420402e01a113984311bf4a1b8de376cac0bdcfaf1b3ac81f13433c7)", True)
+        self.assertIsNotNone(desc)
+        self.assertEqual(desc.wpkh, True)
+        self.assertEqual(desc.sh_wpkh, None)
+        self.assertEqual(desc.origin_fingerprint, "00000001")
+        self.assertEqual(desc.origin_path, "/84'/1'/0'/0/0")
+        self.assertEqual(desc.base_key, "0297dc3f4420402e01a113984311bf4a1b8de376cac0bdcfaf1b3ac81f13433c7")
+        self.assertEqual(desc.path_suffix, None)
+        self.assertEqual(desc.testnet, True)
+        self.assertEqual(desc.m_path, "m/84'/1'/0'/0/0")
+
+    def test_parse_descriptor_with_key_at_end_without_origin(self):
+        desc = Descriptor.parse("wpkh(0297dc3f4420402e01a113984311bf4a1b8de376cac0bdcfaf1b3ac81f13433c7)", True)
+        self.assertIsNotNone(desc)
+        self.assertEqual(desc.wpkh, True)
+        self.assertEqual(desc.sh_wpkh, None)
+        self.assertEqual(desc.origin_fingerprint, None)
+        self.assertEqual(desc.origin_path, None)
+        self.assertEqual(desc.base_key, "0297dc3f4420402e01a113984311bf4a1b8de376cac0bdcfaf1b3ac81f13433c7")
+        self.assertEqual(desc.path_suffix, None)
+        self.assertEqual(desc.testnet, True)
+        self.assertEqual(desc.m_path, None)
+
+    def test_parse_empty_descriptor(self):
+        desc = Descriptor.parse("", True)
+        self.assertIsNone(desc)
+
+    def test_parse_descriptor_replace_h(self):
+        desc = Descriptor.parse("wpkh([00000001/84h/1h/0']tpubD6NzVbkrYhZ4WaWSyoBvQwbpLkojyoTZPRsgXELWz3Popb3qkjcJyJUGLnL4qHHoQvao8ESaAstxYSnhyswJ76uZPStJRJCTKvosUCJZL5B/0/0)", True)
+        self.assertIsNotNone(desc)
+        self.assertEqual(desc.origin_path, "/84'/1'/0'")
+
+    def test_parse_descriptor_request(self):
+        desc = Descriptor.parse("wpkh([00000001/84'/1'/0']/0/*)", True)
+        self.assertIsNotNone(desc)
+        self.assertEqual(desc.origin_fingerprint, "00000001")
+        self.assertEqual(desc.origin_path, "/84'/1'/0'")
+        self.assertEqual(desc.base_key, None)
+        self.assertEqual(desc.path_suffix, "/0/*")
+        self.assertEqual(desc.m_path, "m/84'/1'/0'/0/*")
+        self.assertEqual(desc.m_path_base, "m/84'/1'/0'")
+
+    def test_serialize_descriptor_with_origin(self):
+        descriptor = "wpkh([00000001/84'/1'/0']tpubD6NzVbkrYhZ4WaWSyoBvQwbpLkojyoTZPRsgXELWz3Popb3qkjcJyJUGLnL4qHHoQvao8ESaAstxYSnhyswJ76uZPStJRJCTKvosUCJZL5B/0/0)"
+        desc = Descriptor.parse(descriptor, True)
+        self.assertEqual(desc.serialize(), descriptor)
+
+    def test_serialize_descriptor_without_origin(self):
+        descriptor = "wpkh(tpubD6NzVbkrYhZ4WaWSyoBvQwbpLkojyoTZPRsgXELWz3Popb3qkjcJyJUGLnL4qHHoQvao8ESaAstxYSnhyswJ76uZPStJRJCTKvosUCJZL5B/0/0)"
+        desc = Descriptor.parse(descriptor, True)
+        self.assertEqual(desc.serialize(), descriptor)
+
+    def test_serialize_descriptor_with_key_at_end_with_origin(self):
+        descriptor = "wpkh([00000001/84'/1'/0'/0/0]0297dc3f4420402e01a113984311bf4a1b8de376cac0bdcfaf1b3ac81f13433c7)"
+        desc = Descriptor.parse(descriptor, True)
+        self.assertEqual(desc.serialize(), descriptor)
+
+if __name__ == "__main__":
+    unittest.main()

test/test_device.py

@@ -380,15 +380,16 @@ def tearDown(self):
         self.emulator.stop()
 
     def test_display_address_bad_args(self):
-        result = process_commands(self.dev_args + ['displayaddress', '--sh_wpkh', '--wpkh', 'm/49h/1h/0h/0/0'])
+        result = process_commands(self.dev_args + ['displayaddress', '--sh_wpkh', '--wpkh', '--path', 'm/49h/1h/0h/0/0'])
         self.assertIn('error', result)
         self.assertIn('code', result)
         self.assertEqual(result['code'], -7)
 
     def test_display_address(self):
-        process_commands(self.dev_args + ['displayaddress', 'm/44h/1h/0h/0/0'])
-        process_commands(self.dev_args + ['displayaddress', '--sh_wpkh', 'm/49h/1h/0h/0/0'])
-        process_commands(self.dev_args + ['displayaddress', '--wpkh', 'm/84h/1h/0h/0/0'])
+        process_commands(self.dev_args + ['displayaddress', '--path', 'm/44h/1h/0h/0/0'])
+        process_commands(self.dev_args + ['displayaddress', '--sh_wpkh', '--path', 'm/49h/1h/0h/0/0'])
+        process_commands(self.dev_args + ['displayaddress', '--wpkh', '--path', 'm/84h/1h/0h/0/0'])
+        process_commands(self.dev_args + ['displayaddress', '--desc', 'wpkh(m/84h/1h/0h/0/0)'])
 
 class TestSignMessage(DeviceTestCase):
     def setUp(self):