fix: Python 3 leftovers in protobuf_json and keepkeyctl

keepkeyctl

@@ -29,7 +29,7 @@ import binascii
 import argparse
 import json
 import base64
-import urllib
+import urllib.request
 import tempfile
 
 from keepkeylib.client import KeepKeyClient, KeepKeyClientVerbose, KeepKeyDebuglinkClient, KeepKeyDebuglinkClientVerbose
@@ -180,7 +180,7 @@ class Commands(object):
     def ethereum_get_address(self, args):
         address_n = self.client.expand_path(args.n)
         address = self.client.ethereum_get_address(address_n, args.show_display)
-        return "0x%s" % (binascii.hexlify(address),)
+        return "0x%s" % (address.hex(),)
 
     def ethereum_sign_msg(self, args):
         n = self.client.expand_path(args.n)
@@ -220,16 +220,16 @@ class Commands(object):
             gas_limit = int(gas_limit)
 
         if args.to.startswith('0x') or args.to.startswith('0X'):
-            to_address = args.to[2:].decode('hex')
+            to_address = bytes.fromhex(args.to[2:])
         else:
-            to_address = args.to.decode('hex')
+            to_address = bytes.fromhex(args.to)
 
         nonce = args.nonce
         if nonce:
             nonce = int(nonce)
 
         address_n = self.client.expand_path(args.n)
-        address = "0x%s" % (binascii.hexlify(self.client.ethereum_get_address(address_n)),)
+        address = "0x%s" % (self.client.ethereum_get_address(address_n).hex(),)
 
         if gas_price is None or gas_limit is None or nonce is None:
             host, port = args.host.split(':')
@@ -266,7 +266,7 @@ class Commands(object):
 
         transaction = rlp.encode(
             (nonce, gas_price, gas_limit, to_address, value, data) + sig)
-        tx_hex = '0x%s' % binascii.hexlify(transaction)
+        tx_hex = '0x%s' % transaction.hex()
 
         if args.publish:
             tx_hash = eth.eth_sendRawTransaction(tx_hex)
@@ -304,7 +304,7 @@ class Commands(object):
     def eos_get_public_key(self, args):
         address_n = self.client.expand_path(args.n)
         res = self.client.eos_get_public_key(address_n, args.show_display)
-        return "WIF: %s\nRaw: %s" % (str(res.wif_public_key), binascii.hexlify(res.raw_public_key))
+        return "WIF: %s\nRaw: %s" % (str(res.wif_public_key), res.raw_public_key.hex())
 
     def eos_sign_tx(self, args):
         import keepkeylib.eos as eos
@@ -328,7 +328,7 @@ class Commands(object):
 
     def nano_sign_tx(self, args):
         def decode_hex(value):
-            return value.decode('hex') if value else None
+            return bytes.fromhex(value) if value else None
 
         address_n = self.client.expand_path(args.n)
         link_recipient_n = None
@@ -349,8 +349,8 @@ class Commands(object):
         )
 
         return "Block hash: %s\nSignature: %s" % (
-            res.block_hash.encode('hex'),
-            res.signature.encode('hex'),
+            res.block_hash.hex(),
+            res.signature.hex(),
         )
 
     def cosmos_get_address(self, args):
@@ -427,7 +427,7 @@ class Commands(object):
         return signed
 
     def get_entropy(self, args):
-        return binascii.hexlify(self.client.get_entropy(args.size))
+        return self.client.get_entropy(args.size).hex()
 
     def get_features(self, args):
         return self.client.features
@@ -489,7 +489,7 @@ class Commands(object):
         output = {
             'message': args.message,
             'address': ret.address,
-            'signature': base64.b64encode(ret.signature)
+            'signature': base64.b64encode(ret.signature).decode('ascii')
         }
         return output
 
@@ -499,19 +499,18 @@ class Commands(object):
 
     def encrypt_keyvalue(self, args):
         address_n = self.client.expand_path(args.n)
-        if sys.version_info > (3,):
-            ret = self.client.encrypt_keyvalue(address_n, args.key, bytes(args.value, 'utf-8'))
-        else:
-            ret = self.client.encrypt_keyvalue(address_n, args.key, args.value)
-        return binascii.hexlify(ret)
+        ret = self.client.encrypt_keyvalue(address_n, args.key, bytes(args.value, 'utf-8'))
+        return ret.hex()
 
     def decrypt_keyvalue(self, args):
         address_n = self.client.expand_path(args.n)
