Validate GBL magic and size before erasing the app slot

src/silabs_ble_ota/_const.py

@@ -10,6 +10,17 @@
 OTA_CONTROL_UUID = "f7bf3564-fb6d-4e53-88a4-5e37e0326063"  # write 0x00 = start, 0x03 = finalize
 OTA_DATA_UUID = "984227f3-34fc-4045-a5d0-2c581f81a153"  # GBL data sink
 
+# Every GBL image begins with the header tag GBL_TAG_ID_HEADER_V3 (0x03A617EB),
+# stored little-endian, so the first four bytes on disk are EB 17 A6 03. Checking
+# for it up front rejects a truncated download, an HTML error page, a .hex, or an
+# nRF .zip *before* the AppLoader erases the (single-bank, in-place) app slot.
+GBL_MAGIC = b"\xeb\x17\xa6\x03"
+
+# A valid GBL carries at least a header tag (tag + length + version + type) and an
+# end tag (tag + length + CRC) — well under 64 bytes. Anything smaller cannot be a
+# real firmware image, so reject it before erasing rather than bricking the device.
+MIN_GBL_SIZE = 64
+
 CHUNK_SIZE = 244  # bytes per data write
 APPLOADER_BOOT_DELAY = 6.0  # seconds to wait before the first connect attempt
 CONNECT_ATTEMPTS = 5  # establish_connection retries while the AppLoader boots

src/silabs_ble_ota/ota.py

@@ -19,6 +19,8 @@
     CONGESTION_RETRIES,
     CONNECT_ATTEMPTS,
     FAST_WINDOW,
+    GBL_MAGIC,
+    MIN_GBL_SIZE,
     OTA_CONTROL_UUID,
     OTA_DATA_UUID,
     WINDOW,
