tezos: added tezos cryptocurrency

Signed-off-by: Adrian Matejov <adrian.matejov@simplestaking.com>

src/apps/tezos/__init__.py

@@ -0,0 +1,26 @@
+from trezor.messages.MessageType import TezosGetAddress, TezosGetPublicKey, TezosSignTx
+from trezor.wire import protobuf_workflow, register
+
+
+def dispatch_TezosGetAddress(*args, **kwargs):
+    from .get_address import tezos_get_address
+
+    return tezos_get_address(*args, **kwargs)
+
+
+def dispatch_TezosSignTx(*args, **kwargs):
+    from .sign_tx import tezos_sign_tx
+
+    return tezos_sign_tx(*args, **kwargs)
+
+
+def dispatch_TezosGetPublicKey(*args, **kwargs):
+    from .get_public_key import tezos_get_public_key
+
+    return tezos_get_public_key(*args, **kwargs)
+
+
+def boot():
+    register(TezosGetAddress, protobuf_workflow, dispatch_TezosGetAddress)
+    register(TezosSignTx, protobuf_workflow, dispatch_TezosSignTx)
+    register(TezosGetPublicKey, protobuf_workflow, dispatch_TezosGetPublicKey)

src/apps/tezos/get_address.py

@@ -0,0 +1,31 @@
+from trezor.crypto import hashlib
+from trezor.messages.TezosAddress import TezosAddress
+
+from apps.common import seed
+from apps.common.display_address import show_address, show_qr
+from apps.tezos.helpers import (
+    b58cencode,
+    get_address_prefix,
+    get_curve_module,
+    get_curve_name,
+)
+
+
+async def tezos_get_address(ctx, msg):
+    address_n = msg.address_n or ()
+    curve = msg.curve or 0
+    node = await seed.derive_node(ctx, address_n, get_curve_name(curve))
+
+    sk = node.private_key()
+    pk = get_curve_module(curve).publickey(sk)
+    pkh = hashlib.blake2b(pk, outlen=20).digest()
+    address = b58cencode(pkh, prefix=get_address_prefix(curve))
+
+    if msg.show_display:
+        while True:
+            if await show_address(ctx, address):
+                break
+            if await show_qr(ctx, address):
+                break
+
+    return TezosAddress(address=address)

src/apps/tezos/get_public_key.py

@@ -0,0 +1,36 @@
+from trezor import ui
+from trezor.messages import ButtonRequestType
+from trezor.messages.TezosPublicKey import TezosPublicKey
+from trezor.ui.text import Text
+from trezor.utils import chunks
+
+from apps.common import seed
+from apps.common.confirm import require_confirm
+from apps.tezos.helpers import (
+    b58cencode,
+    get_curve_module,
+    get_curve_name,
+    get_pk_prefix,
+)
+
+
+async def tezos_get_public_key(ctx, msg):
+    address_n = msg.address_n or ()
+    curve = msg.curve or 0
+    node = await seed.derive_node(ctx, address_n, get_curve_name(curve))
+
+    sk = node.private_key()
+    pk = get_curve_module(curve).publickey(sk)
+    pk_prefixed = b58cencode(pk, prefix=get_pk_prefix(curve))
+
+    if msg.show_display:
+        await _show_tezos_pubkey(ctx, pk_prefixed)
+
+    return TezosPublicKey(public_key=pk_prefixed)
+
+
+async def _show_tezos_pubkey(ctx, pubkey):
+    lines = chunks(pubkey, 18)
+    text = Text("Confirm public key", ui.ICON_RECEIVE, icon_color=ui.GREEN)
+    text.mono(*lines)
+    return await require_confirm(ctx, text, code=ButtonRequestType.PublicKey)

src/apps/tezos/helpers.py

