Attempt to fix corrupted coordinator NVRAM during startup

Author: puddly

tests/application/test_startup.py

@@ -216,7 +216,7 @@ async def test_auto_form_unnecessary(device, make_application, mocker):
 
 
 @pytest.mark.parametrize("device", EMPTY_DEVICES)
-async def test_auto_form_necessary(device, make_application, mocker):
+async def test_auto_form_necessary(device, make_application):
     app, znp_server = make_application(server_cls=device)
 
     assert app.channel is None
@@ -238,3 +238,35 @@ async def test_auto_form_necessary(device, make_application, mocker):
     assert nvram[OsalNvIds.ZDO_DIRECT_CB] == t.Bool(True).serialize()
 
     await app.shutdown()
+
+
+@pytest.mark.parametrize("device", [FormedLaunchpadCC26X2R1])
+async def test_corrupted_nvram_startup_fixes(device, make_application, caplog):
+    app, znp_server = make_application(server_cls=device)
+
+    # NIB has a weird NWK update ID and is 110 bytes instead of 116??
+    znp_server._nvram[ExNvIds.LEGACY][OsalNvIds.NIB] = bytes.fromhex(
+        "100502330f33001e0000000105018f000700020d1e0000000b00000000000000000000005a9608"
+        "00000800000f0f04000100000001000000001dbb9f21004b120001000000000000000000000000"
+        "0000000000000000000000000000000000000000000000000f050001780a0100"
+    )
+
+    # One byte too short: should be 20 bytes, not 19
+    znp_server._nvram[ExNvIds.TCLK_TABLE][0x0000] = bytes.fromhex(
+        "21000000000000000000000000000000ff0000"
+    )
+
+    await app.startup()
+
+    assert "Correcting invalid NIB" in caplog.text
+    assert "Correcting invalid TCLK_TABLE" in caplog.text
+
+    # Correct settings are loaded from NIB. Confirmed with Wireshark.
+    assert app.nwk_update_id == 0
+    assert app.channel == 11
+    assert app.channels == t.Channels.from_channel_list([11])
+    assert app.extended_pan_id == t.EUI64.convert("00:12:4b:00:21:9f:bb:1d")
+    assert app.nwk == 0x0000
+    assert app.pan_id == 0x965A
+
+    await app.shutdown()

zigpy_znp/zigbee/application.py

@@ -26,6 +26,7 @@
 import zigpy_znp.config as conf
 import zigpy_znp.commands as c
 from zigpy_znp.api import ZNP
+from zigpy_znp.znp.utils import fix_misaligned_coordinator_nvram
 from zigpy_znp.exceptions import CommandNotRecognized, InvalidCommandResponse
 from zigpy_znp.types.nvids import OsalNvIds
 from zigpy_znp.zigbee.zdo_converters import ZDO_CONVERTERS
@@ -313,7 +314,15 @@ async def _startup(self, auto_form=False, force_form=False, read_only=False):
         if self.znp_config[conf.CONF_LED_MODE] is not None:
             await self._set_led_mode(led=0xFF, mode=self.znp_config[conf.CONF_LED_MODE])
 
-        await self._load_network_info()
+        try:
+            await self._load_network_info()
+        except ValueError:
+            LOGGER.warning(
+                "Failed to read network information. Attempting to fix NVRAM.",
+                exc_info=True,
+            )
+            await fix_misaligned_coordinator_nvram(self._znp)
+            await self._load_network_info()
 
         # Add the coordinator as a zigpy device. We do this up here because
         # `self._register_endpoint()` adds endpoints to this device object.

zigpy_znp/znp/utils.py

@@ -1,9 +1,12 @@
+import logging
 import dataclasses
 
 import zigpy_znp.types as t
 from zigpy_znp.exceptions import CommandNotRecognized
 from zigpy_znp.types.nvids import ExNvIds, OsalNvIds
 
+LOGGER = logging.getLogger(__name__)
+
 
 @dataclasses.dataclass(frozen=True)
 class NetworkInfo:
@@ -22,6 +25,57 @@ def replace(self, **kwargs):
         return dataclasses.replace(self, **kwargs)
 
 
+async def fix_misaligned_coordinator_nvram(znp) -> None:
+    """
+    Some users have coordinators with broken alignment in NVRAM:
+
+        "TCLK_TABLE": {
+            "0x0000": "21000000000000000000000000000000ff0000",  <-- missing a byte??
+            "0x0001": "00000000000000000000000000000000ff000000",
+
+    These issues need to be corrected before zigpy-znp can continue working with NVRAM.
+    """
+
+    if znp.version < 3.30:
+        return
+
+    try:
+        nib_data = await znp.nvram.osal_read(OsalNvIds.NIB, item_type=t.Bytes)
+        znp.nvram.deserialize(nib_data, item_type=t.NIB)
+    except KeyError:
+        pass
+    except ValueError:
+        LOGGER.warning("Correcting invalid NIB alignment: %s", nib_data)
+
+        nib = znp.nvram.deserialize(nib_data + b"\xFF" * 6, item_type=t.NIB)
+
+        if nib.nwkUpdateId == 0xFF:
+            nib.nwkUpdateId = 0
+
+        await znp.nvram.osal_write(OsalNvIds.NIB, nib, create=True)
+
+    offset = 0
+
+    async for data in znp.nvram.read_table(
+        item_id=ExNvIds.TCLK_TABLE,
+        item_type=t.Bytes,
+    ):
+        try:
+            znp.nvram.deserialize(data, item_type=t.TCLKDevEntry)
+        except ValueError:
+            LOGGER.warning(
+                "Correcting invalid TCLK_TABLE[0x%04X] entry: %s", offset, data
+            )
+
+            entry = znp.nvram.deserialize(data + b"\x00", item_type=t.TCLKDevEntry)
+
+            await znp.nvram.write(
+                item_id=ExNvIds.TCLK_TABLE, sub_id=offset, value=entry, create=True
+            )
+
+        offset += 1
+
+
 async def load_network_info(znp) -> NetworkInfo:
     """
     Loads low-level network information from NVRAM.