-        if sys.version_info > (3,):
-            ret = self.client.decrypt_keyvalue(address_n, args.key, bytes.fromhex(args.value))
-        else:
-            ret = self.client.decrypt_keyvalue(address_n, args.key, args.value.decode("hex"))
-        return ret
+        ret = self.client.decrypt_keyvalue(address_n, args.key, bytes.fromhex(args.value))
+        # encrypt_keyvalue encodes the value as UTF-8; return text so that it
+        # prints and serializes with -j, and hex for anything else.
+        try:
+            return ret.decode('utf-8')
+        except UnicodeDecodeError:
+            return ret.hex()
 
     def firmware_update(self, args):
         if not args.file and not args.url:
@@ -521,11 +520,11 @@ class Commands(object):
             fp = open(args.file, 'rb')
         elif args.url:
             print("Downloading from", args.url)
-            resp = urllib.urlretrieve(args.url)
-            fp = open(resp[0], 'r')
-            urllib.urlcleanup() # We still keep file pointer open
+            resp = urllib.request.urlretrieve(args.url)
+            fp = open(resp[0], 'rb')
+            urllib.request.urlcleanup() # We still keep file pointer open
 
-        if fp.read(8) == '54525a52':
+        if fp.read(8) == b'54525a52':
             print("Converting firmware to binary")
 
             fp.seek(0)

keepkeylib/protobuf_json.py

@@ -44,11 +44,19 @@
 import json
 from google.protobuf.descriptor import FieldDescriptor as FD
 import binascii
-import types_pb2 as types
+from . import types_pb2 as types
 
 class ParseError(Exception): pass
 
 
+def _is_repeated(field):
+    # FieldDescriptor.label was removed in protobuf 7; is_repeated replaces it
+    try:
+        return field.is_repeated
+    except AttributeError:
+        return field.label == FD.LABEL_REPEATED
+
+
 def json2pb(pb, js):
     ''' convert JSON string to google.protobuf.descriptor instance '''
     for field in pb.DESCRIPTOR.fields:
@@ -61,7 +69,7 @@ def json2pb(pb, js):
         else:
             raise ParseError("Field %s.%s of type '%d' is not supported" % (pb.__class__.__name__, field.name, field.type, ))
         value = js[field.name]
-        if field.label == FD.LABEL_REPEATED:
+        if _is_repeated(field):
             pb_value = getattr(pb, field.name, None)
             for v in value:
                 if field.type == FD.TYPE_MESSAGE:
@@ -91,7 +99,7 @@ def pb2json(pb):
             ftype = _ftype2js[field.type]
         else:
             raise ParseError("Field %s.%s of type '%d' is not supported" % (pb.__class__.__name__, field.name, field.type, ))
-        if field.label == FD.LABEL_REPEATED:
+        if _is_repeated(field):
             js_value = []
             for v in value:
                 js_value.append(ftype(v))
@@ -104,39 +112,39 @@ def pb2json(pb):
 _ftype2js = {
     FD.TYPE_DOUBLE: float,
     FD.TYPE_FLOAT: float,
-    FD.TYPE_INT64: long,
-    FD.TYPE_UINT64: long,
+    FD.TYPE_INT64: int,
+    FD.TYPE_UINT64: int,
     FD.TYPE_INT32: int,
     FD.TYPE_FIXED64: float,
     FD.TYPE_FIXED32: float,
     FD.TYPE_BOOL: bool,
-    FD.TYPE_STRING: unicode,
+    FD.TYPE_STRING: str,
     #FD.TYPE_MESSAGE handled specially
-    FD.TYPE_BYTES: lambda x: binascii.hexlify(x),
+    FD.TYPE_BYTES: lambda x: binascii.hexlify(x).decode('ascii'),
     FD.TYPE_UINT32: int,
     # FD.TYPE_ENUM: handled specially
     FD.TYPE_SFIXED32: float,
     FD.TYPE_SFIXED64: float,
     FD.TYPE_SINT32: int,
-    FD.TYPE_SINT64: long,
+    FD.TYPE_SINT64: int,
 }
 
 _js2ftype = {
     FD.TYPE_DOUBLE: float,
     FD.TYPE_FLOAT: float,
-    FD.TYPE_INT64: long,
-    FD.TYPE_UINT64: long,
+    FD.TYPE_INT64: int,
+    FD.TYPE_UINT64: int,
     FD.TYPE_INT32: int,
     FD.TYPE_FIXED64: float,
     FD.TYPE_FIXED32: float,
     FD.TYPE_BOOL: bool,
-    FD.TYPE_STRING: unicode,
+    FD.TYPE_STRING: str,
     # FD.TYPE_MESSAGE handled specially
     FD.TYPE_BYTES: lambda x: binascii.unhexlify(x),
     FD.TYPE_UINT32: int,
     FD.TYPE_ENUM: lambda x: getattr(types, x),
     FD.TYPE_SFIXED32: float,
     FD.TYPE_SFIXED64: float,
     FD.TYPE_SINT32: int,
-    FD.TYPE_SINT64: long,
+    FD.TYPE_SINT64: int,
 }

