1
0
mirror of https://github.com/trezor/trezor-firmware.git synced 2024-12-24 23:38:09 +00:00

feat(core,python): support for Ethereum EIP1559 transactions

Initial EIP1559 implementation

Fix a few small issues

Progress on Python lib implementation and firmware

Fix RLP length

Start fixing tests

Fix legacy transactions

Simplify API and logic

Add EIP1559 tests

Fix access list formatting

Fix UI visiblity issue

Fix commented out code

fix: correct linting issues

Fix access_list protobuf formatting

Remove unneeded code

Remove dead code

Check tx_type bounds for EIP 2718

Reduce code duplication

Prefer eip2718_type over re-using tx_type

Add more tests

Simplify format_access_list

Simplify sign_tx slightly

Change Access List format and add logic to encode it

Fix a bunch of small PR comments

Fix a linting issue

Move tests out of class and regenerate

Remove copy-pasted comments

Add access list to CLI

Simplify _parse_access_list_item

Fix small mistakes following rebase

Fix linting

Refactor to use a separate message for EIP 1559 tx

Simplify changed legacy code

Fix a few small PR comments

Fix linting

fix(legacy): recognize SignTxEIP1559 on legacy build

Fix PR comments
This commit is contained in:
Frederik Bolding 2021-06-05 02:15:25 +02:00 committed by matejcik
parent 69564a9a79
commit 38fa9197ca
18 changed files with 824 additions and 45 deletions

View File

