Log an error instead of crashing when loading state from a bad CC2531

Some users have been using CC2531 adapters that were flashed with
Z-Stack 3 *without being erased*.  Their TCLK NVRAM entry is too big but
Z-Stack seems to still run.

Author: puddly

tests/api/test_network_state.py

@@ -1,8 +1,16 @@
+import logging
 import dataclasses
 
 import pytest
 
-from ..conftest import ALL_DEVICES, FORMED_DEVICES, BaseZStack1CC2531
+from zigpy_znp.types.nvids import ExNvIds, OsalNvIds
+
+from ..conftest import (
+    ALL_DEVICES,
+    FORMED_DEVICES,
+    BaseZStack1CC2531,
+    FormedZStack3CC2531,
+)
 
 
 @pytest.mark.parametrize("to_device", ALL_DEVICES)
@@ -35,3 +43,19 @@ async def test_state_transfer(from_device, to_device, make_connected_znp):
         assert formed_znp.network_info == empty_znp.network_info
 
     assert formed_znp.node_info == empty_znp.node_info
+
+
+@pytest.mark.parametrize("device", [FormedZStack3CC2531])
+async def test_broken_cc2531_load_state(device, make_connected_znp, caplog):
+    znp, znp_server = await make_connected_znp(server_cls=device)
+
+    # "Bad" TCLK seed is a TCLK from Z-Stack 1 with the first 16 bytes overwritten
+    znp_server._nvram[ExNvIds.LEGACY][
+        OsalNvIds.TCLK_SEED
+    ] += b"liance092\x00\x00\x00\x00\x00\x00\x00"
+
+    caplog.set_level(logging.ERROR)
+    await znp.load_network_info()
+    assert "inconsistent" in caplog.text
+
+    znp.close()

zigpy_znp/api.py

@@ -142,10 +142,31 @@ async def load_network_info(self, *, load_devices=False):
             stack_specific=None,
         )
 
+        tclk_seed = None
+
         if self.version > 1.2:
-            tclk_seed = await self.nvram.osal_read(
-                OsalNvIds.TCLK_SEED, item_type=t.KeyData
-            )
+            try:
+                tclk_seed = await self.nvram.osal_read(
+                    OsalNvIds.TCLK_SEED, item_type=t.KeyData
+                )
+            except ValueError:
+                if self.version != 3.0:
+                    raise
+
+                # CC2531s that have been cross-flashed from 1.2 -> 3.0 can have NVRAM
+                # entries from both. Ignore deserialization length errors for the
+                # trailing data to allow them to be backed up.
+                tclk_seed_value = await self.nvram.osal_read(
+                    OsalNvIds.TCLK_SEED, item_type=t.Bytes
+                )
+
+                tclk_seed = self.nvram.deserialize(
+                    tclk_seed_value, t.KeyData, allow_trailing=True
+                )
+                LOGGER.error(
+                    "Your adapter's NVRAM is inconsistent! Perform a network backup,"
+                    " re-flash its firmware, and restore the backup."
+                )
 
             network_info.stack_specific = {
                 "zstack": {
@@ -155,7 +176,7 @@ async def load_network_info(self, *, load_devices=False):
 
         # This takes a few seconds
         if load_devices:
-            for dev in await security.read_devices(self):
+            for dev in await security.read_devices(self, tclk_seed=tclk_seed):
                 if dev.node_info.nwk == 0xFFFE:
                     dev = dev.replace(
                         node_info=dataclasses.replace(dev.node_info, nwk=None)

zigpy_znp/tools/network_backup.py

@@ -77,12 +77,10 @@ def zigpy_state_to_json_backup(
 
 async def backup_network(znp: ZNP) -> t.JSONType:
     try:
-        await znp.load_network_info()
+        await znp.load_network_info(load_devices=True)
     except ValueError as e:
         raise RuntimeError("Failed to load network info") from e
 
-    await znp.load_network_info(load_devices=True)
-
     obj = zigpy_state_to_json_backup(
         network_info=znp.network_info,
         node_info=znp.node_info,

zigpy_znp/znp/security.py

@@ -253,12 +253,9 @@ async def read_unhashed_link_keys(
         )
 
 
-async def read_devices(znp: ZNP) -> typing.Sequence[StoredDevice]:
-    tclk_seed = None
-
-    if znp.version > 1.2:
-        tclk_seed = await znp.nvram.osal_read(OsalNvIds.TCLK_SEED, item_type=t.KeyData)
-
+async def read_devices(
+    znp: ZNP, *, tclk_seed: t.KeyData | None
+) -> typing.Sequence[StoredDevice]:
     addr_mgr = await read_addr_manager_entries(znp)
     devices = {}