@@ -0,0 +1,92 @@
+from micropython import const
+
+from trezor import wire
+from trezor.crypto import base58
+from trezor.crypto.curve import ed25519, nist256p1, secp256k1
+
+TEZOS_AMOUNT_DIVISIBILITY = const(6)
+
+PREFIXES = {
+    # addresses
+    "tz1": [6, 161, 159],
+    "tz2": [6, 161, 161],
+    "tz3": [6, 161, 164],
+    "KT1": [2, 90, 121],
+    # public keys
+    "edpk": [13, 15, 37, 217],
+    "sppk": [3, 254, 226, 86],
+    "p2pk": [3, 178, 139, 127],
+    # signatures
+    "edsig": [9, 245, 205, 134, 18],
+    "spsig1": [13, 115, 101, 19, 63],
+    "p2sig": [54, 240, 44, 52],
+    # operation hash
+    "o": [5, 116],
+}
+
+TEZOS_CURVES = [
+    {
+        "name": "ed25519",
+        "address_prefix": "tz1",
+        "pk_prefix": "edpk",
+        "sig_prefix": "edsig",
+        "sig_remove_first_byte": False,
+        "module": ed25519,
+    },
+    {
+        "name": "secp256k1",
+        "address_prefix": "tz2",
+        "pk_prefix": "sppk",
+        "sig_prefix": "spsig1",
+        "sig_remove_first_byte": True,
+        "module": secp256k1,
+    },
+    {
+        "name": "nist256p1",
+        "address_prefix": "tz3",
+        "pk_prefix": "p2pk",
+        "sig_prefix": "p2sig",
+        "sig_remove_first_byte": True,
+        "module": nist256p1,
+    },
+]
+
+
+def get_curve_name(index):
+    if 0 <= index < len(TEZOS_CURVES):
+        return TEZOS_CURVES[index]["name"]
+    raise wire.DataError("Invalid type of curve")
+
+
+def get_curve_module(curve):
+    return TEZOS_CURVES[curve]["module"]
+
+
+def get_address_prefix(curve):
+    return TEZOS_CURVES[curve]["address_prefix"]
+
+
+def get_pk_prefix(curve):
+    return TEZOS_CURVES[curve]["pk_prefix"]
+
+
+def get_sig_prefix(curve):
+    return TEZOS_CURVES[curve]["sig_prefix"]
+
+
+def check_sig_remove_first_byte(curve):
+    return TEZOS_CURVES[curve]["sig_remove_first_byte"]
+
+
+def b58cencode(payload, prefix=None):
+    result = payload
+    if prefix is not None:
+        result = bytes(PREFIXES[prefix]) + payload
+    return base58.encode_check(result)
+
+
+def b58cdecode(enc, prefix=None):
+    decoded = base58.decode_check(enc)
+    if prefix is not None:
+        decoded = decoded[len(PREFIXES[prefix]) :]
+    return decoded

src/apps/tezos/layout.py