@@ -65,13 +67,31 @@ async def perform_silabs_ota(
             and corrupt the image. Defaults to ``False``.
 
     Raises:
-        SilabsOTAError: Connection or transfer failed, or the device is not in
-            Silabs OTA mode.
+        SilabsOTAError: ``gbl_bytes`` is not a valid GBL image (bad magic or too
+            small — checked before connecting, so the device is left untouched), the
+            connection or transfer failed, or the device is not in Silabs OTA mode.
     """
     log: LogCallback = on_log or (lambda _: None)
     file_size = len(gbl_bytes)
     window = FAST_WINDOW if fast else WINDOW
 
+    # Validate the image BEFORE connecting/erasing. AppLoader OTA is single-bank and
+    # in-place: writing 0x00 erases the app slot, so a bad image (truncated download,
+    # HTML error page, .hex, nRF .zip) would leave the device app-less. Reject it here
+    # while the app is still intact.
+    if file_size < MIN_GBL_SIZE:
+        raise SilabsOTAError(
+            f"Firmware is too small to be a valid GBL image ({file_size} bytes; "
+            f"expected at least {MIN_GBL_SIZE}). Refusing to erase the device — "
+            "check the firmware download."
+        )
+    if not gbl_bytes.startswith(GBL_MAGIC):
+        raise SilabsOTAError(
+            "Firmware is not a valid GBL image (missing GBL header magic "
+            f"{GBL_MAGIC.hex(' ').upper()}). Refusing to erase the device — the file may "
+            "be a truncated download, an error page, or the wrong format (.hex/.zip)."
+        )
+
     # Brief pause to let the device begin booting into the AppLoader before the
     # first connect. establish_connection then retries the connect itself —
     # failed/incomplete connects are harmless because the AppLoader only arms its

tests/test_ota.py

@@ -11,6 +11,8 @@
     APPLOADER_BOOT_DELAY,
     CHUNK_SIZE,
     FAST_WINDOW,
+    GBL_MAGIC,
+    MIN_GBL_SIZE,
     OTA_CONTROL_UUID,
     OTA_DATA_UUID,
     WINDOW,
@@ -21,6 +23,13 @@
 # ---------------------------------------------------------------------------
 
 
+def _gbl(size: int) -> bytes:
+    """A blob of exactly ``size`` bytes that passes GBL validation: header magic
+    followed by zero padding."""
+    assert size >= MIN_GBL_SIZE
+    return GBL_MAGIC + b"\x00" * (size - len(GBL_MAGIC))
+
+
 def _make_client(char_uuids: list[str]) -> MagicMock:
     """A connected client (as returned by establish_connection) whose services
     expose the given characteristic UUIDs. write_gatt_char/disconnect are async."""
@@ -53,7 +62,7 @@ def _patch_connect(client_or_exc: object) -> object:
 @pytest.mark.asyncio
 async def test_happy_path() -> None:
     """Full transfer: start write, data chunks, finalize write, disconnect."""
-    gbl = bytes(range(256)) * 2  # 512 bytes → 3 chunks of 244/244/24
+    gbl = _gbl(512)  # 512 bytes → 3 chunks of 244/244/24
     client = _make_client([OTA_CONTROL_UUID, OTA_DATA_UUID])
     progress: list[float] = []
     connect = AsyncMock(return_value=client)
@@ -88,7 +97,7 @@ async def test_happy_path() -> None:
 async def test_windowed_flow_control() -> None:
     """Data chunks sync (response=True) every WINDOW chunks and on the last chunk."""
     n_chunks = WINDOW * 2 + 3
-    gbl = b"\x00" * (CHUNK_SIZE * n_chunks)
+    gbl = _gbl(CHUNK_SIZE * n_chunks)
     client = _make_client([OTA_CONTROL_UUID, OTA_DATA_UUID])
 
     with patch("silabs_ble_ota.ota.asyncio.sleep", new=AsyncMock()), _patch_connect(client):
@@ -106,7 +115,7 @@ async def test_fast_mode_uses_larger_window() -> None:
     """fast=True syncs only every FAST_WINDOW chunks (plus the last); the rest are
     write-without-response — the speedup over the proxy-safe every-chunk-acked mode."""
     n_chunks = FAST_WINDOW * 2 + 3
-    gbl = b"\x00" * (CHUNK_SIZE * n_chunks)
+    gbl = _gbl(CHUNK_SIZE * n_chunks)
     client = _make_client([OTA_CONTROL_UUID, OTA_DATA_UUID])
 
     with patch("silabs_ble_ota.ota.asyncio.sleep", new=AsyncMock()), _patch_connect(client):
@@ -135,7 +144,7 @@ async def write(uuid, data, response=False):  # noqa: ARG001
     client.write_gatt_char = AsyncMock(side_effect=write)
 
     with patch("silabs_ble_ota.ota.asyncio.sleep", new=AsyncMock()), _patch_connect(client):
-        await perform_silabs_ota(b"\x00" * (CHUNK_SIZE * 2), _make_ble_device())
+        await perform_silabs_ota(_gbl(CHUNK_SIZE * 2), _make_ble_device())
 
     assert fails["n"] == 2  # first data chunk resent twice then succeeded
 
@@ -147,7 +156,7 @@ async def test_waits_for_apploader_boot() -> None:
     client = _make_client([OTA_CONTROL_UUID, OTA_DATA_UUID])
 
     with patch("silabs_ble_ota.ota.asyncio.sleep", new=sleep_mock), _patch_connect(client):
-        await perform_silabs_ota(b"\x00" * 10, _make_ble_device())
+        await perform_silabs_ota(_gbl(MIN_GBL_SIZE), _make_ble_device())
 
     sleep_mock.assert_awaited_once_with(APPLOADER_BOOT_DELAY)
 
@@ -158,7 +167,7 @@ async def test_missing_control_char_raises() -> None:
     client = _make_client([OTA_DATA_UUID, "some-other-uuid"])
     with patch("silabs_ble_ota.ota.asyncio.sleep", new=AsyncMock()), _patch_connect(client):
         with pytest.raises(SilabsOTAError, match="not in Silabs OTA mode"):
-            await perform_silabs_ota(b"\x00" * 10, _make_ble_device())
+            await perform_silabs_ota(_gbl(MIN_GBL_SIZE), _make_ble_device())
     client.disconnect.assert_awaited_once()
 
 
@@ -168,7 +177,7 @@ async def test_missing_data_char_raises() -> None:
     client = _make_client([OTA_CONTROL_UUID])
     with patch("silabs_ble_ota.ota.asyncio.sleep", new=AsyncMock()), _patch_connect(client):
         with pytest.raises(SilabsOTAError, match="not in Silabs OTA mode"):
-            await perform_silabs_ota(b"\x00" * 10, _make_ble_device())
+            await perform_silabs_ota(_gbl(MIN_GBL_SIZE), _make_ble_device())
 
 
 @pytest.mark.asyncio
@@ -179,7 +188,7 @@ async def test_connect_failure_wrapped() -> None:
         _patch_connect(RuntimeError("connection refused")),
     ):
         with pytest.raises(SilabsOTAError, match="Could not connect to AppLoader"):
-            await perform_silabs_ota(b"\x00" * 10, _make_ble_device())
+            await perform_silabs_ota(_gbl(MIN_GBL_SIZE), _make_ble_device())
 
 
 @pytest.mark.asyncio
@@ -189,7 +198,7 @@ async def test_transfer_error_wrapped() -> None:
     client.write_gatt_char = AsyncMock(side_effect=RuntimeError("gatt write failed"))
     with patch("silabs_ble_ota.ota.asyncio.sleep", new=AsyncMock()), _patch_connect(client):
         with pytest.raises(SilabsOTAError, match="Silabs OTA failed"):
-            await perform_silabs_ota(b"\x00" * 10, _make_ble_device())
+            await perform_silabs_ota(_gbl(MIN_GBL_SIZE), _make_ble_device())
     client.disconnect.assert_awaited_once()
 
 
@@ -198,7 +207,7 @@ async def test_no_progress_callback() -> None:
     """Works without an on_progress callback."""
     client = _make_client([OTA_CONTROL_UUID, OTA_DATA_UUID])
     with patch("silabs_ble_ota.ota.asyncio.sleep", new=AsyncMock()), _patch_connect(client):
-        await perform_silabs_ota(b"\x00" * 10, _make_ble_device(), on_progress=None)
+        await perform_silabs_ota(_gbl(MIN_GBL_SIZE), _make_ble_device(), on_progress=None)
 
 
 @pytest.mark.asyncio
@@ -207,5 +216,52 @@ async def test_log_callback() -> None:
     client = _make_client([OTA_CONTROL_UUID, OTA_DATA_UUID])
     logs: list[str] = []
     with patch("silabs_ble_ota.ota.asyncio.sleep", new=AsyncMock()), _patch_connect(client):
-        await perform_silabs_ota(b"\x00" * 10, _make_ble_device(), on_log=logs.append)
+        await perform_silabs_ota(_gbl(MIN_GBL_SIZE), _make_ble_device(), on_log=logs.append)
     assert any("AppLoader" in m for m in logs) and any("complete" in m.lower() for m in logs)
+
+
+# ---------------------------------------------------------------------------
+# GBL validation — must happen BEFORE connecting/erasing (single-bank, in-place)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_rejects_wrong_magic_before_erase() -> None:
+    """A blob without the GBL header magic raises before any BLE connect/erase."""
+    connect = AsyncMock()
+    sleep = AsyncMock()
+    not_gbl = b"<html>error</html>" + b"\x00" * MIN_GBL_SIZE  # right size, wrong magic
+    with (
+        patch("silabs_ble_ota.ota.asyncio.sleep", new=sleep),
+        patch("silabs_ble_ota.ota.establish_connection", new=connect),
+    ):
+        with pytest.raises(SilabsOTAError, match="not a valid GBL image"):
+            await perform_silabs_ota(not_gbl, _make_ble_device())
+    connect.assert_not_awaited()  # never connected → device app slot untouched
+    sleep.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_rejects_too_short_before_erase() -> None:
+    """A too-short blob (even with the magic) raises before any BLE connect/erase."""
+    connect = AsyncMock()
+    with (
+        patch("silabs_ble_ota.ota.asyncio.sleep", new=AsyncMock()),
+        patch("silabs_ble_ota.ota.establish_connection", new=connect),
+    ):
+        with pytest.raises(SilabsOTAError, match="too small"):
+            await perform_silabs_ota(GBL_MAGIC, _make_ble_device())
+        with pytest.raises(SilabsOTAError, match="too small"):
+            await perform_silabs_ota(b"", _make_ble_device())
+    connect.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_valid_magic_passes_validation() -> None:
+    """A blob with the correct magic and sane size passes validation and transfers."""
+    client = _make_client([OTA_CONTROL_UUID, OTA_DATA_UUID])
+    with patch("silabs_ble_ota.ota.asyncio.sleep", new=AsyncMock()), _patch_connect(client):
+        await perform_silabs_ota(_gbl(MIN_GBL_SIZE), _make_ble_device())
+    # 0x00 start-erase was written → validation let the transfer proceed.
+    control = [c for c in client.write_gatt_char.call_args_list if c.args[0] == OTA_CONTROL_UUID]
+    assert control[0].args[1] == bytearray([0x00])