Merge pull request #23 from blooo-io/feat/make-it-work-on-bagl

LDG-512: Implement signTransferWithSchedule method

ledger_app.toml

@@ -1,7 +1,7 @@
 [app]
 build_directory = "./"
 sdk = "C"
-devices = ["nanox", "nanos+"]
+devices = ["nanos+", "nanox"]
 
 [tests]
 unit_directory = "./unit-tests/"

tests/application_client/boilerplate_command_sender.py

@@ -1,10 +1,10 @@
 from enum import IntEnum
-from typing import Generator, List, Optional
+from typing import Generator, Optional
 from contextlib import contextmanager
 
 from ragger.backend.interface import BackendInterface, RAPDU
 from ragger.bip import pack_derivation_path
-
+from utils import split_message
 
 MAX_APDU_LEN: int = 255
 
@@ -18,6 +18,8 @@ class P1(IntEnum):
     # Parameter 1 for screen confirmation for GET_PUBLIC_KEY.
     P1_CONFIRM = 0x00
     P1_NO_CONFIRM = 0x01
+    # Parameter 1 for the scheduled amounts
+    P1_SCHEDULED_AMOUNTS = 0x01
     # Basic P1 for all instructions
     P1_NONE = 0x00
 
@@ -72,10 +74,6 @@ class Errors(IntEnum):
     SW_SIGNATURE_FAIL = 0xB008
 
 
-def split_message(message: bytes, max_size: int) -> List[bytes]:
-    return [message[x : x + max_size] for x in range(0, len(message), max_size)]
-
-
 class BoilerplateCommandSender:
     def __init__(self, backend: BackendInterface) -> None:
         self.backend = backend
@@ -200,6 +198,35 @@ def sign_simple_transfer_with_memo(
         ) as response:
             yield response
 
+    @contextmanager
+    def sign_tx_with_schedule_part_1(
+        self, path: str, header_and_to_address: bytes, num_pairs: int
+    ) -> Generator[None, None, None]:
+        # Send the derivation path, the header, the to address and the number of pairs
+        data = pack_derivation_path(path)
+        data += header_and_to_address
+        data += num_pairs.to_bytes(1, byteorder="big")
+        print("km------------data", data.hex())
+        with self.backend.exchange_async(
+            cla=CLA,
+            ins=InsType.SIGN_TRANSFER_WITH_SCHEDULE,
+            p1=P1.P1_NONE,
+            p2=P2.P2_NONE,
+            data=data,
+        ) as response:
+            yield response
+
+    @contextmanager
+    def sign_tx_with_schedule_part_2(self, data: bytes) -> Generator[None, None, None]:
+        with self.backend.exchange_async(
+            cla=CLA,
+            ins=InsType.SIGN_TRANSFER_WITH_SCHEDULE,
+            p1=P1.P1_SCHEDULED_AMOUNTS,
+            p2=P2.P2_NONE,
+            data=data,
+        ) as response:
+            yield response
+
     # @contextmanager
     # def sign_tx(
     #     self, path: str, tx_type_ins: InsType, transaction: bytes

tests/test_sign_cmd.py

@@ -12,7 +12,9 @@
 )
 from ragger.error import ExceptionRAPDU
 from ragger.navigator import NavInsID