@@ -0,0 +1,71 @@
+from trezor import ui
+from trezor.messages import ButtonRequestType
+from trezor.ui.text import Text
+from trezor.utils import chunks, format_amount
+
+from apps.common.confirm import *
+from apps.tezos.helpers import TEZOS_AMOUNT_DIVISIBILITY
+
+
+async def require_confirm_tx(ctx, to, value):
+    text = Text("Confirm sending", ui.ICON_SEND, icon_color=ui.GREEN)
+    text.bold(format_tezos_amount(value))
+    text.normal("to")
+    text.mono(*split_address(to))
+    return await require_confirm(ctx, text, ButtonRequestType.SignTx)
+
+
+async def require_confirm_fee(ctx, value, fee):
+    text = Text("Confirm transaction", ui.ICON_SEND, icon_color=ui.GREEN)
+    text.normal("Amount:")
+    text.bold(format_tezos_amount(value))
+    text.normal("Fee:")
+    text.bold(format_tezos_amount(fee))
+    await require_hold_to_confirm(ctx, text, ButtonRequestType.SignTx)
+
+
+async def require_confirm_origination(ctx, address):
+    text = Text("Confirm origination", ui.ICON_SEND, icon_color=ui.ORANGE)
+    text.normal("Address:")
+    text.mono(*split_address(address))
+    return await require_confirm(ctx, text, ButtonRequestType.SignTx)
+
+
+async def require_confirm_origination_fee(ctx, balance, fee):
+    text = Text("Confirm origination", ui.ICON_SEND, icon_color=ui.ORANGE)
+    text.normal("Balance:")
+    text.bold(format_tezos_amount(balance))
+    text.normal("Fee:")
+    text.bold(format_tezos_amount(fee))
+    await require_hold_to_confirm(ctx, text, ButtonRequestType.SignTx)
+
+
+async def require_confirm_delegation_baker(ctx, baker):
+    text = Text("Confirm delegation", ui.ICON_SEND, icon_color=ui.BLUE)
+    text.normal("Baker address:")
+    text.mono(*split_address(baker))
+    return await require_confirm(ctx, text, ButtonRequestType.SignTx)
+
+
+async def require_confirm_set_delegate(ctx, fee):
+    text = Text("Confirm delegation", ui.ICON_SEND, icon_color=ui.BLUE)
+    text.normal("Fee:")
+    text.bold(format_tezos_amount(fee))
+    await require_hold_to_confirm(ctx, text, ButtonRequestType.SignTx)
+
+
+async def require_confirm_register_delegate(ctx, address, fee):
+    text = Text("Register delegate", ui.ICON_SEND, icon_color=ui.BLUE)
+    text.bold("Fee: " + format_tezos_amount(fee))
+    text.normal("Address:")
+    text.mono(*split_address(address))
+    await require_hold_to_confirm(ctx, text, ButtonRequestType.SignTx)
+
+
+def split_address(address):
+    return chunks(address, 18)
+
+
+def format_tezos_amount(value):
+    formatted_value = format_amount(value, TEZOS_AMOUNT_DIVISIBILITY)
+    return formatted_value + " XTZ"

src/apps/tezos/sign_tx.py