@ -66,7 +66,33 @@ message EthereumSignTx {
optional bytes data_initial_chunk = 7; // The initial data chunk (<= 1024 bytes)
optional uint32 data_length = 8; // Length of transaction payload
optional uint32 chain_id = 9; // Chain Id for EIP 155
optional uint32 tx_type = 10; // (only for Wanchain)
optional uint32 tx_type = 10; // Used for Wanchain
}
/**
* Request: Ask device to sign EIP1559 transaction
* Note: the first at most 1024 bytes of data MUST be transmitted as part of this message.
* @start
* @next EthereumTxRequest
* @next Failure
*/
message EthereumSignTxEIP1559 {
repeated uint32 address_n = 1; // BIP-32 path to derive the key from master node
required bytes nonce = 2; // <=256 bit unsigned big endian
required bytes max_gas_fee = 3; // <=256 bit unsigned big endian (in wei)
required bytes max_priority_fee = 4; // <=256 bit unsigned big endian (in wei)
required bytes gas_limit = 5; // <=256 bit unsigned big endian
optional string to = 6 [default='']; // recipient address
required bytes value = 7; // <=256 bit unsigned big endian (in wei)
optional bytes data_initial_chunk = 8 [default='']; // The initial data chunk (<= 1024 bytes)
required uint32 data_length = 9; // Length of transaction payload
required uint32 chain_id = 10; // Chain Id for EIP 155
repeated EthereumAccessList access_list = 11; // Access List
message EthereumAccessList {
required string address = 1;
repeated bytes storage_keys = 2;
}
}
/**

View File

@ -174,6 +174,7 @@ enum MessageType {
MessageType_EthereumGetAddress = 56 [(wire_in) = true];
MessageType_EthereumAddress = 57 [(wire_out) = true];
MessageType_EthereumSignTx = 58 [(wire_in) = true];
MessageType_EthereumSignTxEIP1559 = 452 [(wire_in) = true];
MessageType_EthereumTxRequest = 59 [(wire_out) = true];
MessageType_EthereumTxAck = 60 [(wire_in) = true];
MessageType_EthereumSignMessage = 64 [(wire_in) = true];

View File

@ -0,0 +1 @@
Support for Ethereum EIP1559 transactions

View File

@ -496,6 +496,8 @@ if utils.BITCOIN_ONLY:
import apps.ethereum.sign_message
apps.ethereum.sign_tx
import apps.ethereum.sign_tx
apps.ethereum.sign_tx_eip1559
import apps.ethereum.sign_tx_eip1559
apps.ethereum.tokens
import apps.ethereum.tokens
apps.ethereum.verify_message

View File

@ -1,4 +1,5 @@
from trezor import wire
from trezor.messages import EthereumSignTxEIP1559
from apps.common import paths
from apps.common.keychain import get_keychain
@ -72,7 +73,9 @@ def _schemas_from_chain_id(msg: EthereumSignTx) -> Iterable[paths.PathSchema]:
if info is None:
# allow Ethereum or testnet paths for unknown networks
slip44_id = (60, 1)
elif networks.is_wanchain(msg.chain_id, msg.tx_type):
elif not EthereumSignTxEIP1559.is_type_of(msg) and networks.is_wanchain(
msg.chain_id, msg.tx_type
):
slip44_id = (networks.SLIP44_WANCHAIN,)
elif info.slip44 != 60 and info.slip44 != 1:
# allow cross-signing with Ethereum unless it's testnet

View File

@ -3,7 +3,12 @@ from ubinascii import hexlify
from trezor import ui
from trezor.enums import ButtonRequestType
from trezor.strings import format_amount
from trezor.ui.layouts import confirm_address, confirm_blob, confirm_output
from trezor.ui.layouts import (
confirm_address,
confirm_amount,
confirm_blob,
confirm_output,
)
from trezor.ui.layouts.tt.altcoin import confirm_total_ethereum
from . import networks, tokens
@ -36,6 +41,29 @@ async def require_confirm_fee(
)
async def require_confirm_eip1559_fee(
ctx, max_priority_fee, max_gas_fee, gas_limit, chain_id
):
await confirm_amount(
ctx,
title="Confirm fee",
description="Maximum fee per gas",
amount=format_ethereum_amount(max_gas_fee, None, chain_id),
)
await confirm_amount(
ctx,
title="Confirm fee",
description="Priority fee per gas",
amount=format_ethereum_amount(max_priority_fee, None, chain_id),
)
await confirm_amount(
ctx,
title="Confirm fee",
description="Maximum fee",
amount=format_ethereum_amount(max_gas_fee * gas_limit, None, chain_id),
)
async def require_confirm_unknown_token(ctx, address_bytes):
contract_address_hex = "0x" + hexlify(address_bytes).decode()
await confirm_address(

View File

@ -23,30 +23,15 @@ MAX_CHAIN_ID = 2147483629
@with_keychain_from_chain_id
async def sign_tx(ctx, msg, keychain):
msg = sanitize(msg)
check(msg)
await paths.validate_path(ctx, keychain, msg.address_n)
# Handle ERC20s
token, address_bytes, recipient, value = await handle_erc20(ctx, msg)
data_total = msg.data_length
# detect ERC - 20 token
token = None
address_bytes = recipient = address.bytes_from_address(msg.to)
value = int.from_bytes(msg.value, "big")
if (
len(msg.to) in (40, 42)
and len(msg.value) == 0
and data_total == 68
and len(msg.data_initial_chunk) == 68
and msg.data_initial_chunk[:16]
== b"\xa9\x05\x9c\xbb\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
):
token = tokens.token_by_chain_address(msg.chain_id, address_bytes)
recipient = msg.data_initial_chunk[16:36]
value = int.from_bytes(msg.data_initial_chunk[36:68], "big")
if token is tokens.UNKNOWN_TOKEN:
await require_confirm_unknown_token(ctx, address_bytes)
await require_confirm_tx(ctx, recipient, value, msg.chain_id, token, msg.tx_type)
if token is None and msg.data_length > 0:
await require_confirm_data(ctx, msg.data_initial_chunk, data_total)
@ -99,6 +84,28 @@ async def sign_tx(ctx, msg, keychain):
return result
async def handle_erc20(ctx, msg):
token = None
address_bytes = recipient = address.bytes_from_address(msg.to)
value = int.from_bytes(msg.value, "big")
if (
len(msg.to) in (40, 42)
and len(msg.value) == 0
and msg.data_length == 68
and len(msg.data_initial_chunk) == 68
and msg.data_initial_chunk[:16]
== b"\xa9\x05\x9c\xbb\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
):
token = tokens.token_by_chain_address(msg.chain_id, address_bytes)
recipient = msg.data_initial_chunk[16:36]
value = int.from_bytes(msg.data_initial_chunk[36:68], "big")
if token is tokens.UNKNOWN_TOKEN:
await require_confirm_unknown_token(ctx, address_bytes)
return token, address_bytes, recipient, value
def get_total_length(msg: EthereumSignTx, data_total: int) -> int:
length = 0
if msg.tx_type is not None:
@ -120,6 +127,7 @@ def get_total_length(msg: EthereumSignTx, data_total: int) -> int:
length += rlp.header_length(data_total, msg.data_initial_chunk)
length += data_total
return length
@ -157,9 +165,14 @@ def check(msg: EthereumSignTx):
if msg.tx_type not in [1, 6, None]:
raise wire.DataError("tx_type out of bounds")
if msg.chain_id < 0:
raise wire.DataError("chain_id out of bounds")
check_data(msg)
# safety checks
if not check_gas(msg) or not check_to(msg):
raise wire.DataError("Safety check failed")
def check_data(msg: EthereumSignTx):
if msg.data_length > 0:
if not msg.data_initial_chunk:
raise wire.DataError("Data length provided, but no initial chunk")
@ -170,10 +183,6 @@ def check(msg: EthereumSignTx):
if len(msg.data_initial_chunk) > msg.data_length:
raise wire.DataError("Invalid size of initial chunk")
# safety checks
if not check_gas(msg) or not check_to(msg):
raise wire.DataError("Safety check failed")
def check_gas(msg: EthereumSignTx) -> bool:
if msg.gas_price is None or msg.gas_limit is None:

View File

@ -0,0 +1,154 @@
from trezor import wire
from trezor.crypto import rlp
from trezor.crypto.curve import secp256k1
from trezor.crypto.hashlib import sha3_256
from trezor.messages import EthereumAccessList, EthereumSignTxEIP1559, EthereumTxRequest
from trezor.utils import HashWriter
from apps.common import paths
from . import address
from .keychain import with_keychain_from_chain_id
from .layout import (
require_confirm_data,
require_confirm_eip1559_fee,
require_confirm_tx,
)
from .sign_tx import check_data, check_to, handle_erc20, sanitize, send_request_chunk
TX_TYPE = 2
def access_list_item_length(item: EthereumAccessList) -> int:
address_length = rlp.length(address.bytes_from_address(item.address))
keys_length = rlp.length(item.storage_keys)
return (
rlp.header_length(address_length + keys_length) + address_length + keys_length
)
def access_list_length(access_list: list[EthereumAccessList]) -> int:
payload_length = sum(access_list_item_length(i) for i in access_list)
return rlp.header_length(payload_length) + payload_length
def write_access_list(w: HashWriter, access_list: list[EthereumAccessList]) -> None:
payload_length = sum(access_list_item_length(i) for i in access_list)
rlp.write_header(w, payload_length, rlp.LIST_HEADER_BYTE)
for item in access_list:
address_bytes = address.bytes_from_address(item.address)
address_length = rlp.length(address_bytes)
keys_length = rlp.length(item.storage_keys)
rlp.write_header(w, address_length + keys_length, rlp.LIST_HEADER_BYTE)
rlp.write(w, address_bytes)
rlp.write(w, item.storage_keys)
@with_keychain_from_chain_id
async def sign_tx_eip1559(ctx, msg, keychain):
msg = sanitize(msg)
check(msg)
await paths.validate_path(ctx, keychain, msg.address_n)
# Handle ERC20s
token, address_bytes, recipient, value = await handle_erc20(ctx, msg)
data_total = msg.data_length
await require_confirm_tx(ctx, recipient, value, msg.chain_id, token)
if token is None and msg.data_length > 0:
await require_confirm_data(ctx, msg.data_initial_chunk, data_total)
await require_confirm_eip1559_fee(
ctx,
int.from_bytes(msg.max_priority_fee, "big"),
int.from_bytes(msg.max_gas_fee, "big"),
int.from_bytes(msg.gas_limit, "big"),
msg.chain_id,
)
data = bytearray()
data += msg.data_initial_chunk
data_left = data_total - len(msg.data_initial_chunk)
total_length = get_total_length(msg, data_total)
sha = HashWriter(sha3_256(keccak=True))
rlp.write(sha, TX_TYPE)
rlp.write_header(sha, total_length, rlp.LIST_HEADER_BYTE)
for field in (
msg.chain_id,
msg.nonce,
msg.max_priority_fee,
msg.max_gas_fee,
msg.gas_limit,
address_bytes,
msg.value,
):
rlp.write(sha, field)
if data_left == 0:
rlp.write(sha, data)
else:
rlp.write_header(sha, data_total, rlp.STRING_HEADER_BYTE, data)
sha.extend(data)
while data_left > 0:
resp = await send_request_chunk(ctx, data_left)
data_left -= len(resp.data_chunk)
sha.extend(resp.data_chunk)
write_access_list(sha, msg.access_list)
digest = sha.get_digest()
result = sign_digest(msg, keychain, digest)
return result
def get_total_length(msg: EthereumSignTxEIP1559, data_total: int) -> int:
length = 0
for item in (
msg.nonce,
msg.gas_limit,
address.bytes_from_address(msg.to),
msg.value,
msg.chain_id,
msg.max_gas_fee,
msg.max_priority_fee,
):
length += rlp.length(item)
length += rlp.header_length(data_total, msg.data_initial_chunk)
length += data_total
length += access_list_length(msg.access_list)
return length
def sign_digest(msg: EthereumSignTxEIP1559, keychain, digest):
node = keychain.derive(msg.address_n)
signature = secp256k1.sign(
node.private_key(), digest, False, secp256k1.CANONICAL_SIG_ETHEREUM
)
req = EthereumTxRequest()
req.signature_v = signature[0] - 27
req.signature_r = signature[1:33]
req.signature_s = signature[33:]
return req
def check(msg: EthereumSignTxEIP1559):
check_data(msg)
if not check_to(msg):
raise wire.DataError("Safety check failed")

View File

@ -103,6 +103,8 @@ def find_message_handler_module(msg_type: int) -> str:
return "apps.ethereum.get_public_key"
elif msg_type == MessageType.EthereumSignTx:
return "apps.ethereum.sign_tx"
elif msg_type == MessageType.EthereumSignTxEIP1559:
return "apps.ethereum.sign_tx_eip1559"
elif msg_type == MessageType.EthereumSignMessage:
return "apps.ethereum.sign_message"
elif msg_type == MessageType.EthereumVerifyMessage:

View File

@ -91,6 +91,7 @@ if not utils.BITCOIN_ONLY:
EthereumGetAddress = 56
EthereumAddress = 57
EthereumSignTx = 58
EthereumSignTxEIP1559 = 452
EthereumTxRequest = 59
EthereumTxAck = 60
EthereumSignMessage = 64

View File

@ -96,6 +96,7 @@ if TYPE_CHECKING:
EthereumGetAddress = 56
EthereumAddress = 57
EthereumSignTx = 58
EthereumSignTxEIP1559 = 452
EthereumTxRequest = 59
EthereumTxAck = 60
EthereumSignMessage = 64

View File

@ -2962,6 +2962,40 @@ if TYPE_CHECKING:
def is_type_of(cls, msg: protobuf.MessageType) -> TypeGuard["EthereumSignTx"]:
return isinstance(msg, cls)
class EthereumSignTxEIP1559(protobuf.MessageType):
address_n: "list[int]"
nonce: "bytes"
max_gas_fee: "bytes"
max_priority_fee: "bytes"
gas_limit: "bytes"
to: "str"
value: "bytes"
data_initial_chunk: "bytes"
data_length: "int"
chain_id: "int"
access_list: "list[EthereumAccessList]"
def __init__(
self,
*,
nonce: "bytes",
max_gas_fee: "bytes",
max_priority_fee: "bytes",
gas_limit: "bytes",
value: "bytes",
data_length: "int",
chain_id: "int",
address_n: "list[int] | None" = None,
access_list: "list[EthereumAccessList] | None" = None,
to: "str | None" = None,
data_initial_chunk: "bytes | None" = None,
) -> None:
pass
@classmethod
def is_type_of(cls, msg: protobuf.MessageType) -> TypeGuard["EthereumSignTxEIP1559"]:
return isinstance(msg, cls)
class EthereumTxRequest(protobuf.MessageType):
data_length: "int | None"
signature_v: "int | None"
@ -3046,6 +3080,22 @@ if TYPE_CHECKING:
def is_type_of(cls, msg: protobuf.MessageType) -> TypeGuard["EthereumVerifyMessage"]:
return isinstance(msg, cls)
class EthereumAccessList(protobuf.MessageType):
address: "str"
storage_keys: "list[bytes]"
def __init__(
self,
*,
address: "str",
storage_keys: "list[bytes] | None" = None,
) -> None:
pass
@classmethod
def is_type_of(cls, msg: protobuf.MessageType) -> TypeGuard["EthereumAccessList"]:
return isinstance(msg, cls)
class LiskGetAddress(protobuf.MessageType):
address_n: "list[int]"
show_display: "bool | None"

View File

@ -0,0 +1 @@
Support for Ethereum EIP1559 transactions

View File

@ -17,6 +17,7 @@
import re
import sys
from decimal import Decimal
from typing import List
import click
@ -74,6 +75,25 @@ def _amount_to_int(ctx, param, value):
raise click.BadParameter("Amount not understood")
def _parse_access_list(ctx, param, value):
try:
return [_parse_access_list_item(val) for val in value]
except Exception:
raise click.BadParameter("Access List format invalid")
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]
return ethereum.messages.EthereumAccessList(address, storage_keys_bytes)
except Exception:
raise click.BadParameter("Access List format invalid")
def _list_units(ctx, param, value):
if not value or ctx.resilient_parsing:
return
@ -107,6 +127,14 @@ def _erc20_contract(w3, token_address, to_address, amount):
return contract.encodeABI("transfer", [to_address, amount])
def _format_access_list(access_list: List[ethereum.messages.EthereumAccessList]):
mapped = map(
lambda item: [_decode_hex(item.address), item.storage_keys],
access_list,
)
return list(mapped)
#####################
#
# commands start here
@ -166,8 +194,22 @@ def get_public_node(client, address, show_display):
)
@click.option("-d", "--data", help="Data as hex string, e.g. 0x12345678")
@click.option("-p", "--publish", is_flag=True, help="Publish transaction via RPC")
@click.option("-x", "--tx-type", type=int, help="TX type (used only for Wanchain)")
@click.option("-x", "--tx-type", type=int, help="TX type")
@click.option("-t", "--token", help="ERC20 token address")
@click.option(
"-a",
"--access-list",
help="Access List",
callback=_parse_access_list,
multiple=True,
)
@click.option("--max-gas-fee", help="Max Gas Fee (EIP1559)", callback=_amount_to_int)
@click.option(
"--max-priority-fee",
help="Max Priority Fee (EIP1559)",
callback=_amount_to_int,
)
@click.option("-e", "--eip2718-type", type=int, help="EIP2718 tx type")
@click.option(
"--list-units",
is_flag=True,
@ -192,6 +234,10 @@ def sign_tx(
to_address,
tx_type,
token,
max_gas_fee,
max_priority_fee,
access_list,
eip2718_type,
):
"""Sign (and optionally publish) Ethereum transaction.
@ -216,9 +262,11 @@ def sign_tx(
click.echo(" pip install web3 rlp")
sys.exit(1)
is_eip1559 = eip2718_type == 2
w3 = web3.Web3()
if (
any(x is None for x in (gas_price, gas_limit, nonce))
(not is_eip1559 and gas_price is None)
or any(x is None for x in (gas_limit, nonce))
or publish
and not w3.isConnected()
):
@ -246,7 +294,7 @@ def sign_tx(
else:
data = b""
if gas_price is None:
if gas_price is None and not is_eip1559:
gas_price = w3.eth.gasPrice
if gas_limit is None:
@ -262,27 +310,61 @@ def sign_tx(
if nonce is None:
nonce = w3.eth.getTransactionCount(from_address)
sig = ethereum.sign_tx(
client,
n=address_n,
tx_type=tx_type,
nonce=nonce,
gas_price=gas_price,
gas_limit=gas_limit,
to=to_address,
value=amount,
data=data,
chain_id=chain_id,
sig = (
ethereum.sign_tx_eip1559(
client,
n=address_n,
nonce=nonce,
gas_limit=gas_limit,
to=to_address,
value=amount,
data=data,
chain_id=chain_id,
max_gas_fee=max_gas_fee,
max_priority_fee=max_priority_fee,
access_list=access_list,
)
if is_eip1559
else ethereum.sign_tx(
client,
n=address_n,
tx_type=tx_type,
nonce=nonce,
gas_price=gas_price,
gas_limit=gas_limit,
to=to_address,
value=amount,
data=data,
chain_id=chain_id,
)
)
to = _decode_hex(to_address)
if tx_type is None:
if is_eip1559:
transaction = rlp.encode(
(
chain_id,
nonce,
max_priority_fee,
max_gas_fee,
gas_limit,
to,
amount,
data,
_format_access_list(access_list) if access_list is not None else [],
)
+ sig
)
elif tx_type is None:
transaction = rlp.encode((nonce, gas_price, gas_limit, to, amount, data) + sig)
else:
transaction = rlp.encode(
(tx_type, nonce, gas_price, gas_limit, to, amount, data) + sig
)
tx_hex = "0x%s" % transaction.hex()
tx_hex = "0x%s%s" % (
str(eip2718_type).zfill(2) if eip2718_type is not None else "",
transaction.hex(),
)
if publish:
tx_hash = w3.eth.sendRawTransaction(tx_hex).hex()

View File

@ -55,7 +55,7 @@ def sign_tx(
msg = messages.EthereumSignTx(
address_n=n,
nonce=int_to_big_endian(nonce),
gas_price=int_to_big_endian(gas_price),
gas_price=int_to_big_endian(gas_price) if gas_price is not None else None,
gas_limit=int_to_big_endian(gas_limit),
value=int_to_big_endian(value),
to=to,
@ -83,6 +83,47 @@ def sign_tx(
return response.signature_v, response.signature_r, response.signature_s
@session
def sign_tx_eip1559(
client,
n,
*,
nonce,
gas_limit,
to,
value,
data=b"",
chain_id,
max_gas_fee,
max_priority_fee,
access_list=()
):
length = len(data)
data, chunk = data[1024:], data[:1024]
msg = messages.EthereumSignTxEIP1559(
address_n=n,
nonce=int_to_big_endian(nonce),
gas_limit=int_to_big_endian(gas_limit),
value=int_to_big_endian(value),
to=to,
chain_id=chain_id,
max_gas_fee=int_to_big_endian(max_gas_fee),
max_priority_fee=int_to_big_endian(max_priority_fee),
access_list=access_list,
data_length=length,
data_initial_chunk=chunk,
)
response = client.call(msg)
while response.data_length is not None:
data_length = response.data_length
data, chunk = data[data_length:], data[:data_length]
response = client.call(messages.EthereumTxAck(data_chunk=chunk))
return response.signature_v, response.signature_r, response.signature_s
@expect(messages.EthereumMessageSignature)
def sign_message(client, n, message):
message = normalize_nfc(message)

View File

@ -117,6 +117,7 @@ class MessageType(IntEnum):
EthereumGetAddress = 56
EthereumAddress = 57
EthereumSignTx = 58
EthereumSignTxEIP1559 = 452
EthereumTxRequest = 59
EthereumTxAck = 60
EthereumSignMessage = 64
@ -3971,6 +3972,50 @@ class EthereumSignTx(protobuf.MessageType):
self.tx_type = tx_type
class EthereumSignTxEIP1559(protobuf.MessageType):
MESSAGE_WIRE_TYPE = 452
FIELDS = {
1: protobuf.Field("address_n", "uint32", repeated=True, required=False),
2: protobuf.Field("nonce", "bytes", repeated=False, required=True),
3: protobuf.Field("max_gas_fee", "bytes", repeated=False, required=True),
4: protobuf.Field("max_priority_fee", "bytes", repeated=False, required=True),
5: protobuf.Field("gas_limit", "bytes", repeated=False, required=True),
6: protobuf.Field("to", "string", repeated=False, required=False),
7: protobuf.Field("value", "bytes", repeated=False, required=True),
8: protobuf.Field("data_initial_chunk", "bytes", repeated=False, required=False),
9: protobuf.Field("data_length", "uint32", repeated=False, required=True),
10: protobuf.Field("chain_id", "uint32", repeated=False, required=True),
11: protobuf.Field("access_list", "EthereumAccessList", repeated=True, required=False),
}
def __init__(
self,
*,
nonce: "bytes",
max_gas_fee: "bytes",
max_priority_fee: "bytes",
gas_limit: "bytes",
value: "bytes",
data_length: "int",
chain_id: "int",
address_n: Optional[List["int"]] = None,
access_list: Optional[List["EthereumAccessList"]] = None,
to: Optional["str"] = '',
data_initial_chunk: Optional["bytes"] = b'',
) -> None:
self.address_n = address_n if address_n is not None else []
self.access_list = access_list if access_list is not None else []
self.nonce = nonce
self.max_gas_fee = max_gas_fee
self.max_priority_fee = max_priority_fee
self.gas_limit = gas_limit
self.value = value
self.data_length = data_length
self.chain_id = chain_id
self.to = to
self.data_initial_chunk = data_initial_chunk
class EthereumTxRequest(protobuf.MessageType):
MESSAGE_WIRE_TYPE = 59
FIELDS = {
@ -4062,6 +4107,23 @@ class EthereumVerifyMessage(protobuf.MessageType):
self.address = address
class EthereumAccessList(protobuf.MessageType):
MESSAGE_WIRE_TYPE = None
FIELDS = {
1: protobuf.Field("address", "string", repeated=False, required=True),
2: protobuf.Field("storage_keys", "bytes", repeated=True, required=False),
}
def __init__(
self,
*,
address: "str",
storage_keys: Optional[List["bytes"]] = None,
) -> None:
self.storage_keys = storage_keys if storage_keys is not None else []
self.address = address
class LiskGetAddress(protobuf.MessageType):
MESSAGE_WIRE_TYPE = 114
FIELDS = {

View File

@ -0,0 +1,308 @@
# 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, messages
from trezorlib.tools import parse_path
TO_ADDR = "0x1d1c328764a41bda0492b66baa30c4a339ff85ef"
pytestmark = [pytest.mark.altcoin, pytest.mark.ethereum, pytest.mark.skip_t1]
def test_ethereum_signtx_nodata(client):
with client:
sig_v, sig_r, sig_s = ethereum.sign_tx_eip1559(
client,
n=parse_path("44'/60'/0'/0/100"),
nonce=0,
gas_limit=20,
to=TO_ADDR,
chain_id=1,
value=10,
max_gas_fee=20,
max_priority_fee=1,
)
assert sig_v == 1
assert (
sig_r.hex()
== "2ceeaabc994fbce2fbd66551f9d48fc711c8db2a12e93779eeddede11e41f636"
)
assert (
sig_s.hex()
== "2db4a9ecc73da91206f84397ae9287a399076fdc01ed7f3c6554b1c57c39bf8c"
)
def test_ethereum_signtx_data(client):
with client:
sig_v, sig_r, sig_s = ethereum.sign_tx_eip1559(
client,
n=parse_path("44'/60'/0'/0/0"),
nonce=0,
gas_limit=20,
chain_id=1,
to=TO_ADDR,
value=10,
data=b"abcdefghijklmnop" * 16,
max_gas_fee=20,
max_priority_fee=1,
)
assert sig_v == 0
assert (
sig_r.hex()
== "8e4361e40e76a7cab17e0a982724bbeaf5079cd02d50c20d431ba7dde2404ea4"
)
assert (
sig_s.hex()
== "411930f091bb508e593e22a9ee45bd4d9eeb504ac398123aec889d5951bdebc3"
)
with client:
sig_v, sig_r, sig_s = ethereum.sign_tx_eip1559(
client,
n=parse_path("44'/60'/0'/0/0"),
nonce=123456,
gas_limit=20000,
to=TO_ADDR,
chain_id=1,
value=12345678901234567890,
data=b"ABCDEFGHIJKLMNOP" * 256 + b"!!!",
max_gas_fee=20,
max_priority_fee=1,
)
assert sig_v == 0
assert (
sig_r.hex()
== "2e4f4c0e7c4e51270b891480060712e9d3bcab01e8ad0fadf2dfddd71504ca94"
)
assert (
sig_s.hex()
== "2599beb32757a144dedc82b79153c21269c9939a9245342bcf35764115b62bc1"
)
def test_ethereum_signtx_access_list(client):
with client:
sig_v, sig_r, sig_s = ethereum.sign_tx_eip1559(
client,
n=parse_path("44'/60'/0'/0/100"),
nonce=0,
gas_limit=20,
to=TO_ADDR,
chain_id=1,
value=10,
max_gas_fee=20,
max_priority_fee=1,
access_list=[
messages.EthereumAccessList(
address="0xde0b295669a9fd93d5f28d9ec85e40f4cb697bae",
storage_keys=[
bytes.fromhex(
"0000000000000000000000000000000000000000000000000000000000000003"
),
bytes.fromhex(
"0000000000000000000000000000000000000000000000000000000000000007"
),
],
)
],
)
assert sig_v == 1
assert (
sig_r.hex()
== "9f8763f3ff8d4d409f6b96bc3f1d84dd504e2c667b162778508478645401f121"
)
assert (
sig_s.hex()
== "51e30b68b9091cf8138c07380c4378c2711779b68b2e5264d141479f13a12f57"
)
def test_ethereum_signtx_access_list_larger(client):
with client:
sig_v, sig_r, sig_s = ethereum.sign_tx_eip1559(
client,
n=parse_path("44'/60'/0'/0/100"),
nonce=0,
gas_limit=20,
to=TO_ADDR,
chain_id=1,
value=10,
max_gas_fee=20,
max_priority_fee=1,
access_list=[
messages.EthereumAccessList(
address="0xde0b295669a9fd93d5f28d9ec85e40f4cb697bae",
storage_keys=[
bytes.fromhex(
"0000000000000000000000000000000000000000000000000000000000000003"
),
bytes.fromhex(
"0000000000000000000000000000000000000000000000000000000000000007"
),
],
),
messages.EthereumAccessList(
address="0xbb9bc244d798123fde783fcc1c72d3bb8c189413",
storage_keys=[
bytes.fromhex(
"0000000000000000000000000000000000000000000000000000000000000006"
),
bytes.fromhex(
"0000000000000000000000000000000000000000000000000000000000000007"
),
bytes.fromhex(
"0000000000000000000000000000000000000000000000000000000000000009"
),
],
),
],
)
assert sig_v == 1
assert (
sig_r.hex()
== "718a3a30827c979975c846d2f60495310c4959ee3adce2d89e0211785725465c"
)
assert (
sig_s.hex()
== "7d0ea2a28ef5702ca763c1f340427c0020292ffcbb4553dd1c8ea8e2b9126dbc"
)
def test_ethereum_signtx_known_erc20_token(client):
with client:
data = bytearray()
# method id signalizing `transfer(address _to, uint256 _value)` function
data.extend(bytes.fromhex("a9059cbb"))
# 1st function argument (to - the receiver)
data.extend(
bytes.fromhex(
"000000000000000000000000574bbb36871ba6b78e27f4b4dcfb76ea0091880b"
)
)
# 2nd function argument (value - amount to be transferred)
data.extend(
bytes.fromhex(
"000000000000000000000000000000000000000000000000000000000bebc200"
)
)
sig_v, sig_r, sig_s = ethereum.sign_tx_eip1559(
client,
n=parse_path("44'/60'/0'/0/0"),
nonce=0,
max_gas_fee=20,
max_priority_fee=1,
gas_limit=20,
# ADT token address
to="0xd0d6d6c5fe4a677d343cc433536bb717bae167dd",
chain_id=1,
# value needs to be 0, token value is set in the contract (data)
value=0,
data=data,
)
assert sig_v == 1
assert (
sig_r.hex()
== "94d67bacb7966f881339d91103f5d738d9c491fff4c01a6513c554ab15e86cc0"
)
assert (
sig_s.hex()
== "405bd19a7bf4ae62d41fcb7844e36c786b106b456185c3d0877a7ce7eab6c751"
)
def test_ethereum_signtx_unknown_erc20_token(client):
with client:
data = bytearray()
# method id signalizing `transfer(address _to, uint256 _value)` function
data.extend(bytes.fromhex("a9059cbb"))
# 1st function argument (to - the receiver)
data.extend(
bytes.fromhex(
"000000000000000000000000574bbb36871ba6b78e27f4b4dcfb76ea0091880b"
)
)
# 2nd function argument (value - amount to be transferred)
data.extend(
bytes.fromhex(
"0000000000000000000000000000000000000000000000000000000000000123"
)
)
# since this token is unknown trezor should display "unknown token value"
sig_v, sig_r, sig_s = ethereum.sign_tx_eip1559(
client,
n=parse_path("44'/60'/0'/0/1"),
nonce=0,
max_gas_fee=20,
max_priority_fee=1,
gas_limit=20,
# unknown token address (Grzegorz Brzęczyszczykiewicz Token)
to="0xfc6b5d6af8a13258f7cbd0d39e11b35e01a32f93",
chain_id=1,
# value needs to be 0, token value is set in the contract (data)
value=0,
data=data,
)
assert sig_v == 1
assert (
sig_r.hex()
== "e631b56bcc596844cb8686b2046e36cf33634aa396e7e1ea94a97aac02c18bda"
)
assert (
sig_s.hex()
== "399bff8752539176c4b2f1d5d2a8f6029f79841d28802149ab339a033ffe4c1f"
)
def test_ethereum_signtx_large_chainid(client):
with client:
sig_v, sig_r, sig_s = ethereum.sign_tx_eip1559(
client,
n=parse_path("44'/60'/0'/0/100"),
nonce=0,
gas_limit=20,
to=TO_ADDR,
chain_id=3125659152, # Pirl chain id, doesn't support EIP1559 at this time, but chosen for large chain id
value=10,
max_gas_fee=20,
max_priority_fee=1,
)
assert sig_v == 0
assert (
sig_r.hex()
== "07f8c967227c5a190cb90525c3387691a426fe61f8e0503274280724060ea95c"
)
assert (
sig_s.hex()
== "0bf83eaf74e24aa9146b23e06f9edec6e25acb81d3830e8d146b9e7b6923ad1e"
)

View File

@ -235,6 +235,13 @@
"test_msg_ethereum_signtx_eip155.py::test_chain_ids[609112567-60-sig6]": "c8e01d20eccadcca4f05e4e8351c3bfc38d0fdbe4a61f63dfd74e065faea86e7",
"test_msg_ethereum_signtx_eip155.py::test_chain_ids[61-61-sig3]": "cd5f04cc7b055503e83f0538709a7ac577445c6089ead12f1fc3a3c45ad96419",
"test_msg_ethereum_signtx_eip155.py::test_with_data": "670913def0b7268671258f70dfbdc794a8405e1e432e423aae10b616028f3db9",
"test_msg_ethereum_signtx_eip1559.py::test_ethereum_signtx_access_list": "0d73046641e1d148ed01a36314509c7d38284c37d98b8a16b92f59cea543055c",
"test_msg_ethereum_signtx_eip1559.py::test_ethereum_signtx_access_list_larger": "0d73046641e1d148ed01a36314509c7d38284c37d98b8a16b92f59cea543055c",
"test_msg_ethereum_signtx_eip1559.py::test_ethereum_signtx_data": "17f52fa7d2edc36201050df432398839b2c27e82d6e01a2feafd58bbf5de19aa",
"test_msg_ethereum_signtx_eip1559.py::test_ethereum_signtx_known_erc20_token": "ca5c69923a79c1112bbdd81159adfbc389e1538c1776c141ce28fc72d0feeb81",
"test_msg_ethereum_signtx_eip1559.py::test_ethereum_signtx_large_chainid": "95ea813ceeb0782ff3554eafd463831f60490aa418385fc9cf113140dfa8c75c",
"test_msg_ethereum_signtx_eip1559.py::test_ethereum_signtx_nodata": "0d73046641e1d148ed01a36314509c7d38284c37d98b8a16b92f59cea543055c",
"test_msg_ethereum_signtx_eip1559.py::test_ethereum_signtx_unknown_erc20_token": "a25e1171b7dc9bdcfcfe02bf10e7ce8bfa02a0ec9b57fc3dce2b591cdec798e5",
"test_msg_ethereum_verifymessage.py-test_verify": "f97754c9168c436209997dc17b4cf4fa0b55171fb2dca4b06256ed87fd910fdc",
"test_msg_ethereum_verifymessage.py-test_verify_invalid": "5a80508a71a9ef64f94762b07636f90e464832f0f4a3102af8fa1a8c69e94586",
"test_msg_getaddress.py-test_bch": "5a80508a71a9ef64f94762b07636f90e464832f0f4a3102af8fa1a8c69e94586",