EBL parsing (#11)

* Parse EBL images

* Get flashing working for HUSBZB-1

* Send a newline before `EBL info`

* Delay for 4s before launching bootloader communication

* Use correct data type when comparing regex match

* Remove `setup.cfg`

* Handle invalid firmware with a better error

* Remove all Ember/Gecko bootloader logic

* Increase EZSP bootloader launch delay to 5s

* Use millisecond logging

* Make 5s a constant

tests/test_firmware.py

@@ -0,0 +1,55 @@
+import pathlib
+
+import pytest
+
+from universal_silabs_flasher import firmware
+from universal_silabs_flasher.common import Version
+
+FIRMWARES_DIR = pathlib.Path(__file__).parent / "firmwares"
+
+
+def test_firmware_ebl_valid():
+    data = (FIRMWARES_DIR / "ncp-uart-sw-6.4.1.ebl").read_bytes()
+    fw = firmware.parse_firmware_image(data)
+
+    assert isinstance(fw, firmware.EBLImage)
+    assert fw.serialize() == data
+
+    with pytest.raises(KeyError):
+        fw.get_nabucasa_metadata()
+
+
+def test_firmware_gbl_valid_with_metadata():
+    data = (
+        FIRMWARES_DIR / "NabuCasa_SkyConnect_RCP_v4.1.3_rcp-uart-hw-802154_115200.gbl"
+    ).read_bytes()
+    fw = firmware.parse_firmware_image(data)
+
+    assert isinstance(fw, firmware.GBLImage)
+    assert fw.serialize() == data
+    assert fw.get_nabucasa_metadata() == firmware.NabuCasaMetadata(
+        metadata_version=1,
+        sdk_version=Version("4.1.3"),
+        ezsp_version=None,
+        fw_type=firmware.FirmwareImageType.RCP_UART_802154,
+        ot_rcp_version=None,
+        baudrate=None,
+        original_json={
+            "metadata_version": 1,
+            "sdk_version": "4.1.3",
+            "fw_type": "rcp-uart-802154",
+        },
+    )
+
+
+def test_firmware_gbl_valid_no_metadata():
+    data = (
+        FIRMWARES_DIR / "NabuCasa_EZSP_v6.10.3.0_PB32_ncp-uart-hw_115200.gbl"
+    ).read_bytes()
+    fw = firmware.parse_firmware_image(data)
+
+    assert isinstance(fw, firmware.GBLImage)
+    assert fw.serialize() == data
+
+    with pytest.raises(KeyError):
+        fw.get_nabucasa_metadata()

universal_silabs_flasher/common.py

@@ -57,6 +57,18 @@ def crc16_kermit(data: bytes) -> int:
     return CRC_KERMIT.checksum(data)
 
 
+def pad_to_multiple(data: bytes, multiple: int, padding: bytes) -> bytes:
+    assert len(padding) == 1
+
+    if len(data) % multiple == 0:
+        return data
+
+    num_complete_blocks = len(data) // multiple
+    padded_size = multiple * (num_complete_blocks + 1)
+
+    return data + padding * (padded_size - len(data))
+
+
 class BufferTooShort(Exception):
     """Protocol buffer requires more data to parse a packet."""
 

universal_silabs_flasher/cpc_types.py

@@ -1,10 +1,5 @@
 import zigpy.types as zigpy_t
 
-
-class enum32(zigpy_t.enum_factory(zigpy_t.uint32_t)):  # type: ignore
-    pass
-
-
 FLAG = zigpy_t.uint8_t(0x14)
 
 
@@ -50,7 +45,7 @@ class UnnumberedFrameCommandId(zigpy_t.enum8):
     INVALID = 0xFF
 
 
-class PropertyId(enum32):
+class PropertyId(zigpy_t.enum32):
     LAST_STATUS = 0x0000
     PROTOCOL_VERSION = 0x0001
     CAPABILITIES = 0x0002
@@ -326,12 +321,12 @@ class PropertyId(enum32):
     ENDPOINT_STATES = 0x1100
 
 
-class RebootMode(enum32):
+class RebootMode(zigpy_t.enum32):
     APPLICATION = 0
     BOOTLOADER = 1
 
 
-class Status(enum32):
+class Status(zigpy_t.enum32):
     # Operation has completed successfully.
     OK = 0
     # Operation has failed for some undefined reason.

universal_silabs_flasher/gbl.py → universal_silabs_flasher/firmware.py

@@ -5,18 +5,18 @@
 import logging
 import typing
 
-from zigpy.ota.validators import parse_silabs_gbl
+import zigpy.types as zigpy_t
+from zigpy.ota.validators import ValidationError, parse_silabs_ebl, parse_silabs_gbl
 
-from .common import Version
+from .common import Version, pad_to_multiple
 from .const import FirmwareImageType
-from .cpc_types import enum32
 
 _LOGGER = logging.getLogger(__name__)
 
 NABUCASA_METADATA_VERSION = 1
 
 
-class TagId(enum32):
+class GBLTagId(zigpy_t.enum32):
     # First tag in the file. The header tag contains the version number of the GBL file
     # specification, and flags indicating the type of GBL file – whether it is signed
     # or encrypted.
@@ -49,6 +49,19 @@ class TagId(enum32):
     END = 0xFC0404FC
 
 
+class EBLTagId(zigpy_t.enum16):
+    # TODO: flip the endianness
+    HEADER = 0x0000
+    PROG = 0x01FE
+    MFGPROG = 0xFE02
+    ERASEPROG = 0x03FD
+    END = 0x04FC
+    ENC_HEADER = 0x05FB
+    ENC_INIT = 0x06FA
+    ENC_EBL_DATA = 0x07F9
+    ENC_MAC = 0x09F7
+
+
 @dataclasses.dataclass(frozen=True)
 class NabuCasaMetadata:
     metadata_version: int
@@ -105,9 +118,25 @@ def from_json(cls, obj: dict[str, typing.Any]) -> NabuCasaMetadata:
 
 
 @dataclasses.dataclass(frozen=True)
-class GBLImage:
-    tags: list[tuple[TagId, bytes]]
+class FirmwareImage:
+    tags: list[tuple[GBLTagId, bytes]]
+
+    @classmethod
+    def from_bytes(cls, data: bytes) -> FirmwareImage:
+        raise NotImplementedError()
+
+    def serialize(self) -> bytes:
+        raise NotImplementedError()
+
+    def get_first_tag(self, tag_id: GBLTagId) -> bytes:
+        try:
+            return next(v for t, v in self.tags if t == tag_id)
+        except StopIteration:
+            raise KeyError(f"No tag with id {tag_id!r} exists")
+
 
+@dataclasses.dataclass(frozen=True)
+class GBLImage(FirmwareImage):
     @classmethod
     def from_bytes(cls, data: bytes) -> GBLImage:
         if isinstance(data, memoryview):
@@ -116,26 +145,62 @@ def from_bytes(cls, data: bytes) -> GBLImage:
         tags = []
 
         for tag_bytes, value in parse_silabs_gbl(data):
-            tag, _ = TagId.deserialize(tag_bytes)
+            tag, _ = GBLTagId.deserialize(tag_bytes)
             tags.append((tag, value))
 
         return cls(tags=tags)
 
     def serialize(self) -> bytes:
-        return b"".join(
-            [
-                tag_id.serialize() + len(value).to_bytes(4, "little") + value
-                for tag_id, value in self.tags
-            ]
+        return pad_to_multiple(
+            b"".join(
+                [
+                    tag_id.serialize() + len(value).to_bytes(4, "little") + value
+                    for tag_id, value in self.tags
+                ]
+            ),
+            4,
+            b"\xFF",
         )
 
-    def get_first_tag(self, tag_id: TagId) -> bytes:
-        try:
-            return next(v for t, v in self.tags if t == tag_id)
-        except StopIteration:
-            raise KeyError(f"No tag with id {tag_id!r} exists")
-
     def get_nabucasa_metadata(self) -> NabuCasaMetadata:
-        metadata = self.get_first_tag(TagId.METADATA)
+        metadata = self.get_first_tag(GBLTagId.METADATA)
 
         return NabuCasaMetadata.from_json(json.loads(metadata))
+
+
+@dataclasses.dataclass(frozen=True)
+class EBLImage(FirmwareImage):
+    @classmethod
+    def from_bytes(cls, data: bytes) -> EBLImage:
+        tags = []
+
+        for tag_bytes, value in parse_silabs_ebl(data):
+            tag, _ = EBLTagId.deserialize(tag_bytes)
+            tags.append((tag, value))
+
+        return cls(tags=tags)
+
+    def serialize(self) -> bytes:
+        return pad_to_multiple(
+            b"".join(
+                [
+                    tag_id.serialize() + len(value).to_bytes(2, "big") + value
+                    for tag_id, value in self.tags
+                ]
+            ),
+            64,
+            b"\xFF",
+        )
+
+    def get_nabucasa_metadata(self) -> NabuCasaMetadata:
+        raise KeyError("Metadata not supported for EBL")
+
+
+def parse_firmware_image(data: bytes) -> FirmwareImage:
+    for fw_cls in [GBLImage, EBLImage]:
+        try:
+            return fw_cls.from_bytes(data)
+        except ValidationError:
+            pass
+
+    raise ValueError("Unknown firmware image type")

universal_silabs_flasher/flash.py

@@ -24,8 +24,8 @@
     ApplicationType,
     ResetTarget,
 )
+from .firmware import FirmwareImageType, parse_firmware_image
 from .flasher import Flasher
-from .gbl import FirmwareImageType, GBLImage
 from .xmodemcrc import BLOCK_SIZE as XMODEM_BLOCK_SIZE, ReceiverCancelled
 
 patch_pyserial_asyncio()
@@ -156,7 +156,15 @@ def main(
     probe_method: list[ApplicationType],
     bootloader_reset: str | None,
 ) -> None:
-    coloredlogs.install(level=LOG_LEVELS[min(len(LOG_LEVELS) - 1, verbose)])
+    coloredlogs.install(
+        fmt=(
+            "%(asctime)s.%(msecs)03d"
+            " %(hostname)s"
+            " %(name)s"
+            " %(levelname)s %(message)s"
+        ),
+        level=LOG_LEVELS[min(len(LOG_LEVELS) - 1, verbose)],
+    )
 
     # Override all application baudrates if a specific value is provided
     if ctx.get_parameter_source("baudrate") != click.core.ParameterSource.DEFAULT:
@@ -203,19 +211,19 @@ async def dump_gbl_metadata(ctx: click.Context, firmware: typing.BinaryIO) -> No
     firmware.close()
 
     try:
-        gbl_image = GBLImage.from_bytes(firmware_data)
+        fw_image = parse_firmware_image(firmware_data)
     except zigpy.ota.validators.ValidationError as e:
         raise click.ClickException(
-            f"{firmware.name!r} does not appear to be a valid GBL image: {e!r}"
+            f"{firmware.name!r} does not appear to be a valid firmware image: {e!r}"
         )
 
     try:
-        metadata = gbl_image.get_nabucasa_metadata()
+        metadata = fw_image.get_nabucasa_metadata()
     except KeyError:
         metadata_obj = None
     else:
         metadata_obj = metadata.original_json
-        _LOGGER.info("Extracted GBL metadata: %s", metadata)
+        _LOGGER.info("Extracted firmware metadata: %s", metadata)
 
     print(json.dumps(metadata_obj))
 
@@ -279,14 +287,14 @@ async def flash(
     firmware.close()
 
     try:
-        gbl_image = GBLImage.from_bytes(firmware_data)
-    except zigpy.ota.validators.ValidationError as e:
+        fw_image = parse_firmware_image(firmware_data)
+    except (zigpy.ota.validators.ValidationError, ValueError) as e:
         raise click.ClickException(
-            f"{firmware.name!r} does not appear to be a valid GBL image: {e!r}"
+            f"{firmware.name!r} does not appear to be a valid firmware image: {e!r}"
         )
 
     try:
-        metadata = gbl_image.get_nabucasa_metadata()
+        metadata = fw_image.get_nabucasa_metadata()
     except KeyError:
         metadata = None
     else:
@@ -413,7 +421,7 @@ async def flash(
     with pbar:
         try:
             await flasher.flash_firmware(
-                gbl_image,
+                fw_image,
                 run_firmware=True,
                 progress_callback=lambda current, _: pbar.update(XMODEM_BLOCK_SIZE),
             )