Extending and fixing transaction tests

src/monero_crypto.c

@@ -1270,6 +1270,8 @@ int monero_rng_mod_order(unsigned char *r, size_t r_len) {
     unsigned char rnd[32 + 8];
     int error;
     cx_rng(rnd, 32 + 8);
+    /* To uncomment in order to freeze the secret for test purposes */
+    /* memset(rnd, 0xBB, 32 + 8);                      */
     error = cx_math_modm_no_throw(rnd, 32 + 8, (unsigned char *)C_ED25519_ORDER, 32);
     if (error) {
         return SW_SECURITY_INTERNAL;

tests/monero_client/monero_cmd.py

@@ -27,6 +27,7 @@
 from .utils.varint import encode_varint
 from .utils.utils import get_nano_review_instructions
 from pathlib import Path
+from ragger.firmware import Firmware
 from ragger.navigator import NavInsID, NavIns
 
 PROTOCOL_VERSION: int = 3
@@ -223,13 +224,13 @@ def prefix_hash_init(self, backend, test_name, firmware, navigator, version: int
             encode_varint(timelock)
         ])
 
-        if firmware.device == "nanos":
+        if firmware == Firmware.NANOS:
             instructions = get_nano_review_instructions(1)
-        elif firmware.device.startswith("nano"):
+        elif firmware.is_nano:
             instructions = get_nano_review_instructions(1)
         else:
             instructions = [
-                NavIns(NavInsID.SWIPE_CENTER_TO_LEFT)
+                NavInsID.SWIPE_CENTER_TO_LEFT
             ]
         with self.device.send_async(cla=PROTOCOL_VERSION,
                                     ins=ins,
@@ -314,7 +315,7 @@ def blind(self,
                         MoneroCryptoCmd.HMAC_KEY,
                         Type.AMOUNT_KEY),
             mask,
-            amount.to_bytes(32, byteorder="big")
+            amount.to_bytes(32, byteorder="little")
         ])
 
         self.device.send(cla=PROTOCOL_VERSION,
@@ -386,13 +387,13 @@ def validate_prehash_init(self,
         # txntype is skipped in the app
         payload: bytes = struct.pack("B", txntype) + encode_varint(txnfee)
 
-        if firmware.device == "nanos":
+        if firmware == Firmware.NANOS:
             instructions = get_nano_review_instructions(1)
-        elif firmware.device.startswith("nano"):
+        elif firmware.is_nano:
             instructions = get_nano_review_instructions(1)
         else:
             instructions = [
-                NavIns(NavInsID.USE_CASE_REVIEW_TAP)
+                NavInsID.USE_CASE_REVIEW_TAP
             ]
 
         with self.device.send_async(cla=PROTOCOL_VERSION,
@@ -402,7 +403,7 @@ def validate_prehash_init(self,
                                     option=0,
                                     payload=payload):
 
-            if firmware.device.startswith("nano"):
+            if firmware.is_nano:
                 navigator.navigate_and_compare(TESTS_ROOT_DIR,
                                                test_name + "_prehash_init",
                                                instructions)
@@ -450,17 +451,28 @@ def validate_prehash_update(self,
             blinded_amount
         ))
 
-        if firmware.device == "nanos":
-            instructions = get_nano_review_instructions(7)
-        elif firmware.device.startswith("nano"):
-            instructions = get_nano_review_instructions(3)
+        if firmware == Firmware.NANOS:
+            if is_last:
+                instructions = get_nano_review_instructions(1)
+            else:
+                instructions = get_nano_review_instructions(7)
+        elif firmware.is_nano:
+            if is_last:
+                instructions = get_nano_review_instructions(1)
+            else:
+                instructions = get_nano_review_instructions(3)
         else:
-            instructions = [
+            if is_last:
+                instructions = [
+                    NavInsID.USE_CASE_REVIEW_TAP,
+                    NavInsID.USE_CASE_REVIEW_CONFIRM
+                ]
 
-                NavIns(NavInsID.SWIPE_CENTER_TO_LEFT),
-                NavIns(NavInsID.USE_CASE_REVIEW_TAP),
-                NavIns(NavInsID.USE_CASE_REVIEW_CONFIRM)
-            ]
+            else:
+                instructions = [
+                    NavInsID.SWIPE_CENTER_TO_LEFT,
+                    NavInsID.SWIPE_CENTER_TO_LEFT,
+                ]
 
         backend.wait_for_text_on_screen("Processing")
         with self.device.send_async(cla=PROTOCOL_VERSION,
@@ -470,11 +482,15 @@ def validate_prehash_update(self,
                                     option=(0 if is_last else 0x80) | (
                                         0x02 if is_short else 0),
                                     payload=payload):
-
-            navigator.navigate_and_compare(TESTS_ROOT_DIR,
-                                           test_name + "_prehash_update",
-                                           instructions,
-                                           screen_change_after_last_instruction=False, timeout=10000)
+            if is_last:
+                navigator.navigate_and_compare(TESTS_ROOT_DIR,
+                                               test_name + "_prehash_update_last",
+                                               instructions,
+                                               screen_change_after_last_instruction=False, timeout=10000)
+            else:
+                navigator.navigate_and_compare(TESTS_ROOT_DIR,
+                                               test_name + "_prehash_update_first",
+                                               instructions)
 
         sw, response = self.device.async_response()  # type: int, bytes
 
@@ -527,4 +543,5 @@ def validate_prehash_finalize(self,
         if not sw & 0x9000:
             raise DeviceError(error_code=sw, ins=ins, message="P1=3 (finalize)")
 
-        assert len(response) == 32
+        if is_last:
+            assert len(response) == 32

tests/test_sig.py

@@ -37,14 +37,16 @@ def state():
         )
 
         return {"sender": sender,
-                "receiver": receiver,
-                "amount": 11**3,
+                "receiver_number": 2,
+                "receiver": [receiver, sender],
+                "is_change_addr": [False, True],
+                "amount": [int(69.36*1000000000000), int(0.55*1000000000000)],
                 "tx_pub_key": None,
                 "_tx_priv_key": None,
-                "_ak_amount": [],
-                "blinded_amount": [],
-                "blinded_mask": [],
-                "y": []}
+                "_ak_amount": [[], []],
+                "blinded_amount": [[], []],
+                "blinded_mask": [[], []],
+                "y": [[], []]}
 
     @staticmethod
     def test_set_sig(monero):
@@ -70,17 +72,18 @@ def test_open_tx(monero, state):
 
     @staticmethod
     def test_gen_txout_keys(monero, state):
-        _ak_amount, out_ephemeral_pub_key = monero.gen_txout_keys(
-            _tx_priv_key=state["_tx_priv_key"],
-            tx_pub_key=state["tx_pub_key"],
-            dst_pub_view_key=state["receiver"].public_view_key,
-            dst_pub_spend_key=state["receiver"].public_spend_key,
-            output_index=0,
-            is_change_addr=False,
-            is_subaddress=False
-        )  # type: bytes, bytes
-
-        state["_ak_amount"].append(_ak_amount)  # _ak_amount_t
+        for index in range(state["receiver_number"]):
+            _ak_amount, out_ephemeral_pub_key = monero.gen_txout_keys(
+                _tx_priv_key=state["_tx_priv_key"],
+                tx_pub_key=state["tx_pub_key"],
+                dst_pub_view_key=state["receiver"][index].public_view_key,
+                dst_pub_spend_key=state["receiver"][index].public_spend_key,
+                output_index=index,
+                is_change_addr=state["is_change_addr"][index],
+                is_subaddress=False
+            )  # type: bytes, bytes
+
+            state["_ak_amount"][index].append(_ak_amount)  # _ak_amount_t
 
     @staticmethod
     def test_prefix_hash(monero, backend, navigator, firmware, test_name):
@@ -99,40 +102,42 @@ def test_prefix_hash(monero, backend, navigator, firmware, test_name):
 
     @staticmethod
     def test_gen_commitment_mask(monero, state):
-        assert len(state["_ak_amount"]) != 0
-
-        s: bytes = monero.gen_commitment_mask(state["_ak_amount"][0])
-        state["y"].append(s)  # y_t
+        for index in range(state["receiver_number"]):
+            assert len(state["_ak_amount"][index]) != 0
+            s: bytes = monero.gen_commitment_mask(state["_ak_amount"][index][0])
+            state["y"][index].append(s)  # y_t
 
     @staticmethod
     def test_blind(monero, state):
-        assert len(state["y"]) != 0
-        assert len(state["_ak_amount"]) != 0
-
-        blinded_mask, blinded_amount = monero.blind(
-            _ak_amount=state["_ak_amount"][0],
-            mask=state["y"][0],
-            amount=state["amount"],
-            is_short=False
-        )  # type: bytes, bytes
-
-        mask, amount = monero.unblind(
-            _ak_amount=state["_ak_amount"][0],
-            blinded_mask=blinded_mask,
-            blinded_amount=blinded_amount,
-            is_short=False
-        )  # type: bytes, bytes
-
-        assert state["y"][0] == mask
-        # assert state["amount"] == int.from_bytes(amount, byteorder="big")
-
-        state["blinded_mask"].append(blinded_mask)
-        state["blinded_amount"].append(blinded_amount)
-
-        assert len(state["y"]) != 0
-        assert len(state["_ak_amount"]) != 0
-        assert len(state["blinded_amount"]) != 0
-        assert len(state["blinded_mask"]) != 0
+        for index in range(state["receiver_number"]):
+            assert len(state["y"][index]) != 0
+            assert len(state["_ak_amount"][index]) != 0
+
+            print(f'state["amount"][index] = {state["amount"][index]}')
+            blinded_mask, blinded_amount = monero.blind(
+                _ak_amount=state["_ak_amount"][index][0],
+                mask=state["y"][index][0],
+                amount=state["amount"][index],
+                is_short=True
+            )  # type: bytes, bytes
+
+            mask, amount = monero.unblind(
+                _ak_amount=state["_ak_amount"][index][0],
+                blinded_mask=blinded_mask,
+                blinded_amount=blinded_amount,
+                is_short=True
+            )  # type: bytes, bytes
+
+            assert state["y"][index][0] == mask
+            assert state["amount"][index] == int.from_bytes(amount, byteorder="little")
+
+            state["blinded_mask"][index].append(blinded_mask)
+            state["blinded_amount"][index].append(blinded_amount)
+
+            assert len(state["y"][index]) != 0
+            assert len(state["_ak_amount"][index]) != 0
+            assert len(state["blinded_amount"][index]) != 0
+            assert len(state["blinded_mask"][index]) != 0
 
     @staticmethod
     def test_validate(monero, backend, navigator, firmware, test_name, state):
@@ -148,37 +153,38 @@ def test_validate(monero, backend, navigator, firmware, test_name, state):
                                      txntype=0,
                                      txnfee=fee)
 
-        monero.validate_prehash_update(
-            backend,
-            test_name,
-            firmware,
-            navigator,
-            index=1,
-            is_short=False,
-            is_change_addr=False,
-            is_subaddress=False,
-            dst_pub_view_key=state["receiver"].public_view_key,
-            dst_pub_spend_key=state["receiver"].public_spend_key,
-            _ak_amount=state["_ak_amount"][0],
-            commitment=bytes.fromhex(32*"00"),
-            blinded_amount=state["blinded_amount"][0],
-            blinded_mask=state["blinded_mask"][0],
-            is_last=True
-        )
-
-        monero.validate_prehash_finalize(
-            index=1,
-            is_short=False,
-            is_change_addr=False,
-            is_subaddress=False,
-            dst_pub_view_key=state["receiver"].public_view_key,
-            dst_pub_spend_key=state["receiver"].public_spend_key,
-            _ak_amount=state["_ak_amount"][0],
-            commitment=bytes.fromhex(32*"00"),
-            blinded_amount=state["blinded_amount"][0],
-            blinded_mask=state["blinded_mask"][0],
-            is_last=True
-        )
+        for index in range(state["receiver_number"]):
+            monero.validate_prehash_update(
+                backend,
+                test_name,
+                firmware,
+                navigator,
+                index=index+1,
+                is_short=True,
+                is_change_addr=state["is_change_addr"][index],
+                is_subaddress=False,
+                dst_pub_view_key=state["receiver"][index].public_view_key,
+                dst_pub_spend_key=state["receiver"][index].public_spend_key,
+                _ak_amount=state["_ak_amount"][index][0],
+                commitment=bytes.fromhex(32*"00"),
+                blinded_amount=state["blinded_amount"][index][0],
+                blinded_mask=state["blinded_mask"][index][0],
+                is_last=True if index == (state["receiver_number"] - 1) else False
+            )
+        for index in range(state["receiver_number"]):
+            monero.validate_prehash_finalize(
+                index=index+1,
+                is_short=False,
+                is_change_addr=False,
+                is_subaddress=False,
+                dst_pub_view_key=state["receiver"][index].public_view_key,
+                dst_pub_spend_key=state["receiver"][index].public_spend_key,
+                _ak_amount=state["_ak_amount"][index][0],
+                commitment=bytes.fromhex(32*"00"),
+                blinded_amount=state["blinded_amount"][index][0],
+                blinded_mask=state["blinded_mask"][index][0],
+                is_last=True if index == (state["receiver_number"] - 1) else False
+            )
 
     @staticmethod
     def test_close_tx(monero):