WIP - verify version and data hash

core/src/apps/management/change_language.py

@@ -68,9 +68,20 @@ class TranslationsHeader:
         except EOFError:
             raise DataError("Invalid header data")
 
+    def version_tuple(self) -> tuple[int, int, int]:
+        try:
+            version_parts = self.version.split(".")
+            major = int(version_parts[0])
+            minor = int(version_parts[1])
+            patch = int(version_parts[2])
+            return major, minor, patch
+        except (ValueError, IndexError):
+            raise DataError("Invalid header version")
+
 
 async def change_language(msg: ChangeLanguage) -> Success:
-    from trezor import translations
+    from trezor import translations, utils
+    from trezor.crypto.hashlib import sha256
     from trezor.messages import Success
 
     data_length = msg.data_length  # local_cache_attribute
@@ -86,35 +97,47 @@ async def change_language(msg: ChangeLanguage) -> Success:
     if data_length < _HEADER_SIZE:
         raise DataError("Translations too short")
 
-    data_left = data_length
-    offset = 0
-
     # Getting and parsing the header
-    header_data = await get_data_chunk(_HEADER_SIZE, offset)
+    header_data = await get_data_chunk(_HEADER_SIZE, 0)
     header = TranslationsHeader.from_bytes(header_data)
 
+    # Verifying header information
+    # TODO: verify the header signature (signature of sha256(header))
     if header.data_length + _HEADER_SIZE != data_length:
         raise DataError("Invalid header data length")
-
-    # TODO: verify the hash of the data (get all of them and hash them)
-    # TODO: verify the header signature (signature of sha256(header))
+    # TODO: how to handle the version updates - numbers have to be bumped in cs.json and others
+    # (or have this logic in a separate blob-creating tool)
+    if header.version_tuple() != (
+        utils.VERSION_MAJOR,
+        utils.VERSION_MINOR,
+        utils.VERSION_PATCH,
+    ):
+        raise DataError("Invalid translations version")
 
     # Confirm with user and wipe old data
     await _require_confirm_change_language(header.language)
     translations.wipe()
 
     # Write the header
-    translations.write(header_data, offset)
-    offset += len(header_data)
-    data_left -= len(header_data)
+    translations.write(header_data, 0)
 
     # Requesting the data in chunks and saving them
+    # Also checking the hash of the data for consistency
+    data_left = data_length - len(header_data)
+    offset = len(header_data)
+    hash_writer = utils.HashWriter(sha256())
     while data_left > 0:
         data_chunk = await get_data_chunk(data_left, offset)
         translations.write(data_chunk, offset)
+        hash_writer.write(data_chunk)
         data_left -= len(data_chunk)
         offset += len(data_chunk)
 
+    # When the data do not match the hash, wipe all the written translations
+    if hash_writer.get_digest() != header.data_hash:
+        translations.wipe()
+        raise DataError("Invalid data hash")
+
     return Success(message="Language changed")
 
 

python/src/trezorlib/translations.py

@@ -14,17 +14,17 @@ HeaderData = Dict[str, str]
 
 def blob_from_file(file: TextIO) -> bytes:
     data = json.load(file)
-    return _blob_from_dict(data)
+    return blob_from_dict(data)
 
 
 def blob_from_url(url: str) -> bytes:
     r = requests.get(url)
     r.raise_for_status()
     data = r.json()
-    return _blob_from_dict(data)
+    return blob_from_dict(data)
 
 
-def _blob_from_dict(data: Dict[str, Any]) -> bytes:
+def blob_from_dict(data: Dict[str, Any]) -> bytes:
     header: HeaderData = data["header"]
     translations: TranslationData = data["translations"]
     return _blob_from_data(header, translations)

tests/device_tests/test_language.py

@@ -14,6 +14,7 @@
 # You should have received a copy of the License along with this library.
 # If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
 
+import json
 from contextlib import contextmanager
 from pathlib import Path
 from typing import Generator
@@ -72,9 +73,7 @@ def test_change_language_errors(client: Client):
         assert client.features.language == "en-US"
 
         # TODO: invalid header
-        # TODO: invalid data hash
         # TODO: invalid signature
-        # TODO: invalid data-length
 
         # Translations too short
         with pytest.raises(
@@ -90,6 +89,48 @@ def test_change_language_errors(client: Client):
             device.change_language(client, language_data=(MAX_DATA_LENGTH + 1) * b"a")
         assert client.features.language == "en-US"
 
+        # Invalid header data length
+        with pytest.raises(
+            exceptions.TrezorFailure, match="Invalid header data length"
+        ), client:
+            with open(CS_JSON, "r") as f:
+                device.change_language(
+                    client, language_data=translations.blob_from_file(f) + b"abc"
+                )
+        assert client.features.language == "en-US"
+
+        # Invalid data hash
+        with pytest.raises(exceptions.TrezorFailure, match="Invalid data hash"), client:
+            with open(CS_JSON, "r") as f:
+                device.change_language(
+                    client, language_data=translations.blob_from_file(f)[:-4] + b"abcd"
+                )
+        assert client.features.language == "en-US"
+
+        # Invalid translations version
+        with pytest.raises(
+            exceptions.TrezorFailure, match="Invalid translations version"
+        ), client:
+            with open(CS_JSON, "r") as f:
+                data = json.load(f)
+            data["header"]["version"] = "3.5.4"
+            device.change_language(
+                client, language_data=translations.blob_from_dict(data)
+            )
+        assert client.features.language == "en-US"
+
+        # Invalid header version
+        with pytest.raises(
+            exceptions.TrezorFailure, match="Invalid header version"
+        ), client:
+            with open(CS_JSON, "r") as f:
+                data = json.load(f)
+            data["header"]["version"] = "ABC.XYZ.DEF"
+            device.change_language(
+                client, language_data=translations.blob_from_dict(data)
+            )
+        assert client.features.language == "en-US"
+
 
 def test_full_language_change(client: Client):
     with _set_english_return_back(client) as client: