Merge pull request #583 from LedgerHQ/feat/apa/eip712_improvements

EIP-712 filtering improvements

client/src/ledger_app_clients/ethereum/client.py

@@ -106,8 +106,20 @@ def eip712_filtering_message_info(self, name: str, filters_count: int, sig: byte
                                                                                     filters_count,
                                                                                     sig))
 
-    def eip712_filtering_show_field(self, name: str, sig: bytes):
-        return self._exchange_async(self._cmd_builder.eip712_filtering_show_field(name, sig))
+    def eip712_filtering_amount_join_token(self, token_idx: int, sig: bytes):
+        return self._exchange_async(self._cmd_builder.eip712_filtering_amount_join_token(token_idx,
+                                                                                         sig))
+
+    def eip712_filtering_amount_join_value(self, token_idx: int, name: str, sig: bytes):
+        return self._exchange_async(self._cmd_builder.eip712_filtering_amount_join_value(token_idx,
+                                                                                         name,
+                                                                                         sig))
+
+    def eip712_filtering_datetime(self, name: str, sig: bytes):
+        return self._exchange_async(self._cmd_builder.eip712_filtering_datetime(name, sig))
+
+    def eip712_filtering_raw(self, name: str, sig: bytes):
+        return self._exchange_async(self._cmd_builder.eip712_filtering_raw(name, sig))
 
     def sign(self,
              bip32_path: str,

client/src/ledger_app_clients/ethereum/command_builder.py

@@ -41,8 +41,11 @@ class P2Type(IntEnum):
     LEGACY_IMPLEM = 0x00
     NEW_IMPLEM = 0x01
     FILTERING_ACTIVATE = 0x00
-    FILTERING_CONTRACT_NAME = 0x0f
-    FILTERING_FIELD_NAME = 0xff
+    FILTERING_MESSAGE_INFO = 0x0f
+    FILTERING_DATETIME = 0xfc
+    FILTERING_TOKEN_ADDR_CHECK = 0xfd
+    FILTERING_AMOUNT_FIELD = 0xfe
+    FILTERING_RAW = 0xff
 
 
 class CommandBuilder:
@@ -62,17 +65,11 @@ def _serialize(self,
         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))
