From 5ad2eb74a03e96afe8de7fd5112040646e6bd4c5 Mon Sep 17 00:00:00 2001
From: matejcik <ja@matejcik.cz>
Date: Thu, 23 Aug 2018 13:05:41 +0200
Subject: [PATCH] coin_info: shuffle knowledge about duplicates, validation and
 support information

---
 tools/coin_info.py     | 134 ++++++++++++++++---------
 tools/cointool.py      | 220 +++++++++++++++++++++++++++++++++++------
 tools/requirements.txt |   3 +
 3 files changed, 280 insertions(+), 77 deletions(-)

diff --git a/tools/coin_info.py b/tools/coin_info.py
index 4c3061d72f..f6f2cf43a9 100755
--- a/tools/coin_info.py
+++ b/tools/coin_info.py
@@ -261,6 +261,8 @@ def _load_misc():
 # ====== support info ======
 
 RELEASES_URL = "https://wallet.trezor.io/data/firmware/{}/releases.json"
+MISSING_SUPPORT_MEANS_NO = ("connect", "webwallet")
+VERSIONED_SUPPORT_INFO = ("trezor1", "trezor2")
 
 
 def get_support_data():
@@ -280,6 +282,10 @@ def latest_releases():
     return latest
 
 
+def is_token(coin):
+    return coin["key"].startswith("erc20:")
+
+
 def support_info_single(support_data, coin):
     """Extract a support dict from `support.json` data.
 
@@ -287,7 +293,7 @@ def support_info_single(support_data, coin):
     top-level key.
 
     The support value for each device is determined in order of priority:
-    * if the coin is marked as duplicate, all support values are `None`
+    * if the coin is a duplicate ERC20 token, all support values are `None`
     * if the coin has an entry in `unsupported`, its support is `None`
     * if the coin has an entry in `supported` its support is that entry
       (usually a version string, or `True` for connect/webwallet)
@@ -297,12 +303,14 @@ def support_info_single(support_data, coin):
     key = coin["key"]
     dup = coin.get("duplicate")
     for device, values in support_data.items():
-        if dup:
+        if dup and is_token(coin):
             support_value = None
         elif key in values["unsupported"]:
             support_value = None
         elif key in values["supported"]:
             support_value = values["supported"][key]
+        elif device in MISSING_SUPPORT_MEANS_NO:
+            support_value = None
         else:
             support_value = "soon"
         support_info[device] = support_value
@@ -337,43 +345,6 @@ def support_info(coins):
 # ====== data cleanup functions ======
 
 
-def find_address_collisions(coins):
-    """Detects collisions in:
-    - SLIP44 path prefixes
-    - address type numbers, both for p2pkh and p2sh
-    """
-    slip44 = defaultdict(list)
-    at_p2pkh = defaultdict(list)
-    at_p2sh = defaultdict(list)
-
-    for coin in coins:
-        name = coin["name"]
-        s = coin["slip44"]
-        # ignore m/1 testnets
-        if not (name.endswith("Testnet") and s == 1):
-            slip44[s].append(name)
-
-        # skip address types on cashaddr currencies
-        if coin["cashaddr_prefix"]:
-            continue
-
-        at_p2pkh[coin["address_type"]].append(name)
-        at_p2sh[coin["address_type_p2sh"]].append(name)
-
-    def prune(d):
-        ret = d.copy()
-        for key in d:
-            if len(d[key]) < 2:
-                del ret[key]
-        return ret
-
-    return dict(
-        slip44=prune(slip44),
-        address_type=prune(at_p2pkh),
-        address_type_p2sh=prune(at_p2sh),
-    )
-
-
 def _ensure_mandatory_values(coins):
     """Checks that every coin has the mandatory fields: name, shortcut, key"""
     for coin in coins:
@@ -381,9 +352,33 @@ def _ensure_mandatory_values(coins):
             raise ValueError(coin)
 
 
+def symbol_from_shortcut(shortcut):
+    symsplit = shortcut.split(" ", maxsplit=1)
+    return symsplit[0], symsplit[1] if len(symsplit) > 1 else ""
+
+
 def mark_duplicate_shortcuts(coins):
     """Finds coins with identical `shortcut`s.
     Updates their keys and sets a `duplicate` field.
