From 7bf1c858978d1f0c7f098b69918e62632eb2ec4d Mon Sep 17 00:00:00 2001 From: Martin Novak Date: Fri, 9 Sep 2022 09:41:19 +0200 Subject: [PATCH] feat(core,common): Eth definitions signed using Merkle tree --- common/protob/messages-ethereum.proto | 2 +- common/protob/protocol.md | 26 +- common/tools/ethereum_definitions.py | 174 +++++++----- common/tools/merkle_tree.py | 79 ++++++ core/src/apps/ethereum/definitions.py | 55 +++- core/src/apps/ethereum/sign_tx.py | 6 +- core/src/apps/ethereum/tokens.py | 1 - core/src/apps/ethereum/tokens.py.mako | 1 - core/tests/ethereum_common.py | 82 ++++-- core/tests/test_apps.ethereum.definitions.py | 274 ++++++++++++------- core/tests/test_apps.ethereum.tokens.py | 4 + 11 files changed, 489 insertions(+), 215 deletions(-) create mode 100755 common/tools/merkle_tree.py diff --git a/common/protob/messages-ethereum.proto b/common/protob/messages-ethereum.proto index e250ce026c..5eeb71baca 100644 --- a/common/protob/messages-ethereum.proto +++ b/common/protob/messages-ethereum.proto @@ -52,7 +52,7 @@ message EthereumAddress { /** * Ethereum definitions type enum. - * Used to check the encoded EthereumNetworkInfo and/or EthereumTokenInfo message. + * Used to check the encoded EthereumNetworkInfo or EthereumTokenInfo message. */ enum EthereumDefinitionType { NETWORK = 0; diff --git a/common/protob/protocol.md b/common/protob/protocol.md index 45ecd8961c..ed7db9642c 100644 --- a/common/protob/protocol.md +++ b/common/protob/protocol.md @@ -21,16 +21,24 @@ Following packets has the following structure: | 1 | 63 | uint8_t[63] | following bytes of message encoded in Protocol Buffers (padded with zeroes if shorter) | ## Ethereum network and token definitions -Ethereum network and token definitions could be sent from host to device. Definitions are generated in our CI job and are provided -on publicly accessible website - on `data.trezor.io`. To ensure, that the definitions send to device are genuine, every generated -definition is signed and checked in FW. -Every definition (network/token) is encoded and has special format: +Ethereum network and token definitions could be sent from host to device. External definitions are generated from time to time +and provided on publicly accessible website - `data.trezor.io` # TODO: update url. For more info look +at the [definitions README.md](../defs/README.md#eth-and-erc20). +To ensure that the definitions send to device are genuine we use Merkle tree with a combination of signed Merkle tree root hash. +Every definition is then checked and verified at FW side on receiving. + + +Definitions (network/token) are binary encoded and have a special format: 1. prefix: - 1. part: format version of the definition (UTF-8 string `trzd` + version number, padded with zeroes if shorter, 8 bytes) - 2. part: type of data (unsigned integer, 1 byte) - 3. part: data version of the definition (unsigned integer, 4 bytes) -2. payload: serialized form of protobuf message EthereumNetworkInfo and/or EthereumTokenInfo (N bytes) -3. suffix: signature of prefix+payload (plain bits, 64 bytes) + 1. format version of the definition (UTF-8 string `trzd` + version number, padded with zeroes if shorter, 8 bytes) + 2. type of data (unsigned integer, 1 byte) + 3. data version of the definition (unsigned integer, 4 bytes) + 4. length of the encoded protobuf message - payload length in bytes (unsigned integer, 2 bytes) +2. payload: serialized form of protobuf message EthereumNetworkInfo or EthereumTokenInfo (N bytes) +3. suffix: + 1. length of the Merkle tree proof - number of hashes in the proof (unsigned integer, 1 byte) + 2. proof - individual hashes used to compute Merkle tree root hash (plain bits, N*32 bytes) + 3. signed Merkle tree root hash (plain bits, 64 bytes) ## Adding new message diff --git a/common/tools/ethereum_definitions.py b/common/tools/ethereum_definitions.py index 4a167a273b..04a2780481 100755 --- a/common/tools/ethereum_definitions.py +++ b/common/tools/ethereum_definitions.py @@ -10,7 +10,7 @@ import pathlib import re import shutil from collections import defaultdict -from typing import Any, Dict, List, TextIO, Tuple +from typing import Any, Dict, List, TextIO, Tuple, cast import click import ed25519 @@ -24,9 +24,9 @@ from coin_info import ( Coins, _load_builtin_erc20_tokens, _load_builtin_ethereum_networks, - coin_info, load_json, ) +from merkle_tree import MerkleTree from trezorlib import protobuf from trezorlib.messages import ( EthereumDefinitionType, @@ -677,19 +677,38 @@ def check_definitions_list( _set_definition_metadata(definition) -def _load_prepared_definitions(): - # set keys for the networks and tokens - # networks.append(cast(Coin, network)) - # key=f"eth:{shortcut}", +def _load_prepared_definitions(definitions_file: pathlib.Path) -> tuple[list[dict], list[dict]]: + if not definitions_file.is_file(): + click.ClickException(f"File {definitions_file} with prepared definitions does not exists or is not a file.") - # token.update( - # chain=chain, - # chain_id=network["chain_id"], - # address_bytes=bytes.fromhex(token["address"][2:]), - # shortcut=token["symbol"], - # key=f"erc20:{chain}:{token['symbol']}", - # ) - pass + prepared_definitions_data = load_json(definitions_file) + try: + networks_data = prepared_definitions_data["networks"] + tokens_data = prepared_definitions_data["tokens"] + except KeyError: + click.ClickException(f"File with prepared definitions is not complete. Whole \"networks\" and/or \"tokens\" section are missing.") + + networks: Coins = [] + for network_data in networks_data: + network_data.update( + chain_id=str(network_data["chain_id"]), + key=f"eth:{network_data['shortcut']}", + ) + networks.append(cast(Coin, network_data)) + + tokens: Coins = [] + + for token in tokens_data: + token.update( + chain_id=str(token["chain_id"]), + address=token["address"].lower(), + address_bytes=bytes.fromhex(token["address"][2:]), + symbol=token["shortcut"], + key=f"erc20:{token['chain']}:{token['shortcut']}", + ) + tokens.append(cast(Coin, token)) + + return networks, tokens # ====== coindefs generators ====== @@ -723,7 +742,10 @@ def serialize_eth_info( buf = io.BytesIO() protobuf.dump_message(buf, info) - ser += buf.getvalue() + msg = buf.getvalue() + # write the length of encoded protobuf message + ser += len(msg).to_bytes(2, "big") + ser += msg return ser @@ -930,9 +952,6 @@ def prepare_definitions( f.write("\n") -# TODO: separate function to generate built-in defs??? - - @cli.command() @click.option( "-o", @@ -946,7 +965,14 @@ def prepare_definitions( type=click.File(mode="r"), help="Private key (text, hex formated) to use to sign data. Could be also loaded from `PRIVATE_KEY` env variable. Provided file is preffered over env variable.", ) -def coindefs(outdir: pathlib.Path, privatekey: TextIO) -> None: +@click.option( + "-d", + "--deffile", + type=click.Path(resolve_path=True, dir_okay=False, path_type=pathlib.Path), + default="./definitions-latest.json", + help="File where the prepared definitions are saved in json format." +) +def sign_definitions(outdir: pathlib.Path, privatekey: TextIO, deffile: pathlib.Path) -> None: """Generate signed Ethereum definitions for python-trezor and others.""" hex_key = None if privatekey is None: @@ -962,68 +988,92 @@ def coindefs(outdir: pathlib.Path, privatekey: TextIO) -> None: sign_key = ed25519.SigningKey(ed25519.from_ascii(hex_key, encoding="hex")) def save_definition(directory: pathlib.Path, keys: list[str], data: bytes): - complete_filename = "_".join(keys) + ".dat" - with open(directory / complete_filename, mode="wb+") as f: + complete_file_path = directory / ("_".join(keys) + ".dat") + + if complete_file_path.exists(): + raise click.ClickException( + f"Definition \"{complete_file_path}\" already generated - attempt to generate another definition." + ) + + directory.mkdir(parents=True, exist_ok=True) + with open(complete_file_path, mode="wb+") as f: f.write(data) - def generate_token_defs(tokens: Coins, path: pathlib.Path): + def generate_token_defs(tokens: Coins): for token in tokens: - if token["address"] is None: + if token["address"] is None or token["chain_id"] is None: continue - # generate definition of the token - keys = ["token", token["address"][2:].lower()] - ser = serialize_eth_info( - eth_info_from_dict(token, EthereumTokenInfo), - EthereumDefinitionType.TOKEN, + # save token definition + save_definition( + outdir / "by_chain_id" / token["chain_id"], + ["token", token["address"][2:].lower()], + token["serialized"], ) - save_definition(path, keys, ser + sign_data(sign_key, ser)) - def generate_network_def(net: Coin, tokens: Coins): - if net["chain_id"] is None: + def generate_network_def(network: Coin): + if network["chain_id"] is None: return - # create path for networks identified by chain ids - network_dir = outdir / "by_chain_id" / str(net["chain_id"]) + # create path for networks identified by chain and slip44 ids + network_dir = outdir / "by_chain_id" / network["chain_id"] + slip44_dir = outdir / "by_slip44" / str(network["slip44"]) + # save network definition + save_definition(network_dir, ["network"], network["serialized"]) + try: - network_dir.mkdir(parents=True) - except FileExistsError: - raise click.ClickException( - f"Network with chain ID {net['chain_id']} already exists - attempt to generate defs for network \"{net['name']}\" ({net['shortcut']})." - ) + # TODO: this way only the first network with given slip is saved - save other networks?? + save_definition(slip44_dir, ["network"], network["serialized"]) + except click.ClickException: + pass - # generate definition of the network - keys = ["network"] - ser = serialize_eth_info( - eth_info_from_dict(net, EthereumNetworkInfo), EthereumDefinitionType.NETWORK - ) - complete_data = ser + sign_data(sign_key, ser) - save_definition(network_dir, keys, complete_data) - - # generate tokens for the network - generate_token_defs(tokens, network_dir) - - # create path for networks identified by slip44 ids - slip44_dir = outdir / "by_slip44" / str(net["slip44"]) - if not slip44_dir.exists(): - slip44_dir.mkdir(parents=True) - # TODO: save only first network?? - save_definition(slip44_dir, keys, complete_data) + # load prepared definitions + networks, tokens = _load_prepared_definitions(deffile) # clear defs directory if outdir.exists(): shutil.rmtree(outdir) outdir.mkdir(parents=True) - all_coins = coin_info.coin_info() + # serialize definitions + definitions_by_serialization: dict[bytes, dict] = dict() + for network in networks: + ser = serialize_eth_info( + eth_info_from_dict(network, EthereumNetworkInfo), EthereumDefinitionType.NETWORK + ) + network["serialized"] = ser + definitions_by_serialization[ser] = network + for token in tokens: + ser = serialize_eth_info( + eth_info_from_dict(token, EthereumTokenInfo), EthereumDefinitionType.TOKEN + ) + token["serialized"] = ser + definitions_by_serialization[ser] = token - # group tokens by their chain_id - token_buckets: CoinBuckets = defaultdict(list) - for token in all_coins.erc20: - token_buckets[token["chain_id"]].append(token) + # build Merkle tree + mt = MerkleTree( + [network["serialized"] for network in networks] + + [token["serialized"] for token in tokens] + ) - for network in all_coins.eth: - generate_network_def(network, token_buckets[network["chain_id"]]) + # sign tree root hash + signed_root_hash = sign_data(sign_key, mt.get_root_hash()) + + # update definitions + for ser, proof in mt.get_proofs().items(): + definition = definitions_by_serialization[ser] + # append number of hashes in proof + definition["serialized"] += len(proof).to_bytes(1, "big") + # append proof itself + for p in proof: + definition["serialized"] += p + # append signed tree root hash + definition["serialized"] += signed_root_hash + + for network in networks: + generate_network_def(network) + + generate_token_defs(tokens) if __name__ == "__main__": diff --git a/common/tools/merkle_tree.py b/common/tools/merkle_tree.py new file mode 100755 index 0000000000..d1c9b63eb0 --- /dev/null +++ b/common/tools/merkle_tree.py @@ -0,0 +1,79 @@ +from typing import Optional, Union + +try: + from trezor.crypto.hashlib import sha256 +except ImportError: + from hashlib import sha256 + + +class Node(): + """ + Single node of Merkle tree. + """ + def __init__(self: "Node", left: Union[bytes, "Node"], right: Optional["Node"] = None) -> None: + self.is_leaf = (left is None) != (right is None) # XOR + if self.is_leaf: + self.raw_value = left + self.hash = None + self.left_child = left + self.right_child = right + self.proof_list: list[bytes] = [] + + def compute_hash(self) -> bytes: + if not self.hash: + if self.is_leaf: + self.hash = sha256(b"\x00" + self.left_child).digest() + else: + left_hash = self.left_child.compute_hash() + right_hash = self.right_child.compute_hash() + hash_a = min(left_hash, right_hash) + hash_b = max(left_hash, right_hash) + self.hash = sha256(b"\x01" + hash_a + hash_b).digest() + + # distribute proof + self.left_child.add_to_proof(right_hash) + self.right_child.add_to_proof(left_hash) + + return self.hash + + def add_to_proof(self, proof: bytes) -> None: + self.proof_list.append(proof) + if not self.is_leaf: + self.left_child.add_to_proof(proof) + self.right_child.add_to_proof(proof) + + +class MerkleTree(): + """ + Simple Merkle tree that implements the building of Merkle tree itself and generate proofs + for leaf nodes. + """ + def __init__(self, values: list[bytes]) -> None: + self.leaves = [Node(v) for v in values] + + # build the tree + actual_level = [n for n in self.leaves] + while len(actual_level) > 1: + # build one level of the tree + next_level = [] + while len(actual_level) // 2: + left_node = actual_level.pop(0) + right_node = actual_level.pop(0) + next_level.append(Node(left_node, right_node)) + + if len(actual_level) == 1: + # odd number of nodes on actual level so last node will be "joined" on another level + next_level.append(actual_level.pop(0)) + + # switch levels and continue + actual_level = next_level + + # set root and compute hash + self.root_node = actual_level[0] + self.root_node.compute_hash() + + def get_proofs(self) -> dict[bytes, list[bytes]]: + return {n.raw_value: n.proof_list for n in self.leaves} + + def get_root_hash(self) -> bytes: + return self.root_node.hash diff --git a/core/src/apps/ethereum/definitions.py b/core/src/apps/ethereum/definitions.py index 7d6acaf654..ab6fddbe77 100644 --- a/core/src/apps/ethereum/definitions.py +++ b/core/src/apps/ethereum/definitions.py @@ -27,15 +27,31 @@ if __debug__: class EthereumDefinitionParser: def __init__(self, definition_bytes: bytes) -> None: - if len(definition_bytes) <= (8 + 1 + 4 + 64): - raise wire.DataError("Received Ethereum definition is probably malformed (too few data).") + actual_position = 0 - self.format_version: str = definition_bytes[:8].rstrip(b'\0').decode("utf-8") - self.definition_type: int = definition_bytes[8] - self.data_version: int = int.from_bytes(definition_bytes[9:13], 'big') - self.clean_payload = definition_bytes[13:-64] - self.payload = definition_bytes[:-64] - self.signature = definition_bytes[-64:] + try: + # prefix + self.format_version = definition_bytes[:8].rstrip(b'\0').decode("utf-8") + self.definition_type: int = definition_bytes[8] + self.data_version = int.from_bytes(definition_bytes[9:13], 'big') + self.payload_length_in_bytes = int.from_bytes(definition_bytes[13:15], 'big') + actual_position += 8 + 1 + 4 + 2 + + # payload + self.payload = definition_bytes[actual_position:(actual_position + self.payload_length_in_bytes)] + self.payload_with_prefix = definition_bytes[:(actual_position + self.payload_length_in_bytes)] + actual_position += self.payload_length_in_bytes + + # suffix - Merkle tree proof and signed root hash + self.proof_length: int = definition_bytes[actual_position] + actual_position += 1 + self.proof: list[bytes] = [] + for _ in range(self.proof_length): + self.proof.append(definition_bytes[actual_position:(actual_position + 32)]) + actual_position += 32 + self.signed_tree_root = definition_bytes[actual_position:(actual_position + 64)] + except IndexError: + raise wire.DataError("Invalid Ethereum definition.") def decode_definition( @@ -56,19 +72,32 @@ def decode_definition( if parsed_definition.data_version < MIN_DATA_VERSION: raise wire.DataError("Used Ethereum definition data version too low.") - # at the end verify the signature - if not ed25519.verify(DEFINITIONS_PUBLIC_KEY, parsed_definition.signature, parsed_definition.payload): + # at the end verify the signature - compute Merkle tree root hash using provided leaf data and proof + def compute_mt_root_hash(data: bytes, proof: list[bytes]) -> bytes: + from trezor.crypto.hashlib import sha256 + hash = sha256(b"\x00" + data).digest() + for p in proof: + hash_a = min(hash, p) + hash_b = max(hash, p) + hash = sha256(b"\x01" + hash_a + hash_b).digest() + + return hash + + # verify Merkle proof + root_hash = compute_mt_root_hash(parsed_definition.payload_with_prefix, parsed_definition.proof) + + if not ed25519.verify(DEFINITIONS_PUBLIC_KEY, parsed_definition.signed_tree_root, root_hash): error_msg = wire.DataError("Ethereum definition signature is invalid.") if __debug__: # check against dev key - if not ed25519.verify(DEFINITIONS_DEV_PUBLIC_KEY, parsed_definition.signature, parsed_definition.payload): + if not ed25519.verify(DEFINITIONS_DEV_PUBLIC_KEY, parsed_definition.signed_tree_root, root_hash): raise error_msg else: raise error_msg # decode it if it's OK if expected_type == EthereumDefinitionType.NETWORK: - info = protobuf.decode(parsed_definition.clean_payload, EthereumNetworkInfo, True) + info = protobuf.decode(parsed_definition.payload, EthereumNetworkInfo, True) # TODO: temporarily convert to internal class if info is not None: @@ -81,7 +110,7 @@ def decode_definition( rskip60=info.rskip60 ) else: - info = protobuf.decode(parsed_definition.clean_payload, EthereumTokenInfo, True) + info = protobuf.decode(parsed_definition.payload, EthereumTokenInfo, True) # TODO: temporarily convert to internal class if info is not None: diff --git a/core/src/apps/ethereum/sign_tx.py b/core/src/apps/ethereum/sign_tx.py index 74a7df0e9f..dce8cf4421 100644 --- a/core/src/apps/ethereum/sign_tx.py +++ b/core/src/apps/ethereum/sign_tx.py @@ -12,7 +12,7 @@ if TYPE_CHECKING: from trezor.messages import EthereumSignTx, EthereumTxAck from trezor.wire import Context - from .keychain import EthereumSignTxAny + from .keychain import MsgInKeychainChainIdDefs from . import tokens, definitions @@ -100,7 +100,7 @@ async def sign_tx( async def handle_erc20( - ctx: Context, msg: EthereumSignTxAny, token_dict: dict[bytes, tokens.TokenInfo] + ctx: Context, msg: MsgInKeychainChainIdDefs, token_dict: dict[bytes, tokens.TokenInfo] ) -> tuple[tokens.TokenInfo | None, bytes, bytes, int]: from .layout import require_confirm_unknown_token from . import tokens @@ -185,7 +185,7 @@ def _sign_digest( return req -def check_common_fields(msg: EthereumSignTxAny) -> None: +def check_common_fields(msg: MsgInKeychainChainIdDefs) -> None: data_length = msg.data_length # local_cache_attribute if data_length > 0: diff --git a/core/src/apps/ethereum/tokens.py b/core/src/apps/ethereum/tokens.py index 1ac6c67eff..4de7234431 100644 --- a/core/src/apps/ethereum/tokens.py +++ b/core/src/apps/ethereum/tokens.py @@ -35,7 +35,6 @@ class TokenInfo: UNKNOWN_TOKEN = TokenInfo("Wei UNKN", 0, b"", 0) -# TODO: delete completely def token_by_chain_address(chain_id: int, address: bytes) -> TokenInfo: for addr, symbol, decimal in _token_iterator(chain_id): if address == addr: diff --git a/core/src/apps/ethereum/tokens.py.mako b/core/src/apps/ethereum/tokens.py.mako index 9936565420..2053bc9dc9 100644 --- a/core/src/apps/ethereum/tokens.py.mako +++ b/core/src/apps/ethereum/tokens.py.mako @@ -44,7 +44,6 @@ class TokenInfo: UNKNOWN_TOKEN = TokenInfo("Wei UNKN", 0, b"", 0) -# TODO: delete completely def token_by_chain_address(chain_id: int, address: bytes) -> TokenInfo: for addr, symbol, decimal in _token_iterator(chain_id): if address == addr: diff --git a/core/tests/ethereum_common.py b/core/tests/ethereum_common.py index 4aa5f1a5ec..3c44418f46 100644 --- a/core/tests/ethereum_common.py +++ b/core/tests/ethereum_common.py @@ -1,12 +1,13 @@ +from typing import Iterator from trezor.utils import ensure -from ubinascii import unhexlify # noqa: F401 +from ubinascii import hexlify, unhexlify # noqa: F401 from trezor import messages from apps.ethereum import networks, tokens EXPECTED_FORMAT_VERSION = 1 -EXPECTED_DATA_VERSION = 1657791892 # unix epoch time Thu Jul 14 2022 09:44:52 GMT+0000 +EXPECTED_DATA_VERSION = 1663054984 # unix epoch time class InfoWithDefinition(): def __init__(self, definition, info): @@ -14,12 +15,12 @@ class InfoWithDefinition(): self.info = info -# definitions created by `common/tools/cointool.py` and copied here +# definitions created by `common/tools/ethereum_definitions.py` and copied here NETWORKS = { # chain_id: network info with encoded definition # Ethereum 1: InfoWithDefinition( - definition=unhexlify("74727a64310000000062d6384d0801103c1a034554482208457468657265756d28003342fb0073eb26285e8b50f402a346fee0d37b98721c516d13864043e54d7e33e31d82004024495138bc72b946b080f5d319ea41c8ba53aaffbf7988419f860e"), + definition=None, # built-in definitions are not encoded info=networks.NetworkInfo( chain_id=1, slip44=60, @@ -28,20 +29,20 @@ NETWORKS = { rskip60=False, ), ), - # Expanse - 2: InfoWithDefinition( - definition=unhexlify("74727a64310000000062d6384d080210281a03455850220f457870616e7365204e6574776f726b2800b5f8aba1a056398b340fc16d66ca845db1cbd258cb1a560b6d1d227655ff1c88cd785a3005c31cc2a1da3b20edc69c09d470d0ebabb19f52e3cfb7139dac9104"), + # Rinkeby + 4: InfoWithDefinition( + definition=unhexlify("74727a643100000000632034880015080410011a047452494e220752696e6b65627928000e8cc47ed4e657d9a9b98e1dd02164320c54a9724e17f91d1d79f6760169582c98ec70ca6f4e94d27e574175c59d2ae04e0cd30b65fb19acd8d2c5fb90bcb7db96f6102e4182c0cef5f412ac3c5fa94f9505b4df2633a0f7bdffa309588d722415624adeb8f329b1572ff9dfc81fbc86e61f1fcb2369f51ba85ea765c908ac254ba996f842a6277583f8d02f149c78bc0eeb8f3d41240403f85785dc3a3925ea768d76aae12342c8a24de223c1ea75e5f07f6b94b8f22189413631eed3c9a362b4501f68b645aa487b9d159a8161404a218507641453ebf045cec56710bb7d873e102777695b56903766e1af16f95576ec4f41874bdaf80cec02ee067d30e721515564d4f30fa74a6c61eb784ea65cc881ead7af2ffac02d5bf1fe1a756918fe37b74828a24b640025cd79443ada60063e3034444fc49ed6055dbba6a09fa4484c42cb85abb49103dc8c781c8f190c4632e2dec30081770448021313955dbb49e8a02fd49b34d030280452fe0a5c3bcba4958bc287c67e12519be4f4aec7ab0c8e574e53a663f635f75508f23d92c77b2147f29feb79c38d0f793fba295aae605c7e8226523edefc6ad1eefe088e5b8376028bf90116ece4fb876510b4ae1c89686dbcaacbbac8225baba429ca376fafac50f4bd1ff4ce1c61dd53318d0718bf513ea6f770cce81e07a653622e4dbd03bdaa570bfe43219eb0d4fab725c9a8da04"), info=networks.NetworkInfo( - chain_id=2, - slip44=40, - shortcut="EXP", - name="Expanse Network", + chain_id=4, + slip44=1, + shortcut="tRIN", + name="Rinkeby", rskip60=False, ), ), # Ubiq 8: InfoWithDefinition( - definition=unhexlify("74727a64310000000062d6384d0808106c1a035542512204556269712800505661237b6e6c5ff7a7b4b6e9f5d6dbf70055507bb6ca48e1260e76a2a01a1788038d80588d643da576183842d6367d7f1c9fefc15d56fd8b15ee8f1165bb0c"), + definition=unhexlify("74727a6431000000006320348800110808106c1a0355425122045562697128000e5641d82e3622b4e6addd4354efd933cf15947d1d608a60d324d1156b5a4999f70c41beb85bd866aa3059123447dfeef2e1b6c009b66ac8d04ebbca854ad30049edbbb2fbfda3bfedc6fdb4a76f1db8a4f210bd89d3c3ec1761157b0ec2b13e2f624adeb8f329b1572ff9dfc81fbc86e61f1fcb2369f51ba85ea765c908ac254ba996f842a6277583f8d02f149c78bc0eeb8f3d41240403f85785dc3a3925ea768d76aae12342c8a24de223c1ea75e5f07f6b94b8f22189413631eed3c9a362b4501f68b645aa487b9d159a8161404a218507641453ebf045cec56710bb7d873e102777695b56903766e1af16f95576ec4f41874bdaf80cec02ee067d30e721515564d4f30fa74a6c61eb784ea65cc881ead7af2ffac02d5bf1fe1a756918fe37b74828a24b640025cd79443ada60063e3034444fc49ed6055dbba6a09fa4484c42cb85abb49103dc8c781c8f190c4632e2dec30081770448021313955dbb49e8a02fd49b34d030280452fe0a5c3bcba4958bc287c67e12519be4f4aec7ab0c8e574e53a663f635f75508f23d92c77b2147f29feb79c38d0f793fba295aae605c7e8226523edefc6ad1eefe088e5b8376028bf90116ece4fb876510b4ae1c89686dbcaacbbac8225baba429ca376fafac50f4bd1ff4ce1c61dd53318d0718bf513ea6f770cce81e07a653622e4dbd03bdaa570bfe43219eb0d4fab725c9a8da04"), info=networks.NetworkInfo( chain_id=8, slip44=108, @@ -52,7 +53,7 @@ NETWORKS = { ), # Ethereum Classic 61: InfoWithDefinition( - definition=unhexlify("74727a64310000000062d6384d083d103d1a034554432210457468657265756d20436c61737369632800095cb5be721429440fd9856d7c35acc7751c9cb1d006189a78f5693654b36c39bb4ab5add8d5cd6af3c345a8aa20307f79567c78c4f5940e3f16221a0c5bd30f"), + definition=unhexlify("74727a64310000000063203488001d083d103d1a034554432210457468657265756d20436c617373696328000e6b891a57fe4c38c54b475f22f0d9242dd8ddab0b4f360bd86e37e2e8b79de5ef29237436351f7bc924cd110716b5adde7c28c03d76ac83b091dbce1b5d7d0edbddb221bd894806f7ea1b195443176e06830a83c0204e33f19c51d2fccc3a9f80ac2cca38822db998ddf76778dada240d39b3c6193c6335d7c693dea90d19a41f86855375c2f48c18cdc012ccac771aa316d776c8721c2b1f6d5980808337dfdae13b5be07e3cbc3526119b88c5eb44be0b1dab1094a5ec5215b47daf91736d16501f68b645aa487b9d159a8161404a218507641453ebf045cec56710bb7d873e102777695b56903766e1af16f95576ec4f41874bdaf80cec02ee067d30e721515564d4f30fa74a6c61eb784ea65cc881ead7af2ffac02d5bf1fe1a756918fe37b74828a24b640025cd79443ada60063e3034444fc49ed6055dbba6a09fa4484c42cb85abb49103dc8c781c8f190c4632e2dec30081770448021313955dbb49e8a02fd49b34d030280452fe0a5c3bcba4958bc287c67e12519be4f4aec7ab0c8e574e53a663f635f75508f23d92c77b2147f29feb79c38d0f793fba295aae605c7e8226523edefc6ad1eefe088e5b8376028bf90116ece4fb876510b4ae1c89686dbcaacbbac8225baba429ca376fafac50f4bd1ff4ce1c61dd53318d0718bf513ea6f770cce81e07a653622e4dbd03bdaa570bfe43219eb0d4fab725c9a8da04"), info=networks.NetworkInfo( chain_id=61, slip44=61, @@ -62,21 +63,23 @@ NETWORKS = { ), ), } + + SLIP44_TO_CHAIN_ID_MAP = { - 40: [2], + 1: [4], 60: [1], 61: [61], 108: [8], } -# definitions created by `common/tools/cointool.py` and copied here +# definitions created by `common/tools/ethereum_definitions.py` and copied here TOKENS = { # chain_id: { address: token info with encoded definition, address: token info with encoded definition,... } 1: { # AAVE "7fc66500c84a76ad7e9c93437bfc5ac33e2ddae9": InfoWithDefinition( - definition=unhexlify("74727a64310000000162d6384d0a044141564510121a147fc66500c84a76ad7e9c93437bfc5ac33e2ddae920012a04416176652d5458fe6726f8dfad0af3d1bd0959199da25dba4f429c14517f54acdd4b4117b676f7467bb34fe9a64dbd95d1a875085774df7b30fdf09f0b33f4e52b2d900b"), + definition=None, # built-in definitions are not encoded info=tokens.TokenInfo( symbol="AAVE", decimals=18, @@ -84,16 +87,26 @@ TOKENS = { chain_id=1, ), ), - }, - 61: { - # BEC - "085fb4f24031eaedbc2b611aa528f22343eb52db": InfoWithDefinition( - definition=unhexlify("74727a64310000000162d68cd70a0342454310081a14085fb4f24031eaedbc2b611aa528f22343eb52db203d2a03424543194264a2b334a93b8592997d311d65f1e0840c150eab051fd93cce8997d5200705e5d256abfcdd3c7b68fc5c72ebe6e32c5205d8f60c47d0986d313a4a92c80a"), + # TrueAUD + "00006100f7090010005f1bd7ae6122c3c2cf0090": InfoWithDefinition( + definition=unhexlify("74727a6431000000016320348800290a045441554410121a1400006100f7090010005f1bd7ae6122c3c2cf009020012a07547275654155440e310dad13f7d3012903a9a457134c9f38c62c04370cb92c7a528838e30a032dffbceeaa2aa849e590c4e6dbc69b0ea5359f3527b95b56ab59a33dc584105b35ea7c06afc296cc1c1e58cc3d6b461631c4c770b9409837ab3d29bc1b666fb9cf5245c4c218b0e9521c185d102f596905ba860e6f56a0a8b394f943855c74eea6fcac87210a9988ac02803f4cc61cf78e7e2409175a75f4f3a82eb84b1f2d1ea8177d5dccd62949d80d7942105e22a452be01859fe816736e803b120fb9bcc0c1117180dbda19e1ad1aafb9b9f1555c75275820bf7c1e568bcb265bdc4dfdae0511782026e11a151f6894d11128327c8c42958c9ae900af970fec13a11ffdeba6ac10733ca55a906142e0b9130312e8e85606108612581aca9087c452f38f14185db74828a24b640025cd79443ada60063e3034444fc49ed6055dbba6a09fa4484c42cb85abb49103dc8c781c8f190c4632e2dec30081770448021313955dbb49e8a02fd49b34d030280452fe0a5c3bcba4958bc287c67e12519be4f4aec7ab0c8e574e53a663f635f75508f23d92c77b2147f29feb79c38d0f793fba295aae605c7e8226523edefc6ad1eefe088e5b8376028bf90116ece4fb876510b4ae1c89686dbcaacbbac8225baba429ca376fafac50f4bd1ff4ce1c61dd53318d0718bf513ea6f770cce81e07a653622e4dbd03bdaa570bfe43219eb0d4fab725c9a8da04"), info=tokens.TokenInfo( - symbol="BEC", - decimals=8, - address=unhexlify("085fb4f24031eaedbc2b611aa528f22343eb52db"), - chain_id=61, + symbol="TAUD", + decimals=18, + address=unhexlify("00006100f7090010005f1bd7ae6122c3c2cf0090"), + chain_id=1, + ), + ), + }, + 4: { + # Karma Token + "275a5b346599b56917e7b1c9de019dcf9ead861a": InfoWithDefinition( + definition=unhexlify("74727a64310000000163203488002b0a024b4310121a14275a5b346599b56917e7b1c9de019dcf9ead861a20042a0b4b61726d6120546f6b656e0e2b3cb176ff5a2cf431620c1a7eee9aa297f5de36d29ae6d423166cf7391e41c5826c57f30b11421a4bf10f336f12050f6d959e02bfb17a8ce7ae15087d4f083124c0cebed2ce45b15b2608b1a8f0ee443e8c4f33111d880a6a3c09a77c627f82d68b62a1bd39975b2a2c86f196b9a3dcb62bdc3554fbf85b75331bc0d39f23a46f5ed91f208757d1136bb20b3618294fbfb0a826e9c09e392fe8109181bc6c28cad78db1987947f461bfc1042b88a91d6d61297d0cf194dfeea981b4515c2ed09dc2966671f5c715c64ceb25e53e1df3c7234e3e0ddf0dcd54d40fde0c51903685f9dc7fa69c71184f17af852e74490ea7286e89a0aa4770629664f7dd8eab8c4e009ff4c24682f85f7e01d4e10ae5c06212d5a4f43bac2b4f0e79383666ef12054ddbf757809aa6b446d65f7fd1bdd76fb1d7770398bd17af50635027e680801d244bd7b4f14c57edc3cd961722315e076120bf1d35db8520edb812bfbb5bab8ff57cc2dc1b3d1f9d95b33dba5d759aef1123f2ef346b6328973fba204fd745e644c8e492f9a76c0019b2cf21715fba682b46b9c58013e0b0927e5272c808a67e8226523edefc6ad1eefe088e5b8376028bf90116ece4fb876510b4ae1c89686dbcaacbbac8225baba429ca376fafac50f4bd1ff4ce1c61dd53318d0718bf513ea6f770cce81e07a653622e4dbd03bdaa570bfe43219eb0d4fab725c9a8da04"), + info=tokens.TokenInfo( + symbol="KC", + decimals=18, + address=unhexlify("275a5b346599b56917e7b1c9de019dcf9ead861a"), + chain_id=4, ), ), }, @@ -187,3 +200,22 @@ def get_ethereum_encoded_definition(chain_id: int | None = None, slip44: int | N encoded_network=get_ethereum_encoded_network_definition(chain_id, slip44), encoded_token=None, # TODO ) + + +def builtin_networks_iterator() -> Iterator[networks.NetworkInfo]: + """Mockup function replaces original function from core/src/apps/ethereum/networks.py used to get built-in network definitions.""" + for _, network in NETWORKS.items(): + if network.definition is None: + yield network.info + + +def builtin_token_by_chain_address(chain_id: int, address: bytes) -> tokens.TokenInfo: + """Mockup function replaces original function from core/src/apps/ethereum/tokens.py used to get built-in token definitions.""" + address_str = hexlify(address).decode('hex') + try: + if TOKENS[chain_id][address_str].definition is None: + return TOKENS[chain_id][address_str].info + except KeyError: + pass + + return tokens.UNKNOWN_TOKEN diff --git a/core/tests/test_apps.ethereum.definitions.py b/core/tests/test_apps.ethereum.definitions.py index 48147fa9a7..b0ea4ab646 100644 --- a/core/tests/test_apps.ethereum.definitions.py +++ b/core/tests/test_apps.ethereum.definitions.py @@ -23,26 +23,38 @@ if not utils.BITCOIN_ONLY: @unittest.skipUnless(not utils.BITCOIN_ONLY, "altcoin") class TestEthereumDefinitionParser(unittest.TestCase): def setUp(self): + # prefix self.format_version = b'trzd1' + b'\x00' * 3 - self.definition_type = b'\x02' - self.data_version = b'\x00\x00\x00\x03' - self.clean_payload = b'\x04' # optional length - self.payload = self.format_version + self.definition_type + self.data_version + self.clean_payload - self.signature = b'\x00' * 63 + b'\x05' - self.definition = self.payload + self.signature + self.definition_type = b'\x01' + self.data_version = b'\x00\x00\x00\x02' + self.payload_length_in_bytes = b'\x00\x03' + self.prefix = self.format_version + self.definition_type + self.data_version + self.payload_length_in_bytes + + # payload + self.payload = b'\x00\x00\x04' # optional length + self.payload_with_prefix = self.prefix + self.payload + + # suffix - Merkle tree proof and signed root hash + self.proof_length = b'\x01' + self.proof = b'\x00' * 31 + b'\x06' + self.signed_tree_root = b'\x00' * 63 + b'\x07' + self.definition = self.payload_with_prefix + self.proof_length + self.proof + self.signed_tree_root def test_short_message(self): with self.assertRaises(wire.DataError): - dfs.EthereumDefinitionParser(b'\x00' * (len(self.definition) - 1)) + dfs.EthereumDefinitionParser(b'\x00') def test_ok_message(self): parser = dfs.EthereumDefinitionParser(self.definition) self.assertEqual(parser.format_version, self.format_version.rstrip(b'\0').decode("utf-8")) self.assertEqual(parser.definition_type, int.from_bytes(self.definition_type, 'big')) self.assertEqual(parser.data_version, int.from_bytes(self.data_version, 'big')) - self.assertEqual(parser.clean_payload, self.clean_payload) + self.assertEqual(parser.payload_length_in_bytes, int.from_bytes(self.payload_length_in_bytes, 'big')) self.assertEqual(parser.payload, self.payload) - self.assertEqual(parser.signature, self.signature) + self.assertEqual(parser.payload_with_prefix, self.payload_with_prefix) + self.assertEqual(parser.proof_length, int.from_bytes(self.proof_length, 'big')) + self.assertEqual(parser.proof, [self.proof]) + self.assertEqual(parser.signed_tree_root, self.signed_tree_root) @unittest.skipUnless(not utils.BITCOIN_ONLY, "altcoin") @@ -53,38 +65,38 @@ class TestDecodeDefinition(unittest.TestCase): # successful decode network def test_network_definition(self): - eth_network = get_ethereum_network_info_with_definition(chain_id=1) - self.assertEqual(dfs.decode_definition(eth_network.definition, EthereumDefinitionType.NETWORK), eth_network.info) + rinkeby_network = get_ethereum_network_info_with_definition(chain_id=4) + self.assertEqual(dfs.decode_definition(rinkeby_network.definition, EthereumDefinitionType.NETWORK), rinkeby_network.info) # successful decode token def test_token_definition(self): - # AAVE - eth_token = get_ethereum_token_info_with_definition(chain_id=1, token_address="7fc66500c84a76ad7e9c93437bfc5ac33e2ddae9") - self.assertEqual(dfs.decode_definition(eth_token.definition, EthereumDefinitionType.TOKEN), eth_token.info) + # Karma Token + kc_token = get_ethereum_token_info_with_definition(chain_id=4) + self.assertEqual(dfs.decode_definition(kc_token.definition, EthereumDefinitionType.TOKEN), kc_token.info) def test_invalid_data(self): - eth_network = get_ethereum_network_info_with_definition(chain_id=1) + rinkeby_network = get_ethereum_network_info_with_definition(chain_id=4) invalid_dataset = [] # mangle signature - invalid_dataset.append(bytearray(eth_network.definition)) + invalid_dataset.append(bytearray(rinkeby_network.definition)) invalid_dataset[-1][-1] += 1 # mangle payload - invalid_dataset.append(bytearray(eth_network.definition)) - invalid_dataset[-1][-65] += 1 + invalid_dataset.append(bytearray(rinkeby_network.definition)) + invalid_dataset[-1][16] += 1 # wrong format version - invalid_dataset.append(bytearray(eth_network.definition)) + invalid_dataset.append(bytearray(rinkeby_network.definition)) invalid_dataset[-1][:5] = b'trzd2' # change "trzd1" to "trzd2" # wrong definition type - invalid_dataset.append(bytearray(eth_network.definition)) + invalid_dataset.append(bytearray(rinkeby_network.definition)) invalid_dataset[-1][8] += 1 # wrong data format version - invalid_dataset.append(bytearray(eth_network.definition)) + invalid_dataset.append(bytearray(rinkeby_network.definition)) invalid_dataset[-1][13] += 1 for data in invalid_dataset: @@ -92,15 +104,17 @@ class TestDecodeDefinition(unittest.TestCase): dfs.decode_definition(bytes(data), EthereumDefinitionType.NETWORK) def test_wrong_requested_type(self): - eth_network = get_ethereum_network_info_with_definition(chain_id=1) + rinkeby_network = get_ethereum_network_info_with_definition(chain_id=4) with self.assertRaises(wire.DataError): - dfs.decode_definition(eth_network.definition, EthereumDefinitionType.TOKEN) + dfs.decode_definition(rinkeby_network.definition, EthereumDefinitionType.TOKEN) @unittest.skipUnless(not utils.BITCOIN_ONLY, "altcoin") class TestGetNetworkDefiniton(unittest.TestCase): def setUp(self): self.addTypeEqualityFunc(networks.NetworkInfo, equalNetworkInfo) + # use mockup function for built-in networks + networks._networks_iterator = builtin_networks_iterator def test_get_network_definition(self): eth_network = get_ethereum_network_info_with_definition(chain_id=1) @@ -125,11 +139,11 @@ class TestGetNetworkDefiniton(unittest.TestCase): # reference chain_id is used to check the encoded network chain_id - so in case they do not equal # error is raised with self.assertRaises(wire.DataError): - dfs._get_network_definiton(ubiq_network.definition, ubiq_network.info.chain_id + 1) + dfs._get_network_definiton(ubiq_network.definition, ubiq_network.info.chain_id + 9999) def test_invalid_encoded_definition(self): - eth_network = get_ethereum_network_info_with_definition(chain_id=1) - definition = bytearray(eth_network.definition) + rinkeby_network = get_ethereum_network_info_with_definition(chain_id=4) + definition = bytearray(rinkeby_network.definition) # mangle signature - this should have the same effect as it has in "decode_definition" function definition[-1] += 1 with self.assertRaises(wire.DataError): @@ -140,41 +154,43 @@ class TestGetNetworkDefiniton(unittest.TestCase): class TestGetTokenDefiniton(unittest.TestCase): def setUp(self): self.addTypeEqualityFunc(tokens.TokenInfo, equalTokenInfo) + # use mockup function for built-in tokens + tokens.token_by_chain_address = builtin_token_by_chain_address def test_get_token_definition(self): - eth_token = get_ethereum_token_info_with_definition(chain_id=1) - self.assertEqual(dfs._get_token_definiton(None, eth_token.info.chain_id, eth_token.info.address), eth_token.info) + aave_token = get_ethereum_token_info_with_definition(chain_id=1, token_address="7fc66500c84a76ad7e9c93437bfc5ac33e2ddae9") + self.assertEqual(dfs._get_token_definiton(None, aave_token.info.chain_id, aave_token.info.address), aave_token.info) def test_built_in_preference(self): - eth_token = get_ethereum_token_info_with_definition(chain_id=1) - eth_classic_token = get_ethereum_token_info_with_definition(chain_id=61) - self.assertEqual(dfs._get_token_definiton(eth_classic_token.definition, eth_token.info.chain_id, eth_token.info.address), eth_token.info) + aave_token = get_ethereum_token_info_with_definition(chain_id=1, token_address="7fc66500c84a76ad7e9c93437bfc5ac33e2ddae9") + taud_token = get_ethereum_token_info_with_definition(chain_id=1, token_address="00006100f7090010005f1bd7ae6122c3c2cf0090") + self.assertEqual(dfs._get_token_definiton(taud_token.definition, aave_token.info.chain_id, aave_token.info.address), aave_token.info) def test_no_built_in(self): - eth_classic_token = get_ethereum_token_info_with_definition(chain_id=61) + kc_token = get_ethereum_token_info_with_definition(chain_id=4) # use provided (encoded) definition - self.assertEqual(dfs._get_token_definiton(eth_classic_token.definition, eth_classic_token.info.chain_id, eth_classic_token.info.address), eth_classic_token.info) + self.assertEqual(dfs._get_token_definiton(kc_token.definition, kc_token.info.chain_id, kc_token.info.address), kc_token.info) # here the results should be the same as above - self.assertEqual(dfs._get_token_definiton(eth_classic_token.definition, None, eth_classic_token.info.address), eth_classic_token.info) - self.assertEqual(dfs._get_token_definiton(eth_classic_token.definition, eth_classic_token.info.chain_id, None), eth_classic_token.info) - self.assertEqual(dfs._get_token_definiton(eth_classic_token.definition, None, None), eth_classic_token.info) + self.assertEqual(dfs._get_token_definiton(kc_token.definition, None, kc_token.info.address), kc_token.info) + self.assertEqual(dfs._get_token_definiton(kc_token.definition, kc_token.info.chain_id, None), kc_token.info) + self.assertEqual(dfs._get_token_definiton(kc_token.definition, None, None), kc_token.info) # nothing should be found - self.assertEqual(dfs._get_token_definiton(None, eth_classic_token.info.chain_id, eth_classic_token.info.address), tokens.UNKNOWN_TOKEN) - self.assertEqual(dfs._get_token_definiton(None, None, eth_classic_token.info.address), tokens.UNKNOWN_TOKEN) - self.assertEqual(dfs._get_token_definiton(None, eth_classic_token.info.chain_id, None), tokens.UNKNOWN_TOKEN) + self.assertEqual(dfs._get_token_definiton(None, kc_token.info.chain_id, kc_token.info.address), tokens.UNKNOWN_TOKEN) + self.assertEqual(dfs._get_token_definiton(None, None, kc_token.info.address), tokens.UNKNOWN_TOKEN) + self.assertEqual(dfs._get_token_definiton(None, kc_token.info.chain_id, None), tokens.UNKNOWN_TOKEN) # reference chain_id and/or token address is used to check the encoded token chain_id/address - so in case they do not equal # tokens.UNKNOWN_TOKEN is returned - self.assertEqual(dfs._get_token_definiton(eth_classic_token.definition, eth_classic_token.info.chain_id + 1, eth_classic_token.info.address + b"\x00"), tokens.UNKNOWN_TOKEN) - self.assertEqual(dfs._get_token_definiton(eth_classic_token.definition, eth_classic_token.info.chain_id, eth_classic_token.info.address + b"\x00"), tokens.UNKNOWN_TOKEN) - self.assertEqual(dfs._get_token_definiton(eth_classic_token.definition, eth_classic_token.info.chain_id + 1, eth_classic_token.info.address), tokens.UNKNOWN_TOKEN) - self.assertEqual(dfs._get_token_definiton(eth_classic_token.definition, None, eth_classic_token.info.address + b"\x00"), tokens.UNKNOWN_TOKEN) - self.assertEqual(dfs._get_token_definiton(eth_classic_token.definition, eth_classic_token.info.chain_id + 1, None), tokens.UNKNOWN_TOKEN) + self.assertEqual(dfs._get_token_definiton(kc_token.definition, kc_token.info.chain_id + 1, kc_token.info.address + b"\x00"), tokens.UNKNOWN_TOKEN) + self.assertEqual(dfs._get_token_definiton(kc_token.definition, kc_token.info.chain_id, kc_token.info.address + b"\x00"), tokens.UNKNOWN_TOKEN) + self.assertEqual(dfs._get_token_definiton(kc_token.definition, kc_token.info.chain_id + 1, kc_token.info.address), tokens.UNKNOWN_TOKEN) + self.assertEqual(dfs._get_token_definiton(kc_token.definition, None, kc_token.info.address + b"\x00"), tokens.UNKNOWN_TOKEN) + self.assertEqual(dfs._get_token_definiton(kc_token.definition, kc_token.info.chain_id + 1, None), tokens.UNKNOWN_TOKEN) def test_invalid_encoded_definition(self): - eth_token = get_ethereum_token_info_with_definition(chain_id=1) - definition = bytearray(eth_token.definition) + kc_token = get_ethereum_token_info_with_definition(chain_id=4) + definition = bytearray(kc_token.definition) # mangle signature - this should have the same effect as it has in "decode_definition" function definition[-1] += 1 with self.assertRaises(wire.DataError): @@ -182,10 +198,13 @@ class TestGetTokenDefiniton(unittest.TestCase): @unittest.skipUnless(not utils.BITCOIN_ONLY, "altcoin") -class TestEthereumDefinitons(unittest.TestCase): +class TestEthereumDefinitions(unittest.TestCase): def setUp(self): self.addTypeEqualityFunc(networks.NetworkInfo, equalNetworkInfo) self.addTypeEqualityFunc(tokens.TokenInfo, equalTokenInfo) + # use mockup functions for built-in definitions + networks._networks_iterator = builtin_networks_iterator + tokens.token_by_chain_address = builtin_token_by_chain_address def get_and_compare_ethereum_definitions( self, @@ -208,34 +227,38 @@ class TestEthereumDefinitons(unittest.TestCase): self.assertDictEqual(definitions.token_dict, ref_token_dict) def test_get_definitions(self): + # built-in eth_network = get_ethereum_network_info_with_definition(chain_id=1) - eth_token = get_ethereum_token_info_with_definition(chain_id=1) + aave_token = get_ethereum_token_info_with_definition(chain_id=1, token_address="7fc66500c84a76ad7e9c93437bfc5ac33e2ddae9") + # not built-in + rinkeby_network = get_ethereum_network_info_with_definition(chain_id=4) + kc_token = get_ethereum_token_info_with_definition(chain_id=4) # these variations should have the same result - successfully load built-in or encoded network/token calls_params = [ - (None, None, eth_network.info.chain_id, eth_token.info.address), - (eth_network.definition, None, eth_network.info.chain_id, eth_token.info.address), - (None, eth_token.definition, eth_network.info.chain_id, eth_token.info.address), - (eth_network.definition, eth_token.definition, eth_network.info.chain_id, eth_token.info.address), - (eth_network.definition, eth_token.definition, None, eth_token.info.address), - (eth_network.definition, eth_token.definition, eth_network.info.chain_id, None), - (eth_network.definition, eth_token.definition, None, None), + (None, None, eth_network.info.chain_id, aave_token.info.address, eth_network.info, aave_token.info), + (rinkeby_network.definition, None, eth_network.info.chain_id, aave_token.info.address, eth_network.info, aave_token.info), + (None, kc_token.definition, eth_network.info.chain_id, aave_token.info.address, eth_network.info, aave_token.info), + (rinkeby_network.definition, kc_token.definition, eth_network.info.chain_id, aave_token.info.address, eth_network.info, aave_token.info), + (rinkeby_network.definition, kc_token.definition, None, kc_token.info.address, rinkeby_network.info, kc_token.info), + (rinkeby_network.definition, kc_token.definition, rinkeby_network.info.chain_id, None, rinkeby_network.info, kc_token.info), + (rinkeby_network.definition, kc_token.definition, None, None, rinkeby_network.info, kc_token.info), ] for params in calls_params: - self.get_and_compare_ethereum_definitions(*(params + (eth_network.info, eth_token.info))) + self.get_and_compare_ethereum_definitions(*params) def test_no_network_or_token(self): - eth_network = get_ethereum_network_info_with_definition(chain_id=1) - eth_token = get_ethereum_token_info_with_definition(chain_id=1) + rinkeby_network = get_ethereum_network_info_with_definition(chain_id=4) + kc_token = get_ethereum_token_info_with_definition(chain_id=4) calls_params = [ # without network there should be no token loaded - (None, eth_token.definition, None, eth_token.info.address, None, None), - (None, eth_token.definition, 0, eth_token.info.address, None, None), # non-existing chain_id + (None, kc_token.definition, None, kc_token.info.address, None, None), + (None, kc_token.definition, 0, kc_token.info.address, None, None), # non-existing chain_id # also without token there should be no token loaded - (eth_network.definition, None, eth_network.info.chain_id, None, eth_network.info, None), - (eth_network.definition, None, eth_network.info.chain_id, eth_token.info.address + b"\x00", eth_network.info, None), # non-existing token address + (rinkeby_network.definition, None, rinkeby_network.info.chain_id, None, rinkeby_network.info, None), + (rinkeby_network.definition, None, rinkeby_network.info.chain_id, kc_token.info.address + b"\x00", rinkeby_network.info, None), # non-existing token address ] for params in calls_params: self.get_and_compare_ethereum_definitions(*params) @@ -246,6 +269,9 @@ class TestGetDefinitonsFromMsg(unittest.TestCase): def setUp(self): self.addTypeEqualityFunc(networks.NetworkInfo, equalNetworkInfo) self.addTypeEqualityFunc(tokens.TokenInfo, equalTokenInfo) + # use mockup functions for built-in definitions + networks._networks_iterator = builtin_networks_iterator + tokens.token_by_chain_address = builtin_token_by_chain_address def get_and_compare_ethereum_definitions( self, @@ -265,8 +291,12 @@ class TestGetDefinitonsFromMsg(unittest.TestCase): self.assertDictEqual(definitions.token_dict, ref_token_dict) def test_get_definitions_SignTx_messages(self): + # built-in eth_network = get_ethereum_network_info_with_definition(chain_id=1) - eth_token = get_ethereum_token_info_with_definition(chain_id=1) + aave_token = get_ethereum_token_info_with_definition(chain_id=1, token_address="7fc66500c84a76ad7e9c93437bfc5ac33e2ddae9") + # not built-in + rinkeby_network = get_ethereum_network_info_with_definition(chain_id=4) + kc_token = get_ethereum_token_info_with_definition(chain_id=4) def create_EthereumSignTx_msg(**kwargs): return EthereumSignTx( @@ -287,62 +317,106 @@ class TestGetDefinitonsFromMsg(unittest.TestCase): ) # both network and token should be loaded - messages = [ - create_EthereumSignTx_msg( - chain_id=eth_network.info.chain_id, - to=hexlify(eth_token.info.address), - definitions=EthereumEncodedDefinitions(eth_network.definition, eth_token.definition), + params_set = [ + ( + create_EthereumSignTx_msg( + chain_id=rinkeby_network.info.chain_id, + to=hexlify(kc_token.info.address), + definitions=EthereumEncodedDefinitions( + encoded_network=rinkeby_network.definition, + encoded_token=kc_token.definition, + ), + ), + rinkeby_network.info, + kc_token.info, ), - create_EthereumSignTx_msg( - chain_id=eth_network.info.chain_id, - to=hexlify(eth_token.info.address), + ( + create_EthereumSignTx_msg( + chain_id=eth_network.info.chain_id, + to=hexlify(aave_token.info.address), + ), + eth_network.info, + aave_token.info, ), - create_EthereumSignTxEIP1559_msg( - chain_id=eth_network.info.chain_id, - to=hexlify(eth_token.info.address), - definitions=EthereumEncodedDefinitions(eth_network.definition, eth_token.definition), + ( + create_EthereumSignTxEIP1559_msg( + chain_id=rinkeby_network.info.chain_id, + to=hexlify(kc_token.info.address), + definitions=EthereumEncodedDefinitions( + encoded_network=rinkeby_network.definition, + encoded_token=kc_token.definition, + ), + ), + rinkeby_network.info, + kc_token.info, ), - create_EthereumSignTxEIP1559_msg( - chain_id=eth_network.info.chain_id, - to=hexlify(eth_token.info.address), + ( + create_EthereumSignTxEIP1559_msg( + chain_id=eth_network.info.chain_id, + to=hexlify(aave_token.info.address), + ), + eth_network.info, + aave_token.info, ), ] - for msg in messages: - self.get_and_compare_ethereum_definitions(msg, eth_network.info, eth_token.info) + for params in params_set: + self.get_and_compare_ethereum_definitions(*params) # missing "to" parameter in messages should lead to no token is loaded if none was provided - messages = [ - create_EthereumSignTx_msg( - chain_id=eth_network.info.chain_id, - definitions=EthereumEncodedDefinitions(eth_network.definition, None), + params_set = [ + ( + create_EthereumSignTx_msg( + chain_id=rinkeby_network.info.chain_id, + definitions=EthereumEncodedDefinitions( + encoded_network=rinkeby_network.definition, + encoded_token=None, + ), + ), + rinkeby_network.info, + None, ), - create_EthereumSignTx_msg( - chain_id=eth_network.info.chain_id, + ( + create_EthereumSignTx_msg( + chain_id=eth_network.info.chain_id, + ), + eth_network.info, + None, ), - create_EthereumSignTxEIP1559_msg( - chain_id=eth_network.info.chain_id, - definitions=EthereumEncodedDefinitions(eth_network.definition, None), + ( + create_EthereumSignTxEIP1559_msg( + chain_id=rinkeby_network.info.chain_id, + definitions=EthereumEncodedDefinitions( + encoded_network=rinkeby_network.definition, + encoded_token=None + ), + ), + rinkeby_network.info, + None, ), - create_EthereumSignTxEIP1559_msg( - chain_id=eth_network.info.chain_id, + ( + create_EthereumSignTxEIP1559_msg( + chain_id=eth_network.info.chain_id, + ), + eth_network.info, + None, ), ] - for msg in messages: - self.get_and_compare_ethereum_definitions(msg, eth_network.info, None) + for params in params_set: + self.get_and_compare_ethereum_definitions(*params) def test_other_messages(self): - eth_network = get_ethereum_network_info_with_definition(chain_id=1) + rinkeby_network = get_ethereum_network_info_with_definition(chain_id=4) # only network should be loaded messages = [ - EthereumGetAddress(encoded_network=eth_network.definition), - EthereumGetPublicKey(encoded_network=eth_network.definition), - EthereumSignMessage(message=b'', encoded_network=eth_network.definition), - EthereumSignTypedData(primary_type="", encoded_network=eth_network.definition), - EthereumVerifyMessage(signature=b'', message=b'', address="", encoded_network=eth_network.definition), + EthereumGetAddress(encoded_network=rinkeby_network.definition), + EthereumGetPublicKey(encoded_network=rinkeby_network.definition), + EthereumSignMessage(message=b'', encoded_network=rinkeby_network.definition), + EthereumSignTypedData(primary_type="", encoded_network=rinkeby_network.definition), + EthereumVerifyMessage(signature=b'', message=b'', address="", encoded_network=rinkeby_network.definition), ] for msg in messages: - self.get_and_compare_ethereum_definitions(msg, eth_network.info, None) + self.get_and_compare_ethereum_definitions(msg, rinkeby_network.info, None) # neither network nor token should be loaded messages = [ diff --git a/core/tests/test_apps.ethereum.tokens.py b/core/tests/test_apps.ethereum.tokens.py index c9294503ef..ded1c5310d 100644 --- a/core/tests/test_apps.ethereum.tokens.py +++ b/core/tests/test_apps.ethereum.tokens.py @@ -2,10 +2,14 @@ from common import * if not utils.BITCOIN_ONLY: from apps.ethereum import tokens + from ethereum_common import builtin_token_by_chain_address @unittest.skipUnless(not utils.BITCOIN_ONLY, "altcoin") class TestEthereumTokens(unittest.TestCase): + def setUp(self): + # use mockup function for built-in tokens + tokens.token_by_chain_address = builtin_token_by_chain_address def test_token_by_chain_address(self):