-from utils import navigate_until_text_and_compare
+from utils import navigate_until_text_and_compare, instructions_builder, split_message
+
+MAX_SCHEDULE_PAIRS_IN_ONE_APDU: int = (250 // 16) * 16
 
 
 # In this test we send to the device a transaction to sign and validate it on screen
@@ -130,3 +132,83 @@ def test_sign_tx_simple_transfer_with_memo_legacy_path(
         == "a588094eef4ed6053df2ab4b851bc5ec09b311c204d2fa94a9c7d7c8feebf74de26d2d2a547f18c4e959b24388394305ebd3dca99653de1cb1aa689bb6674207"
     )
     # assert check_signature_validity(public_key, der_sig, transaction)
+
+
+@pytest.mark.active_test_scope
+def test_sign_tx_transfer_with_schedule_legacy_path(
+    backend, firmware, navigator, default_screenshot_path, test_name
+):
+    # Initialize the command sender client
+    client = BoilerplateCommandSender(backend)
+    # Define the path for the transaction
+    path = "m/1105/0/0/0/0/2/0/0"
+
+    # Create the transaction that will be sent to the device for signing
+    header_and_to_address = "20a845815bd43a1999e90fbf971537a70392eb38f89e6bd32b3dd70e1a9551d7000000000000000a0000000000000064000000290000000063de5da71320a845815bd43a1999e90fbf971537a70392eb38f89e6bd32b3dd70e1a9551d7"
+    header_and_to_address = bytes.fromhex(header_and_to_address)
+
+    # Define the schedule pairs
+    pairs = [
+        "0000017a396883d90000000005f5e100",
+        "0000017a396883d90000000005f5e100",
+        "0000017a396883d90000000005f5e100",
+        "0000017a396883d90000000005f5e100",
+        "0000017a396883d90000000005f5e100",
+    ]
+    joined_pairs = bytes.fromhex("".join(pairs))
+
+    # Ensure pairs are a multiple of 16 bytes
+    if len(joined_pairs) % 16 != 0:
+        raise ValueError("Pairs must be a multiple of 16 bytes")
+
+    # Split the pairs into chunks for APDU transmission
+    pairs_chunk = split_message(joined_pairs, MAX_SCHEDULE_PAIRS_IN_ONE_APDU)
+
+    # Send the first part of the transaction signing request
+    with client.sign_tx_with_schedule_part_1(
+        path=path,
+        header_and_to_address=header_and_to_address,
+        num_pairs=len(pairs),
+    ):
+        # Navigate and compare screenshots for validation
+        navigator.navigate_and_compare(
+            default_screenshot_path,
+            test_name,
+            instructions_builder(5, backend),
+            10,
+            True,
+            False,
+        )
+
+    # Process each chunk of pairs
+    screenshots_so_far = 6
+    for chunk in pairs_chunk:
+        number_of_pairs_in_chunk = len(chunk) // 16
+        # Create the instructions to validate each pair
+        instructions = []
+        for _ in range(number_of_pairs_in_chunk):
+            instructions.extend(instructions_builder(2, backend))
+
+        # Send the second part of the transaction signing request
+        with client.sign_tx_with_schedule_part_2(data=chunk):
+            # Navigate and compare screenshots for validation
+            navigator.navigate_and_compare(
+                default_screenshot_path,
+                test_name,
+                instructions,
+                10,
+                False,
+                False,
+                screenshots_so_far,
+            )
+        screenshots_so_far += number_of_pairs_in_chunk * 3
+
+    # The device as yielded the result, parse it and ensure that the signature is correct
+    response = client.get_async_response().data
+    response_hex = response.hex()
+    print("km------------response", response_hex)
+    # TODO: verify the signature
+    assert (
+        response_hex
+        == "e22fa38f78a79db71e84376c4eec2382166cdc412994207e7631b0ba3828f069b17b6f30351a64c50e5efacec3fe25161e9f7131e0235cd740739b24e0b06308"
+    )

tests/utils.py

@@ -1,13 +1,18 @@
 from hashlib import sha256
-from sha3 import keccak_256
+from sha3 import keccak_256  # type: ignore
+from typing import List
 
-from ecdsa.curves import SECP256k1
-from ecdsa.keys import VerifyingKey
-from ecdsa.util import sigdecode_der
+from ecdsa.curves import SECP256k1  # type: ignore
+from ecdsa.keys import VerifyingKey  # type: ignore
+from ecdsa.util import sigdecode_der  # type: ignore
 
 from ragger.navigator import NavInsID
 
 
+def split_message(message: bytes, max_size: int) -> List[bytes]:
+    return [message[x : x + max_size] for x in range(0, len(message), max_size)]
+
+
 # Check if a signature of a given message is valid
 def check_signature_validity(
     public_key: bytes, signature: bytes, message: bytes
@@ -38,7 +43,13 @@ def instructions_builder(
 
 
 def navigate_until_text_and_compare(
-    firmware, navigator, text: str, screenshot_path: str, test_name: str
+    firmware,
+    navigator,
+    text: str,
+    screenshot_path: str,
+    test_name: str,
+    screen_change_before_first_instruction: bool = True,
+    screen_change_after_last_instruction: bool = True,
 ):
     """Navigate through device screens until specified text is found and compare screenshots.