diff --git a/keepkeyctl b/keepkeyctl index 47b89e35..0c8775cc 100755 --- a/keepkeyctl +++ b/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) diff --git a/keepkeylib/protobuf_json.py b/keepkeylib/protobuf_json.py index 908b9065..46178a0f 100644 --- a/keepkeylib/protobuf_json.py +++ b/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,33 +112,33 @@ 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, @@ -138,5 +146,5 @@ def pb2json(pb): FD.TYPE_SFIXED32: float, FD.TYPE_SFIXED64: float, FD.TYPE_SINT32: int, - FD.TYPE_SINT64: long, + FD.TYPE_SINT64: int, } diff --git a/tests/unit/test_keepkeyctl.py b/tests/unit/test_keepkeyctl.py new file mode 100644 index 00000000..07d540d3 --- /dev/null +++ b/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()