chore(common/tools): CoinDef generator revived

common/tools/README.md

@@ -21,8 +21,7 @@ the following commands:
 * **`check`**: check validity of json definitions and associated data. Used in CI.
 * **`dump`**: dump coin information, including support status, in JSON format. Various
   filtering options are available, check help for details.
-* **`coindefs`**: generate signed protobuf descriptions of coins. This is for future use
+* **`coindefs`**: generate signed protobuf definitions for Ethereum networks (chains) and tokens.
-  and could allow us to not need to store coin data in Trezor itself.
 
 Use `cointool.py command --help` to get more information on each command.
 

common/tools/cointool.py

@@ -1,21 +1,26 @@
 #!/usr/bin/env python3
 from __future__ import annotations
 
+import ed25519
 import fnmatch
 import glob
+import io
 import json
 import logging
 import os
+import pathlib
 import re
+import shutil
 import sys
 from collections import defaultdict
 from hashlib import sha256
-from typing import Any, Callable, Iterator, TextIO, cast
+from typing import Any, Callable, Dict, Iterator, TextIO, cast
 
 import click
 
 import coin_info
 from coin_info import Coin, CoinBuckets, Coins, CoinsInfo, FidoApps, SupportInfo
+from trezorlib import protobuf
 
 try:
     import termcolor
@@ -580,6 +585,48 @@ def check_fido(apps: FidoApps) -> bool:
     return check_passed
 
 
+# ====== coindefs generators ======
+from trezorlib.messages import EthereumDefinitionType, EthereumNetworkInfo, EthereumTokenInfo
+import time
+
+FORMAT_VERSION = "trzd1"
+FORMAT_VERSION_BYTES = FORMAT_VERSION.encode("utf-8").ljust(8, b'\0')
+DATA_VERSION_BYTES = int(time.time()).to_bytes(4, "big")
+
+
+def eth_info_from_dict(coin: Coin, msg_type: EthereumNetworkInfo | EthereumTokenInfo) -> EthereumNetworkInfo | EthereumTokenInfo:
+    attributes: Dict[str, Any] = dict()
+    for field in msg_type.FIELDS.values():
+        val = coin.get(field.name)
+
+        if field.name in ("chain_id", "slip44"):
+            attributes[field.name] = int(val)
+        elif msg_type == EthereumTokenInfo and field.name == "address":
+            attributes[field.name] = coin.get("address_bytes")
+        else:
+            attributes[field.name] = val
+
+    proto = msg_type(**attributes)
+
+    return proto
+
+
+def serialize_eth_info(info: EthereumNetworkInfo | EthereumTokenInfo, data_type_num: EthereumDefinitionType) -> bytes:
+    ser = FORMAT_VERSION_BYTES
+    ser += data_type_num.to_bytes(1, "big")
+    ser += DATA_VERSION_BYTES
+
+    buf = io.BytesIO()
+    protobuf.dump_message(buf, info)
+    ser += buf.getvalue()
+
+    return ser
+
+
+def sign_data(sign_key: ed25519.SigningKey, data: bytes) -> bytes:
+    return sign_key.sign(data)
+
+
 # ====== click command handlers ======
 
 
@@ -867,6 +914,87 @@ def dump(
         outfile.write("\n")
 
 
+@cli.command()
+@click.option("-o", "--outdir", type=click.Path(resolve_path=True, file_okay=False, path_type=pathlib.Path), default="./definitions-latest")
+@click.option(
+    "-k", "--privatekey",
+    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):
+    """Generate signed Ethereum definitions for python-trezor and others."""
+    hex_key = None
+    if privatekey is None:
+        # load from env variable
+        hex_key = os.getenv("PRIVATE_KEY", default=None)
+    else:
+        with privatekey:
+            hex_key = privatekey.readline()
+
+    if hex_key is None:
+        raise click.ClickException("No private key for signing was provided.")
+
+    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:
+            f.write(data)
+
+    def generate_token_defs(tokens: Coins, path: pathlib.Path):
+        for token in tokens:
+            if token['address'] 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_definition(path, keys, ser + sign_data(sign_key, ser))
+
+    def generate_network_def(net: Coin, tokens: Coins):
+        if net['chain_id'] is None:
+            return
+
+        # create path for networks identified by chain ids
+        network_dir = outdir / "by_chain_id" / str(net['chain_id'])
+        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']}).")
+
+        # 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)
+
+    # clear defs directory
+    if outdir.exists():
+        shutil.rmtree(outdir)
+    outdir.mkdir(parents=True)
+
+    all_coins = coin_info.coin_info()
+
+    # group tokens by their chain_id
+    token_buckets: CoinBuckets = defaultdict(list)
+    for token in all_coins.erc20:
+        token_buckets[token['chain_id']].append(token)
+
+    for network in all_coins.eth:
+        generate_network_def(network, token_buckets[network['chain_id']])
+
+
+
 @cli.command()
 # fmt: off
 @click.argument("paths", metavar="[path]...", nargs=-1)