feat(core/ethereum): EIP-712

Based on original contribution by Max Kupriianov <xlab@hey.com>

Implemented EIP-712 typed data signatures in Ethereum app.

Add eth_abi into pyproject deps

device test for EIP 712

fixed hex decoding for address

fixup! fixed hex decoding for address

code quality, more pythonic code, removing unused imports

running black and isort on changed files

trezorctl file input for EIP 712 data signing

fixup! code quality, more pythonic code, removing unused imports

fixup! fixup! code quality, more pythonic code, removing unused imports

necessary changes after rebase to master

unit tests for sign_typed_data.py

new protobuf messages, working for nonarray types

simplified and verified solution for our simple data

support for simple arrays, without their confirmation

reverting protobuf value messages to bytes, appropriate changes

showing arrays in Trezor, code quality improvements

data validation on Trezor, minor improvements

using custom types for storing type data instead of dicts, addressing feedback from review

moving helper functions to its own file, tests for decode_data

additional overall tests

support for arrays of structs

adding support for metamask_v4_compat variable

using HashWriter object to collect the final hash continously

minor improvements in code quality

validate_field_type function

streaming values from client without saving them, missing UI

prototype of streamed UI using confirm_properties

accounting for bytes in data, more data types in integration tests

rebase on master, using f-strings

minor fixes and improvements from code review

StructHasher class for the whole hashing process

mypy and style changes

asking users whether to show structs and arrays

protobuf descriptions to fix make defs_check

unifying comments, mypy fix

unit tests for StructHasher class

UI fixtures, skipping device tests for T1

addressing majority of code review comments about code quality and structure

changing file structure - layouts, helpers, sign_typed_data

decode_data renaming and docstring, renaming unit test file

using tuples instead of lists in elifs

layout improvements

excluding core/src/apps/common/confirm.py file from the PR

True/False returning layout with Show more button

code review layout improvements

forgotten br_type argument to should_show_more
pull/1894/head
grdddj 3 years ago committed by matejcik
parent 082b7ecd91
commit 73238135d6

@ -8,7 +8,7 @@ error = False
MYDIR = os.path.dirname(__file__)
EXPECTED_PREFIX_RE = re.compile(r"messages-(\w+)\.proto")
EXPECTED_PREFIX_RE = re.compile(r"messages-(\w+)(?:-.*)?\.proto")
for fn in sorted(glob(os.path.join(MYDIR, "messages-*.proto"))):
with open(fn, "rt") as f:

@ -0,0 +1,98 @@
syntax = "proto2";
package hw.trezor.messages.ethereum_eip712;
// Sugar for easier handling in Java
option java_package = "com.satoshilabs.trezor.lib.protobuf";
option java_outer_classname = "TrezorMessageEthereumEIP712";
// Separated from messages-ethereum.proto as it is not implemented on T1 side
// and defining all the messages and fields could be even impossible as recursive
// messages are used here
/**
* Request: Ask device to sign typed data
* @start
* @next EthereumTypedDataStructRequest
* @next EthereumTypedDataValueRequest
* @next EthereumTypedDataSignature
* @next Failure
*/
message EthereumSignTypedData {
repeated uint32 address_n = 1; // BIP-32 path to derive the key from master node
required string primary_type = 2; // name of the root message struct
optional bool metamask_v4_compat = 3 [default=true]; // use MetaMask v4 (see https://github.com/MetaMask/eth-sig-util/issues/106)
}
/**
* Response: Device asks for type information about a struct.
* @next EthereumTypedDataStructAck
*/
message EthereumTypedDataStructRequest {
required string name = 1; // name of the requested struct
}
/**
* Request: Type information about a struct.
* @next EthereumTypedDataStructRequest
*/
message EthereumTypedDataStructAck {
repeated EthereumStructMember members = 1;
message EthereumStructMember {
required EthereumFieldType type = 1;
required string name = 2;
}
message EthereumFieldType {
required EthereumDataType data_type = 1;
optional uint32 size = 2; // for integer types: size in bytes (uint8 has size 1, uint256 has size 32)
// for bytes types: size in bytes, or unset for dynamic
// for arrays: size in elements, or unset for dynamic
// for structs: number of members
// for string, bool and address: unset
optional EthereumFieldType entry_type = 3; // for array types, type of single entry
optional string struct_name = 4; // for structs: its name
}
enum EthereumDataType {
UINT = 1;
INT = 2;
BYTES = 3;
STRING = 4;
BOOL = 5;
ADDRESS = 6;
ARRAY = 7;
STRUCT = 8;
}
}
/**
* Response: Device asks for data at the specific member path.
* @next EthereumTypedDataValueAck
*/
message EthereumTypedDataValueRequest {
repeated uint32 member_path = 1; // member path requested by device
}
/**
* Request: Single value of a specific atomic field.
* @next EthereumTypedDataValueRequest
*/
message EthereumTypedDataValueAck {
required bytes value = 1;
// * atomic types: value of the member.
// Length must match the `size` of the corresponding field type, unless the size is dynamic.
// * array types: number of elements, encoded as uint16.
// * struct types: undefined, Trezor will not query a struct field.
}
/**
* Response: Signed typed data
* @end
*/
message EthereumTypedDataSignature {
required bytes signature = 1; // signature of the typed data
required string address = 2; // address used to sign the typed data
}

