#!/usr/bin/env python3 from __future__ import annotations import json import logging import re from collections import OrderedDict, defaultdict from pathlib import Path from typing import Dict # for python38 support, must be used in type aliases from typing import List # for python38 support, must be used in type aliases from typing import Any, Callable, Iterable, Iterator, cast from typing_extensions import ( # for python37 support, is not present in typing there Literal, TypedDict, ) try: import requests except ImportError: requests = None log = logging.getLogger(__name__) ROOT = Path(__file__).resolve().parent.parent DEFS_DIR = ROOT / "defs" class SupportItemBool(TypedDict): supported: dict[str, bool] unsupported: dict[str, bool] class SupportItemVersion(TypedDict): supported: dict[str, str] unsupported: dict[str, str] class SupportData(TypedDict): connect: SupportItemBool suite: SupportItemBool t1b1: SupportItemVersion t2t1: SupportItemVersion t2b1: SupportItemVersion class SupportInfoItem(TypedDict): connect: bool suite: bool t1b1: Literal[False] | str t2t1: Literal[False] | str t2b1: Literal[False] | str SupportInfo = Dict[str, SupportInfoItem] class Coin(TypedDict): # Necessary fields for BTC - from BTC_CHECKS coin_name: str coin_shortcut: str coin_label: str website: str github: str maintainer: str curve_name: str address_type: int address_type_p2sh: int maxfee_kb: int minfee_kb: int hash_genesis_block: str xprv_magic: int xpub_magic: int xpub_magic_segwit_p2sh: int xpub_magic_segwit_native: int slip44: int segwit: bool decred: bool fork_id: int force_bip143: bool default_fee_b: dict[str, int] dust_limit: int blocktime_seconds: int signed_message_header: str uri_prefix: str min_address_length: int max_address_length: int bech32_prefix: str cashaddr_prefix: str # Other fields optionally coming from JSON links: dict[str, str] curve: str decimals: int # Mandatory fields added later in coin.update() name: str shortcut: str key: str icon: str # Special ETH fields coingecko_id: str chain: str chain_id: int url: str # Special erc20 fields symbol: str address: str address_bytes: bytes dup_key_nontoken: bool # Special NEM fields ticker: str # Fields that are being created unsupported: bool duplicate: bool support: SupportInfoItem is_testnet: bool # Backend-oriented fields blockchain_link: dict[str, Any] blockbook: list[str] bitcore: list[str] Coins = List[Coin] CoinBuckets = Dict[str, Coins] class FidoApp(TypedDict): name: str webauthn: list[str] u2f: list[dict[str, str]] use_sign_count: bool use_self_attestation: bool use_compact: bool no_icon: bool key: str icon: str FidoApps = List[FidoApp] def load_json(*path: str | Path) -> Any: """Convenience function to load a JSON file from DEFS_DIR.""" if len(path) == 1 and isinstance(path[0], Path): file = path[0] else: file = Path(DEFS_DIR, *path) return json.loads(file.read_text(), object_pairs_hook=OrderedDict) def get_btc_testnet_status(name: str) -> bool: return any((mark in name.lower()) for mark in ("testnet", "regtest")) # ====== CoinsInfo ====== class CoinsInfo(Dict[str, Coins]): """Collection of information about all known kinds of coins. It contains the following lists: `bitcoin` for btc-like coins, `eth` for ethereum networks, `erc20` for ERC20 tokens, `nem` for NEM mosaics, `misc` for other networks. Accessible as a dict or by attribute: `info["misc"] == info.misc` """ def as_list(self) -> Coins: return sum(self.values(), []) def as_dict(self) -> dict[str, Coin]: return {coin["key"]: coin for coin in self.as_list()} def __getattr__(self, attr: str) -> Coins: if attr in self: return self[attr] else: raise AttributeError(attr) # ====== coin validation ====== def check_type( val: Any, types: type | tuple[type, ...], nullable: bool = False, empty: bool = False, regex: str | None = None, choice: list[str] | None = None, ) -> None: # check nullable if val is None: if nullable: return else: raise ValueError("Missing required value") # check type if not isinstance(val, types): raise TypeError(f"Wrong type (expected: {types})") # check empty if isinstance(val, (list, dict)) and not empty and not val: raise ValueError("Empty collection") if isinstance(val, str) and not empty and not val: raise ValueError("Empty string") # check regex if regex is not None: if types is not str: raise TypeError("Wrong type for regex check") assert isinstance(val, str) if not re.search(regex, val): raise ValueError(f"Value does not match regex {regex}") # check choice if choice is not None and val not in choice: choice_str = ", ".join(choice) raise ValueError(f"Value not allowed, use one of: {choice_str}") def check_key( key: str, types: type | tuple[type, ...], optional: bool = False, **kwargs: Any ) -> Callable[[Coin], None]: def do_check(coin: Coin) -> None: if key not in coin: if optional: return else: raise KeyError(f"{key}: Missing key") try: check_type(coin[key], types, **kwargs) except Exception as e: raise ValueError(f"{key}: {e}") from e return do_check BTC_CHECKS = [ check_key("coin_name", str, regex=r"^[A-Z]"), check_key("coin_shortcut", str, regex=r"^t?[A-Z]{3,}$"), check_key("coin_label", str, regex=r"^x?[A-Z]"), check_key("website", str, regex=r"^https://.*[^/]$"), check_key("github", str, regex=r"^https://git(hu|la)b.com/.*[^/]$"), check_key("maintainer", str), check_key( "curve_name", str, choice=[ "secp256k1", "secp256k1_decred", "secp256k1_groestl", "secp256k1_smart", ], ), check_key("address_type", int), check_key("address_type_p2sh", int), check_key("maxfee_kb", int), check_key("minfee_kb", int), check_key("hash_genesis_block", str, regex=r"^[0-9a-f]{64}$"), check_key("xprv_magic", int), check_key("xpub_magic", int), check_key("xpub_magic_segwit_p2sh", int, nullable=True), check_key("xpub_magic_segwit_native", int, nullable=True), check_key("slip44", int), check_key("segwit", bool), check_key("decred", bool), check_key("fork_id", int, nullable=True), check_key("force_bip143", bool), check_key("default_fee_b", dict), check_key("dust_limit", int), check_key("blocktime_seconds", int), check_key("signed_message_header", str), check_key("uri_prefix", str, regex=r"^[a-z-\.\+]+$"), check_key("min_address_length", int), check_key("max_address_length", int), check_key("bech32_prefix", str, regex=r"^[a-z-\.\+]+$", nullable=True), check_key("cashaddr_prefix", str, regex=r"^[a-z-\.\+]+$", nullable=True), ] def validate_btc(coin: Coin) -> list[str]: errors: list[str] = [] for check in BTC_CHECKS: try: check(coin) except Exception as e: errors.append(str(e)) magics: list[int] = [ coin[k] for k in ( "xprv_magic", "xpub_magic", "xpub_magic_segwit_p2sh", "xpub_magic_segwit_native", ) if coin[k] is not None ] # each of those must be unique # therefore length of list == length of set of unique values if len(magics) != len(set(magics)): errors.append("XPUB/XPRV magic numbers must be unique") if coin["address_type"] == coin["address_type_p2sh"]: errors.append("address_type must be distinct from address_type_p2sh") if not coin["maxfee_kb"] >= coin["minfee_kb"]: errors.append("max fee must not be smaller than min fee") if not coin["max_address_length"] >= coin["min_address_length"]: errors.append("max address length must not be smaller than min address length") if coin["is_testnet"] and coin["slip44"] != 1: errors.append("testnet coins must use slip44 coin type 1") if coin["segwit"]: if coin["bech32_prefix"] is None: errors.append("bech32_prefix must be defined for segwit-enabled coin") if coin["xpub_magic_segwit_p2sh"] is None: errors.append( "xpub_magic_segwit_p2sh must be defined for segwit-enabled coin" ) else: if coin["bech32_prefix"] is not None: errors.append("bech32_prefix must not be defined for segwit-disabled coin") if coin["xpub_magic_segwit_p2sh"] is not None: errors.append( "xpub_magic_segwit_p2sh must not be defined for segwit-disabled coin" ) return errors # ======= Coin json loaders ======= def _load_btc_coins() -> Coins: """Load btc-like coins from `bitcoin/*.json`""" coins: Coins = [] for file in DEFS_DIR.glob("bitcoin/*.json"): coin: Coin = load_json(file) coin.update( name=coin["coin_label"], shortcut=coin["coin_shortcut"], key=f"bitcoin:{coin['coin_shortcut']}", icon=str(file.with_suffix(".png")), is_testnet=get_btc_testnet_status(coin["coin_label"]), ) coins.append(coin) return coins def _load_builtin_ethereum_networks() -> Coins: """Load ethereum networks from `ethereum/networks.json`""" chains_data = load_json("ethereum", "networks.json") networks: Coins = [] for chain_data in chains_data: chain_data["key"] = f"eth:{chain_data['shortcut']}:{chain_data['chain_id']}" # is_testnet is present in the JSON networks.append(cast(Coin, chain_data)) return networks def _load_builtin_erc20_tokens() -> Coins: """Load ERC20 tokens from `ethereum/tokens.json`.""" tokens_data = load_json("ethereum", "tokens.json") all_tokens: Coins = [] for chain_id_and_chain, tokens in tokens_data.items(): chain_id, chain = chain_id_and_chain.split(";", maxsplit=1) for token in tokens: token.update( chain=chain, chain_id=int(chain_id), address=token["address"].lower(), address_bytes=bytes.fromhex(token["address"][2:]), symbol=token["shortcut"], key=f"erc20:{chain}:{token['shortcut']}", is_testnet=False, ) all_tokens.append(cast(Coin, token)) return all_tokens def _load_nem_mosaics() -> Coins: """Loads NEM mosaics from `nem/nem_mosaics.json`""" mosaics: Coins = load_json("nem/nem_mosaics.json") for mosaic in mosaics: shortcut = mosaic["ticker"].strip() mosaic.update( shortcut=shortcut, key=f"nem:{shortcut}", is_testnet=False, ) return mosaics def _load_misc() -> Coins: """Loads miscellaneous networks from `misc/misc.json`""" others: Coins = load_json("misc/misc.json") for other in others: other.update( key=f"misc:{other['shortcut']}", is_testnet=False, ) return others def _load_fido_apps() -> FidoApps: """Load FIDO apps from `fido/*.json`""" apps: FidoApps = [] for file in sorted(DEFS_DIR.glob("fido/*.json")): app_name = file.stem.lower() app = load_json(file) app.setdefault("use_sign_count", None) app.setdefault("use_self_attestation", None) app.setdefault("use_compact", None) app.setdefault("u2f", []) app.setdefault("webauthn", []) icon_file = file.with_suffix(".png") if not icon_file.exists(): icon_path = None else: icon_path = str(icon_file) app.update(key=app_name, icon=icon_path) apps.append(app) return apps # ====== support info ====== RELEASES_URL = "https://data.trezor.io/firmware/{}/releases.json" MISSING_SUPPORT_MEANS_NO = ("connect", "suite") VERSIONED_SUPPORT_INFO = ("T1B1", "T2T1", "T2B1") def get_support_data() -> SupportData: """Get raw support data from `support.json`.""" return load_json("support.json") def latest_releases() -> dict[str, Any]: """Get latest released firmware versions for all models""" if not requests: raise RuntimeError("requests library is required for getting release info") latest: dict[str, Any] = {} for model in VERSIONED_SUPPORT_INFO: # TODO: support new UPPERCASE model names in RELEASES_URL url_model = model.lower() # need to be e.g. t1b1 for now releases = requests.get(RELEASES_URL.format(url_model)).json() latest[model] = max(tuple(r["version"]) for r in releases) return latest def support_info_single(support_data: SupportData, coin: Coin) -> SupportInfoItem: """Extract a support dict from `support.json` data. Returns a dict of support values for each "device", i.e., `support.json` top-level key. The support value for each device is determined in order of priority: * if the coin has an entry in `unsupported`, its support is `False` * if the coin has an entry in `supported` its support is that entry (usually a version string, or `True` for connect/suite) * if the coin doesn't have an entry, its support status is `None` """ support_info_item = {} key = coin["key"] for device, values in support_data.items(): assert isinstance(values, dict) if key in values["unsupported"]: support_value: Any = False elif key in values["supported"]: support_value = values["supported"][key] elif device in MISSING_SUPPORT_MEANS_NO: support_value = False else: support_value = None support_info_item[device] = support_value return cast(SupportInfoItem, support_info_item) def support_info(coins: Iterable[Coin] | CoinsInfo | dict[str, Coin]) -> SupportInfo: """Generate Trezor support information. Takes a collection of coins and generates a support-info entry for each. The support-info is a dict with keys based on `support.json` keys. These are usually: "T1B1", "T2T1", "T2B1", "connect" and "suite". The `coins` argument can be a `CoinsInfo` object, a list or a dict of coin items. Support information is taken from `support.json`. """ if isinstance(coins, CoinsInfo): coins = coins.as_list() elif isinstance(coins, dict): coins = coins.values() support_data = get_support_data() support: SupportInfo = {} for coin in coins: support[coin["key"]] = support_info_single(support_data, coin) return support # ====== data cleanup functions ====== def _ensure_mandatory_values(coins: Coins) -> None: """Checks that every coin has the mandatory fields: name, shortcut, key""" for coin in coins: if not all(coin.get(k) for k in ("name", "shortcut", "key")): raise ValueError(coin) def symbol_from_shortcut(shortcut: str) -> tuple[str, str]: symsplit = shortcut.split(" ", maxsplit=1) return symsplit[0], symsplit[1] if len(symsplit) > 1 else "" def mark_duplicate_shortcuts(coins: Coins) -> CoinBuckets: """Finds coins with identical symbols and sets their `duplicate` field. "Symbol" here means the first part of `shortcut` (separated by space), so, e.g., "BTL (Battle)" and "BTL (Bitlle)" have the same symbol "BTL". The result of this function is a dictionary of _buckets_, each of which is indexed by the duplicated symbol, or `_override`. The `_override` bucket will contain all coins that are set to `true` in `duplicity_overrides.json`. Each coin in every bucket will have its "duplicate" property set to True, unless it's explicitly marked as `false` in `duplicity_overrides.json`. """ dup_symbols: CoinBuckets = defaultdict(list) for coin in coins: symbol, _ = symbol_from_shortcut(coin["shortcut"].lower()) dup_symbols[symbol].append(coin) dup_symbols = {k: v for k, v in dup_symbols.items() if len(v) > 1} # mark duplicate symbols for values in dup_symbols.values(): for coin in values: coin["duplicate"] = True return dup_symbols def apply_duplicity_overrides(coins: Coins) -> Coins: overrides = load_json("duplicity_overrides.json") override_bucket: Coins = [] for coin in coins: override_value = overrides.get(coin["key"]) if override_value is True: override_bucket.append(coin) if override_value is not None: coin["duplicate"] = override_value return override_bucket def find_duplicate_keys(all_coins: Coins) -> None: dups: CoinBuckets = defaultdict(list) for coin in all_coins: dups[coin["key"]].append(coin) for coins in dups.values(): if len(coins) <= 1: continue coin = coins[0] raise ValueError(f"Duplicate key {coin['key']}") def fill_blockchain_links(all_coins: CoinsInfo) -> None: blockchain_links = load_json("blockchain_link.json") for coins in all_coins.values(): for coin in coins: link = blockchain_links.get(coin["key"]) coin["blockchain_link"] = link if link and link["type"] == "blockbook": coin["blockbook"] = link["url"] else: coin["blockbook"] = [] def _btc_sort_key(coin: Coin) -> str: if coin["name"] in ("Bitcoin", "Testnet", "Regtest"): return "000000" + coin["name"] else: return coin["name"] def collect_coin_info() -> CoinsInfo: """Returns all definition as dict organized by coin type. `coins` for btc-like coins, `eth` for ethereum networks, `erc20` for ERC20 tokens, `nem` for NEM mosaics, `misc` for other networks. """ all_coins = CoinsInfo( bitcoin=_load_btc_coins(), eth=_load_builtin_ethereum_networks(), erc20=_load_builtin_erc20_tokens(), nem=_load_nem_mosaics(), misc=_load_misc(), ) for coins in all_coins.values(): _ensure_mandatory_values(coins) fill_blockchain_links(all_coins) return all_coins def sort_coin_infos(all_coins: CoinsInfo) -> None: for k, coins in all_coins.items(): if k == "bitcoin": coins.sort(key=_btc_sort_key) elif k == "nem": # do not sort nem pass elif k == "eth": # sort ethereum networks by chain_id coins.sort(key=lambda c: c["chain_id"]) else: coins.sort(key=lambda c: c["key"].upper()) def coin_info_with_duplicates() -> tuple[CoinsInfo, CoinBuckets]: """Collects coin info, detects duplicates but does not remove them. Returns the CoinsInfo object and duplicate buckets. """ all_coins = collect_coin_info() coin_list = all_coins.as_list() # generate duplicity buckets based on shortcuts buckets = mark_duplicate_shortcuts(all_coins.as_list()) # ensure the whole list has unique keys find_duplicate_keys(coin_list) # apply duplicity overrides buckets["_override"] = apply_duplicity_overrides(coin_list) sort_coin_infos(all_coins) return all_coins, buckets def coin_info() -> CoinsInfo: """Collects coin info, fills out support info and returns the result. Does not auto-delete duplicates. This should now be based on support info. """ all_coins, _ = coin_info_with_duplicates() return all_coins def fido_info() -> FidoApps: """Returns info about known FIDO/U2F apps.""" return _load_fido_apps() def search(coins: CoinsInfo | Coins, keyword: str) -> Iterator[Any]: kwl = keyword.lower() if isinstance(coins, CoinsInfo): coins = coins.as_list() for coin in coins: key = coin["key"].lower() name = coin["name"].lower() shortcut = coin["shortcut"].lower() symbol, suffix = symbol_from_shortcut(shortcut) if ( kwl == key or kwl in name or kwl == shortcut or kwl == symbol or kwl in suffix ): yield coin