@@ -0,0 +1,159 @@
+import ustruct
+
+from trezor import wire
+from trezor.crypto import hashlib
+from trezor.messages import TezosContractType
+from trezor.messages.TezosSignedTx import TezosSignedTx
+
+from apps.common import seed
+from apps.tezos.helpers import (
+    b58cencode,
+    check_sig_remove_first_byte,
+    get_address_prefix,
+    get_curve_module,
+    get_curve_name,
+    get_sig_prefix,
+)
+from apps.tezos.layout import *
+
+
+async def tezos_sign_tx(ctx, msg):
+    address_n = msg.address_n or ()
+    curve = msg.curve or 0
+
+    node = await seed.derive_node(ctx, address_n, get_curve_name(curve))
+
+    if msg.transaction is not None:
+        to = _get_address_from_contract(msg.transaction.destination)
+        await require_confirm_tx(ctx, to, msg.transaction.amount)
+        await require_confirm_fee(ctx, msg.transaction.amount, msg.transaction.fee)
+
+    elif msg.origination is not None:
+        source = _get_address_from_contract(msg.origination.source)
+        await require_confirm_origination(ctx, source)
+        await require_confirm_origination_fee(
+            ctx, msg.origination.balance, msg.origination.fee
+        )
+
+    elif msg.delegation is not None:
+        source = _get_address_from_contract(msg.delegation.source)
+        delegate = _get_address_by_tag(msg.delegation.delegate)
+
+        if source != delegate:
+            to = _get_address_by_tag(msg.delegation.delegate)
+            await require_confirm_delegation_baker(ctx, to)
+            await require_confirm_set_delegate(ctx, msg.delegation.fee)
+        # if account registers itself as a delegate
+        else:
+            await require_confirm_register_delegate(ctx, source, msg.delegation.fee)
+
+    else:
+        raise wire.DataError("Invalid operation")
+
+    opbytes = _get_operation_bytes(msg)
+
+    watermark = bytes([3])
+    wm_opbytes = watermark + opbytes
+    wm_opbytes_hash = hashlib.blake2b(wm_opbytes, outlen=32).digest()
+
+    curve_module = get_curve_module(curve)
+    signature = curve_module.sign(node.private_key(), wm_opbytes_hash)
+
+    if check_sig_remove_first_byte(curve):
+        signature = signature[1:]
+
+    sig_op_contents = opbytes + signature
+    sig_op_contents_hash = hashlib.blake2b(sig_op_contents, outlen=32).digest()
+    ophash = b58cencode(sig_op_contents_hash, prefix="o")
+
+    sig_prefixed = b58cencode(signature, prefix=get_sig_prefix(curve))
+
+    return TezosSignedTx(
+        signature=sig_prefixed, sig_op_contents=sig_op_contents, operation_hash=ophash
+    )
+
+
+def _get_address_by_tag(address_hash):
+    tag = int(address_hash[0])
+    return b58cencode(address_hash[1:], prefix=get_address_prefix(tag))
+
+
+def _get_address_from_contract(address):
+    if address.tag == TezosContractType.Implicit:
+        return _get_address_by_tag(address.hash)
+
+    elif address.tag == TezosContractType.Originated:
+        return b58cencode(address.hash[:-1], prefix="KT1")
+
+    raise wire.DataError("Invalid contract type")
+
+
+def _get_operation_bytes(msg):
+    result = msg.branch
+
+    # when the account sends first operation in lifetime,
+    # we need to reveal its publickey
+    if msg.reveal is not None:
+        result += _encode_common(msg.reveal, "reveal")
+        result += msg.reveal.public_key
+
+    # transaction operation
+    if msg.transaction is not None:
+        result += _encode_common(msg.transaction, "transaction")
+        result += _encode_zarith(msg.transaction.amount)
+        result += _encode_contract_id(msg.transaction.destination)
+        result += _encode_data_with_bool_prefix(msg.transaction.parameters)
+    # origination operation
+    elif msg.origination is not None:
+        result += _encode_common(msg.origination, "origination")
+        result += msg.origination.manager_pubkey
+        result += _encode_zarith(msg.origination.balance)
+        result += _encode_bool(msg.origination.spendable)
+        result += _encode_bool(msg.origination.delegatable)
+        result += _encode_data_with_bool_prefix(msg.origination.delegate)
+        result += _encode_data_with_bool_prefix(msg.origination.script)
+    # delegation operation
+    elif msg.delegation is not None:
+        result += _encode_common(msg.delegation, "delegation")
+        result += _encode_data_with_bool_prefix(msg.delegation.delegate)
+
+    return bytes(result)
+
+
+def _encode_common(operation, str_operation):
+    operation_tags = {"reveal": 7, "transaction": 8, "origination": 9, "delegation": 10}
+    result = ustruct.pack("<b", operation_tags[str_operation])
+    result += _encode_contract_id(operation.source)
+    result += _encode_zarith(operation.fee)
+    result += _encode_zarith(operation.counter)
+    result += _encode_zarith(operation.gas_limit)
+    result += _encode_zarith(operation.storage_limit)
+    return result
+
+
+def _encode_contract_id(contract_id):
+    return ustruct.pack("<b", contract_id.tag) + contract_id.hash
+
+
+def _encode_bool(boolean):
+    return ustruct.pack("<b", 255) if boolean else ustruct.pack("<b", 0)
+
+
+def _encode_data_with_bool_prefix(data):
+    return _encode_bool(True) + data if data is not None else _encode_bool(False)
+
+
+def _encode_zarith(num):
+    result = bytes()
+
+    while True:
+        byte = num & 127
+        num = num >> 7
+
+        if num == 0:
+            result += ustruct.pack("<b", byte)
+            break
+
+        result += ustruct.pack("<b", 128 | byte)
+
+    return result