@ -185,6 +185,12 @@ enum MessageType {
MessageType_EthereumSignMessage = 64 [(wire_in) = true];
MessageType_EthereumVerifyMessage = 65 [(wire_in) = true];
MessageType_EthereumMessageSignature = 66 [(wire_out) = true];
MessageType_EthereumSignTypedData = 464 [(wire_in) = true];
MessageType_EthereumTypedDataStructRequest = 465 [(wire_out) = true];
MessageType_EthereumTypedDataStructAck = 466 [(wire_in) = true];
MessageType_EthereumTypedDataValueRequest = 467 [(wire_out) = true];
MessageType_EthereumTypedDataValueAck = 468 [(wire_in) = true];
MessageType_EthereumTypedDataSignature = 469 [(wire_out) = true];
// NEM
MessageType_NEMGetAddress = 67 [(wire_in) = true];

@ -0,0 +1,382 @@
{
"setup": {
"mnemonic": "all all all all all all all all all all all all",
"passphrase": ""
},
"tests": [
{
"name": "basic_data",
"parameters": {
"path": "m/44'/60'/0'/0/0",
"metamask_v4_compat": true,
"data": {
"types": {
"EIP712Domain": [
{
"name": "name",
"type": "string"
},
{
"name": "version",
"type": "string"
},
{
"name": "chainId",
"type": "uint256"
},
{
"name": "verifyingContract",
"type": "address"
}
],
"Person": [
{
"name": "name",
"type": "string"
},
{
"name": "wallet",
"type": "address"
}
],
"Mail": [
{
"name": "from",
"type": "Person"
},
{
"name": "to",
"type": "Person"
},
{
"name": "contents",
"type": "string"
}
]
},
"primaryType": "Mail",
"domain": {
"name": "Ether Mail",
"version": "1",
"chainId": 1,
"verifyingContract": "0x1e0Ae8205e9726E6F296ab8869160A6423E2337E"
},
"message": {
"from": {
"name": "Cow",
"wallet": "0xc0004B62C5A39a728e4Af5bee0c6B4a4E54b15ad"
},
"to": {
"name": "Bob",
"wallet": "0x54B0Fa66A065748C40dCA2C7Fe125A2028CF9982"
},
"contents": "Hello, Bob!"
}
}
},
"result": {
"address": "0x73d0385F4d8E00C5e6504C6030F47BF6212736A8",
"sig": "0x2c2d8c7c1facf5bdcd997b5435bb42f3f4170a111ce079c94b5d1e34414f76560c4600d2167568e052ab846555bd590de93bb230987766c636613262eaeb8bdc1c"
}
},
{
"name": "complex_data",
"parameters": {
"path": "m/44'/60'/0'/0/0",
"metamask_v4_compat": true,
"data": {
"types": {
"EIP712Domain": [
{
"name": "name",
"type": "string"
},
{
"name": "version",
"type": "string"
},
{
"name": "chainId",
"type": "uint256"
},
{
"name": "verifyingContract",
"type": "address"
},
{
"name": "salt",
"type": "bytes32"
}
],
"Person": [
{
"name": "name",
"type": "string"
},
{
"name": "wallet",
"type": "address"
},
{
"name": "married",
"type": "bool"
},
{
"name": "kids",
"type": "uint8"
},
{
"name": "karma",
"type": "int16"
},
{
"name": "secret",
"type": "bytes"
},
{
"name": "small_secret",
"type": "bytes16"
},
{
"name": "pets",
"type": "string[]"
},
{
"name": "two_best_friends",
"type": "string[2]"
}
],
"Mail": [
{
"name": "from",
"type": "Person"
},
{
"name": "to",
"type": "Person"
},
{
"name": "messages",
"type": "string[]"
}
]
},
"primaryType": "Mail",
"domain": {
"name": "Ether Mail",
"version": "1",
"chainId": 1,
"verifyingContract": "0x1e0Ae8205e9726E6F296ab8869160A6423E2337E",
"salt": "0xca92da1a6e91d9358328d2f2155af143a7cb74b81a3a4e3e57e2191823dbb56c"
},
"message": {
"from": {
"name": "Amy",
"wallet": "0xc0004B62C5A39a728e4Af5bee0c6B4a4E54b15ad",
"married": true,
"kids": 2,
"karma": 4,
"secret": "0x62c5a39a728e4af5bee0c6b462c5a39a728e4af5bee0c6b462c5a39a728e4af5bee0c6b462c5a39a728e4af5bee0c6b4",
"small_secret": "0x5ccf0e54367104795a47bc0481645d9e",
"pets": [
"parrot"
],
"two_best_friends": [
"Carl",
"Denis"
]
},
"to": {
"name": "Bob",
"wallet": "0x54B0Fa66A065748C40dCA2C7Fe125A2028CF9982",
"married": false,
"kids": 0,
"karma": -4,
"secret": "0x7fe125a2028cf97fe125a2028cf97fe125a2028cf97fe125a2028cf97fe125a2028cf97fe125a2028cf97fe125a2028cf9",
"small_secret": "0xa5e5c47b64775abc476d2962403258de",
"pets": [
"dog",
"cat"
],
"two_best_friends": [
"Emil",
"Franz"
]
},
"messages": [
"Hello, Bob!",
"How are you?",
"Hope you're fine"
]
}
}
},
"result": {
"address": "0x73d0385F4d8E00C5e6504C6030F47BF6212736A8",
"sig": "0xf0a187388b33f17885c915173f38bd613d2ce4346acadfc390b2bae4c6def03667ceac155b5398bd8be326386e841e8820c5254f389a09d6d95ac72e2f6e19e61c"
}
},
{
"name": "struct_list_v4",
"parameters": {
"path": "m/44'/60'/0'/0/0",
"metamask_v4_compat": true,
"data": {
"types": {
"EIP712Domain": [
{
"name": "name",
"type": "string"
},
{
"name": "version",
"type": "string"
},
{
"name": "chainId",
"type": "uint256"
},
{
"name": "verifyingContract",
"type": "address"
}
],
"Person": [
{
"name": "name",
"type": "string"
},
{
"name": "wallet",
"type": "address"
}
],
"Mail": [
{
"name": "from",
"type": "Person"
},
{
"name": "to",
"type": "Person[]"
},
{
"name": "contents",
"type": "string"
}
]
},
"primaryType": "Mail",
"domain": {
"name": "Ether Mail",
"version": "1",
"chainId": 1,
"verifyingContract": "0x1e0Ae8205e9726E6F296ab8869160A6423E2337E"
},
"message": {
"from": {
"name": "Cow",
"wallet": "0xc0004B62C5A39a728e4Af5bee0c6B4a4E54b15ad"
},
"to": [
{
"name": "Bob",
"wallet": "0x54B0Fa66A065748C40dCA2C7Fe125A2028CF9982"
},
{
"name": "Dave",
"wallet": "0x73d0385F4d8E00C5e6504C6030F47BF6212736A8"
}
],
"contents": "Hello, guys!"
}
}
},
"result": {
"address": "0x73d0385F4d8E00C5e6504C6030F47BF6212736A8",
"sig": "0x61d4a929f8513b6327c5eae227d65c394c3857904de483a2191095e2ec35a9ea2ecaf1a461332a6f4847679018848612b35c94150d9be8870ffad01fcbe72cf71c"
}
},
{
"name": "struct_list_non_v4",
"parameters": {
"path": "m/44'/60'/0'/0/0",
"metamask_v4_compat": false,
"data": {
"types": {
"EIP712Domain": [
{
"name": "name",
"type": "string"
},
{
"name": "version",
"type": "string"
},
{
"name": "chainId",
"type": "uint256"
},
{
"name": "verifyingContract",
"type": "address"
}
],
"Person": [
{
"name": "name",
"type": "string"
},
{
"name": "wallet",
"type": "address"
}
],
"Mail": [
{
"name": "from",
"type": "Person"
},
{
"name": "to",
"type": "Person[]"
},
{
"name": "contents",
"type": "string"
}
]
},
"primaryType": "Mail",
"domain": {
"name": "Ether Mail",
"version": "1",
"chainId": 1,
"verifyingContract": "0x1e0Ae8205e9726E6F296ab8869160A6423E2337E"
},
"message": {
"from": {
"name": "Cow",
"wallet": "0xc0004B62C5A39a728e4Af5bee0c6B4a4E54b15ad"
},
"to": [
{
"name": "Bob",
"wallet": "0x54B0Fa66A065748C40dCA2C7Fe125A2028CF9982"
},
{
"name": "Dave",
"wallet": "0x73d0385F4d8E00C5e6504C6030F47BF6212736A8"
}
],
"contents": "Hello, guys!"
}
}
},
"result": {
"address": "0x73d0385F4d8E00C5e6504C6030F47BF6212736A8",
"sig": "0xba6658fd95d8f6048150c8ac64a596d974184522d1069237a57d0e170835fff661ff6f10c5049906a8a508c18d58145dcff91508e70e7e3c186193e3e3bb7dd61b"
}
}
]
}

@ -0,0 +1 @@
Ethereum - support for EIP712 - signing typed data

@ -400,6 +400,8 @@ if not utils.BITCOIN_ONLY:
import trezor.enums.CardanoTxSigningMode
trezor.enums.CardanoTxWitnessType
import trezor.enums.CardanoTxWitnessType
trezor.enums.EthereumDataType
import trezor.enums.EthereumDataType
trezor.enums.NEMImportanceTransferMode
import trezor.enums.NEMImportanceTransferMode
trezor.enums.NEMModificationType
@ -500,12 +502,12 @@ if not utils.BITCOIN_ONLY:
import apps.eos.writers
apps.ethereum
import apps.ethereum
apps.ethereum.address
import apps.ethereum.address
apps.ethereum.get_address
import apps.ethereum.get_address
apps.ethereum.get_public_key
import apps.ethereum.get_public_key
apps.ethereum.helpers
import apps.ethereum.helpers
apps.ethereum.keychain
import apps.ethereum.keychain
apps.ethereum.layout
@ -518,6 +520,8 @@ if not utils.BITCOIN_ONLY:
import apps.ethereum.sign_tx
apps.ethereum.sign_tx_eip1559
import apps.ethereum.sign_tx_eip1559
apps.ethereum.sign_typed_data
import apps.ethereum.sign_typed_data
apps.ethereum.tokens
import apps.ethereum.tokens
apps.ethereum.verify_message

@ -1,6 +1,8 @@
from ubinascii import hexlify, unhexlify
from trezor import wire
from trezor.enums import EthereumDataType
from trezor.messages import EthereumFieldType
if False:
from .networks import NetworkInfo
@ -53,3 +55,71 @@ def bytes_from_address(address: str) -> bytes:
raise wire.ProcessError("Ethereum: Invalid address length")
def get_type_name(field: EthereumFieldType) -> str:
"""Create a string from type definition (like uint256 or bytes16)."""
data_type = field.data_type
size = field.size
TYPE_TRANSLATION_DICT = {
EthereumDataType.UINT: "uint",
EthereumDataType.INT: "int",
EthereumDataType.BYTES: "bytes",
EthereumDataType.STRING: "string",
EthereumDataType.BOOL: "bool",
EthereumDataType.ADDRESS: "address",
}
if data_type == EthereumDataType.STRUCT:
assert field.struct_name is not None # validate_field_type
return field.struct_name
elif data_type == EthereumDataType.ARRAY:
assert field.entry_type is not None # validate_field_type
type_name = get_type_name(field.entry_type)
if size is None:
return f"{type_name}[]"
else:
return f"{type_name}[{size}]"
elif data_type in (EthereumDataType.UINT, EthereumDataType.INT):
assert size is not None # validate_field_type
return TYPE_TRANSLATION_DICT[data_type] + str(size * 8)
elif data_type == EthereumDataType.BYTES:
if size:
return TYPE_TRANSLATION_DICT[data_type] + str(size)
else:
return TYPE_TRANSLATION_DICT[data_type]
else:
# all remaining types can use the name directly
# if the data_type is left out, this will raise KeyError
return TYPE_TRANSLATION_DICT[data_type]
def decode_typed_data(data: bytes, type_name: str) -> str:
"""Used by sign_typed_data module to show data to user."""
if type_name.startswith("bytes"):
return hexlify(data).decode()
elif type_name == "string":
return data.decode()
elif type_name == "address":
return address_from_bytes(data)
elif type_name == "bool":
return "true" if data == b"\x01" else "false"
elif type_name.startswith("uint"):
return str(int.from_bytes(data, "big"))
elif type_name.startswith("int"):
# Micropython does not implement "signed" arg in int.from_bytes()
return str(from_bytes_bigendian_signed(data))
raise ValueError # Unsupported data type for direct field decoding
def from_bytes_bigendian_signed(b: bytes) -> int:
negative = b[0] & 0x80
if negative:
neg_b = bytearray(b)
for i in range(len(neg_b)):
neg_b[i] = ~neg_b[i] & 0xFF
result = int.from_bytes(neg_b, "big")
return -result - 1
else:
return int.from_bytes(b, "big")

@ -14,6 +14,7 @@ if False:
EthereumSignMessage,
EthereumSignTx,
EthereumSignTxEIP1559,
EthereumSignTypedData,
)
from apps.common.keychain import MsgOut, Handler, HandlerWithKeychain
@ -23,6 +24,7 @@ if False:
EthereumGetPublicKey,
EthereumSignTx,
EthereumSignMessage,
EthereumSignTypedData,
]
MsgIn = TypeVar("MsgIn", bound=EthereumMessages)