+
+    The logic is a little crazy.
+
+    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`. These
+    will _always_ be marked as duplicate (and later possibly deleted if they're ERC20).
+
+    The rest will disambiguate based on the full shortcut.
+    (i.e., when `shortcut` is `BTL (Battle)`, the `symbol` is just `BTL`).
+    If _all tokens_ in the bucket have shortcuts with distinct suffixes, e.g.,
+    `CAT (BitClave)` and `CAT (Blockcat)`, we DO NOT mark them as duplicate.
+    These will then be supported and included in outputs.
+
+    If even one token in the bucket _does not_ have a distinct suffix, e.g.,
+    `MIT` and `MIT (Mychatcoin)`, the whole bucket is marked as duplicate.
+
+    If a token is set to `false` in `duplicity_overrides.json`, it will NOT
+    be marked as duplicate in this step, even if it is part of a "bad" bucket.
     """
     dup_symbols = defaultdict(list)
     dup_keys = defaultdict(list)
@@ -392,26 +387,50 @@ def mark_duplicate_shortcuts(coins):
         return {k: v for k, v in dups.items() if len(v) > 1}
 
     for coin in coins:
-        symsplit = coin["shortcut"].split(" ", maxsplit=1)
-        symbol = symsplit[0]
+        symbol, _ = symbol_from_shortcut(coin["shortcut"])
         dup_symbols[symbol].append(coin)
         dup_keys[coin["key"]].append(coin)
 
     dup_symbols = dups_only(dup_symbols)
     dup_keys = dups_only(dup_keys)
 
-    # mark duplicate symbols
-    for values in dup_symbols.values():
-        for coin in values:
-            coin["duplicate"] = True
-
-    # deduplicate keys
+    # first deduplicate keys so that we can identify overrides
     for values in dup_keys.values():
         for i, coin in enumerate(values):
-            # presumably only duplicate symbols can have duplicate keys
-            assert coin.get("duplicate")
             coin["key"] += f":{i}"
 
+    # load overrides and put them into their own bucket
+    overrides = load_json("duplicity_overrides.json")
+    override_bucket = []
+    for coin in coins:
+        if overrides.get(coin["key"], False):
+            coin["duplicate"] = True
+            override_bucket.append(coin)
+
+    # mark duplicate symbols
+    for values in dup_symbols.values():
+        splits = (symbol_from_shortcut(coin["shortcut"]) for coin in values)
+        suffixes = {suffix for _, suffix in splits}
+        # if 1. all suffixes are distinct and 2. none of them are empty
+        if len(suffixes) == len(values) and all(suffixes):
+            # Allow the whole bucket.
+            # For all intents and purposes these should be considered non-dups
+            # So we won't mark them as dups here
+            # But they still have their own bucket, and also overrides can
+            # explicitly mark them as duplicate one step before, in which case
+            # they *still* keep duplicate status (and possibly are deleted).
+            continue
+
+        nontokens = [coin for coin in values if not is_token(coin)]
+
+        for coin in values:
+            # allow overrides to skip this; if not listed in overrides, assume True
+            is_dup = overrides.get(coin["key"], True)
+            if is_dup:
+                coin["duplicate"] = True
+            # again: still in dups, but not marked as duplicate and not deleted
+
+    dup_symbols["_override"] = override_bucket
     return dup_symbols
 
 
@@ -461,3 +480,20 @@ def get_all(deduplicate=True):
         ]
 
     return all_coins
+
+
+def search(coins, keyword):
+    kwl = keyword.lower()
+    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
diff --git a/tools/cointool.py b/tools/cointool.py
index ee05e61837..6428e71fc6 100755
--- a/tools/cointool.py
+++ b/tools/cointool.py
@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 import io
 import json
+import logging
 import re
 import sys
 import os
@@ -8,6 +9,7 @@ import glob
 import binascii
 import struct
 import zlib
+from collections import defaultdict
 from hashlib import sha256
 
 import click
@@ -15,6 +17,10 @@ import click
 import coin_info
 from coindef import CoinDef
 
+try:
+    import termcolor
+except ImportError:
+    termcolor = None
 
 try:
     import mako
@@ -40,6 +46,36 @@ except ImportError:
     CAN_BUILD_DEFS = False
 
 
+# ======= Crayon colors ======
+USE_COLORS = False
+
+
+def crayon(color, string, bold=False, dim=False):
+    if not termcolor or not USE_COLORS:
+        return string
+    else:
+        if bold:
+            attrs = ["bold"]
+        elif dim:
+            attrs = ["dark"]
+        else:
+            attrs = []
+        return termcolor.colored(string, color, attrs=attrs)
+
+
+def print_log(level, *args, **kwargs):
+    prefix = logging.getLevelName(level)
+    if level == logging.DEBUG:
+        prefix = crayon("blue", prefix, bold=False)
+    elif level == logging.INFO:
+        prefix = crayon("blue", prefix, bold=True)
+    elif level == logging.WARNING:
+        prefix = crayon("red", prefix, bold=False)
+    elif level == logging.ERROR:
+        prefix = crayon("red", prefix, bold=True)
+    print(prefix, *args, **kwargs)
+
+
 # ======= Mako management ======
 
 
@@ -106,46 +142,146 @@ def render_file(src, dst, coins, support_info):
 # ====== validation functions ======
 
 
+def highlight_key(coin, color):
+    keylist = coin["key"].split(":")
+    if keylist[-1].isdigit():
+        keylist[-2] = crayon(color, keylist[-2], bold=True)
+    else:
+        keylist[-1] = crayon(color, keylist[-1], bold=True)
+    key = crayon(color, ":".join(keylist))
+    name = crayon(None, f"({coin['name']})", dim=True)
+    return f"{key} {name}"
+
+
+def find_address_collisions(coins, field):
+    """Detects collisions in a given field. Returns buckets of colliding coins."""
+    collisions = defaultdict(list)
+    for coin in coins:
+        value = coin[field]
+        collisions[value].append(coin)
+    return {k: v for k, v in collisions.items() if len(v) > 1}
+
+
 def check_btc(coins):
     check_passed = True
+    support_infos = coin_info.support_info(coins)
 
     for coin in coins:
         errors = coin_info.validate_btc(coin)
         if errors:
             check_passed = False
-            print("ERR:", "invalid definition for", coin["name"])
+            print_log(logging.ERROR, "invalid definition for", coin["name"])
             print("\n".join(errors))
 
-    collisions = coin_info.find_address_collisions(coins)
-    # warning only
-    for key, dups in collisions.items():
-        if dups:
-            print("WARN: collisions found in", key)
-            for k, v in dups.items():
-                print("-", k, ":", ", ".join(map(str, v)))
+    def collision_str(bucket):
+        coin_strings = []
+        for coin in bucket:
+            name = coin["name"]
+            prefix = ""
+            if name.endswith("Testnet"):
+                color = "green"
+            elif name == "Bitcoin":
+                color = "red"
+            elif coin.get("unsupported"):
+                color = "grey"
+                prefix = crayon("blue", "(X)", bold=True)
+            else:
+                color = "blue"
+            hl = highlight_key(coin, color)
+            coin_strings.append(prefix + hl)
+        return ", ".join(coin_strings)
+
+    def print_collision_buckets(buckets, prefix):
+        failed = False
+        for key, bucket in buckets.items():
+            mainnets = [c for c in bucket if not c["name"].endswith("Testnet")]
+
+            have_bitcoin = False
+            for coin in mainnets:
+                if coin["name"] == "Bitcoin":
+                    have_bitcoin = True
+                if all(v is None for k,v in support_infos[coin["key"]].items()):
+                    coin["unsupported"] = True
+
+            supported_mainnets = [c for c in mainnets if not c.get("unsupported")]
+
+            if len(mainnets) > 1:
+                if len(supported_mainnets) > 1:
+                    if have_bitcoin:
+                        level = logging.ERROR
+                        failed = True
+                    else:
+                        level = logging.WARNING
+                else:
+                    level = logging.INFO
+                print_log(level, f"prefix {key}:", collision_str(bucket))
+
+        return failed
+
+    # slip44 collisions
+    print("Checking SLIP44 prefix collisions...")
+    slip44 = find_address_collisions(coins, "slip44")
+    if print_collision_buckets(slip44, "key"):
+        check_passed = False
+
+    nocashaddr = [coin for coin in coins if not coin.get("cashaddr_prefix")]
+    
+    print("Checking address_type collisions...")
+    address_type = find_address_collisions(nocashaddr, "address_type")
+    if print_collision_buckets(address_type, "address type"):
+        check_passed = False
+
+    print("Checking address_type_p2sh collisions...")
+    address_type_p2sh = find_address_collisions(nocashaddr, "address_type_p2sh")
+    # we ignore failed checks on P2SH, because reasons
+    print_collision_buckets(address_type_p2sh, "address type")
 
     return check_passed
 
 
-def check_dups(buckets):
-    check_passed = True
-    for bucket in buckets.values():
-        nontokens = [coin for coin in bucket if not coin["key"].startswith("erc20")]
-        token_list = [coin["key"] for coin in bucket if coin["key"].startswith("erc20")]
-        if not nontokens:
-            continue
-        if len(nontokens) == 1:
-            coin = nontokens[0]
-            print(
-                f"Coin {coin['key']} ({coin['name']}) is duplicate with",
-                ", ".join(token_list),
-                "and that is OK.",
-            )
+def check_dups(buckets, show_tok_notok, show_erc20):
+    def coin_str(coin):
+        if coin_info.is_token(coin):
+            color = "cyan"
         else:
-            nontoken_list = [f"{coin['key']} ({coin['name']})" for coin in nontokens]
-            print("Duplicate shortcuts for", ", ".join(nontoken_list))
+            color = "red"
+        highlighted = highlight_key(coin, color)
+        if not coin.get("duplicate"):
+            prefix = crayon("green", "*", bold=True)
+        else:
+            prefix = ""
+        return f"{prefix}{highlighted}"
+
+    check_passed = True
+
+    for symbol in sorted(buckets.keys()):
+        bucket = buckets[symbol]
+        if not bucket:
+            continue
+
+        nontokens = [coin for coin in bucket if not coin_info.is_token(coin)]
+
+        # string generation
+        dup_str = ", ".join(coin_str(coin) for coin in bucket)
+        if not nontokens:
+            level = logging.DEBUG
+        elif len(nontokens) == 1:
+            level = logging.INFO
+        else:
+            level = logging.ERROR
             check_passed = False
 
+        # deciding whether to print
+        if not nontokens and not show_erc20:
+            continue
+        if len(nontokens) == 1 and not show_tok_notok:
+            continue
+
+        if symbol == "_override":
+            print_log(level, "force-set duplicates:", dup_str)
+        else:
+            print_log(level, f"duplicate symbol {symbol}:", dup_str)
+
     return check_passed
 
 
@@ -250,8 +386,15 @@ def sign(data):
 
 
 @click.group()
-def cli():
-    pass
+@click.option(
+    "--colors/--no-colors",
+    "-c/-C",
+    default=sys.stdout.isatty(),
+    help="Force colored output on/off",
+)
+def cli(colors):
+    global USE_COLORS
+    USE_COLORS = colors
 
 
 @cli.command()
@@ -259,12 +402,24 @@ def cli():
 @click.option("--missing-support/--no-missing-support", "-s", default=False, help="Fail if support info for a coin is missing")
 @click.option("--backend/--no-backend", "-b", default=False, help="Check blockbook/bitcore responses")
 @click.option("--icons/--no-icons", default=True, help="Check icon files")
+@click.option("-d", "--show-duplicates", type=click.Choice(("all", "nontoken", "errors")),
+    default="errors", help="How much information about duplicate shortcuts should be shown.")
 # fmt: on
-def check(missing_support, backend, icons):
+def check(missing_support, backend, icons, show_duplicates):
     """Validate coin definitions.
 
     Checks that every btc-like coin is properly filled out, reports address collisions
     and missing support information.
+
+    The `--show-duplicates` option can be set to:
+    * all: all shortcut collisions are shown, including colliding ERC20 tokens
+    * nontoken: only collisions that affect non-ERC20 coins are shown
+    * errors: only collisions between non-ERC20 tokens are shown. This is the default,
+      as a collision between two or more non-ERC20 tokens is an error.
+
+    In the output, duplicate ERC tokens will be shown in cyan; duplicate non-tokens
+    in red. An asterisk (*) next to symbol name means that even though it was detected
+    as duplicate, it is still included in results.
     """
     if backend and requests is None:
         raise click.ClickException("You must install requests for backend check")
@@ -286,8 +441,17 @@ def check(missing_support, backend, icons):
     # if not check_support(defs, support_data, fail_missing=missing_support):
     #     all_checks_passed = False
 
+    if show_duplicates == "all":
+        show_tok_notok = True
+        show_erc20 = True
+    elif show_duplicates == "nontoken":
+        show_tok_notok = True
+        show_erc20 = False
+    else:
+        show_tok_notok = False
+        show_erc20 = False
     print("Checking unexpected duplicates...")
-    if not check_dups(buckets):
+    if not check_dups(buckets, show_tok_notok, show_erc20):
         all_checks_passed = False
 
     if icons:
diff --git a/tools/requirements.txt b/tools/requirements.txt
index f910733abd..59cab2694a 100644
--- a/tools/requirements.txt
+++ b/tools/requirements.txt
@@ -11,3 +11,6 @@ requests>=2.19
 # for rendering templates:
 Mako>=1.0.7
 munch>=2.3.2
+
+# for pretty colors in checks
+termcolor >= 0.1.2