src/main.py

@@ -18,6 +18,7 @@ import apps.nem
 import apps.stellar
 import apps.ripple
 import apps.cardano
+import apps.tezos
 
 if __debug__:
     import apps.debug
@@ -34,6 +35,7 @@ apps.nem.boot()
 apps.stellar.boot()
 apps.ripple.boot()
 apps.cardano.boot()
+apps.tezos.boot()
 if __debug__:
     apps.debug.boot()
 else:

src/trezor/messages/TezosCurveType.py

@@ -0,0 +1,5 @@
+# Automatically generated by pb2py
+# fmt: off
+Ed25519 = 0
+Secp256k1 = 1
+P256 = 2

src/trezor/messages/TezosGetAddress.py

@@ -13,13 +13,16 @@ class TezosGetAddress(p.MessageType):
     MESSAGE_WIRE_TYPE = 150
     FIELDS = {
         1: ('address_n', p.UVarintType, p.FLAG_REPEATED),
-        2: ('show_display', p.BoolType, 0),
+        2: ('curve', p.UVarintType, 0),  # default=Ed25519
+        3: ('show_display', p.BoolType, 0),
     }
 
     def __init__(
         self,
         address_n: List[int] = None,
+        curve: int = None,
         show_display: bool = None,
     ) -> None:
         self.address_n = address_n if address_n is not None else []
+        self.curve = curve
         self.show_display = show_display

src/trezor/messages/TezosGetPublicKey.py

@@ -13,13 +13,16 @@ class TezosGetPublicKey(p.MessageType):
     MESSAGE_WIRE_TYPE = 154
     FIELDS = {
         1: ('address_n', p.UVarintType, p.FLAG_REPEATED),
-        2: ('show_display', p.BoolType, 0),
+        2: ('curve', p.UVarintType, 0),  # default=Ed25519
+        3: ('show_display', p.BoolType, 0),
     }
 
     def __init__(
         self,
         address_n: List[int] = None,
+        curve: int = None,
         show_display: bool = None,
     ) -> None:
         self.address_n = address_n if address_n is not None else []
+        self.curve = curve
         self.show_display = show_display

tests/test_apps.tezos.address.py

@@ -0,0 +1,42 @@
+from common import *
+from ubinascii import unhexlify
+from trezor.messages import TezosContractType
+from trezor.messages.TezosContractID import TezosContractID
+from apps.tezos.sign_tx import _get_address_from_contract
+
+
+class TestTezosAddress(unittest.TestCase):
+
+    def test_get_address_from_contract(self):
+        contracts = [
+            TezosContractID(
+                tag=TezosContractType.Implicit,
+                hash=unhexlify("0090ec585b4d5fa39b20213e46b232cc57a4cfab4b")
+            ),
+            TezosContractID(
+                tag=TezosContractType.Implicit,
+                hash=unhexlify("017dfb3fef44082eca8cd3eccebd77db44633ffc9e")
+            ),
+            TezosContractID(
+                tag=TezosContractType.Implicit,
+                hash=unhexlify("02c1fc1b7e503825068ff4fe2f8916f98af981eab1")
+            ),
+            TezosContractID(
+                tag=TezosContractType.Originated,
+                hash=unhexlify("65671dedc69669f066f45d586a2ecdeddacc95af00")
+            )
+        ]
+
+        outputs = [
+            "tz1YrK8Hqt6GAPYRHAaeJmhETYyPSQCHTrkj",
+            "tz2KoN7TFjhp96V2XikqYSGyDmVVUHXvkzko",
+            "tz3e1k3QzCwEbRZrfHCwT3Npvw1rezmMQArY",
+            "KT1HpwLq2AjZgEQspiSnYmdtaHy4NgXw6BDC"
+        ]
+
+        for i, contract in enumerate(contracts):
+            self.assertEqual(_get_address_from_contract(contract), outputs[i])
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_apps.tezos.encode.py