+                               name.encode())
 
     def eip712_send_struct_def_struct_field(self,
                                             field_type: EIP712FieldType,
@@ -88,7 +85,7 @@ def eip712_send_struct_def_struct_field(self,
         data.append(typedesc)
         if field_type == EIP712FieldType.CUSTOM:
             data.append(len(type_name))
-            data += self._string_to_bytes(type_name)
+            data += type_name.encode()
         if type_size is not None:
             data.append(type_size)
         if len(array_levels) > 0:
@@ -98,7 +95,7 @@ def eip712_send_struct_def_struct_field(self,
                 if level is not None:
                     data.append(level)
         data.append(len(key_name))
-        data += self._string_to_bytes(key_name)
+        data += key_name.encode()
         return self._serialize(InsType.EIP712_SEND_STRUCT_DEF,
                                P1Type.COMPLETE_SEND,
                                P2Type.STRUCT_FIELD,
@@ -108,7 +105,7 @@ 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))
+                               name.encode())
 
     def eip712_send_struct_impl_array(self, size: int) -> bytes:
         data = bytearray()
@@ -162,33 +159,61 @@ def eip712_filtering_activate(self):
     def _eip712_filtering_send_name(self, name: str, sig: bytes) -> bytes:
         data = bytearray()
         data.append(len(name))
-        data += self._string_to_bytes(name)
+        data += name.encode()
         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 += name.encode()
         data.append(filters_count)
         data.append(len(sig))
         data += sig
         return self._serialize(InsType.EIP712_SEND_FILTERING,
                                P1Type.COMPLETE_SEND,
-                               P2Type.FILTERING_CONTRACT_NAME,
+                               P2Type.FILTERING_MESSAGE_INFO,
+                               data)
+
+    def eip712_filtering_amount_join_token(self, token_idx: int, sig: bytes) -> bytes:
+        data = bytearray()
+        data.append(token_idx)
+        data.append(len(sig))
+        data += sig
+        return self._serialize(InsType.EIP712_SEND_FILTERING,
+                               P1Type.COMPLETE_SEND,
+                               P2Type.FILTERING_TOKEN_ADDR_CHECK,
+                               data)
+
+    def eip712_filtering_amount_join_value(self, token_idx: int, name: str, sig: bytes) -> bytes:
+        data = bytearray()
+        data.append(len(name))
+        data += name.encode()
+        data.append(token_idx)
+        data.append(len(sig))
+        data += sig
+        return self._serialize(InsType.EIP712_SEND_FILTERING,
+                               P1Type.COMPLETE_SEND,
+                               P2Type.FILTERING_AMOUNT_FIELD,
                                data)
 
-    def eip712_filtering_show_field(self, name: str, sig: bytes) -> bytes:
+    def eip712_filtering_datetime(self, name: str, sig: bytes) -> bytes:
+        return self._serialize(InsType.EIP712_SEND_FILTERING,
+                               P1Type.COMPLETE_SEND,
+                               P2Type.FILTERING_DATETIME,
+                               self._eip712_filtering_send_name(name, sig))
+
+    def eip712_filtering_raw(self, name: str, sig: bytes) -> bytes:
         return self._serialize(InsType.EIP712_SEND_FILTERING,
                                P1Type.COMPLETE_SEND,
-                               P2Type.FILTERING_FIELD_NAME,
+                               P2Type.FILTERING_RAW,
                                self._eip712_filtering_send_name(name, sig))
 
     def set_external_plugin(self, plugin_name: str, contract_address: bytes, selector: bytes, sig: bytes) -> bytes:
         data = bytearray()
         data.append(len(plugin_name))
-        data += self._string_to_bytes(plugin_name)
+        data += plugin_name.encode()
         data += contract_address
         data += selector
         data += sig

client/src/ledger_app_clients/ethereum/eip712/InputData.py

@@ -165,10 +165,7 @@ def encode_bool(value: str, typesize: int) -> bytes:
 
 
 def encode_string(value: str, typesize: int) -> bytes:
-    data = bytearray()
-    for char in value:
-        data.append(ord(char))
-    return data
+    return value.encode()
 
 
 def encode_bytes_fix(value: str, typesize: int) -> bytes:
@@ -203,7 +200,17 @@ def send_struct_impl_field(value, field):
     if filtering_paths:
         path = ".".join(current_path)
         if path in filtering_paths.keys():
-            send_filtering_show_field(filtering_paths[path])
+            if filtering_paths[path]["type"] == "amount_join_token":
+                send_filtering_amount_join_token(filtering_paths[path]["token"])
+            elif filtering_paths[path]["type"] == "amount_join_value":
+                send_filtering_amount_join_value(filtering_paths[path]["token"],
+                                                 filtering_paths[path]["name"])
+            elif filtering_paths[path]["type"] == "datetime":
+                send_filtering_datetime(filtering_paths[path]["name"])
+            elif filtering_paths[path]["type"] == "raw":
+                send_filtering_raw(filtering_paths[path]["name"])
+            else:
+                assert False
 
     with app_client.eip712_send_struct_impl_struct_field(data):
         enable_autonext()
@@ -254,42 +261,82 @@ def send_struct_impl(structs, data, structname):
     return True
 
 
+def start_signature_payload(ctx: dict, magic: int) -> bytearray:
+    to_sign = bytearray()
+    # magic number so that signature for one type of filter can't possibly be
+    # valid for another, defined in APDU specs
+    to_sign.append(magic)
+    to_sign += ctx["chainid"]
+    to_sign += ctx["caddr"]
+    to_sign += ctx["schema_hash"]
+    return to_sign
+
+
 # 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 = start_signature_payload(sig_ctx, 183)
     to_sign.append(filters_count)
-    for char in display_name:
-        to_sign.append(ord(char))
+    to_sign += display_name.encode()
 
     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()
 
 
+def send_filtering_amount_join_token(token_idx: int):
+    global sig_ctx
+
+    path_str = ".".join(current_path)
+
+    to_sign = start_signature_payload(sig_ctx, 11)
+    to_sign += path_str.encode()
+    to_sign.append(token_idx)
+    sig = keychain.sign_data(keychain.Key.CAL, to_sign)
+    with app_client.eip712_filtering_amount_join_token(token_idx, sig):
+        pass
+
+
+def send_filtering_amount_join_value(token_idx: int, display_name: str):
+    global sig_ctx
+
+    path_str = ".".join(current_path)
+
+    to_sign = start_signature_payload(sig_ctx, 22)
+    to_sign += path_str.encode()
+    to_sign += display_name.encode()
+    to_sign.append(token_idx)
+    sig = keychain.sign_data(keychain.Key.CAL, to_sign)
+    with app_client.eip712_filtering_amount_join_value(token_idx, display_name, sig):
+        pass
+
+
+def send_filtering_datetime(display_name: str):
+    global sig_ctx
+
+    path_str = ".".join(current_path)
+
+    to_sign = start_signature_payload(sig_ctx, 33)
+    to_sign += path_str.encode()
+    to_sign += display_name.encode()
+    sig = keychain.sign_data(keychain.Key.CAL, to_sign)
+    with app_client.eip712_filtering_datetime(display_name, sig):
+        pass
+
+
 # ledgerjs doesn't actually sign anything, and instead uses already pre-computed signatures
-def send_filtering_show_field(display_name):
+def send_filtering_raw(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))
+    to_sign = start_signature_payload(sig_ctx, 72)
+    to_sign += path_str.encode()
+    to_sign += display_name.encode()
     sig = keychain.sign_data(keychain.Key.CAL, to_sign)
-    with app_client.eip712_filtering_show_field(display_name, sig):
+    with app_client.eip712_filtering_raw(display_name, sig):
         pass
 
 
@@ -300,6 +347,12 @@ def prepare_filtering(filtr_data, message):
         filtering_paths = filtr_data["fields"]
     else:
         filtering_paths = {}
+    if "tokens" in filtr_data:
+        for token in filtr_data["tokens"]:
+            app_client.provide_token_metadata(token["ticker"],
+                                              bytes.fromhex(token["addr"][2:]),
+                                              token["decimals"],
+                                              token["chain_id"])
 
 
 def handle_optional_domain_values(domain):