[add] Python client packaging first draft
This commit is contained in:
Lucas PASCAL
@@ -1,143 +0,0 @@
|
||||
from enum import IntEnum, auto
|
||||
from typing import Optional
|
||||
from ragger.backend import BackendInterface
|
||||
from ragger.utils import RAPDU
|
||||
from .command_builder import CommandBuilder
|
||||
from .eip712 import EIP712FieldType
|
||||
from .tlv import format_tlv
|
||||
from pathlib import Path
|
||||
import keychain
|
||||
import rlp
|
||||
|
||||
|
||||
ROOT_SCREENSHOT_PATH = Path(__file__).parent.parent
|
||||
WEI_IN_ETH = 1e+18
|
||||
|
||||
|
||||
class StatusWord(IntEnum):
|
||||
OK = 0x9000
|
||||
ERROR_NO_INFO = 0x6a00
|
||||
INVALID_DATA = 0x6a80
|
||||
INSUFFICIENT_MEMORY = 0x6a84
|
||||
INVALID_INS = 0x6d00
|
||||
INVALID_P1_P2 = 0x6b00
|
||||
CONDITION_NOT_SATISFIED = 0x6985
|
||||
REF_DATA_NOT_FOUND = 0x6a88
|
||||
|
||||
class DOMAIN_NAME_TAG(IntEnum):
|
||||
STRUCTURE_TYPE = 0x01
|
||||
STRUCTURE_VERSION = 0x02
|
||||
CHALLENGE = 0x12
|
||||
SIGNER_KEY_ID = 0x13
|
||||
SIGNER_ALGO = 0x14
|
||||
SIGNATURE = 0x15
|
||||
DOMAIN_NAME = 0x20
|
||||
COIN_TYPE = 0x21
|
||||
ADDRESS = 0x22
|
||||
|
||||
|
||||
class EthAppClient:
|
||||
def __init__(self, client: BackendInterface):
|
||||
self._client = client
|
||||
self._cmd_builder = CommandBuilder()
|
||||
|
||||
def _send(self, payload: bytearray):
|
||||
return self._client.exchange_async_raw(payload)
|
||||
|
||||
def response(self) -> RAPDU:
|
||||
return self._client._last_async_response
|
||||
|
||||
def eip712_send_struct_def_struct_name(self, name: str):
|
||||
return self._send(self._cmd_builder.eip712_send_struct_def_struct_name(name))
|
||||
|
||||
def eip712_send_struct_def_struct_field(self,
|
||||
field_type: EIP712FieldType,
|
||||
type_name: str,
|
||||
type_size: int,
|
||||
array_levels: [],
|
||||
key_name: str):
|
||||
return self._send(self._cmd_builder.eip712_send_struct_def_struct_field(
|
||||
field_type,
|
||||
type_name,
|
||||
type_size,
|
||||
array_levels,
|
||||
key_name))
|
||||
|
||||
def eip712_send_struct_impl_root_struct(self, name: str):
|
||||
return self._send(self._cmd_builder.eip712_send_struct_impl_root_struct(name))
|
||||
|
||||
def eip712_send_struct_impl_array(self, size: int):
|
||||
return self._send(self._cmd_builder.eip712_send_struct_impl_array(size))
|
||||
|
||||
def eip712_send_struct_impl_struct_field(self, raw_value: bytes):
|
||||
chunks = self._cmd_builder.eip712_send_struct_impl_struct_field(raw_value)
|
||||
for chunk in chunks[:-1]:
|
||||
with self._send(chunk):
|
||||
pass
|
||||
return self._send(chunks[-1])
|
||||
|
||||
def eip712_sign_new(self, bip32_path: str, verbose: bool):
|
||||
return self._send(self._cmd_builder.eip712_sign_new(bip32_path))
|
||||
|
||||
def eip712_sign_legacy(self,
|
||||
bip32_path: str,
|
||||
domain_hash: bytes,
|
||||
message_hash: bytes):
|
||||
return self._send(self._cmd_builder.eip712_sign_legacy(bip32_path,
|
||||
domain_hash,
|
||||
message_hash))
|
||||
|
||||
def eip712_filtering_activate(self):
|
||||
return self._send(self._cmd_builder.eip712_filtering_activate())
|
||||
|
||||
def eip712_filtering_message_info(self, name: str, filters_count: int, sig: bytes):
|
||||
return self._send(self._cmd_builder.eip712_filtering_message_info(name, filters_count, sig))
|
||||
|
||||
def eip712_filtering_show_field(self, name: str, sig: bytes):
|
||||
return self._send(self._cmd_builder.eip712_filtering_show_field(name, sig))
|
||||
|
||||
def send_fund(self,
|
||||
bip32_path: str,
|
||||
nonce: int,
|
||||
gas_price: int,
|
||||
gas_limit: int,
|
||||
to: bytes,
|
||||
amount: float,
|
||||
chain_id: int):
|
||||
data = list()
|
||||
data.append(nonce)
|
||||
data.append(gas_price)
|
||||
data.append(gas_limit)
|
||||
data.append(to)
|
||||
data.append(int(amount * WEI_IN_ETH))
|
||||
data.append(bytes())
|
||||
data.append(chain_id)
|
||||
data.append(bytes())
|
||||
data.append(bytes())
|
||||
|
||||
chunks = self._cmd_builder.sign(bip32_path, rlp.encode(data))
|
||||
for chunk in chunks[:-1]:
|
||||
with self._send(chunk):
|
||||
pass
|
||||
return self._send(chunks[-1])
|
||||
|
||||
def get_challenge(self):
|
||||
return self._send(self._cmd_builder.get_challenge())
|
||||
|
||||
def provide_domain_name(self, challenge: int, name: str, addr: bytes):
|
||||
payload = format_tlv(DOMAIN_NAME_TAG.STRUCTURE_TYPE, 3) # TrustedDomainName
|
||||
payload += format_tlv(DOMAIN_NAME_TAG.STRUCTURE_VERSION, 1)
|
||||
payload += format_tlv(DOMAIN_NAME_TAG.SIGNER_KEY_ID, 0) # test key
|
||||
payload += format_tlv(DOMAIN_NAME_TAG.SIGNER_ALGO, 1) # secp256k1
|
||||
payload += format_tlv(DOMAIN_NAME_TAG.CHALLENGE, challenge)
|
||||
payload += format_tlv(DOMAIN_NAME_TAG.COIN_TYPE, 0x3c) # ETH in slip-44
|
||||
payload += format_tlv(DOMAIN_NAME_TAG.DOMAIN_NAME, name)
|
||||
payload += format_tlv(DOMAIN_NAME_TAG.ADDRESS, addr)
|
||||
payload += format_tlv(DOMAIN_NAME_TAG.SIGNATURE,
|
||||
keychain.sign_data(keychain.Key.DOMAIN_NAME, payload))
|
||||
|
||||
chunks = self._cmd_builder.provide_domain_name(payload)
|
||||
for chunk in chunks[:-1]:
|
||||
with self._send(chunk):
|
||||
pass
|
||||
return self._send(chunks[-1])
|
||||
@@ -1,201 +0,0 @@
|
||||
from enum import IntEnum, auto
|
||||
from typing import Iterator, Optional
|
||||
from .eip712 import EIP712FieldType
|
||||
from ragger.bip import pack_derivation_path
|
||||
import struct
|
||||
|
||||
class InsType(IntEnum):
|
||||
SIGN = 0x04
|
||||
EIP712_SEND_STRUCT_DEF = 0x1a
|
||||
EIP712_SEND_STRUCT_IMPL = 0x1c
|
||||
EIP712_SEND_FILTERING = 0x1e
|
||||
EIP712_SIGN = 0x0c
|
||||
GET_CHALLENGE = 0x20
|
||||
PROVIDE_DOMAIN_NAME = 0x22
|
||||
|
||||
class P1Type(IntEnum):
|
||||
COMPLETE_SEND = 0x00
|
||||
PARTIAL_SEND = 0x01
|
||||
SIGN_FIRST_CHUNK = 0x00
|
||||
SIGN_SUBSQT_CHUNK = 0x80
|
||||
|
||||
class P2Type(IntEnum):
|
||||
STRUCT_NAME = 0x00
|
||||
STRUCT_FIELD = 0xff
|
||||
ARRAY = 0x0f
|
||||
LEGACY_IMPLEM = 0x00
|
||||
NEW_IMPLEM = 0x01
|
||||
FILTERING_ACTIVATE = 0x00
|
||||
FILTERING_CONTRACT_NAME = 0x0f
|
||||
FILTERING_FIELD_NAME = 0xff
|
||||
|
||||
class CommandBuilder:
|
||||
_CLA: int = 0xE0
|
||||
|
||||
def _serialize(self,
|
||||
ins: InsType,
|
||||
p1: int,
|
||||
p2: int,
|
||||
cdata: bytearray = bytes()) -> bytes:
|
||||
|
||||
header = bytearray()
|
||||
header.append(self._CLA)
|
||||
header.append(ins)
|
||||
header.append(p1)
|
||||
header.append(p2)
|
||||
header.append(len(cdata))
|
||||
return header + cdata
|
||||
|
||||
def _string_to_bytes(self, string: str) -> bytes:
|
||||
data = bytearray()
|
||||
for char in string:
|
||||
data.append(ord(char))
|
||||
return data
|
||||
|
||||
def eip712_send_struct_def_struct_name(self, name: str) -> bytes:
|
||||
return self._serialize(InsType.EIP712_SEND_STRUCT_DEF,
|
||||
P1Type.COMPLETE_SEND,
|
||||
P2Type.STRUCT_NAME,
|
||||
self._string_to_bytes(name))
|
||||
|
||||
def eip712_send_struct_def_struct_field(self,
|
||||
field_type: EIP712FieldType,
|
||||
type_name: str,
|
||||
type_size: int,
|
||||
array_levels: [],
|
||||
key_name: str) -> bytes:
|
||||
data = bytearray()
|
||||
typedesc = 0
|
||||
typedesc |= (len(array_levels) > 0) << 7
|
||||
typedesc |= (type_size != None) << 6
|
||||
typedesc |= field_type
|
||||
data.append(typedesc)
|
||||
if field_type == EIP712FieldType.CUSTOM:
|
||||
data.append(len(type_name))
|
||||
data += self._string_to_bytes(type_name)
|
||||
if type_size != None:
|
||||
data.append(type_size)
|
||||
if len(array_levels) > 0:
|
||||
data.append(len(array_levels))
|
||||
for level in array_levels:
|
||||
data.append(0 if level == None else 1)
|
||||
if level != None:
|
||||
data.append(level)
|
||||
data.append(len(key_name))
|
||||
data += self._string_to_bytes(key_name)
|
||||
return self._serialize(InsType.EIP712_SEND_STRUCT_DEF,
|
||||
P1Type.COMPLETE_SEND,
|
||||
P2Type.STRUCT_FIELD,
|
||||
data)
|
||||
|
||||
def eip712_send_struct_impl_root_struct(self, name: str) -> bytes:
|
||||
return self._serialize(InsType.EIP712_SEND_STRUCT_IMPL,
|
||||
P1Type.COMPLETE_SEND,
|
||||
P2Type.STRUCT_NAME,
|
||||
self._string_to_bytes(name))
|
||||
|
||||
def eip712_send_struct_impl_array(self, size: int) -> bytes:
|
||||
data = bytearray()
|
||||
data.append(size)
|
||||
return self._serialize(InsType.EIP712_SEND_STRUCT_IMPL,
|
||||
P1Type.COMPLETE_SEND,
|
||||
P2Type.ARRAY,
|
||||
data)
|
||||
|
||||
def eip712_send_struct_impl_struct_field(self, data: bytearray) -> Iterator[bytes]:
|
||||
chunks = list()
|
||||
# Add a 16-bit integer with the data's byte length (network byte order)
|
||||
data_w_length = bytearray()
|
||||
data_w_length.append((len(data) & 0xff00) >> 8)
|
||||
data_w_length.append(len(data) & 0x00ff)
|
||||
data_w_length += data
|
||||
while len(data_w_length) > 0:
|
||||
p1 = P1Type.PARTIAL_SEND if len(data_w_length) > 0xff else P1Type.COMPLETE_SEND
|
||||
chunks.append(self._serialize(InsType.EIP712_SEND_STRUCT_IMPL,
|
||||
p1,
|
||||
P2Type.STRUCT_FIELD,
|
||||
data_w_length[:0xff]))
|
||||
data_w_length = data_w_length[0xff:]
|
||||
return chunks
|
||||
|
||||
def eip712_sign_new(self, bip32_path: str) -> bytes:
|
||||
data = pack_derivation_path(bip32_path)
|
||||
return self._serialize(InsType.EIP712_SIGN,
|
||||
P1Type.COMPLETE_SEND,
|
||||
P2Type.NEW_IMPLEM,
|
||||
data)
|
||||
|
||||
def eip712_sign_legacy(self,
|
||||
bip32_path: str,
|
||||
domain_hash: bytes,
|
||||
message_hash: bytes) -> bytes:
|
||||
data = pack_derivation_path(bip32_path)
|
||||
data += domain_hash
|
||||
data += message_hash
|
||||
return self._serialize(InsType.EIP712_SIGN,
|
||||
P1Type.COMPLETE_SEND,
|
||||
P2Type.LEGACY_IMPLEM,
|
||||
data)
|
||||
|
||||
def eip712_filtering_activate(self):
|
||||
return self._serialize(InsType.EIP712_SEND_FILTERING,
|
||||
P1Type.COMPLETE_SEND,
|
||||
P2Type.FILTERING_ACTIVATE,
|
||||
bytearray())
|
||||
|
||||
def _eip712_filtering_send_name(self, name: str, sig: bytes) -> bytes:
|
||||
data = bytearray()
|
||||
data.append(len(name))
|
||||
data += self._string_to_bytes(name)
|
||||
data.append(len(sig))
|
||||
data += sig
|
||||
return data
|
||||
|
||||
def eip712_filtering_message_info(self, name: str, filters_count: int, sig: bytes) -> bytes:
|
||||
data = bytearray()
|
||||
data.append(len(name))
|
||||
data += self._string_to_bytes(name)
|
||||
data.append(filters_count)
|
||||
data.append(len(sig))
|
||||
data += sig
|
||||
return self._serialize(InsType.EIP712_SEND_FILTERING,
|
||||
P1Type.COMPLETE_SEND,
|
||||
P2Type.FILTERING_CONTRACT_NAME,
|
||||
data)
|
||||
|
||||
def eip712_filtering_show_field(self, name: str, sig: bytes) -> bytes:
|
||||
return self._serialize(InsType.EIP712_SEND_FILTERING,
|
||||
P1Type.COMPLETE_SEND,
|
||||
P2Type.FILTERING_FIELD_NAME,
|
||||
self._eip712_filtering_send_name(name, sig))
|
||||
|
||||
def sign(self, bip32_path: str, rlp_data: bytes) -> list[bytes]:
|
||||
apdus = list()
|
||||
payload = pack_derivation_path(bip32_path)
|
||||
payload += rlp_data
|
||||
p1 = P1Type.SIGN_FIRST_CHUNK
|
||||
while len(payload) > 0:
|
||||
apdus.append(self._serialize(InsType.SIGN,
|
||||
p1,
|
||||
0x00,
|
||||
payload[:0xff]))
|
||||
payload = payload[0xff:]
|
||||
p1 = P1Type.SIGN_SUBSQT_CHUNK
|
||||
return apdus
|
||||
|
||||
def get_challenge(self) -> bytes:
|
||||
return self._serialize(InsType.GET_CHALLENGE, 0x00, 0x00)
|
||||
|
||||
def provide_domain_name(self, tlv_payload: bytes) -> list[bytes]:
|
||||
chunks = list()
|
||||
payload = struct.pack(">H", len(tlv_payload))
|
||||
payload += tlv_payload
|
||||
p1 = 1
|
||||
while len(payload) > 0:
|
||||
chunks.append(self._serialize(InsType.PROVIDE_DOMAIN_NAME,
|
||||
p1,
|
||||
0x00,
|
||||
payload[:0xff]))
|
||||
payload = payload[0xff:]
|
||||
p1 = 0
|
||||
return chunks
|
||||
@@ -1,11 +0,0 @@
|
||||
from enum import IntEnum, auto
|
||||
|
||||
class EIP712FieldType(IntEnum):
|
||||
CUSTOM = 0,
|
||||
INT = auto()
|
||||
UINT = auto()
|
||||
ADDRESS = auto()
|
||||
BOOL = auto()
|
||||
STRING = auto()
|
||||
FIX_BYTES = auto()
|
||||
DYN_BYTES = auto()
|
||||
@@ -1,14 +0,0 @@
|
||||
def signature(data: bytes) -> tuple[bytes, bytes, bytes]:
|
||||
assert len(data) == (1 + 32 + 32)
|
||||
|
||||
v = data[0:1]
|
||||
data = data[1:]
|
||||
r = data[0:32]
|
||||
data = data[32:]
|
||||
s = data[0:32]
|
||||
|
||||
return v, r, s
|
||||
|
||||
def challenge(data: bytes) -> int:
|
||||
assert len(data) == 4
|
||||
return int.from_bytes(data, "big")
|
||||
@@ -1,63 +0,0 @@
|
||||
from enum import Enum, auto
|
||||
from typing import List
|
||||
from ragger.firmware import Firmware
|
||||
from ragger.navigator import Navigator, NavInsID, NavIns
|
||||
|
||||
class SettingID(Enum):
|
||||
BLIND_SIGNING = auto()
|
||||
DEBUG_DATA = auto()
|
||||
NONCE = auto()
|
||||
VERBOSE_EIP712 = auto()
|
||||
VERBOSE_ENS = auto()
|
||||
|
||||
def get_device_settings(device: str) -> list[SettingID]:
|
||||
if device == "nanos":
|
||||
return [
|
||||
SettingID.BLIND_SIGNING,
|
||||
SettingID.DEBUG_DATA,
|
||||
SettingID.NONCE
|
||||
]
|
||||
if (device == "nanox") or (device == "nanosp") or (device == "stax"):
|
||||
return [
|
||||
SettingID.BLIND_SIGNING,
|
||||
SettingID.DEBUG_DATA,
|
||||
SettingID.NONCE,
|
||||
SettingID.VERBOSE_EIP712,
|
||||
SettingID.VERBOSE_ENS
|
||||
]
|
||||
return []
|
||||
|
||||
settings_per_page = 3
|
||||
|
||||
def get_setting_position(device: str, setting: NavInsID) -> tuple[int, int]:
|
||||
screen_height = 672 # px
|
||||
header_height = 85 # px
|
||||
footer_height = 124 # px
|
||||
usable_height = screen_height - (header_height + footer_height)
|
||||
setting_height = usable_height // settings_per_page
|
||||
index_in_page = get_device_settings(device).index(setting) % settings_per_page
|
||||
return 350, header_height + (setting_height * index_in_page) + (setting_height // 2)
|
||||
|
||||
def settings_toggle(fw: Firmware, nav: Navigator, to_toggle: list[SettingID]):
|
||||
moves = list()
|
||||
settings = get_device_settings(fw.device)
|
||||
# Assume the app is on the home page
|
||||
if fw.device.startswith("nano"):
|
||||
moves += [NavInsID.RIGHT_CLICK] * 2
|
||||
moves += [NavInsID.BOTH_CLICK]
|
||||
for setting in settings:
|
||||
if setting in to_toggle:
|
||||
moves += [NavInsID.BOTH_CLICK]
|
||||
moves += [NavInsID.RIGHT_CLICK]
|
||||
moves += [NavInsID.BOTH_CLICK] # Back
|
||||
else:
|
||||
moves += [NavInsID.USE_CASE_HOME_SETTINGS]
|
||||
moves += [NavInsID.USE_CASE_SETTINGS_NEXT]
|
||||
for setting in settings:
|
||||
setting_idx = settings.index(setting)
|
||||
if (setting_idx > 0) and (setting_idx % settings_per_page) == 0:
|
||||
moves += [NavInsID.USE_CASE_SETTINGS_NEXT]
|
||||
if setting in to_toggle:
|
||||
moves += [NavIns(NavInsID.TOUCH, get_setting_position(fw.device, setting))]
|
||||
moves += [NavInsID.USE_CASE_SETTINGS_MULTI_PAGE_EXIT]
|
||||
nav.navigate(moves, screen_change_before_first_instruction=False)
|
||||
@@ -1,25 +0,0 @@
|
||||
from typing import Any
|
||||
|
||||
def der_encode(value: int) -> bytes:
|
||||
# max() to have minimum length of 1
|
||||
value_bytes = value.to_bytes(max(1, (value.bit_length() + 7) // 8), 'big')
|
||||
if value >= 0x80:
|
||||
value_bytes = (0x80 | len(value_bytes)).to_bytes(1, 'big') + value_bytes
|
||||
return value_bytes
|
||||
|
||||
def format_tlv(tag: int, value: Any) -> bytes:
|
||||
if isinstance(value, int):
|
||||
# max() to have minimum length of 1
|
||||
value = value.to_bytes(max(1, (value.bit_length() + 7) // 8), 'big')
|
||||
elif isinstance(value, str):
|
||||
value = value.encode()
|
||||
|
||||
if not isinstance(value, bytes):
|
||||
print("Unhandled TLV formatting for type : %s" % (type(value)))
|
||||
return None
|
||||
|
||||
tlv = bytearray()
|
||||
tlv += der_encode(tag)
|
||||
tlv += der_encode(len(value))
|
||||
tlv += value
|
||||
return tlv
|
||||
@@ -1,397 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import json
|
||||
import sys
|
||||
import re
|
||||
import hashlib
|
||||
from app.client import EthAppClient, EIP712FieldType
|
||||
import keychain
|
||||
from typing import Callable
|
||||
import signal
|
||||
|
||||
# global variables
|
||||
app_client: EthAppClient = None
|
||||
filtering_paths = None
|
||||
current_path = list()
|
||||
sig_ctx = {}
|
||||
|
||||
autonext_handler: Callable = None
|
||||
|
||||
|
||||
|
||||
|
||||
# From a string typename, extract the type and all the array depth
|
||||
# Input = "uint8[2][][4]" | "bool"
|
||||
# Output = ('uint8', [2, None, 4]) | ('bool', [])
|
||||
def get_array_levels(typename):
|
||||
array_lvls = list()
|
||||
regex = re.compile(r"(.*)\[([0-9]*)\]$")
|
||||
|
||||
while True:
|
||||
result = regex.search(typename)
|
||||
if not result:
|
||||
break
|
||||
typename = result.group(1)
|
||||
|
||||
level_size = result.group(2)
|
||||
if len(level_size) == 0:
|
||||
level_size = None
|
||||
else:
|
||||
level_size = int(level_size)
|
||||
array_lvls.insert(0, level_size)
|
||||
return (typename, array_lvls)
|
||||
|
||||
|
||||
# From a string typename, extract the type and its size
|
||||
# Input = "uint64" | "string"
|
||||
# Output = ('uint', 64) | ('string', None)
|
||||
def get_typesize(typename):
|
||||
regex = re.compile(r"^(\w+?)(\d*)$")
|
||||
result = regex.search(typename)
|
||||
typename = result.group(1)
|
||||
typesize = result.group(2)
|
||||
if len(typesize) == 0:
|
||||
typesize = None
|
||||
else:
|
||||
typesize = int(typesize)
|
||||
return (typename, typesize)
|
||||
|
||||
|
||||
|
||||
def parse_int(typesize):
|
||||
return (EIP712FieldType.INT, int(typesize / 8))
|
||||
|
||||
def parse_uint(typesize):
|
||||
return (EIP712FieldType.UINT, int(typesize / 8))
|
||||
|
||||
def parse_address(typesize):
|
||||
return (EIP712FieldType.ADDRESS, None)
|
||||
|
||||
def parse_bool(typesize):
|
||||
return (EIP712FieldType.BOOL, None)
|
||||
|
||||
def parse_string(typesize):
|
||||
return (EIP712FieldType.STRING, None)
|
||||
|
||||
def parse_bytes(typesize):
|
||||
if typesize != None:
|
||||
return (EIP712FieldType.FIX_BYTES, typesize)
|
||||
return (EIP712FieldType.DYN_BYTES, None)
|
||||
|
||||
# set functions for each type
|
||||
parsing_type_functions = {};
|
||||
parsing_type_functions["int"] = parse_int
|
||||
parsing_type_functions["uint"] = parse_uint
|
||||
parsing_type_functions["address"] = parse_address
|
||||
parsing_type_functions["bool"] = parse_bool
|
||||
parsing_type_functions["string"] = parse_string
|
||||
parsing_type_functions["bytes"] = parse_bytes
|
||||
|
||||
|
||||
|
||||
def send_struct_def_field(typename, keyname):
|
||||
type_enum = None
|
||||
|
||||
(typename, array_lvls) = get_array_levels(typename)
|
||||
(typename, typesize) = get_typesize(typename)
|
||||
|
||||
if typename in parsing_type_functions.keys():
|
||||
(type_enum, typesize) = parsing_type_functions[typename](typesize)
|
||||
else:
|
||||
type_enum = EIP712FieldType.CUSTOM
|
||||
typesize = None
|
||||
|
||||
with app_client.eip712_send_struct_def_struct_field(type_enum,
|
||||
typename,
|
||||
typesize,
|
||||
array_lvls,
|
||||
keyname):
|
||||
pass
|
||||
return (typename, type_enum, typesize, array_lvls)
|
||||
|
||||
|
||||
|
||||
def encode_integer(value, typesize):
|
||||
data = bytearray()
|
||||
|
||||
# Some are already represented as integers in the JSON, but most as strings
|
||||
if isinstance(value, str):
|
||||
base = 10
|
||||
if value.startswith("0x"):
|
||||
base = 16
|
||||
value = int(value, base)
|
||||
|
||||
if value == 0:
|
||||
data.append(0)
|
||||
else:
|
||||
if value < 0: # negative number, send it as unsigned
|
||||
mask = 0
|
||||
for i in range(typesize): # make a mask as big as the typesize
|
||||
mask = (mask << 8) | 0xff
|
||||
value &= mask
|
||||
while value > 0:
|
||||
data.append(value & 0xff)
|
||||
value >>= 8
|
||||
data.reverse()
|
||||
return data
|
||||
|
||||
def encode_int(value, typesize):
|
||||
return encode_integer(value, typesize)
|
||||
|
||||
def encode_uint(value, typesize):
|
||||
return encode_integer(value, typesize)
|
||||
|
||||
def encode_hex_string(value, size):
|
||||
data = bytearray()
|
||||
value = value[2:] # skip 0x
|
||||
byte_idx = 0
|
||||
while byte_idx < size:
|
||||
data.append(int(value[(byte_idx * 2):(byte_idx * 2 + 2)], 16))
|
||||
byte_idx += 1
|
||||
return data
|
||||
|
||||
def encode_address(value, typesize):
|
||||
return encode_hex_string(value, 20)
|
||||
|
||||
def encode_bool(value, typesize):
|
||||
return encode_integer(value, typesize)
|
||||
|
||||
def encode_string(value, typesize):
|
||||
data = bytearray()
|
||||
for char in value:
|
||||
data.append(ord(char))
|
||||
return data
|
||||
|
||||
def encode_bytes_fix(value, typesize):
|
||||
return encode_hex_string(value, typesize)
|
||||
|
||||
def encode_bytes_dyn(value, typesize):
|
||||
# length of the value string
|
||||
# - the length of 0x (2)
|
||||
# / by the length of one byte in a hex string (2)
|
||||
return encode_hex_string(value, int((len(value) - 2) / 2))
|
||||
|
||||
# set functions for each type
|
||||
encoding_functions = {}
|
||||
encoding_functions[EIP712FieldType.INT] = encode_int
|
||||
encoding_functions[EIP712FieldType.UINT] = encode_uint
|
||||
encoding_functions[EIP712FieldType.ADDRESS] = encode_address
|
||||
encoding_functions[EIP712FieldType.BOOL] = encode_bool
|
||||
encoding_functions[EIP712FieldType.STRING] = encode_string
|
||||
encoding_functions[EIP712FieldType.FIX_BYTES] = encode_bytes_fix
|
||||
encoding_functions[EIP712FieldType.DYN_BYTES] = encode_bytes_dyn
|
||||
|
||||
|
||||
|
||||
def send_struct_impl_field(value, field):
|
||||
# Something wrong happened if this triggers
|
||||
if isinstance(value, list) or (field["enum"] == EIP712FieldType.CUSTOM):
|
||||
breakpoint()
|
||||
|
||||
data = encoding_functions[field["enum"]](value, field["typesize"])
|
||||
|
||||
|
||||
if filtering_paths:
|
||||
path = ".".join(current_path)
|
||||
if path in filtering_paths.keys():
|
||||
send_filtering_show_field(filtering_paths[path])
|
||||
|
||||
with app_client.eip712_send_struct_impl_struct_field(data):
|
||||
enable_autonext()
|
||||
disable_autonext()
|
||||
|
||||
|
||||
|
||||
def evaluate_field(structs, data, field, lvls_left, new_level = True):
|
||||
array_lvls = field["array_lvls"]
|
||||
|
||||
if new_level:
|
||||
current_path.append(field["name"])
|
||||
if len(array_lvls) > 0 and lvls_left > 0:
|
||||
with app_client.eip712_send_struct_impl_array(len(data)):
|
||||
pass
|
||||
idx = 0
|
||||
for subdata in data:
|
||||
current_path.append("[]")
|
||||
if not evaluate_field(structs, subdata, field, lvls_left - 1, False):
|
||||
return False
|
||||
current_path.pop()
|
||||
idx += 1
|
||||
if array_lvls[lvls_left - 1] != None:
|
||||
if array_lvls[lvls_left - 1] != idx:
|
||||
print("Mismatch in array size! Got %d, expected %d\n" %
|
||||
(idx, array_lvls[lvls_left - 1]),
|
||||
file=sys.stderr)
|
||||
return False
|
||||
else:
|
||||
if field["enum"] == EIP712FieldType.CUSTOM:
|
||||
if not send_struct_impl(structs, data, field["type"]):
|
||||
return False
|
||||
else:
|
||||
send_struct_impl_field(data, field)
|
||||
if new_level:
|
||||
current_path.pop()
|
||||
return True
|
||||
|
||||
|
||||
|
||||
def send_struct_impl(structs, data, structname):
|
||||
# Check if it is a struct we don't known
|
||||
if structname not in structs.keys():
|
||||
return False
|
||||
|
||||
struct = structs[structname]
|
||||
for f in struct:
|
||||
if not evaluate_field(structs, data[f["name"]], f, len(f["array_lvls"])):
|
||||
return False
|
||||
return True
|
||||
|
||||
# ledgerjs doesn't actually sign anything, and instead uses already pre-computed signatures
|
||||
def send_filtering_message_info(display_name: str, filters_count: int):
|
||||
global sig_ctx
|
||||
|
||||
to_sign = bytearray()
|
||||
to_sign.append(183)
|
||||
to_sign += sig_ctx["chainid"]
|
||||
to_sign += sig_ctx["caddr"]
|
||||
to_sign += sig_ctx["schema_hash"]
|
||||
to_sign.append(filters_count)
|
||||
for char in display_name:
|
||||
to_sign.append(ord(char))
|
||||
|
||||
sig = keychain.sign_data(keychain.Key.CAL, to_sign)
|
||||
with app_client.eip712_filtering_message_info(display_name, filters_count, sig):
|
||||
enable_autonext()
|
||||
disable_autonext()
|
||||
|
||||
# ledgerjs doesn't actually sign anything, and instead uses already pre-computed signatures
|
||||
def send_filtering_show_field(display_name):
|
||||
global sig_ctx
|
||||
|
||||
path_str = ".".join(current_path)
|
||||
|
||||
to_sign = bytearray()
|
||||
to_sign.append(72)
|
||||
to_sign += sig_ctx["chainid"]
|
||||
to_sign += sig_ctx["caddr"]
|
||||
to_sign += sig_ctx["schema_hash"]
|
||||
for char in path_str:
|
||||
to_sign.append(ord(char))
|
||||
for char in display_name:
|
||||
to_sign.append(ord(char))
|
||||
sig = keychain.sign_data(keychain.Key.CAL, to_sign)
|
||||
with app_client.eip712_filtering_show_field(display_name, sig):
|
||||
pass
|
||||
|
||||
def read_filtering_file(domain, message, filtering_file_path):
|
||||
data_json = None
|
||||
with open(filtering_file_path) as data:
|
||||
data_json = json.load(data)
|
||||
return data_json
|
||||
|
||||
def prepare_filtering(filtr_data, message):
|
||||
global filtering_paths
|
||||
|
||||
if "fields" in filtr_data:
|
||||
filtering_paths = filtr_data["fields"]
|
||||
else:
|
||||
filtering_paths = {}
|
||||
|
||||
def handle_optional_domain_values(domain):
|
||||
if "chainId" not in domain.keys():
|
||||
domain["chainId"] = 0
|
||||
if "verifyingContract" not in domain.keys():
|
||||
domain["verifyingContract"] = "0x0000000000000000000000000000000000000000"
|
||||
|
||||
def init_signature_context(types, domain):
|
||||
global sig_ctx
|
||||
|
||||
handle_optional_domain_values(domain)
|
||||
caddr = domain["verifyingContract"]
|
||||
if caddr.startswith("0x"):
|
||||
caddr = caddr[2:]
|
||||
sig_ctx["caddr"] = bytearray.fromhex(caddr)
|
||||
chainid = domain["chainId"]
|
||||
sig_ctx["chainid"] = bytearray()
|
||||
for i in range(8):
|
||||
sig_ctx["chainid"].append(chainid & (0xff << (i * 8)))
|
||||
sig_ctx["chainid"].reverse()
|
||||
schema_str = json.dumps(types).replace(" ","")
|
||||
schema_hash = hashlib.sha224(schema_str.encode())
|
||||
sig_ctx["schema_hash"] = bytearray.fromhex(schema_hash.hexdigest())
|
||||
|
||||
|
||||
def next_timeout(_signum: int, _frame):
|
||||
autonext_handler()
|
||||
|
||||
def enable_autonext():
|
||||
seconds = 1/4
|
||||
if app_client._client.firmware.device == 'stax': # Stax Speculos is slow
|
||||
interval = seconds * 3
|
||||
else:
|
||||
interval = seconds
|
||||
signal.setitimer(signal.ITIMER_REAL, seconds, interval)
|
||||
|
||||
def disable_autonext():
|
||||
signal.setitimer(signal.ITIMER_REAL, 0, 0)
|
||||
|
||||
|
||||
def process_file(aclient: EthAppClient,
|
||||
input_file_path: str,
|
||||
filtering_file_path = None,
|
||||
autonext: Callable = None) -> bool:
|
||||
global sig_ctx
|
||||
global app_client
|
||||
global autonext_handler
|
||||
|
||||
app_client = aclient
|
||||
with open(input_file_path, "r") as data:
|
||||
data_json = json.load(data)
|
||||
domain_typename = "EIP712Domain"
|
||||
message_typename = data_json["primaryType"]
|
||||
types = data_json["types"]
|
||||
domain = data_json["domain"]
|
||||
message = data_json["message"]
|
||||
|
||||
if autonext:
|
||||
autonext_handler = autonext
|
||||
signal.signal(signal.SIGALRM, next_timeout)
|
||||
|
||||
if filtering_file_path:
|
||||
init_signature_context(types, domain)
|
||||
filtr = read_filtering_file(domain, message, filtering_file_path)
|
||||
|
||||
# send types definition
|
||||
for key in types.keys():
|
||||
with app_client.eip712_send_struct_def_struct_name(key):
|
||||
pass
|
||||
for f in types[key]:
|
||||
(f["type"], f["enum"], f["typesize"], f["array_lvls"]) = \
|
||||
send_struct_def_field(f["type"], f["name"])
|
||||
|
||||
if filtering_file_path:
|
||||
with app_client.eip712_filtering_activate():
|
||||
pass
|
||||
prepare_filtering(filtr, message)
|
||||
|
||||
# send domain implementation
|
||||
with app_client.eip712_send_struct_impl_root_struct(domain_typename):
|
||||
enable_autonext()
|
||||
disable_autonext()
|
||||
if not send_struct_impl(types, domain, domain_typename):
|
||||
return False
|
||||
|
||||
if filtering_file_path:
|
||||
if filtr and "name" in filtr:
|
||||
send_filtering_message_info(filtr["name"], len(filtering_paths))
|
||||
else:
|
||||
send_filtering_message_info(domain["name"], len(filtering_paths))
|
||||
|
||||
# send message implementation
|
||||
with app_client.eip712_send_struct_impl_root_struct(message_typename):
|
||||
enable_autonext()
|
||||
disable_autonext()
|
||||
if not send_struct_impl(types, message, message_typename):
|
||||
return False
|
||||
|
||||
return True
|
||||
@@ -1,27 +0,0 @@
|
||||
import os
|
||||
import hashlib
|
||||
from ecdsa.util import sigencode_der
|
||||
from ecdsa import SigningKey
|
||||
from enum import Enum, auto
|
||||
|
||||
# Private key PEM files have to be named the same (lowercase) as their corresponding enum entries
|
||||
# Example: for an entry in the Enum named DEV, its PEM file must be at keychain/dev.pem
|
||||
class Key(Enum):
|
||||
CAL = auto()
|
||||
DOMAIN_NAME = auto()
|
||||
|
||||
_keys: dict[Key, SigningKey] = dict()
|
||||
|
||||
# Open the corresponding PEM file and load its key in the global dict
|
||||
def _init_key(key: Key):
|
||||
global _keys
|
||||
with open("%s/keychain/%s.pem" % (os.path.dirname(__file__), key.name.lower())) as pem_file:
|
||||
_keys[key] = SigningKey.from_pem(pem_file.read(), hashlib.sha256)
|
||||
assert (key in _keys) and (_keys[key] != None)
|
||||
|
||||
# Generate a SECP256K1 signature of the given data with the given key
|
||||
def sign_data(key: Key, data: bytes) -> bytes:
|
||||
global _keys
|
||||
if key not in _keys:
|
||||
_init_key(key)
|
||||
return _keys[key].sign_deterministic(data, sigencode=sigencode_der)
|
||||
@@ -1,8 +0,0 @@
|
||||
-----BEGIN EC PARAMETERS-----
|
||||
BgUrgQQACg==
|
||||
-----END EC PARAMETERS-----
|
||||
-----BEGIN EC PRIVATE KEY-----
|
||||
MHQCAQEEIHoMkoRaNq0neb1TxRBor4WouV8PQqJf02sg4eh768LpoAcGBSuBBAAK
|
||||
oUQDQgAETMqPrUlqpQQKAKfrL1zDuFN22IuhR6fXBUqZxkBWGIf+F6CW42w7Ujsk
|
||||
Tz4v9/hAribE53rTvHOa9d5vLXentg==
|
||||
-----END EC PRIVATE KEY-----
|
||||
@@ -1,8 +0,0 @@
|
||||
-----BEGIN EC PARAMETERS-----
|
||||
BgUrgQQACg==
|
||||
-----END EC PARAMETERS-----
|
||||
-----BEGIN EC PRIVATE KEY-----
|
||||
MHQCAQEEIHfwyko1dEHTTQ7es7EUy2ajZo1IRRcEC8/9b+MDOzUaoAcGBSuBBAAK
|
||||
oUQDQgAEuR++wXPjukpxTgFOvIJ7b4man6f0rHac3ihDF6APT2UPCfCapP9aMXYC
|
||||
Vf5d/IETKbO1C+mRlPyhFhnmXy7f6g==
|
||||
-----END EC PRIVATE KEY-----
|
||||
@@ -1,4 +1,4 @@
|
||||
ragger[speculos]
|
||||
pytest
|
||||
ecdsa
|
||||
simple-rlp
|
||||
./client/
|
||||
|
||||
@@ -1,12 +1,16 @@
|
||||
import pytest
|
||||
from ragger.error import ExceptionRAPDU
|
||||
from ragger.firmware import Firmware
|
||||
from pathlib import Path
|
||||
from ragger.backend import BackendInterface
|
||||
from ragger.firmware import Firmware
|
||||
from ragger.error import ExceptionRAPDU
|
||||
from ragger.navigator import Navigator, NavInsID
|
||||
from app.client import EthAppClient, StatusWord, ROOT_SCREENSHOT_PATH
|
||||
from app.settings import SettingID, settings_toggle
|
||||
import app.response_parser as ResponseParser
|
||||
import struct
|
||||
|
||||
import ledger_app_clients.ethereum.response_parser as ResponseParser
|
||||
from ledger_app_clients.ethereum.client import EthAppClient, StatusWord
|
||||
from ledger_app_clients.ethereum.settings import SettingID, settings_toggle
|
||||
|
||||
|
||||
ROOT_SCREENSHOT_PATH = Path(__file__).parent
|
||||
|
||||
# Values used across all tests
|
||||
CHAIN_ID = 1
|
||||
@@ -73,7 +77,6 @@ def test_send_fund_wrong_challenge(firmware: Firmware,
|
||||
backend: BackendInterface,
|
||||
navigator: Navigator):
|
||||
app_client = EthAppClient(backend)
|
||||
caught = False
|
||||
challenge = common(app_client)
|
||||
|
||||
try:
|
||||
|
||||
@@ -1,37 +1,42 @@
|
||||
import pytest
|
||||
import os
|
||||
import fnmatch
|
||||
from typing import List
|
||||
from ragger.firmware import Firmware
|
||||
from ragger.backend import BackendInterface
|
||||
from ragger.navigator import Navigator, NavInsID
|
||||
from app.client import EthAppClient
|
||||
from app.settings import SettingID, settings_toggle
|
||||
from eip712 import InputData
|
||||
from pathlib import Path
|
||||
from configparser import ConfigParser
|
||||
import app.response_parser as ResponseParser
|
||||
from functools import partial
|
||||
import os
|
||||
import pytest
|
||||
import time
|
||||
from configparser import ConfigParser
|
||||
from functools import partial
|
||||
from pathlib import Path
|
||||
from ragger.backend import BackendInterface
|
||||
from ragger.firmware import Firmware
|
||||
from ragger.navigator import Navigator, NavInsID
|
||||
from typing import List
|
||||
|
||||
import ledger_app_clients.ethereum.response_parser as ResponseParser
|
||||
from ledger_app_clients.ethereum.client import EthAppClient
|
||||
from ledger_app_clients.ethereum.eip712 import InputData
|
||||
from ledger_app_clients.ethereum.settings import SettingID, settings_toggle
|
||||
|
||||
|
||||
BIP32_PATH = "m/44'/60'/0'/0/0"
|
||||
|
||||
|
||||
def input_files() -> List[str]:
|
||||
files = []
|
||||
for file in os.scandir("%s/eip712/input_files" % (os.path.dirname(__file__))):
|
||||
for file in os.scandir("%s/eip712_input_files" % (os.path.dirname(__file__))):
|
||||
if fnmatch.fnmatch(file, "*-data.json"):
|
||||
files.append(file.path)
|
||||
return sorted(files)
|
||||
|
||||
|
||||
@pytest.fixture(params=input_files())
|
||||
def input_file(request) -> str:
|
||||
return Path(request.param)
|
||||
|
||||
|
||||
@pytest.fixture(params=[True, False])
|
||||
def verbose(request) -> bool:
|
||||
return request.param
|
||||
|
||||
|
||||
@pytest.fixture(params=[False, True])
|
||||
def filtering(request) -> bool:
|
||||
return request.param
|
||||
|
||||
Reference in New Issue
Block a user