@@ -0,0 +1,79 @@
+from common import *
+from ubinascii import unhexlify
+
+from trezor.messages import TezosContractType
+from trezor.messages.TezosContractID import TezosContractID
+
+from apps.tezos.helpers import b58cencode, b58cdecode
+from apps.tezos.sign_tx import (
+    _encode_zarith,
+    _encode_data_with_bool_prefix,
+    _encode_bool,
+    _encode_contract_id
+)
+
+
+class TestTezosEncoding(unittest.TestCase):
+
+    def test_tezos_encode_zarith(self):
+        inputs = [2000000, 159066, 200, 60000, 157000000, 0]
+        outputs = ["80897a", "dada09", "c801", "e0d403", "c0c2ee4a", "00"]
+
+        for i, o in zip(inputs, outputs):
+            self.assertEqual(_encode_zarith(i), unhexlify(o))
+
+    def test_tezos_encode_data_with_bool_prefix(self):
+        self.assertEqual(_encode_data_with_bool_prefix(None), bytes([0]))
+
+        data = "afffeb1dc3c0"
+        self.assertEqual(_encode_data_with_bool_prefix(unhexlify(data)),
+                         unhexlify("ff" + data))
+
+    def test_tezos_encode_bool(self):
+        self.assertEqual(_encode_bool(True), bytes([255]))
+        self.assertEqual(_encode_bool(False), bytes([0]))
+
+    def test_tezos_encode_contract_id(self):
+        implicit = TezosContractID(
+            tag=TezosContractType.Implicit,
+            hash=unhexlify("00101368afffeb1dc3c089facbbe23f5c30b787ce9")
+        )
+        self.assertEqual(_encode_contract_id(implicit),
+                         unhexlify("0000101368afffeb1dc3c089facbbe23f5c30b787ce9"))
+
+        originated = TezosContractID(
+            tag=TezosContractType.Originated,
+            hash=unhexlify("65671dedc69669f066f45d586a2ecdeddacc95af00")
+        )
+        self.assertEqual(_encode_contract_id(originated),
+                         unhexlify("0165671dedc69669f066f45d586a2ecdeddacc95af00"))
+
+    def test_tezos_b58cencode(self):
+        pkh = unhexlify("101368afffeb1dc3c089facbbe23f5c30b787ce9")
+
+        self.assertEqual(b58cencode(pkh, prefix="tz1"),
+                         "tz1M72kkAJrntPtayM4yU4CCwQPLSdpEgRrn")
+        self.assertEqual(b58cencode(pkh, prefix="tz2"),
+                         "tz29nEixktH9p9XTFX7p8hATUyeLxXEz96KR")
+        self.assertEqual(b58cencode(pkh, prefix="tz3"),
+                         "tz3Mo3gHekQhCmykfnC58ecqJLXrjMKzkF2Q")
+        self.assertEqual(b58cencode(pkh), "2U14dJ6ED97bBHDZTQWA6umVL8SAVefXj")
+
+    def test_tezos_b58cdecode(self):
+        pkh = unhexlify("101368afffeb1dc3c089facbbe23f5c30b787ce9")
+
+        address = "tz1M72kkAJrntPtayM4yU4CCwQPLSdpEgRrn"
+        self.assertEqual(b58cdecode(address, prefix="tz1"), pkh)
+
+        address = "tz29nEixktH9p9XTFX7p8hATUyeLxXEz96KR"
+        self.assertEqual(b58cdecode(address, prefix="tz2"), pkh)
+
+        address = "tz3Mo3gHekQhCmykfnC58ecqJLXrjMKzkF2Q"
+        self.assertEqual(b58cdecode(address, prefix="tz3"), pkh)
+
+        address = "2U14dJ6ED97bBHDZTQWA6umVL8SAVefXj"
+        self.assertEqual(b58cdecode(address), pkh)
+
+
+if __name__ == '__main__':
+    unittest.main()