tests/unit/test_keepkeyctl.py

@@ -0,0 +1,144 @@
+import argparse
+import importlib.machinery
+import importlib.util
+import json
+import os
+import tempfile
+import types
+import unittest
+from unittest import mock
+
+KEEPKEYCTL = os.path.join(os.path.dirname(__file__), '..', '..', 'keepkeyctl')
+
+
+def load_keepkeyctl():
+    loader = importlib.machinery.SourceFileLoader('keepkeyctl', KEEPKEYCTL)
+    spec = importlib.util.spec_from_loader('keepkeyctl', loader)
+    module = importlib.util.module_from_spec(spec)
+    loader.exec_module(module)
+    return module
+
+
+keepkeyctl = load_keepkeyctl()
+
+
+class KeepkeyctlCommandsTestCase(unittest.TestCase):
+    """Run the keepkeyctl command handlers against a mocked client."""
+
+    def setUp(self):
+        self.client = mock.MagicMock()
+        self.client.expand_path.return_value = [44, 60, 0, 0, 0]
+        self.cmds = keepkeyctl.Commands(self.client)
+
+    def test_ethereum_get_address(self):
+        self.client.ethereum_get_address.return_value = bytes.fromhex('ab' * 20)
+        args = argparse.Namespace(n="m/44'/60'/0'/0/0", show_display=False)
+        self.assertEqual(self.cmds.ethereum_get_address(args), '0x' + 'ab' * 20)
+
+    def test_ethereum_sign_tx(self):
+        fake_rlp = types.SimpleNamespace(encode=mock.Mock(return_value=b'\xf8\x01'))
+        fake_ethjsonrpc = types.ModuleType('ethjsonrpc')
+        fake_ethjsonrpc.EthJsonRpc = mock.Mock()
+        fake_utils = types.ModuleType('ethjsonrpc.utils')
+        fake_utils.hex_to_dec = mock.Mock()
+        self.client.ethereum_get_address.return_value = bytes.fromhex('cd' * 20)
+        self.client.ethereum_sign_tx.return_value = (27, b'r', b's')
+        args = argparse.Namespace(
+            n="m/44'/60'/0'/0/0", value='1', gas_price='2', gas='21000',
+            to='0x' + 'ab' * 20, nonce='5', data='0x1234', chain_id=1,
+            publish=False, host='localhost:8545')
+        with mock.patch.dict('sys.modules', {
+                'rlp': fake_rlp,
+                'ethjsonrpc': fake_ethjsonrpc,
+                'ethjsonrpc.utils': fake_utils}):
+            out = self.cmds.ethereum_sign_tx(args)
+        self.assertEqual(out, 'Signed raw transaction: 0xf801')
+        kwargs = self.client.ethereum_sign_tx.call_args.kwargs
+        self.assertEqual(kwargs['to'], bytes.fromhex('ab' * 20))
+        self.assertEqual(kwargs['data'], b'\x12\x34')
+        fake_ethjsonrpc.EthJsonRpc.assert_not_called()
+
+        args.to = 'ab' * 20
+        with mock.patch.dict('sys.modules', {
+                'rlp': fake_rlp,
+                'ethjsonrpc': fake_ethjsonrpc,
+                'ethjsonrpc.utils': fake_utils}):
+            self.cmds.ethereum_sign_tx(args)
+        kwargs = self.client.ethereum_sign_tx.call_args.kwargs
+        self.assertEqual(kwargs['to'], bytes.fromhex('ab' * 20))
+
+    def test_eos_get_public_key(self):
+        self.client.eos_get_public_key.return_value = types.SimpleNamespace(
+            wif_public_key='EOS5abc', raw_public_key=b'\x02\x03')
+        args = argparse.Namespace(n="m/44'/194'/0'/0/0", show_display=False)
+        self.assertEqual(self.cmds.eos_get_public_key(args),
+                         'WIF: EOS5abc\nRaw: 0203')
+
+    def test_nano_sign_tx(self):
+        self.client.nano_sign_tx.return_value = types.SimpleNamespace(
+            block_hash=b'\x01\x02', signature=b'\x03\x04')
+        args = argparse.Namespace(
+            n="m/44'/165'/0'", coin='Nano', link_recipient_n=None,
+            top_parent_hash='aa' * 32, top_link=None,
+            top_representative=None, top_balance=None,
+            link_hash='bb' * 32, link_recipient=None,
+            representative='xrb_1', balance=1)
+        out = self.cmds.nano_sign_tx(args)
+        self.assertEqual(out, 'Block hash: 0102\nSignature: 0304')
+        kwargs = self.client.nano_sign_tx.call_args.kwargs
+        self.assertEqual(kwargs['grandparent_hash'], b'\xaa' * 32)
+        self.assertIsNone(kwargs['parent_link'])
+        self.assertEqual(kwargs['link_hash'], b'\xbb' * 32)
+
+    def test_get_entropy(self):
+        self.client.get_entropy.return_value = b'\x00\xff'
+        out = self.cmds.get_entropy(argparse.Namespace(size=2))
+        self.assertEqual(out, '00ff')
+        json.dumps(out)
+
+    def test_sign_message(self):
+        self.client.sign_message.return_value = types.SimpleNamespace(
+            address='1abc', signature=b'\x01\x02\x03')
+        args = argparse.Namespace(n="m/44'/0'/0'/0/0", coin='Bitcoin', message='hi')
+        out = self.cmds.sign_message(args)
+        self.assertEqual(out['signature'], 'AQID')
+        json.dumps(out)
+
+    def test_encrypt_decrypt_keyvalue(self):
+        self.client.encrypt_keyvalue.return_value = b'\xde\xad'
+        args = argparse.Namespace(n="m/10016'/0", key='k', value='plain')
+        self.assertEqual(self.cmds.encrypt_keyvalue(args), 'dead')
+        self.assertEqual(self.client.encrypt_keyvalue.call_args.args[2], b'plain')
+
+        self.client.decrypt_keyvalue.return_value = b'plain'
+        args = argparse.Namespace(n="m/10016'/0", key='k', value='dead')
+        self.assertEqual(self.cmds.decrypt_keyvalue(args), 'plain')
+        self.client.decrypt_keyvalue.return_value = b'\xff\xfe'
+        self.assertEqual(self.cmds.decrypt_keyvalue(args), 'fffe')
+        self.assertEqual(self.client.decrypt_keyvalue.call_args.args[2], b'\xde\xad')
+
+    def test_firmware_update_url(self):
+        fd, path = tempfile.mkstemp()
+        try:
+            with os.fdopen(fd, 'wb') as f:
+                f.write(b'KPKY' + b'\x00' * 12)
+            seen = {}
+
+            def firmware_update(fp):
+                seen['data'] = fp.read()
+                fp.close()
+
+            self.client.firmware_update.side_effect = firmware_update
+            args = argparse.Namespace(file=None, url='https://example.invalid/fw.bin')
+            with mock.patch('urllib.request.urlretrieve',
+                            return_value=(path, None)) as urlretrieve, \
+                    mock.patch('urllib.request.urlcleanup'):
+                self.cmds.firmware_update(args)
+            urlretrieve.assert_called_once_with(args.url)
+            self.assertEqual(seen['data'], b'KPKY' + b'\x00' * 12)
+        finally:
+            os.remove(path)
+
+
+if __name__ == '__main__':
+    unittest.main()