@ -1,22 +1,24 @@
from ubinascii import hexlify
from trezor import ui
from trezor.enums import ButtonRequestType
from trezor.strings import format_amount
from trezor.enums import ButtonRequestType, EthereumDataType
from trezor.messages import EthereumFieldType, EthereumStructMember
from trezor.strings import format_amount, format_plural
from trezor.ui.layouts import (
confirm_address,
confirm_amount,
confirm_blob,
confirm_output,
confirm_text,
should_show_more,
)
from trezor.ui.layouts.tt.altcoin import confirm_total_ethereum
from . import networks, tokens
from .address import address_from_bytes
from .helpers import address_from_bytes, decode_typed_data, get_type_name
if False:
from typing import Awaitable
from typing import Awaitable, Iterable, Optional
from trezor.wire import Context
@ -107,6 +109,112 @@ def require_confirm_data(ctx: Context, data: bytes, data_total: int) -> Awaitabl
)
async def confirm_hash(ctx: Context, message_hash: bytes) -> None:
await confirm_blob(
ctx,
"confirm_hash",
title="Confirm hash",
data="0x" + hexlify(message_hash).decode(),
hold=True,
)
async def should_show_domain(ctx: Context, name: bytes, version: bytes) -> bool:
domain_name = decode_typed_data(name, "string")
domain_version = decode_typed_data(version, "string")
para = (
(ui.NORMAL, "Name and version"),
(ui.BOLD, domain_name),
(ui.BOLD, domain_version),
)
return await should_show_more(
ctx,
title="Confirm domain",
para=para,
button_text="Show full domain",
br_type="should_show_domain",
)
async def should_show_struct(
ctx: Context,
description: str,
data_members: list[EthereumStructMember],
title: str = "Confirm struct",
button_text: str = "Show full struct",
) -> bool:
para = (
(ui.BOLD, description),
(
ui.NORMAL,
format_plural("Contains {count} {plural}", len(data_members), "key"),
),
(ui.NORMAL, ", ".join(field.name for field in data_members)),
)
return await should_show_more(
ctx,
title=title,
para=para,
button_text=button_text,
br_type="should_show_struct",
)
async def should_show_array(
ctx: Context,
parent_objects: Iterable[str],
data_type: str,
size: int,
) -> bool:
para = ((ui.NORMAL, format_plural("Array of {count} {plural}", size, data_type)),)
return await should_show_more(
ctx,
title=limit_str(".".join(parent_objects)),
para=para,
button_text="Show full array",
br_type="should_show_array",
)
async def confirm_typed_value(
ctx: Context,
name: str,
value: bytes,
parent_objects: list[str],
field: EthereumFieldType,
array_index: Optional[int] = None,
) -> None:
type_name = get_type_name(field)
if array_index is not None:
title = limit_str(".".join(parent_objects + [name]))
description = f"[{array_index}] ({type_name})"
else:
title = limit_str(".".join(parent_objects))
description = f"{name} ({type_name})"
data = decode_typed_data(value, type_name)
if field.data_type in (EthereumDataType.ADDRESS, EthereumDataType.BYTES):
await confirm_blob(
ctx,
"confirm_typed_value",
title=title,
data=data,
description=description,
ask_pagination=True,
)
else:
await confirm_text(
ctx,
"confirm_typed_value",
title=title,
data=data,
description=description,
)
def format_ethereum_amount(
value: int, token: tokens.TokenInfo | None, chain_id: int
) -> str:
@ -123,3 +231,11 @@ def format_ethereum_amount(
decimals = 0
return f"{format_amount(value, decimals)} {suffix}"
def limit_str(s: str, limit: int = 16) -> str:
"""Shortens string to show the last <limit> characters."""
if len(s) <= limit + 2:
return s
return ".." + s[-limit:]

@ -0,0 +1,517 @@
from trezor import wire
from trezor.crypto.curve import secp256k1
from trezor.crypto.hashlib import sha3_256
from trezor.enums import EthereumDataType
from trezor.messages import (
EthereumFieldType,
EthereumSignTypedData,
EthereumTypedDataSignature,
EthereumTypedDataStructAck,
EthereumTypedDataStructRequest,
EthereumTypedDataValueAck,
EthereumTypedDataValueRequest,
)
from trezor.utils import HashWriter
from apps.common import paths
from .helpers import address_from_bytes, get_type_name
from .keychain import PATTERNS_ADDRESS, with_keychain_from_path
from .layout import (
confirm_hash,
confirm_typed_value,
should_show_array,
should_show_domain,
should_show_struct,
)
if False:
from apps.common.keychain import Keychain
from trezor.wire import Context
# Maximum data size we support
MAX_VALUE_BYTE_SIZE = 1024
@with_keychain_from_path(*PATTERNS_ADDRESS)
async def sign_typed_data(
ctx: Context, msg: EthereumSignTypedData, keychain: Keychain
) -> EthereumTypedDataSignature:
await paths.validate_path(ctx, keychain, msg.address_n)
data_hash = await generate_typed_data_hash(
ctx, msg.primary_type, msg.metamask_v4_compat
)
node = keychain.derive(msg.address_n)
signature = secp256k1.sign(
node.private_key(), data_hash, False, secp256k1.CANONICAL_SIG_ETHEREUM
)
return EthereumTypedDataSignature(
address=address_from_bytes(node.ethereum_pubkeyhash()),
signature=signature[1:] + signature[0:1],
)
async def generate_typed_data_hash(
ctx: Context, primary_type: str, metamask_v4_compat: bool = True
) -> bytes:
"""
Generate typed data hash according to EIP-712 specification
https://eips.ethereum.org/EIPS/eip-712#specification
metamask_v4_compat - a flag that enables compatibility with MetaMask's signTypedData_v4 method
"""
typed_data_envelope = TypedDataEnvelope(
ctx=ctx,
primary_type=primary_type,
metamask_v4_compat=metamask_v4_compat,
)
await typed_data_envelope.collect_types()
name, version = await get_name_and_version_for_domain(ctx, typed_data_envelope)
show_domain = await should_show_domain(ctx, name, version)
domain_separator = await typed_data_envelope.hash_struct(
primary_type="EIP712Domain",
member_path=[0],
show_data=show_domain,
parent_objects=["EIP712Domain"],
)
show_message = await should_show_struct(
ctx,
description=primary_type,
data_members=typed_data_envelope.types[primary_type].members,
title="Confirm message",
button_text="Show full message",
)
message_hash = await typed_data_envelope.hash_struct(
primary_type=primary_type,
member_path=[1],
show_data=show_message,
parent_objects=[primary_type],
)
await confirm_hash(ctx, message_hash)
return keccak256(b"\x19" + b"\x01" + domain_separator + message_hash)
def get_hash_writer() -> HashWriter:
return HashWriter(sha3_256(keccak=True))
def keccak256(message: bytes) -> bytes:
h = get_hash_writer()
h.extend(message)
return h.get_digest()
class TypedDataEnvelope:
"""Encapsulates the type information for the message being hashed and signed."""
def __init__(
self,
ctx: Context,
primary_type: str,
metamask_v4_compat: bool,
) -> None:
self.ctx = ctx
self.primary_type = primary_type
self.metamask_v4_compat = metamask_v4_compat
self.types: dict[str, EthereumTypedDataStructAck] = {}
async def collect_types(self) -> None:
"""Aggregate type collection process for both domain and message data."""
await self._collect_types("EIP712Domain")
await self._collect_types(self.primary_type)
async def _collect_types(self, type_name: str) -> None:
"""Recursively collect types from the client."""
req = EthereumTypedDataStructRequest(name=type_name)
current_type = await self.ctx.call(req, EthereumTypedDataStructAck)
self.types[type_name] = current_type
for member in current_type.members:
validate_field_type(member.type)
if (
member.type.data_type == EthereumDataType.STRUCT
and member.type.struct_name not in self.types
):
assert member.type.struct_name is not None # validate_field_type
await self._collect_types(member.type.struct_name)
async def hash_struct(
self,
primary_type: str,
member_path: list[int],
show_data: bool,
parent_objects: list[str],
) -> bytes:
"""Generate a hash representation of the whole struct."""
w = get_hash_writer()
self.hash_type(w, primary_type)
await self.get_and_encode_data(
w=w,
primary_type=primary_type,
member_path=member_path,
show_data=show_data,
parent_objects=parent_objects,
)
return w.get_digest()
def hash_type(self, w: HashWriter, primary_type: str) -> None:
"""Create a representation of a type."""
result = keccak256(self.encode_type(primary_type))
w.extend(result)
def encode_type(self, primary_type: str) -> bytes:
"""
SPEC:
The type of a struct is encoded as name "(" member₁ "," member₂ "," memberₙ ")"
where each member is written as type " " name
If the struct type references other struct types (and these in turn reference even more struct types),
then the set of referenced struct types is collected, sorted by name and appended to the encoding.
"""
result: list[str] = []
deps: set[str] = set()
self.find_typed_dependencies(primary_type, deps)
deps.remove(primary_type)
for type_name in [primary_type] + sorted(deps):
members = self.types[type_name].members
fields = ",".join(f"{get_type_name(m.type)} {m.name}" for m in members)
result.append(f"{type_name}({fields})")
return "".join(result).encode()
def find_typed_dependencies(
self,
primary_type: str,
results: set[str],
) -> None:
"""Find all types within a type definition object."""
# We already have this type or it is not even a defined type
if (primary_type in results) or (primary_type not in self.types):
return
results.add(primary_type)
# Recursively adding all the children struct types,
# also looking into (even nested) arrays for them
for member in self.types[primary_type].members:
member_type = member.type
while member_type.data_type == EthereumDataType.ARRAY:
assert member_type.entry_type is not None # validate_field_type
member_type = member_type.entry_type
if member_type.data_type == EthereumDataType.STRUCT:
assert member_type.struct_name is not None # validate_field_type
self.find_typed_dependencies(member_type.struct_name, results)
async def get_and_encode_data(
self,
w: HashWriter,
primary_type: str,
member_path: list[int],
show_data: bool,
parent_objects: list[str],
) -> None:
"""
Gradually fetch data from client and encode the whole struct.
SPEC:
The encoding of a struct instance is enc(value₁) enc(value₂) enc(valueₙ),
i.e. the concatenation of the encoded member values in the order that they appear in the type.
Each encoded member value is exactly 32-byte long.
"""
type_members = self.types[primary_type].members
member_value_path = member_path + [0]
current_parent_objects = parent_objects + [""]
for member_index, member in enumerate(type_members):
member_value_path[-1] = member_index
field_name = member.name
field_type = member.type
# Arrays and structs need special recursive handling
if field_type.data_type == EthereumDataType.STRUCT:
assert field_type.struct_name is not None # validate_field_type
struct_name = field_type.struct_name
current_parent_objects[-1] = field_name
if show_data:
show_struct = await should_show_struct(
ctx=self.ctx,
description=struct_name,
data_members=self.types[struct_name].members,
title=".".join(current_parent_objects),
)
else:
show_struct = False
res = await self.hash_struct(
primary_type=struct_name,
member_path=member_value_path,
show_data=show_struct,
parent_objects=current_parent_objects,
)
w.extend(res)
elif field_type.data_type == EthereumDataType.ARRAY:
# Getting the length of the array first, if not fixed
if field_type.size is None:
array_size = await get_array_size(self.ctx, member_value_path)
else:
array_size = field_type.size
assert field_type.entry_type is not None # validate_field_type
entry_type = field_type.entry_type
current_parent_objects[-1] = field_name
if show_data:
show_array = await should_show_array(
ctx=self.ctx,
parent_objects=current_parent_objects,
data_type=get_type_name(entry_type),
size=array_size,
)
else:
show_array = False
arr_w = get_hash_writer()
el_member_path = member_value_path + [0]
for i in range(array_size):
el_member_path[-1] = i
# TODO: we do not support arrays of arrays, check if we should
if entry_type.data_type == EthereumDataType.STRUCT:
assert entry_type.struct_name is not None # validate_field_type
struct_name = entry_type.struct_name
# Metamask V4 implementation has a bug, that causes the
# behavior of structs in array be different from SPEC
# Explanation at https://github.com/MetaMask/eth-sig-util/pull/107
# encode_data() is the way to process structs in arrays, but
# Metamask V4 is using hash_struct() even in this case
if self.metamask_v4_compat:
res = await self.hash_struct(
primary_type=struct_name,
member_path=el_member_path,
show_data=show_array,
parent_objects=current_parent_objects,
)
arr_w.extend(res)
else:
await self.get_and_encode_data(
w=arr_w,
primary_type=struct_name,
member_path=el_member_path,
show_data=show_array,
parent_objects=current_parent_objects,
)
else:
value = await get_value(self.ctx, entry_type, el_member_path)
encode_field(arr_w, entry_type, value)
if show_array:
await confirm_typed_value(
ctx=self.ctx,
name=field_name,
value=value,
parent_objects=parent_objects,
field=entry_type,
array_index=i,
)
w.extend(arr_w.get_digest())
else:
value = await get_value(self.ctx, field_type, member_value_path)
encode_field(w, field_type, value)
if show_data:
await confirm_typed_value(
ctx=self.ctx,
name=field_name,
value=value,
parent_objects=parent_objects,
field=field_type,
)
def encode_field(
w: HashWriter,
field: EthereumFieldType,
value: bytes,
) -> None:
"""
SPEC:
Atomic types:
- Boolean false and true are encoded as uint256 values 0 and 1 respectively
- Addresses are encoded as uint160
- Integer values are sign-extended to 256-bit and encoded in big endian order
- Bytes1 to bytes31 are arrays with a beginning (index 0)
and an end (index length - 1), they are zero-padded at the end to bytes32 and encoded
in beginning to end order
Dynamic types:
- Bytes and string are encoded as a keccak256 hash of their contents
Reference types:
- Array values are encoded as the keccak256 hash of the concatenated
encodeData of their contents
- Struct values are encoded recursively as hashStruct(value)
"""
data_type = field.data_type
if data_type == EthereumDataType.BYTES:
if field.size is None:
w.extend(keccak256(value))
else:
write_rightpad32(w, value)
elif data_type == EthereumDataType.STRING:
w.extend(keccak256(value))
elif data_type == EthereumDataType.INT:
write_leftpad32(w, value, signed=True)
elif data_type in (
EthereumDataType.UINT,
EthereumDataType.BOOL,
EthereumDataType.ADDRESS,
):
write_leftpad32(w, value)
else:
raise ValueError # Unsupported data type for field encoding
def write_leftpad32(w: HashWriter, value: bytes, signed: bool = False) -> None:
assert len(value) <= 32
# Values need to be sign-extended, so accounting for negative ints
if signed and value[0] & 0x80:
pad_value = 0xFF
else:
pad_value = 0x00
for _ in range(32 - len(value)):
w.append(pad_value)
w.extend(value)
def write_rightpad32(w: HashWriter, value: bytes) -> None:
assert len(value) <= 32
w.extend(value)
for _ in range(32 - len(value)):
w.append(0x00)
def validate_value(field: EthereumFieldType, value: bytes) -> None:
"""
Make sure the byte data we receive are not corrupted or incorrect.
Raise wire.DataError if encountering a problem, so clients are notified.
"""
# Checking if the size corresponds to what is defined in types,
# and also setting our maximum supported size in bytes
if field.size is not None:
if len(value) != field.size:
raise wire.DataError("Invalid length")
else:
if len(value) > MAX_VALUE_BYTE_SIZE:
raise wire.DataError(f"Invalid length, bigger than {MAX_VALUE_BYTE_SIZE}")
# Specific tests for some data types
if field.data_type == EthereumDataType.BOOL:
if value not in (b"\x00", b"\x01"):
raise wire.DataError("Invalid boolean value")
elif field.data_type == EthereumDataType.ADDRESS:
if len(value) != 20:
raise wire.DataError("Invalid address")
elif field.data_type == EthereumDataType.STRING:
try:
value.decode()
except UnicodeError:
raise wire.DataError("Invalid UTF-8")
def validate_field_type(field: EthereumFieldType) -> None:
"""
Make sure the field type is consistent with our expectation.
Raise wire.DataError if encountering a problem, so clients are notified.
"""
data_type = field.data_type
# entry_type is only for arrays
if data_type == EthereumDataType.ARRAY:
if field.entry_type is None:
raise wire.DataError("Missing entry_type in array")
# We also need to validate it recursively
validate_field_type(field.entry_type)
else:
if field.entry_type is not None:
raise wire.DataError("Unexpected entry_type in nonarray")
# struct_name is only for structs
if data_type == EthereumDataType.STRUCT:
if field.struct_name is None:
raise wire.DataError("Missing struct_name in struct")
else:
if field.struct_name is not None:
raise wire.DataError("Unexpected struct_name in nonstruct")
# size is special for each type
if data_type == EthereumDataType.STRUCT:
if field.size is None:
raise wire.DataError("Missing size in struct")
elif data_type == EthereumDataType.BYTES:
if field.size is not None and not 1 <= field.size <= 32:
raise wire.DataError("Invalid size in bytes")
elif data_type in (
EthereumDataType.UINT,
EthereumDataType.INT,
):
if field.size is None or not 1 <= field.size <= 32:
raise wire.DataError("Invalid size in int/uint")
elif data_type in (
EthereumDataType.STRING,
EthereumDataType.BOOL,
EthereumDataType.ADDRESS,
):
if field.size is not None:
raise wire.DataError("Unexpected size in str/bool/addr")
async def get_array_size(ctx: Context, member_path: list[int]) -> int:
"""Get the length of an array at specific `member_path` from the client."""
# Field type for getting the array length from client, so we can check the return value
ARRAY_LENGTH_TYPE = EthereumFieldType(data_type=EthereumDataType.UINT, size=2)
length_value = await get_value(ctx, ARRAY_LENGTH_TYPE, member_path)
return int.from_bytes(length_value, "big")
async def get_value(
ctx: Context,
field: EthereumFieldType,
member_value_path: list[int],
) -> bytes:
"""Get a single value from the client and perform its validation."""
req = EthereumTypedDataValueRequest(
member_path=member_value_path,
)
res = await ctx.call(req, EthereumTypedDataValueAck)
value = res.value
validate_value(field=field, value=value)
return value
async def get_name_and_version_for_domain(
ctx: Context, typed_data_envelope: TypedDataEnvelope
) -> tuple[bytes, bytes]:
domain_name = b"unknown"
domain_version = b"unknown"
domain_members = typed_data_envelope.types["EIP712Domain"].members
member_value_path = [0, 0]
for member_index, member in enumerate(domain_members):
member_value_path[-1] = member_index
if member.name == "name":
domain_name = await get_value(ctx, member.type, member_value_path)
elif member.name == "version":
domain_version = await get_value(ctx, member.type, member_value_path)
return domain_name, domain_version

@ -110,6 +110,8 @@ def find_message_handler_module(msg_type: int) -> str:
return "apps.ethereum.sign_message"
elif msg_type == MessageType.EthereumVerifyMessage:
return "apps.ethereum.verify_message"
elif msg_type == MessageType.EthereumSignTypedData:
return "apps.ethereum.sign_typed_data"
# monero
elif msg_type == MessageType.MoneroGetAddress:

@ -0,0 +1,12 @@
# Automatically generated by pb2py
# fmt: off
# isort:skip_file
UINT = 1
INT = 2
BYTES = 3
STRING = 4
BOOL = 5
ADDRESS = 6
ARRAY = 7
STRUCT = 8

@ -97,6 +97,12 @@ if not utils.BITCOIN_ONLY:
EthereumSignMessage = 64
EthereumVerifyMessage = 65
EthereumMessageSignature = 66
EthereumSignTypedData = 464
EthereumTypedDataStructRequest = 465
EthereumTypedDataStructAck = 466
EthereumTypedDataValueRequest = 467
EthereumTypedDataValueAck = 468
EthereumTypedDataSignature = 469
NEMGetAddress = 67
NEMAddress = 68
NEMSignTx = 69

@ -102,6 +102,12 @@ if TYPE_CHECKING:
EthereumSignMessage = 64
EthereumVerifyMessage = 65
EthereumMessageSignature = 66
EthereumSignTypedData = 464
EthereumTypedDataStructRequest = 465
EthereumTypedDataStructAck = 466
EthereumTypedDataValueRequest = 467
EthereumTypedDataValueAck = 468
EthereumTypedDataSignature = 469
NEMGetAddress = 67
NEMAddress = 68
NEMSignTx = 69
@ -419,6 +425,16 @@ if TYPE_CHECKING:
LEFT = 2
RIGHT = 3
class EthereumDataType(IntEnum):
UINT = 1
INT = 2
BYTES = 3
STRING = 4
BOOL = 5
ADDRESS = 6
ARRAY = 7
STRUCT = 8
class NEMMosaicLevy(IntEnum):
MosaicLevy_Absolute = 1
MosaicLevy_Percentile = 2

@ -35,6 +35,7 @@ if TYPE_CHECKING:
from trezor.enums import CardanoTxWitnessType # noqa: F401
from trezor.enums import DebugSwipeDirection # noqa: F401
from trezor.enums import DecredStakingSpendType # noqa: F401
from trezor.enums import EthereumDataType # noqa: F401
from trezor.enums import FailureType # noqa: F401
from trezor.enums import InputScriptType # noqa: F401
from trezor.enums import MessageType # noqa: F401
@ -3237,6 +3238,132 @@ if TYPE_CHECKING:
def is_type_of(cls, msg: protobuf.MessageType) -> TypeGuard["EthereumAccessList"]:
return isinstance(msg, cls)
class EthereumSignTypedData(protobuf.MessageType):
address_n: "list[int]"
primary_type: "str"
metamask_v4_compat: "bool"
def __init__(
self,
*,
primary_type: "str",
address_n: "list[int] | None" = None,
metamask_v4_compat: "bool | None" = None,
) -> None:
pass
@classmethod
def is_type_of(cls, msg: protobuf.MessageType) -> TypeGuard["EthereumSignTypedData"]:
return isinstance(msg, cls)
class EthereumTypedDataStructRequest(protobuf.MessageType):
name: "str"
def __init__(
self,
*,
name: "str",
) -> None:
pass
@classmethod
def is_type_of(cls, msg: protobuf.MessageType) -> TypeGuard["EthereumTypedDataStructRequest"]:
return isinstance(msg, cls)
class EthereumTypedDataStructAck(protobuf.MessageType):
members: "list[EthereumStructMember]"
def __init__(
self,
*,
members: "list[EthereumStructMember] | None" = None,
) -> None:
pass
@classmethod
def is_type_of(cls, msg: protobuf.MessageType) -> TypeGuard["EthereumTypedDataStructAck"]:
return isinstance(msg, cls)
class EthereumTypedDataValueRequest(protobuf.MessageType):
member_path: "list[int]"
def __init__(
self,
*,
member_path: "list[int] | None" = None,
) -> None:
pass
@classmethod
def is_type_of(cls, msg: protobuf.MessageType) -> TypeGuard["EthereumTypedDataValueRequest"]:
return isinstance(msg, cls)
class EthereumTypedDataValueAck(protobuf.MessageType):
value: "bytes"
def __init__(
self,
*,
value: "bytes",
) -> None:
pass
@classmethod
def is_type_of(cls, msg: protobuf.MessageType) -> TypeGuard["EthereumTypedDataValueAck"]:
return isinstance(msg, cls)
class EthereumTypedDataSignature(protobuf.MessageType):
signature: "bytes"
address: "str"
def __init__(
self,
*,
signature: "bytes",
address: "str",
) -> None:
pass
@classmethod
def is_type_of(cls, msg: protobuf.MessageType) -> TypeGuard["EthereumTypedDataSignature"]:
return isinstance(msg, cls)
class EthereumStructMember(protobuf.MessageType):
type: "EthereumFieldType"
name: "str"
def __init__(
self,
*,
type: "EthereumFieldType",
name: "str",
) -> None:
pass
@classmethod
def is_type_of(cls, msg: protobuf.MessageType) -> TypeGuard["EthereumStructMember"]:
return isinstance(msg, cls)
class EthereumFieldType(protobuf.MessageType):
data_type: "EthereumDataType"
size: "int | None"
entry_type: "EthereumFieldType | None"
struct_name: "str | None"
def __init__(
self,
*,
data_type: "EthereumDataType",
size: "int | None" = None,
entry_type: "EthereumFieldType | None" = None,
struct_name: "str | None" = None,
) -> None:
pass
@classmethod
def is_type_of(cls, msg: protobuf.MessageType) -> TypeGuard["EthereumFieldType"]:
return isinstance(msg, cls)
class MoneroTransactionSourceEntry(protobuf.MessageType):
outputs: "list[MoneroOutputEntry]"
real_output: "int | None"

@ -515,7 +515,7 @@ async def should_show_more(
ctx: wire.GenericContext,
title: str,
para: Iterable[tuple[int, str]],
button_text: str = "Show details",
button_text: str = "Show all",
br_type: str = "should_show_more",
br_code: ButtonRequestType = ButtonRequestType.Other,
icon: str = ui.ICON_DEFAULT,

@ -0,0 +1,752 @@
from common import *
from trezor import wire
from trezor.messages import EthereumTypedDataStructAck as ETDSA
from trezor.messages import EthereumStructMember as ESM
from trezor.messages import EthereumFieldType as EFT
from trezor.messages import EthereumTypedDataValueAck
from trezor.enums import EthereumDataType as EDT
if not utils.BITCOIN_ONLY:
from apps.ethereum.sign_typed_data import (
encode_field,
validate_value,
validate_field_type,
keccak256,
TypedDataEnvelope,
)
from apps.ethereum.helpers import (
get_type_name,
decode_typed_data,
)
class MockContext:
"""Simulating the client sending us data values."""
def __init__(self, message_contents: list):
# TODO: it could be worth (for better readability and quicker modification)
# to accept a whole EIP712 JSON object and create the list internally
self.message_contents = message_contents
async def call(self, request, _resp_type) -> bytes:
entry = self.message_contents
for index in request.member_path:
entry = entry[index]
if isinstance(entry, list):
value = len(entry).to_bytes(2, "big")
else:
value = entry
return EthereumTypedDataValueAck(value=value)
# Helper functions from trezorctl to build expected type data structures
# TODO: it could be better to group these functions into a class, to visibly differentiate it
def get_type_definitions(types: dict) -> dict:
result = {}
for struct, fields in types.items():
members = []
for name, type in fields:
field_type = get_field_type(type, types)
struct_member = ESM(
type=field_type,
name=name,
)
members.append(struct_member)
result[struct] = ETDSA(members=members)
return result
def get_field_type(type_name: str, types: dict) -> EFT:
data_type = None
size = None
entry_type = None
struct_name = None
if is_array(type_name):
data_type = EDT.ARRAY
array_size = parse_array_n(type_name)
size = None if array_size == "dynamic" else array_size
member_typename = typeof_array(type_name)
entry_type = get_field_type(member_typename, types)
elif type_name.startswith("uint"):
data_type = EDT.UINT
size = get_byte_size_for_int_type(type_name)
elif type_name.startswith("int"):
data_type = EDT.INT
size = get_byte_size_for_int_type(type_name)
elif type_name.startswith("bytes"):
data_type = EDT.BYTES
size = None if type_name == "bytes" else parse_type_n(type_name)
elif type_name == "string":
data_type = EDT.STRING
elif type_name == "bool":
data_type = EDT.BOOL
elif type_name == "address":
data_type = EDT.ADDRESS
elif type_name in types:
data_type = EDT.STRUCT
size = len(types[type_name])
struct_name = type_name
else:
raise ValueError(f"Unsupported type name: {type_name}")
return EFT(
data_type=data_type,
size=size,
entry_type=entry_type,
struct_name=struct_name,
)
def is_array(type_name: str) -> bool:
return type_name[-1] == "]"
def typeof_array(type_name: str) -> str:
return type_name[: type_name.rindex("[")]
def parse_type_n(type_name: str) -> int:
"""Parse N from type<N>.
Example: "uint256" -> 256
"""
# STRANGE: "ImportError: no module named 're'" in Micropython?
buf = ""
for char in reversed(type_name):
if char.isdigit():
buf += char
else:
return int("".join(reversed(buf)))
def parse_array_n(type_name: str) -> Union[int, str]:
"""Parse N in type[<N>] where "type" can itself be an array type."""
if type_name.endswith("[]"):
return "dynamic"
start_idx = type_name.rindex("[") + 1
return int(type_name[start_idx:-1])
def get_byte_size_for_int_type(int_type: str) -> int:
return parse_type_n(int_type) // 8
types_basic = {
"EIP712Domain": [
("name", "string"),
("version", "string"),
("chainId", "uint256"),
("verifyingContract", "address"),
],
"Person": [
("name", "string"),
("wallet", "address"),
],
"Mail": [
("from", "Person"),
("to", "Person"),
("contents", "string"),
],
}
TYPES_BASIC = get_type_definitions(types_basic)
types_complex = {
"EIP712Domain": [
("name", "string"),
("version", "string"),
("chainId", "uint256"),
("verifyingContract", "address"),
("salt", "bytes32"),
],
"Person": [
("name", "string"),
("wallet", "address"),
("married", "bool"),
("kids", "uint8"),
("karma", "int16"),
("secret", "bytes"),
("small_secret", "bytes16"),
("pets", "string[]"),
("two_best_friends", "string[2]"),
],
"Mail": [
("from", "Person"),
("to", "Person"),
("messages", "string[]"),
],
}
TYPES_COMPLEX = get_type_definitions(types_complex)
DOMAIN_VALUES = [
[
b"Ether Mail",
b"1",
# 1
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01",
# 0x1e0Ae8205e9726E6F296ab8869160A6423E2337E
b"\x1e\n\xe8 ^\x97&\xe6\xf2\x96\xab\x88i\x16\nd#\xe23~",
]
]
MESSAGE_VALUES_BASIC = [
[
[
b"Cow",
b"\xc0\x00Kb\xc5\xa3\x9ar\x8eJ\xf5\xbe\xe0\xc6\xb4\xa4\xe5K\x15\xad",
],
[
b"Bob",
b"T\xb0\xfaf\xa0et\x8c@\xdc\xa2\xc7\xfe\x12Z (\xcf\x99\x82",
],
b"Hello, Bob!",
]
]
MESSAGE_VALUES_COMPLEX = [
[
[
b"Amy",
b"\xc0\x00Kb\xc5\xa3\x9ar\x8eJ\xf5\xbe\xe0\xc6\xb4\xa4\xe5K\x15\xad",
b"\x01",
b"\x02",
b"\x00\x04",
b"b\xc5\xa3\x9ar\x8eJ\xf5\xbe\xe0\xc6\xb4b\xc5\xa3\x9ar\x8eJ\xf5\xbe\xe0\xc6\xb4b\xc5\xa3\x9ar\x8eJ\xf5\xbe\xe0\xc6\xb4b\xc5\xa3\x9ar\x8eJ\xf5\xbe\xe0\xc6\xb4",
b"\\\xcf\x0eT6q\x04yZG\xbc\x04\x81d]\x9e",
[
b"parrot",
],
[
b"Carl",
b"Denis",
]
],
[
b"Bob",
b"T\xb0\xfaf\xa0et\x8c@\xdc\xa2\xc7\xfe\x12Z (\xcf\x99\x82",
b"\x00",
b"\x00",
b"\xff\xfc",
b"\x7f\xe1%\xa2\x02\x8c\xf9\x7f\xe1%\xa2\x02\x8c\xf9\x7f\xe1%\xa2\x02\x8c\xf9\x7f\xe1%\xa2\x02\x8c\xf9\x7f\xe1%\xa2\x02\x8c\xf9\x7f\xe1%\xa2\x02\x8c\xf9\x7f\xe1%\xa2\x02\x8c\xf9",
b"\xa5\xe5\xc4{dwZ\xbcGm)b@2X\xde",
[
b"dog",
b"cat",
],
[
b"Emil",
b"Franz",
]
],
[
b"Hello, Bob!",
b"How are you?",
b"Hope you're fine"
]
]
]
# Object for testing functionality not needing context
# (Each test needs to assign EMPTY_ENVELOPE.types as needed)
EMPTY_ENVELOPE = TypedDataEnvelope(
ctx=None,
primary_type="test",
metamask_v4_compat=True,
)
# TODO: validate it more by some third party app, like signing data by Metamask
# ??? How to approach the testing ???
# - we could copy the most important test cases testing important functionality
# - could also download the eth-sig-util package together witn node.js and validating it
# Testcases are at:
# https://github.com/MetaMask/eth-sig-util/blob/73ace3309bf4b97d901fb66cd61db15eede7afe9/src/sign-typed-data.test.ts
# Worth testing/implementing:
# should encode data with a recursive data type
# should ignore extra unspecified message properties
# should throw an error when an atomic property is set to null
# Missing custom type properties are omitted in V3, but encoded as 0 (bytes32) in V4
@unittest.skipUnless(not utils.BITCOIN_ONLY, "altcoin")
class TestEthereumSignTypedData(unittest.TestCase):
def test_hash_struct(self):
"""These final expected results should be generated by some third party"""
VECTORS = ( # primary_type, data, types, expected
(
# Generated by eth_account
"EIP712Domain",
[
[
b"Ether Mail",
b"1",
# 1
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01",
# 0x1e0Ae8205e9726E6F296ab8869160A6423E2337E
b"\x1e\n\xe8 ^\x97&\xe6\xf2\x96\xab\x88i\x16\nd#\xe23~",
]
],
TYPES_BASIC,
b"\x97\xd6\xf57t\xb8\x10\xfb\xda'\xe0\x91\xc0<jmh\x15\xdd\x12p\xc2\xe6.\x82\xc6\x91|\x1e\xffwK",
),
(
# Taken from https://github.com/ethereum/EIPs/blob/master/assets/eip-712/Example.sol
"EIP712Domain",
[
[
b"Ether Mail",
b"1",
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01",
# 0xCcCCccccCCCCcCCCCCCcCcCccCcCCCcCcccccccC
b"\xcc\xcc\xcc\xcc\xcc\xcc\xcc\xcc\xcc\xcc\xcc\xcc\xcc\xcc\xcc\xcc\xcc\xcc\xcc\xcc",
]
],
TYPES_BASIC,
b"\xf2\xce\xe3u\xfaB\xb4!C\x80@%\xfcD\x9d\xea\xfdP\xcc\x03\x1c\xa2W\xe0\xb1\x94\xa6P\xa9\x12\t\x0f",
),
(
# Taken from https://github.com/ethereum/EIPs/blob/master/assets/eip-712/Example.sol
"Mail",
[
[
[
b"Cow",
# 0xCD2a3d9F938E13CD947Ec05AbC7FE734Df8DD826
b"\xcd*=\x9f\x93\x8e\x13\xcd\x94~\xc0Z\xbc\x7f\xe74\xdf\x8d\xd8&",
],
[
b"Bob",
# 0xbBbBBBBbbBBBbbbBbbBbbbbBBbBbbbbBbBbbBBbB
b"\xbb\xbb\xbb\xbb\xbb\xbb\xbb\xbb\xbb\xbb\xbb\xbb\xbb\xbb\xbb\xbb\xbb\xbb\xbb\xbb",
],
b"Hello, Bob!"
],
],
TYPES_BASIC,
b"\xc5,\x0e\xe5\xd8BdG\x18\x06)\n?,L\xec\xfcT\x90bk\xf9\x12\xd0\x1f$\rz'K7\x1e",
),
(
# Generated by eth_account
"Mail",
MESSAGE_VALUES_BASIC,
TYPES_BASIC,
b"\xeae)\xf0\xee\x9e\xb0\xb2\x07\xb5\xa8\xb0\xeb\xfag=9\x8djx&(\x18\xda\x1d'\x0b\xd18\xf8\x1f\x03",
),
(
# Tested by Metamask's eth_signTypedData_v4
"Mail",
MESSAGE_VALUES_COMPLEX,
TYPES_COMPLEX,
b"\xdb\xaf\xe7F\xb1\xc4~Hp\xf6\xf7r\x05f\r<I\xa9M\xb9\xa8\t9\x80\x9b\xfc\xa7\xbfC\x91\x9d\xf5",
),
)
for primary_type, data, types, expected in VECTORS:
ctx = MockContext(data)
typed_data_envelope = TypedDataEnvelope(
ctx=ctx,
primary_type=primary_type,
metamask_v4_compat=True,
)
typed_data_envelope.types = types
res = await_result(
typed_data_envelope.hash_struct(
primary_type=primary_type,
member_path=[0],
show_data=False,
parent_objects=[primary_type],
)
)
self.assertEqual(res, expected)
def test_get_and_encode_data(self):
VECTORS = ( # primary_type, data, types, expected
(
"EIP712Domain",
DOMAIN_VALUES,
TYPES_BASIC,
b"\xc7\x0e\xf0f8S[H\x81\xfa\xfc\xac\x82\x87\xe2\x10\xe3v\x9f\xf1\xa8\xe9\x1f\x1b\x95\xd6$na\xe4\xd3\xc6\xc8\x9e\xfd\xaaT\xc0\xf2\x0cz\xdfa(\x82\xdf\tP\xf5\xa9Qc~\x03\x07\xcd\xcbLg/)\x8b\x8b\xc6\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x1e\n\xe8 ^\x97&\xe6\xf2\x96\xab\x88i\x16\nd#\xe23~",
),
(
"Mail",
MESSAGE_VALUES_BASIC,
TYPES_BASIC,
b"${\xf6.\x89\xe8\xab\xc6g\x02\x1e\xa5\xe4hf\xad\xbe\xf8\x0f\xb0\x01\xc2\x17-\xf9#\n0A/\x13\xa15'H\x1b_\xa5a5\x06\x04\xa6\rsOI\xee\x90\x7f\x17O[\xa6\xbby\x1a\xabAun\xce~\xd1\xb5\xaa\xdf1T\xa2a\xab\xdd\x90\x86\xfcb{a\xef\xca&\xaeW\x02p\x1d\x05\xcd#\x05\xf7\xc5*/\xc8",
),
(
"Mail",
MESSAGE_VALUES_COMPLEX,
TYPES_COMPLEX,
b"\xcaby\xe0T\nDc\x16\x1f\x8f\x019\x1d@\xbf/\xdb\xd3\xb3\xed\x16:p\xf2\x0c\xa5\xaa5\xd5\xfd.\xaa\xf1q\xbd\x91N\x93\xdfE\x88$o76WR\x9a\xe0\x17\xa6#\x9a6Jy\x94\x0c\xf2\xf8!\x97E.\xf5\xf3=3t\xf4\xb6%\xfd\x1bu\xba\xae\xb1\xfa\xbd\xa5%\x8a\xc2\xa3\x19\x0bbu\xf2\xadzkg\x93",
),
)
for primary_type, data, types, expected in VECTORS:
ctx = MockContext(data)
typed_data_envelope = TypedDataEnvelope(
ctx=ctx,
primary_type=primary_type,
metamask_v4_compat=True,
)
typed_data_envelope.types = types
w = bytearray()
await_result(
typed_data_envelope.get_and_encode_data(
w=w,
primary_type=primary_type,
member_path=[0],
show_data=False,
parent_objects=[primary_type],
)
)
self.assertEqual(w, expected)
def test_encode_type(self):
VECTORS = ( # primary_type, types, expected
(
"EIP712Domain",
TYPES_BASIC,
b"EIP712Domain(string name,string version,uint256 chainId,address verifyingContract)",
),
("Person", TYPES_BASIC, b"Person(string name,address wallet)"),
(
"Mail",
TYPES_BASIC,
b"Mail(Person from,Person to,string contents)Person(string name,address wallet)",
),
(
"Person",
TYPES_COMPLEX,
b"Person(string name,address wallet,bool married,uint8 kids,int16 karma,bytes secret,bytes16 small_secret,string[] pets,string[2] two_best_friends)",
),
(
"Mail",
TYPES_COMPLEX,
b"Mail(Person from,Person to,string[] messages)Person(string name,address wallet,bool married,uint8 kids,int16 karma,bytes secret,bytes16 small_secret,string[] pets,string[2] two_best_friends)",
),
)
for primary_type, types, expected in VECTORS:
EMPTY_ENVELOPE.types = types
res = EMPTY_ENVELOPE.encode_type(primary_type=primary_type)
self.assertEqual(res, expected)
def test_hash_type(self):
VECTORS = ( # primary_type, expected
(
"EIP712Domain",
keccak256(
b"EIP712Domain(string name,string version,uint256 chainId,address verifyingContract)"
),
),
("Person", keccak256(b"Person(string name,address wallet)")),
(
"Mail",
keccak256(
b"Mail(Person from,Person to,string contents)Person(string name,address wallet)"
),
),
)
for primary_type, expected in VECTORS:
w = bytearray()
EMPTY_ENVELOPE.types = TYPES_BASIC
EMPTY_ENVELOPE.hash_type(w=w, primary_type=primary_type)
self.assertEqual(w, expected)
def test_find_typed_dependencies(self):
# We need to be able to recognize dependency even as array of structs
types_dependency_only_as_array = {
"MAIN": [
("jkl", "SECONDARY[]"),
],
"SECONDARY": [
("abc", "string"),
("def", "TERNARY[][]"),
],
"TERNARY": [
("ghi", "string"),
],
}
types_dependency_only_as_array = get_type_definitions(types_dependency_only_as_array)
VECTORS = ( # primary_type, expected, types
(
"EIP712Domain",
{"EIP712Domain"},
TYPES_BASIC
),
(
"Person",
{"Person"},
TYPES_BASIC
),
(
"Mail",
{"Mail", "Person"},
TYPES_BASIC
),
(
"MAIN",
{"MAIN", "SECONDARY", "TERNARY"},
types_dependency_only_as_array
),
(
"UnexistingType",
set(),
TYPES_BASIC
),
)
for primary_type, expected, types in VECTORS:
res = set()
EMPTY_ENVELOPE.types = types
EMPTY_ENVELOPE.find_typed_dependencies(primary_type=primary_type, results=res)
self.assertEqual(res, expected)
def test_encode_field(self):
VECTORS = ( # field, value, expected
(
EFT(data_type=EDT.STRING, size=None),
b"Ether Mail",
keccak256(b"Ether Mail"),
),
(
EFT(data_type=EDT.STRING, size=None),
b"1",
keccak256(b"1"),
),
(
EFT(data_type=EDT.UINT, size=32),
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01",
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01",
),
(
EFT(data_type=EDT.UINT, size=4),
b"\x00\x00\x00\xde",
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xde",
),
(
EFT(data_type=EDT.INT, size=1),
b"\x05",
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x05",
),
(
EFT(data_type=EDT.ADDRESS, size=None),
b"\x1e\n\xe8 ^\x97&\xe6\xf2\x96\xab\x88i\x16\nd#\xe23~",
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x1e\n\xe8 ^\x97&\xe6\xf2\x96\xab\x88i\x16\nd#\xe23~",
),
(
EFT(data_type=EDT.BOOL, size=None),
b"\x01",
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01",
),
(
EFT(data_type=EDT.BOOL, size=None),
b"\x00",
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00",
),
)
for field, value, expected in VECTORS:
# metamask_v4_compat should not have any effect on the
# result for items outside of arrays
for metamask_v4_compat in [True, False]:
w = bytearray()
encode_field(
w=w,
field=field,
value=value,
)
self.assertEqual(w, expected)
def test_validate_value(self):
VECTORS_VALID_INVALID = ( # field, valid_values, invalid_values
(
EFT(data_type=EDT.UINT, size=1),
[b"\xff"],
[b"\xff\xee"],
),
(
EFT(data_type=EDT.BYTES, size=8),
[b"\xff" * 8],
[b"\xff" * 7, b"\xff" * 9],
),
(
EFT(data_type=EDT.BOOL, size=None),
[b"\x00", b"\x01"],
[b"0", b"\x00\x01"],
),
(
EFT(data_type=EDT.STRING, size=None),
[b"\x7f", b"a" * 1024],
[b"\x80", b"a" * 1025],
),
(
EFT(data_type=EDT.ADDRESS, size=None),
[b"T\xb0\xfaf\xa0et\x8c@\xdc\xa2\xc7\xfe\x12Z (\xcf\x99\x82"],
[b"T\xb0\xfaf\xa0et\x8c@\xdc\xa2\xc7\xfe\x12Z (\xcf\x99"],
),
)
for field, valid_values, invalid_values in VECTORS_VALID_INVALID:
for valid_value in valid_values:
validate_value(field=field, value=valid_value)
for invalid_value in invalid_values:
with self.assertRaises(wire.DataError):
validate_value(field=field, value=invalid_value)
def test_validate_field_type(self):
ET = EFT(data_type=EDT.BYTES, size=8)
VECTORS_VALID = ( # valid_field
(EFT(data_type=EDT.STRUCT, struct_name="Person", size=3)),
(EFT(data_type=EDT.ARRAY, entry_type=ET, size=None)),
(EFT(data_type=EDT.ARRAY, entry_type=ET, size=6)),
(EFT(data_type=EDT.BYTES, size=32)),
(EFT(data_type=EDT.BYTES, size=None)),
(EFT(data_type=EDT.UINT, size=8)),
(EFT(data_type=EDT.INT, size=16)),
(EFT(data_type=EDT.STRING)),
(EFT(data_type=EDT.BOOL)),
(EFT(data_type=EDT.ADDRESS)),
)
for valid_field in VECTORS_VALID:
validate_field_type(field=valid_field)
ET = EFT(data_type=EDT.BYTES, size=8)
ET_INVALID = EFT(data_type=EDT.BYTES, size=33)
VECTORS_INVALID = ( # invalid_field
(EFT(data_type=EDT.STRUCT, size=None)),
(EFT(data_type=EDT.STRUCT, struct_name=None)),
(EFT(data_type=EDT.STRUCT, entry_type=ET)),
(EFT(data_type=EDT.ARRAY, struct_name="Person")),
(EFT(data_type=EDT.ARRAY, entry_type=None)),
(EFT(data_type=EDT.ARRAY, entry_type=ET_INVALID)),
(EFT(data_type=EDT.BYTES, struct_name="Person")),
(EFT(data_type=EDT.BYTES, size=0)),
(EFT(data_type=EDT.BYTES, size=33)),
(EFT(data_type=EDT.BYTES, entry_type=ET)),
(EFT(data_type=EDT.UINT, struct_name="Person")),
(EFT(data_type=EDT.UINT, size=None)),
(EFT(data_type=EDT.UINT, size=0)),
(EFT(data_type=EDT.UINT, size=33)),
(EFT(data_type=EDT.UINT, entry_type=ET)),
(EFT(data_type=EDT.INT, struct_name="Person")),
(EFT(data_type=EDT.INT, size=None)),
(EFT(data_type=EDT.INT, size=0)),
(EFT(data_type=EDT.INT, size=33)),
(EFT(data_type=EDT.INT, entry_type=ET)),
(EFT(data_type=EDT.STRING, struct_name="Person")),
(EFT(data_type=EDT.STRING, size=3)),
(EFT(data_type=EDT.STRING, entry_type=ET)),
(EFT(data_type=EDT.BOOL, struct_name="Person")),
(EFT(data_type=EDT.BOOL, size=1)),
(EFT(data_type=EDT.BOOL, size=3)),
(EFT(data_type=EDT.BOOL, entry_type=ET)),
(EFT(data_type=EDT.ADDRESS, struct_name="Person")),
(EFT(data_type=EDT.ADDRESS, size=3)),
(EFT(data_type=EDT.ADDRESS, size=20)),
(EFT(data_type=EDT.ADDRESS, entry_type=ET)),
)
for invalid_field in VECTORS_INVALID:
with self.assertRaises(wire.DataError):
validate_field_type(field=invalid_field)
def test_get_type_name(self):
VECTORS = ( # field, expected
(
EFT(
data_type=EDT.ARRAY,
size=None,
entry_type=EFT(data_type=EDT.UINT, size=32),
),
"uint256[]",
),
(
EFT(
data_type=EDT.ARRAY,
size=None,
entry_type=EFT(
data_type=EDT.ARRAY,
size=None,
entry_type=EFT(
data_type=EDT.BYTES,
size=16
),
)
),
"bytes16[][]",
),
(
EFT(
data_type=EDT.ARRAY,
size=4,
entry_type=EFT(data_type=EDT.STRING, size=None),
),
"string[4]",
),
(
EFT(data_type=EDT.STRUCT, size=2, struct_name="Person"),
"Person",
),
(
EFT(data_type=EDT.STRING, size=None),
"string",
),
(
EFT(data_type=EDT.ADDRESS, size=None),
"address",
),
(
EFT(data_type=EDT.BOOL, size=None),
"bool",
),
(
EFT(data_type=EDT.UINT, size=20),
"uint160",
),
(
EFT(data_type=EDT.INT, size=8),
"int64",
),
(
EFT(data_type=EDT.BYTES, size=8),
"bytes8",
),
(
EFT(data_type=EDT.BYTES, size=None),
"bytes",
),
)
for field, expected in VECTORS:
res = get_type_name(field)
self.assertEqual(res, expected)
def test_decode_typed_data(self):
VECTORS = ( # data, type_name, expected
(b"\x4a\x56", "bytes", "4a56"),
(b"Hello, Bob!", "string", "Hello, Bob!"),
(
b"\x1e\n\xe8 ^\x97&\xe6\xf2\x96\xab\x88i\x16\nd#\xe23~",
"address",
"0x1e0Ae8205e9726E6F296ab8869160A6423E2337E",
),
(b"\x01", "bool", "true"),
(b"\x00", "bool", "false"),
(b"\x3f\x46\xaa", "uint", "4146858"),
(b"\x3f\x46\xaa", "int", "4146858"),
(b"\xff\xf1", "uint", "65521"),
(b"\xff\xf1", "int", "-15"),
)
for data, type_name, expected in VECTORS:
res = decode_typed_data(data, type_name)
self.assertEqual(res, expected)
if __name__ == "__main__":
unittest.main()

@ -0,0 +1 @@
Support for EIP-712 in library and `trezorctl ethereum sign-typed-data`

@ -253,6 +253,7 @@ Ethereum commands.
get-address Get Ethereum address in hex encoding.
get-public-node Get Ethereum public node of given path.
sign-message Sign message with Ethereum address.
sign-typed-data Sign typed data (EIP-712) with Ethereum address.
sign-tx Sign (and optionally publish) Ethereum transaction.
verify-message Verify message signed with Ethereum address.

@ -14,6 +14,7 @@
# You should have received a copy of the License along with this library.
# If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
import json
import re
import sys
from decimal import Decimal
@ -87,7 +88,7 @@ def _parse_access_list_item(value):
try:
arr = value.split(":")
address, storage_keys = arr[0], arr[1:]
storage_keys_bytes = [_decode_hex(key) for key in storage_keys]
storage_keys_bytes = [ethereum.decode_hex(key) for key in storage_keys]
return ethereum.messages.EthereumAccessList(address, storage_keys_bytes)
except Exception:
@ -103,13 +104,6 @@ def _list_units(ctx, param, value):
ctx.exit()
def _decode_hex(value):
if value.startswith("0x") or value.startswith("0X"):
return bytes.fromhex(value[2:])
else:
return bytes.fromhex(value)
def _erc20_contract(w3, token_address, to_address, amount):
min_abi = [
{
@ -129,7 +123,7 @@ def _erc20_contract(w3, token_address, to_address, amount):
def _format_access_list(access_list: List[ethereum.messages.EthereumAccessList]):
mapped = map(
lambda item: [_decode_hex(item.address), item.storage_keys],
lambda item: [ethereum.decode_hex(item.address), item.storage_keys],
access_list,
)
return list(mapped)
@ -289,7 +283,7 @@ def sign_tx(
amount = 0
if data:
data = _decode_hex(data)
data = ethereum.decode_hex(data)
else:
data = b""
@ -338,7 +332,7 @@ def sign_tx(
)
)
to = _decode_hex(to_address)
to = ethereum.decode_hex(to_address)
if is_eip1559:
transaction = rlp.encode(
(
@ -389,6 +383,34 @@ def sign_message(client, address, message):
return output
@cli.command()
@click.option("-n", "--address", required=True, help=PATH_HELP)
@click.option(
"--metamask-v4-compat/--no-metamask-v4-compat",
default=True,
help="Be compatible with Metamask's signTypedData_v4 implementation",
)
@click.argument("file", type=click.File("r"))
@with_client
def sign_typed_data(client, address, metamask_v4_compat, file):
"""Sign typed data (EIP-712) with Ethereum address.
Currently NOT supported:
- arrays of arrays
- recursive structs
"""
address_n = tools.parse_path(address)
data = json.loads(file.read())
ret = ethereum.sign_typed_data(
client, address_n, data, metamask_v4_compat=metamask_v4_compat
)
output = {
"address": ret.address,
"signature": f"0x{ret.signature.hex()}",
}
return output
@cli.command()
@click.argument("address")
@click.argument("signature")
@ -396,5 +418,5 @@ def sign_message(client, address, message):
@with_client
def verify_message(client, address, signature, message):
"""Verify message signed with Ethereum address."""
signature = _decode_hex(signature)
signature = ethereum.decode_hex(signature)
return ethereum.verify_message(client, address, signature, message)

@ -14,14 +14,124 @@
# You should have received a copy of the License along with this library.
# If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
import re
from typing import Any, Dict, List, Union
from . import exceptions, messages
from .tools import expect, normalize_nfc, session
def int_to_big_endian(value):
def int_to_big_endian(value) -> bytes:
return value.to_bytes((value.bit_length() + 7) // 8, "big")
def decode_hex(value: str) -> bytes:
if value.startswith(("0x", "0X")):
return bytes.fromhex(value[2:])
else:
return bytes.fromhex(value)
def sanitize_typed_data(data: dict) -> dict:
"""Remove properties from a message object that are not defined per EIP-712."""
REQUIRED_KEYS = ("types", "primaryType", "domain", "message")
sanitized_data = {key: data[key] for key in REQUIRED_KEYS}
sanitized_data["types"].setdefault("EIP712Domain", [])
return sanitized_data
def is_array(type_name: str) -> bool:
return type_name[-1] == "]"
def typeof_array(type_name: str) -> str:
return type_name[: type_name.rindex("[")]
def parse_type_n(type_name: str) -> int:
"""Parse N from type<N>. Example: "uint256" -> 256."""
return int(re.search(r"\d+$", type_name).group(0))
def parse_array_n(type_name: str) -> Union[int, str]:
"""Parse N in type[<N>] where "type" can itself be an array type."""
if type_name.endswith("[]"):
return "dynamic"
start_idx = type_name.rindex("[") + 1
return int(type_name[start_idx:-1])
def get_byte_size_for_int_type(int_type: str) -> int:
return parse_type_n(int_type) // 8
def get_field_type(type_name: str, types: dict) -> messages.EthereumFieldType:
data_type = None
size = None
entry_type = None
struct_name = None
if is_array(type_name):
data_type = messages.EthereumDataType.ARRAY
array_size = parse_array_n(type_name)
size = None if array_size == "dynamic" else array_size
member_typename = typeof_array(type_name)
entry_type = get_field_type(member_typename, types)
# Not supporting nested arrays currently
if entry_type.data_type == messages.EthereumDataType.ARRAY:
raise NotImplementedError("Nested arrays are not supported")
elif type_name.startswith("uint"):
data_type = messages.EthereumDataType.UINT
size = get_byte_size_for_int_type(type_name)
elif type_name.startswith("int"):
data_type = messages.EthereumDataType.INT
size = get_byte_size_for_int_type(type_name)
elif type_name.startswith("bytes"):
data_type = messages.EthereumDataType.BYTES
size = None if type_name == "bytes" else parse_type_n(type_name)
elif type_name == "string":
data_type = messages.EthereumDataType.STRING
elif type_name == "bool":
data_type = messages.EthereumDataType.BOOL
elif type_name == "address":
data_type = messages.EthereumDataType.ADDRESS
elif type_name in types:
data_type = messages.EthereumDataType.STRUCT
size = len(types[type_name])
struct_name = type_name
else:
raise ValueError(f"Unsupported type name: {type_name}")
return messages.EthereumFieldType(
data_type=data_type,
size=size,
entry_type=entry_type,
struct_name=struct_name,
)
def encode_data(value: Any, type_name: str) -> bytes:
if type_name.startswith("bytes"):
return decode_hex(value)
elif type_name == "string":
return value.encode()
elif type_name.startswith(("int", "uint")):
byte_length = get_byte_size_for_int_type(type_name)
return int(value).to_bytes(
byte_length, "big", signed=type_name.startswith("int")
)
elif type_name == "bool":
if not isinstance(value, bool):
raise ValueError(f"Invalid bool value - {value}")
return int(value).to_bytes(1, "big")
elif type_name == "address":
return decode_hex(value)
# We should be receiving only atomic, non-array types
raise ValueError(f"Unsupported data type for direct field encoding: {type_name}")
# ====== Client functions ====== #
@ -96,7 +206,7 @@ def sign_tx_eip1559(
chain_id,
max_gas_fee,
max_priority_fee,
access_list=()
access_list=(),
):
length = len(data)
data, chunk = data[1024:], data[:1024]
@ -130,6 +240,75 @@ def sign_message(client, n, message):
return client.call(messages.EthereumSignMessage(address_n=n, message=message))
@expect(messages.EthereumTypedDataSignature)
def sign_typed_data(
client, n: List[int], data: Dict[str, Any], *, metamask_v4_compat: bool = True
):
data = sanitize_typed_data(data)
types = data["types"]
request = messages.EthereumSignTypedData(
address_n=n,
primary_type=data["primaryType"],
metamask_v4_compat=metamask_v4_compat,
)
response = client.call(request)
# Sending all the types
while isinstance(response, messages.EthereumTypedDataStructRequest):
struct_name = response.name
members = []
for field in types[struct_name]:
field_type = get_field_type(field["type"], types)
struct_member = messages.EthereumStructMember(
type=field_type,
name=field["name"],
)
members.append(struct_member)
request = messages.EthereumTypedDataStructAck(members=members)
response = client.call(request)
# Sending the whole message that should be signed
while isinstance(response, messages.EthereumTypedDataValueRequest):
root_index = response.member_path[0]
# Index 0 is for the domain data, 1 is for the actual message
if root_index == 0:
member_typename = "EIP712Domain"
member_data = data["domain"]
elif root_index == 1:
member_typename = data["primaryType"]
member_data = data["message"]
else:
client.cancel()
raise exceptions.TrezorException("Root index can only be 0 or 1")
# It can be asking for a nested structure (the member path being [X, Y, Z, ...])
# TODO: what to do when the value is missing (for example in recursive types)?
for index in response.member_path[1:]:
if isinstance(member_data, dict):
member_def = types[member_typename][index]
member_typename = member_def["type"]
member_data = member_data[member_def["name"]]
elif isinstance(member_data, list):
member_typename = typeof_array(member_typename)
member_data = member_data[index]
# If we were asked for a list, first sending its length and we will be receiving
# requests for individual elements later
if isinstance(member_data, list):
# Sending the length as uint16
encoded_data = len(member_data).to_bytes(2, "big")
else:
encoded_data = encode_data(member_data, member_typename)
request = messages.EthereumTypedDataValueAck(value=encoded_data)
response = client.call(request)
return response
def verify_message(client, address, signature, message):
message = normalize_nfc(message)
try:

@ -123,6 +123,12 @@ class MessageType(IntEnum):
EthereumSignMessage = 64
EthereumVerifyMessage = 65
EthereumMessageSignature = 66
EthereumSignTypedData = 464
EthereumTypedDataStructRequest = 465
EthereumTypedDataStructAck = 466
EthereumTypedDataValueRequest = 467
EthereumTypedDataValueAck = 468
EthereumTypedDataSignature = 469
NEMGetAddress = 67
NEMAddress = 68
NEMSignTx = 69
@ -447,6 +453,17 @@ class DebugSwipeDirection(IntEnum):
RIGHT = 3
class EthereumDataType(IntEnum):
UINT = 1
INT = 2
BYTES = 3
STRING = 4
BOOL = 5
ADDRESS = 6
ARRAY = 7
STRUCT = 8
class NEMMosaicLevy(IntEnum):
MosaicLevy_Absolute = 1
MosaicLevy_Percentile = 2
@ -4327,6 +4344,139 @@ class EosActionUnknown(protobuf.MessageType):
self.data_chunk = data_chunk
class EthereumSignTypedData(protobuf.MessageType):
MESSAGE_WIRE_TYPE = 464
FIELDS = {
1: protobuf.Field("address_n", "uint32", repeated=True, required=False),
2: protobuf.Field("primary_type", "string", repeated=False, required=True),
3: protobuf.Field("metamask_v4_compat", "bool", repeated=False, required=False),
}
def __init__(
self,
*,
primary_type: "str",
address_n: Optional[List["int"]] = None,
metamask_v4_compat: Optional["bool"] = True,
) -> None:
self.address_n = address_n if address_n is not None else []
self.primary_type = primary_type
self.metamask_v4_compat = metamask_v4_compat
class EthereumTypedDataStructRequest(protobuf.MessageType):
MESSAGE_WIRE_TYPE = 465
FIELDS = {
1: protobuf.Field("name", "string", repeated=False, required=True),
}
def __init__(
self,
*,
name: "str",
) -> None:
self.name = name
class EthereumTypedDataStructAck(protobuf.MessageType):
MESSAGE_WIRE_TYPE = 466
FIELDS = {
1: protobuf.Field("members", "EthereumStructMember", repeated=True, required=False),
}
def __init__(
self,
*,
members: Optional[List["EthereumStructMember"]] = None,
) -> None:
self.members = members if members is not None else []
class EthereumTypedDataValueRequest(protobuf.MessageType):
MESSAGE_WIRE_TYPE = 467
FIELDS = {
1: protobuf.Field("member_path", "uint32", repeated=True, required=False),
}
def __init__(
self,
*,
member_path: Optional[List["int"]] = None,
) -> None:
self.member_path = member_path if member_path is not None else []
class EthereumTypedDataValueAck(protobuf.MessageType):
MESSAGE_WIRE_TYPE = 468
FIELDS = {
1: protobuf.Field("value", "bytes", repeated=False, required=True),
}
def __init__(
self,
*,
value: "bytes",
) -> None:
self.value = value
class EthereumTypedDataSignature(protobuf.MessageType):
MESSAGE_WIRE_TYPE = 469
FIELDS = {
1: protobuf.Field("signature", "bytes", repeated=False, required=True),
2: protobuf.Field("address", "string", repeated=False, required=True),
}
def __init__(
self,
*,
signature: "bytes",
address: "str",
) -> None:
self.signature = signature
self.address = address
class EthereumStructMember(protobuf.MessageType):
MESSAGE_WIRE_TYPE = None
FIELDS = {
1: protobuf.Field("type", "EthereumFieldType", repeated=False, required=True),
2: protobuf.Field("name", "string", repeated=False, required=True),
}
def __init__(
self,
*,
type: "EthereumFieldType",
name: "str",
) -> None:
self.type = type
self.name = name
class EthereumFieldType(protobuf.MessageType):
MESSAGE_WIRE_TYPE = None
FIELDS = {
1: protobuf.Field("data_type", "EthereumDataType", repeated=False, required=True),
2: protobuf.Field("size", "uint32", repeated=False, required=False),
3: protobuf.Field("entry_type", "EthereumFieldType", repeated=False, required=False),
4: protobuf.Field("struct_name", "string", repeated=False, required=False),
}
def __init__(
self,
*,
data_type: "EthereumDataType",
size: Optional["int"] = None,
entry_type: Optional["EthereumFieldType"] = None,
struct_name: Optional["str"] = None,
) -> None:
self.data_type = data_type
self.size = size
self.entry_type = entry_type
self.struct_name = struct_name
class EthereumGetPublicKey(protobuf.MessageType):
MESSAGE_WIRE_TYPE = 450
FIELDS = {

@ -0,0 +1,144 @@
# This file is part of the Trezor project.
#
# Copyright (C) 2012-2019 SatoshiLabs and contributors
#
# This library is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License version 3
# as published by the Free Software Foundation.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the License along with this library.
# If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
import pytest
from trezorlib import ethereum, exceptions
from trezorlib.tools import parse_path
from ...common import parametrize_using_common_fixtures
SHOW_MORE = (143, 167)
pytestmark = [pytest.mark.altcoin, pytest.mark.ethereum, pytest.mark.skip_t1]
@parametrize_using_common_fixtures("ethereum/sign_typed_data.json")
def test_ethereum_sign_typed_data(client, parameters, result):
with client:
address_n = parse_path(parameters["path"])
ret = ethereum.sign_typed_data(
client,
address_n,
parameters["data"],
metamask_v4_compat=parameters["metamask_v4_compat"],
)
assert ret.address == result["address"]
assert f"0x{ret.signature.hex()}" == result["sig"]
# Being the same as the first object in ethereum/sign_typed_data.json
DATA = {
"types": {
"EIP712Domain": [
{"name": "name", "type": "string"},
{"name": "version", "type": "string"},
{"name": "chainId", "type": "uint256"},
{"name": "verifyingContract", "type": "address"},
],
"Person": [
{"name": "name", "type": "string"},
{"name": "wallet", "type": "address"},
],
"Mail": [
{"name": "from", "type": "Person"},
{"name": "to", "type": "Person"},
{"name": "contents", "type": "string"},
],
},
"primaryType": "Mail",
"domain": {
"name": "Ether Mail",
"version": "1",
"chainId": 1,
"verifyingContract": "0x1e0Ae8205e9726E6F296ab8869160A6423E2337E",
},
"message": {
"from": {"name": "Cow", "wallet": "0xc0004B62C5A39a728e4Af5bee0c6B4a4E54b15ad"},
"to": {"name": "Bob", "wallet": "0x54B0Fa66A065748C40dCA2C7Fe125A2028CF9982"},
"contents": "Hello, Bob!",
},
}
def input_flow_show_more(client):
"""Clicks show_more button wherever possible"""
yield # confirm domain
client.debug.wait_layout()
client.debug.click(SHOW_MORE)
# confirm domain properties
for _ in range(4):
yield
client.debug.press_yes()
yield # confirm message
client.debug.wait_layout()
client.debug.click(SHOW_MORE)
yield # confirm message.from
client.debug.wait_layout()
client.debug.click(SHOW_MORE)
# confirm message.from properties
for _ in range(2):
yield
client.debug.press_yes()
yield # confirm message.to
client.debug.wait_layout()
client.debug.click(SHOW_MORE)
# confirm message.to properties
for _ in range(2):
yield
client.debug.press_yes()
yield # confirm message.contents
client.debug.press_yes()
yield # confirm final hash
client.debug.press_yes()
def input_flow_cancel(client):
"""Clicks cancelling button"""
yield # confirm domain
client.debug.press_no()
def test_ethereum_sign_typed_data_show_more_button(client):
with client:
client.watch_layout()
client.set_input_flow(input_flow_show_more(client))
ethereum.sign_typed_data(
client,
parse_path("m/44'/60'/0'/0/0"),
DATA,
metamask_v4_compat=True,
)
def test_ethereum_sign_typed_data_cancel(client):
with client, pytest.raises(exceptions.Cancelled):
client.watch_layout()
client.set_input_flow(input_flow_cancel(client))
ethereum.sign_typed_data(
client,
parse_path("m/44'/60'/0'/0/0"),
DATA,
metamask_v4_compat=True,
)

@ -192,6 +192,12 @@
"ethereum-test_getpublickey.py::test_ethereum_getpublickey[parameters0-result0]": "095af81ec79e9b510c90d9fa34fed343f3840807190c67bc237af885695ae687",
"ethereum-test_getpublickey.py::test_ethereum_getpublickey[parameters1-result1]": "095af81ec79e9b510c90d9fa34fed343f3840807190c67bc237af885695ae687",
"ethereum-test_getpublickey.py::test_ethereum_getpublickey[parameters2-result2]": "095af81ec79e9b510c90d9fa34fed343f3840807190c67bc237af885695ae687",
"ethereum-test_sign_typed_data.py::test_ethereum_sign_typed_data[basic_data]": "4f512d6beb0222079aaa878a2bcd2c41ac389fdd47edad3e0b0b5779677d8697",
"ethereum-test_sign_typed_data.py::test_ethereum_sign_typed_data[complex_data]": "275a8630a795c966419f6fc6de834bb576bfc3dbc16a1fd7605aa2b8ceae666e",
"ethereum-test_sign_typed_data.py::test_ethereum_sign_typed_data[struct_list_non_v4]": "7dd23b14bd273b937836a24cf056c745e7b3461139f895caefd4624e0d5545f5",
"ethereum-test_sign_typed_data.py::test_ethereum_sign_typed_data[struct_list_v4]": "b7e3475d4906942bc0e8d62203ae91a13ea0d702c3a7a53b9777bea670c4a7f7",
"ethereum-test_sign_typed_data.py::test_ethereum_sign_typed_data_cancel": "08712efae2d007610289bbfb3a8fe6800547e884636c83c5bf0e25f33728789e",
"ethereum-test_sign_typed_data.py::test_ethereum_sign_typed_data_show_more_button": "1adbea797586685ce09aae58b0a2b89e1617e4eaad23a8c1ac6fc10b041e57a5",
"ethereum-test_sign_verify_message.py::test_signmessage[parameters0-result0]": "9e2383084cfa23440e7ff9cf95029c73b851f5732de0cb4fb5e89a0ee024fbed",
"ethereum-test_sign_verify_message.py::test_signmessage[parameters1-result1]": "bd669f3ddc72582d3af7aa2e9757e68008690b78533f40eafacf0d071734c108",
"ethereum-test_sign_verify_message.py::test_signmessage[parameters2-result2]": "6ccc6102d1289bce13e28ec2207c96abe4aa1ac0810a13181a874070e7bcf8f3",

@ -13,6 +13,7 @@ CORE_PROTOBUF_SOURCES="\
$PROTOB/messages-debug.proto \
$PROTOB/messages-eos.proto \
$PROTOB/messages-ethereum.proto \
$PROTOB/messages-ethereum-eip712.proto \
$PROTOB/messages-management.proto \
$PROTOB/messages-monero.proto \
$PROTOB/messages-nem.proto \

Loading…
Cancel
Save