fix(thorchain): expose version-gated send denoms

Forward ThorchainMsgSend.denom on firmware 7.15+, retain the fail-closed RUNE-only path on older firmware, and add offline and device-path coverage. Add regression tests for the reviewed EIP-712 array-order and signed ABI integer fixes.

Author: BitHighlander

keepkeylib/client.py

@@ -1204,27 +1204,36 @@ def thorchain_sign_tx(
                     raise CallException("Thorchain.MsgSend", "Multiple amounts per send msg not supported")
 
                 denom = msg['value']['amount'][0]['denom']
-                # Fail CLOSED on any other denomination, deliberately.
-                #
-                # ThorchainMsgSend carries a `denom` field, but the firmware
-                # this talks to builds its amino sign-doc with the string
-                # "rune" HARDCODED (lib/firmware/thorchain.c) -- only 7.15+
-                # reads a denom and validates it. nanopb SKIPS unknown fields
-                # rather than rejecting them, so forwarding `denom` to older
-                # firmware would be silently ignored and the device would sign
-                # a rune transfer while the host believed it had sent another
-                # asset. Refusing is the only safe answer until the capability
-                # can be detected; do not "fix" this by passing denom through.
-                if denom != 'rune':
-                    raise CallException("Thorchain.MsgSend", "Unsupported denomination: " + denom)
+                firmware_version = (
+                    self.features.major_version,
+                    self.features.minor_version,
+                    self.features.patch_version,
+                )
+                supports_denom = firmware_version >= (7, 15, 0)
+
+                # Older firmware hardcodes "rune" in its amino sign-doc and
+                # nanopb skips the unknown denom field. Sending a non-RUNE denom
+                # there would therefore make the host and device disagree about
+                # what was signed. Preserve the fail-closed legacy behaviour,
+                # while exposing the protocol field on firmware that validates,
+                # displays and commits it to the signature.
+                if denom != 'rune' and not supports_denom:
+                    raise CallException(
+                        "Thorchain.MsgSend",
+                        "Unsupported denomination before firmware 7.15.0: " + denom,
+                    )
+
+                send = thorchain_proto.ThorchainMsgSend(
+                    from_address=msg['value']['from_address'],
+                    to_address=msg['value']['to_address'],
+                    amount=int(msg['value']['amount'][0]['amount']),
+                    address_type=types.SPEND,
+                )
+                if supports_denom:
+                    send.denom = denom
 
                 resp = self.call(thorchain_proto.ThorchainMsgAck(
-                    send=thorchain_proto.ThorchainMsgSend(
-                        from_address=msg['value']['from_address'],
-                        to_address=msg['value']['to_address'],
-                        amount=int(msg['value']['amount'][0]['amount']),
-                        address_type=types.SPEND,
-                    )
+                    send=send
                 ))
 
             elif msg['type'] == "thorchain/MsgDeposit":

tests/test_clearsign_abi.py

@@ -0,0 +1,35 @@
+import unittest
+
+from keepkeylib.clearsign_abi import encode_static_args
+
+
+class TestClearsignAbiSignedIntegers(unittest.TestCase):
+
+    def test_negative_int8_is_sign_extended(self):
+        self.assertEqual(encode_static_args(['int8'], [-1]), b'\xff' * 32)
+        self.assertEqual(
+            encode_static_args(['int8'], [-128]),
+            b'\xff' * 31 + b'\x80',
+        )
+
+    def test_int8_bounds_are_enforced(self):
+        self.assertEqual(
+            encode_static_args(['int8'], [127]),
+            b'\x00' * 31 + b'\x7f',
+        )
+        for value in (-129, 128):
+            with self.assertRaises(AssertionError):
+                encode_static_args(['int8'], [value])
+
+    def test_uint8_keeps_unsigned_bounds(self):
+        self.assertEqual(
+            encode_static_args(['uint8'], [255]),
+            b'\x00' * 31 + b'\xff',
+        )
+        for value in (-1, 256):
+            with self.assertRaises(AssertionError):
+                encode_static_args(['uint8'], [value])
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_msg_eip712_streaming.py

@@ -50,6 +50,56 @@
 SPEC_MESSAGE_HASH = "c52c0ee5d84264471806290a3f2c4cecfc5490626bf912d01f240d7a274b371e"
 
 
+class TestEip712StreamHelpers(unittest.TestCase):
+
+    def test_multidimensional_arrays_are_walked_outermost_first(self):
+        doc = {
+            'types': {
+                'EIP712Domain': [],
+                'Matrix': [{'name': 'values', 'type': 'int16[2][][4]'}],
+            },
+            'primaryType': 'Matrix',
+            'domain': {},
+            'message': {
+                'values': [
+                    [[1, 2]],
+                    [[3, 4], [5, 6]],
+                    [[7, 8], [9, 10], [11, 12]],
+                    [[13, 14]],
+                ],
+            },
+        }
+
+        self.assertEqual(es.resolve_member_path(doc, [1, 0]), ('length', 4))
+        self.assertEqual(es.resolve_member_path(doc, [1, 0, 2]), ('length', 3))
+        self.assertEqual(es.resolve_member_path(doc, [1, 0, 2, 1]), ('length', 2))
+        result = es.resolve_member_path(doc, [1, 0, 2, 1, 0])
+        self.assertEqual(result[0], 'value')
+        self.assertEqual(result[2], 9)
+
+    def test_innermost_fixed_array_length_is_checked(self):
+        doc = {
+            'types': {
+                'EIP712Domain': [],
+                'Matrix': [{'name': 'values', 'type': 'int16[2][][4]'}],
+            },
+            'primaryType': 'Matrix',
+            'domain': {},
+            'message': {
+                'values': [
+                    [[1, 2]],
+                    [[3, 4]],
+                    [[5]],
+                    [[6, 7]],
+                ],
+            },
+        }
+
+        with self.assertRaises(es.Eip712Error) as ctx:
+            es.resolve_member_path(doc, [1, 0, 2, 0])
+        self.assertIn('declares 2 elements', str(ctx.exception))
+
+
 class TestMsgEip712Streaming(common.KeepKeyTest):
 
     def _walk(self, doc, max_steps=400):

tests/test_msg_thorchain_signtx.py

@@ -5,7 +5,9 @@
 from binascii import hexlify, unhexlify
 
 import keepkeylib.messages_pb2 as proto
+import keepkeylib.messages_thorchain_pb2 as thorchain_proto
 import keepkeylib.types_pb2 as proto_types
+from keepkeylib.client import CallException, ProtocolMixin
 from keepkeylib.tools import parse_path
 from keepkeylib.signed_metadata import eth_sighash_legacy, keccak256
 
@@ -28,19 +30,94 @@ def recover_eth_signer(sig_r, sig_s, sig_v, digest, chain_id):
 
 DEFAULT_BIP32_PATH = "m/44h/931h/0h/0/0"
 
-def make_send(from_address, to_address, amount):
+def make_send(from_address, to_address, amount, denom='rune'):
     return {
         'type': 'thorchain/MsgSend',
         'value': {
             'amount': [{
-                'denom': 'rune',
+                'denom': denom,
                 'amount': str(amount),
             }],
             'from_address': from_address,
             'to_address': to_address,
         }
     }
 
+
+class _SessionTransport(object):
+    def session_begin(self):
+        pass
+
+    def session_end(self):
+        pass
+
+
+class _ScriptedThorchainClient(object):
+    thorchain_sign_tx = ProtocolMixin.thorchain_sign_tx
+
+    def __init__(self, version):
+        self.features = proto.Features(
+            major_version=version[0],
+            minor_version=version[1],
+            patch_version=version[2],
+        )
+        self.transport = _SessionTransport()
+        self.responses = [
+            thorchain_proto.ThorchainMsgRequest(),
+            thorchain_proto.ThorchainSignedTx(
+                public_key=b'\x02' + b'\x11' * 32,
+                signature=b'\x22' * 64,
+            ),
+        ]
+        self.sent = []
+
+    def call(self, message):
+        self.sent.append(message)
+        if not self.responses:
+            raise AssertionError('unexpected device call: %s' % type(message))
+        return self.responses.pop(0)
+
+
+class TestThorchainClientDenom(unittest.TestCase):
+    ADDRESS_N = [0x8000002C, 0x800003A3, 0x80000000, 0, 0]
+    FROM = 'thor1ls33ayg26kmltw7jjy55p32ghjna09zp6z69y8'
+    TO = 'thor1jvt443rvhq5h8yrna55yjysvhtju0el7ldnwwy'
+
+    def _sign(self, client, denom):
+        return client.thorchain_sign_tx(
+            address_n=self.ADDRESS_N,
+            account_number=92,
+            chain_id='thorchain',
+            fee=3000,
+            gas=200000,
+            msgs=[make_send(self.FROM, self.TO, 10000, denom=denom)],
+            memo='client denom test',
+            sequence=3,
+            testnet=False,
+        )
+
+    def test_non_rune_denom_is_forwarded_on_7_15(self):
+        client = _ScriptedThorchainClient((7, 15, 0))
+        response = self._sign(client, 'btc/btc')
+
+        self.assertIsInstance(response, thorchain_proto.ThorchainSignedTx)
+        self.assertEqual(client.sent[1].send.denom, 'btc/btc')
+
+    def test_non_rune_denom_is_rejected_before_7_15(self):
+        client = _ScriptedThorchainClient((7, 14, 2))
+
+        with self.assertRaises(CallException) as ctx:
+            self._sign(client, 'btc/btc')
+
+        self.assertIn('before firmware 7.15.0', str(ctx.exception))
+        self.assertEqual(len(client.sent), 1)
+
+    def test_legacy_rune_does_not_send_unknown_field(self):
+        client = _ScriptedThorchainClient((7, 14, 2))
+        self._sign(client, 'rune')
+
+        self.assertFalse(client.sent[1].send.HasField('denom'))
+
 class TestMsgThorChainSignTx(common.KeepKeyTest):
 
     def test_thorchain_sign_tx(self):
@@ -66,6 +143,33 @@ def test_thorchain_sign_tx(self):
         self.assertEqual(hexlify(signature.public_key), "031519713b8b42bdc367112d33132cf14cedf928ac5771d444ba459b9497117ba3")
         return
 
+    def test_thorchain_non_rune_denom_changes_signature(self):
+        """The public helper forwards denom and firmware commits it to sign-doc."""
+        self.requires_fullFeature()
+        self.requires_firmware("7.15.0")
+        self.setup_mnemonic_nopin_nopassphrase()
+        address_n = parse_path(DEFAULT_BIP32_PATH)
+        from_address = self.client.thorchain_get_address(address_n)
+        to_address = "thor1jvt443rvhq5h8yrna55yjysvhtju0el7ldnwwy"
+
+        def sign(denom):
+            return self.client.thorchain_sign_tx(
+                address_n=address_n,
+                account_number=92,
+                chain_id="thorchain",
+                fee=3000,
+                gas=200000,
+                msgs=[make_send(from_address, to_address, 10000, denom=denom)],
+                memo="denom binding",
+                sequence=3,
+                testnet=False,
+            )
+
+        rune = sign('rune')
+        btc = sign('btc/btc')
+        self.assertEqual(hexlify(rune.public_key), hexlify(btc.public_key))
+        self.assertNotEqual(hexlify(rune.signature), hexlify(btc.signature))
+
     def test_sign_btc_eth_swap(self):
         self.requires_fullFeature()
         self.requires_firmware("7.0.2")