From 356575e99a7e3068eb84edf9d1674dbaee81ff04 Mon Sep 17 00:00:00 2001 From: Paulus Schoutsen Date: Mon, 6 Jul 2026 02:22:29 -0400 Subject: [PATCH 1/3] fix(ota): align nRF DFU MAC increment to nrf-ota no-carry convention + add discovery fallback Two DFU-rediscovery issues in the nRF OTA path: - `_increment_mac` used a full 48-bit carry, disagreeing with the Nordic-reference convention (a uint8 increment of the last octet: wrap 0xFF -> 0x00, no carry) that the standalone nrf-ota scanner uses. Align py-opendisplay to the no-carry convention so both implementations agree. - `find_nrf_dfu_device` matched only the MAC+1 address, so a wrong guess meant a silent scan miss. Add a fallback that also matches the Legacy DFU service UUID or a DFU-style advertised name. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_012g2e8mr132vcizx92WsgiR --- src/opendisplay/ota.py | 62 ++++++++++++++++++++-------- tests/unit/test_ota.py | 61 +++++++++++++++++++++++---- tests/unit/test_ota_mac_increment.py | 15 ++++--- 3 files changed, 108 insertions(+), 30 deletions(-) diff --git a/src/opendisplay/ota.py b/src/opendisplay/ota.py index 080b304..439a250 100644 --- a/src/opendisplay/ota.py +++ b/src/opendisplay/ota.py @@ -3,23 +3,42 @@ from __future__ import annotations import asyncio -from collections.abc import Callable +from collections.abc import Callable, Sequence from typing import TYPE_CHECKING from .exceptions import OTAError +# Nordic Legacy DFU GATT service, advertised while in bootloader mode. Kept in +# sync with ``nrf_ota._const.LEGACY_DFU_SERVICE_UUID`` and duplicated here so +# ``find_nrf_dfu_device`` doesn't need the optional ``nrf-ota`` dependency. +LEGACY_DFU_SERVICE_UUID = "00001530-1212-efde-1523-785feabcd123" -def _increment_mac(address: str) -> str: - """Return the BLE MAC address incremented by 1, carrying across octets. - The nRF DFU bootloader advertises at ``original + 1``. Incrementing only the - last octet (``+1 & 0xFF``) fails to carry when it is 0xFF (e.g. ...:FF -> ...:00 - instead of rolling into the previous octet), causing a silent 30 s scan miss. +def _increment_mac(address: str) -> str: + """Return the BLE MAC with its last octet incremented by 1 (no carry). + + Nordic DFU bootloaders advertise at ``original + 1``, incrementing only the + last octet as a ``uint8``: it wraps 0xFF -> 0x00 without carrying into the + previous octet (e.g. ``AA:BB:CC:DD:EE:FF`` -> ``AA:BB:CC:DD:EE:00``). This + matches the standalone ``nrf-ota`` scanner's convention and the documented + Nordic bootloader behaviour. The exact wrap behaviour should be validated on + real hardware; the name/service-UUID fallback in ``find_nrf_dfu_device`` + keeps discovery robust even if this guess is off. """ parts = address.upper().split(":") - mac_int = int("".join(parts), 16) - mac_int = (mac_int + 1) & 0xFFFFFFFFFFFF - return ":".join(f"{(mac_int >> (8 * (5 - i))) & 0xFF:02X}" for i in range(6)) + parts[-1] = f"{(int(parts[-1], 16) + 1) & 0xFF:02X}" + return ":".join(parts) + + +def _is_dfu_advertisement(name: str = "", service_uuids: Sequence[str] = ()) -> bool: + """Return True if *name* or *service_uuids* indicate DFU bootloader mode. + + Mirrors ``nrf_ota.scan._is_dfu_advertisement`` so discovery matches the same + devices regardless of which implementation drives the scan. + """ + return any(x in name.upper() for x in ("ADADFU", "DFUTARG", "DFU")) or any( + LEGACY_DFU_SERVICE_UUID.lower() in s.lower() for s in service_uuids + ) if TYPE_CHECKING: @@ -178,9 +197,11 @@ async def find_nrf_dfu_device(original_address: str) -> BLEDevice | None: """Poll the BLE scanner for an nRF DFU-mode device. Call this after ``OpenDisplayDevice.trigger_dfu_bootloader()`` disconnects. - Checks MAC+1 first (Nordic DFU bootloaders commonly increment the last - byte of the address). Falls back to the original address after 10 s in - case this particular bootloader keeps the same address. + Checks MAC+1 first (Nordic DFU bootloaders increment the last byte of the + address, wrapping without carry). As a fallback it also matches any device + advertising the Legacy DFU service UUID or a DFU-style name, so discovery + still succeeds if the MAC+1 guess is off. Falls back to the original address + after 10 s in case this particular bootloader keeps the same address. Works in both plain bleak environments and HA's cached scanner — in HA, BleakScanner.discover() returns the passive-scan cache, so repeated calls @@ -206,10 +227,17 @@ async def find_nrf_dfu_device(original_address: str) -> BLEDevice | None: if attempt >= 5: candidates.append(original_address.upper()) - devices = await BleakScanner.discover(timeout=0.1) - addr_map = {dev.address.upper(): dev for dev in devices} - for addr in candidates: - if addr in addr_map: - return addr_map[addr] + # return_adv=True gives the advertisement data (service UUIDs + local + # name) needed for the DFU service-UUID/name fallback below. + discovered = await BleakScanner.discover(timeout=0.1, return_adv=True) + for device, adv in discovered.values(): + if device.address.upper() in candidates: + return device + # Fallback: a genuine DFU advertisement is safe to return even before + # the 10 s mark — the stale app-mode entry carries neither the DFU + # service UUID nor a DFU name, so it can't be matched here by mistake. + name = adv.local_name or device.name or "" + if _is_dfu_advertisement(name, adv.service_uuids or []): + return device return None diff --git a/tests/unit/test_ota.py b/tests/unit/test_ota.py index 407cb8c..8030f42 100644 --- a/tests/unit/test_ota.py +++ b/tests/unit/test_ota.py @@ -27,12 +27,25 @@ def _make_ble_device(address: str = "AA:BB:CC:DD:EE:FF") -> MagicMock: # --------------------------------------------------------------------------- -def _make_scanner_device(address: str) -> MagicMock: +def _make_scanner_device(address: str, name: str | None = None) -> MagicMock: dev = MagicMock() dev.address = address + dev.name = name return dev +def _make_adv(local_name: str | None = None, service_uuids: list[str] | None = None) -> MagicMock: + adv = MagicMock() + adv.local_name = local_name + adv.service_uuids = service_uuids or [] + return adv + + +def _discovered(*pairs: tuple[MagicMock, MagicMock]) -> dict[str, tuple[MagicMock, MagicMock]]: + """Build the ``address -> (device, adv)`` dict that discover(return_adv=True) returns.""" + return {dev.address: (dev, adv) for dev, adv in pairs} + + @pytest.mark.asyncio async def test_find_nrf_dfu_device_mac_plus1() -> None: """Device found at MAC+1 on the first attempt.""" @@ -42,7 +55,7 @@ async def test_find_nrf_dfu_device_mac_plus1() -> None: patch("opendisplay.ota.asyncio.sleep", new=AsyncMock()), patch("bleak.BleakScanner") as scanner_cls, ): - scanner_cls.discover = AsyncMock(return_value=[dfu_dev]) + scanner_cls.discover = AsyncMock(return_value=_discovered((dfu_dev, _make_adv()))) result = await find_nrf_dfu_device("AA:BB:CC:DD:EE:01") assert result is dfu_dev @@ -54,9 +67,9 @@ async def test_find_nrf_dfu_device_original_address_after_5_attempts() -> None: original_dev = _make_scanner_device("AA:BB:CC:DD:EE:01") attempt = 0 - async def _discover(timeout: float) -> list[MagicMock]: + async def _discover(timeout: float, return_adv: bool = False) -> dict[str, tuple[MagicMock, MagicMock]]: nonlocal attempt - result = [original_dev] if attempt >= 5 else [] + result = _discovered((original_dev, _make_adv())) if attempt >= 5 else {} attempt += 1 return result @@ -78,7 +91,7 @@ async def test_find_nrf_dfu_device_not_found_returns_none() -> None: patch("opendisplay.ota.asyncio.sleep", new=AsyncMock()), patch("bleak.BleakScanner") as scanner_cls, ): - scanner_cls.discover = AsyncMock(return_value=[]) + scanner_cls.discover = AsyncMock(return_value={}) result = await find_nrf_dfu_device("AA:BB:CC:DD:EE:01") assert result is None @@ -86,19 +99,51 @@ async def test_find_nrf_dfu_device_not_found_returns_none() -> None: @pytest.mark.asyncio async def test_find_nrf_dfu_device_mac_plus1_wraps_ff() -> None: - """MAC+1 carries across octets: EE:FF → EF:00 (not EE:00).""" - dfu_dev = _make_scanner_device("AA:BB:CC:DD:EF:00") + """MAC+1 wraps the last octet without carry: EE:FF → EE:00 (not EF:00).""" + dfu_dev = _make_scanner_device("AA:BB:CC:DD:EE:00") with ( patch("opendisplay.ota.asyncio.sleep", new=AsyncMock()), patch("bleak.BleakScanner") as scanner_cls, ): - scanner_cls.discover = AsyncMock(return_value=[dfu_dev]) + scanner_cls.discover = AsyncMock(return_value=_discovered((dfu_dev, _make_adv()))) result = await find_nrf_dfu_device("AA:BB:CC:DD:EE:FF") assert result is dfu_dev +@pytest.mark.asyncio +async def test_find_nrf_dfu_device_service_uuid_fallback() -> None: + """A device advertising the Legacy DFU service UUID is found even if its MAC doesn't match.""" + dfu_dev = _make_scanner_device("11:22:33:44:55:66") # not MAC+1 of the original + adv = _make_adv(service_uuids=["00001530-1212-efde-1523-785feabcd123"]) + + with ( + patch("opendisplay.ota.asyncio.sleep", new=AsyncMock()), + patch("bleak.BleakScanner") as scanner_cls, + ): + scanner_cls.discover = AsyncMock(return_value=_discovered((dfu_dev, adv))) + result = await find_nrf_dfu_device("AA:BB:CC:DD:EE:01") + + assert result is dfu_dev + + +@pytest.mark.asyncio +async def test_find_nrf_dfu_device_name_fallback() -> None: + """A device with a DFU-style advertised name is found even if its MAC doesn't match.""" + dfu_dev = _make_scanner_device("11:22:33:44:55:66") + adv = _make_adv(local_name="DfuTarg") + + with ( + patch("opendisplay.ota.asyncio.sleep", new=AsyncMock()), + patch("bleak.BleakScanner") as scanner_cls, + ): + scanner_cls.discover = AsyncMock(return_value=_discovered((dfu_dev, adv))) + result = await find_nrf_dfu_device("AA:BB:CC:DD:EE:01") + + assert result is dfu_dev + + # --------------------------------------------------------------------------- # perform_* wrappers — optional-dependency import guards # --------------------------------------------------------------------------- diff --git a/tests/unit/test_ota_mac_increment.py b/tests/unit/test_ota_mac_increment.py index fe7b98b..0f0be65 100644 --- a/tests/unit/test_ota_mac_increment.py +++ b/tests/unit/test_ota_mac_increment.py @@ -1,4 +1,9 @@ -"""Test nRF DFU MAC increment carries across octets (§4).""" +"""Test nRF DFU MAC increment wraps the last octet without carry (§4). + +Nordic DFU bootloaders advertise at ``original + 1``, incrementing only the last +octet as a ``uint8`` (wrap 0xFF -> 0x00, no carry into the previous octet). This +matches the standalone ``nrf-ota`` scanner's convention. +""" from __future__ import annotations @@ -11,10 +16,10 @@ ("addr", "expected"), [ ("AA:BB:CC:DD:EE:01", "AA:BB:CC:DD:EE:02"), - ("AA:BB:CC:DD:EE:FF", "AA:BB:CC:DD:EF:00"), # carry into previous octet - ("AA:BB:CC:DD:FF:FF", "AA:BB:CC:DE:00:00"), # double carry - ("FF:FF:FF:FF:FF:FF", "00:00:00:00:00:00"), # wraps around + ("AA:BB:CC:DD:EE:FF", "AA:BB:CC:DD:EE:00"), # wraps, no carry + ("AA:BB:CC:DD:FF:FF", "AA:BB:CC:DD:FF:00"), # only the last octet wraps + ("FF:FF:FF:FF:FF:FF", "FF:FF:FF:FF:FF:00"), # previous octets untouched ], ) -def test_increment_mac_carries(addr: str, expected: str) -> None: +def test_increment_mac_no_carry(addr: str, expected: str) -> None: assert _increment_mac(addr) == expected From 5d1dde53fe4230bc056aa683e70e414fc514bb0b Mon Sep 17 00:00:00 2001 From: Paulus Schoutsen Date: Mon, 6 Jul 2026 02:25:23 -0400 Subject: [PATCH 2/3] fix(ota): drop name-based selection in find_nrf_dfu_device (match nrf-ota #7) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Selecting a DFU target by a name merely containing "DFU"/"DfuTarg" picks the wrong tag when two devices are in bootloader mode at once — the core of finding MJ-23. Align with nrf-ota #7: select only by MAC+1 OR the Legacy DFU service UUID. Replace `_is_dfu_advertisement` with `_advertises_dfu_service` (UUID-only) and update the fallback test to assert a DfuTarg-named advert with no matching UUID/MAC is NOT selected. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_012g2e8mr132vcizx92WsgiR --- src/opendisplay/ota.py | 38 ++++++++++++++++++++------------------ tests/unit/test_ota.py | 17 +++++++++++------ 2 files changed, 31 insertions(+), 24 deletions(-) diff --git a/src/opendisplay/ota.py b/src/opendisplay/ota.py index 439a250..8c497d7 100644 --- a/src/opendisplay/ota.py +++ b/src/opendisplay/ota.py @@ -30,15 +30,16 @@ def _increment_mac(address: str) -> str: return ":".join(parts) -def _is_dfu_advertisement(name: str = "", service_uuids: Sequence[str] = ()) -> bool: - """Return True if *name* or *service_uuids* indicate DFU bootloader mode. - - Mirrors ``nrf_ota.scan._is_dfu_advertisement`` so discovery matches the same - devices regardless of which implementation drives the scan. +def _advertises_dfu_service(service_uuids: Sequence[str]) -> bool: + """Return True if *service_uuids* contain the Legacy DFU service UUID. + + Mirrors the service-UUID half of ``nrf_ota.scan._is_dfu_advertisement``. + Selection deliberately does NOT match on device name: a name that merely + contains "DFU"/"DfuTarg" would pick the wrong tag when two devices are in + bootloader mode at once (finding MJ-23), so only the unique per-device MAC+1 + or the DFU service UUID are used to select the target. """ - return any(x in name.upper() for x in ("ADADFU", "DFUTARG", "DFU")) or any( - LEGACY_DFU_SERVICE_UUID.lower() in s.lower() for s in service_uuids - ) + return any(LEGACY_DFU_SERVICE_UUID.lower() in s.lower() for s in service_uuids) if TYPE_CHECKING: @@ -199,9 +200,11 @@ async def find_nrf_dfu_device(original_address: str) -> BLEDevice | None: Call this after ``OpenDisplayDevice.trigger_dfu_bootloader()`` disconnects. Checks MAC+1 first (Nordic DFU bootloaders increment the last byte of the address, wrapping without carry). As a fallback it also matches any device - advertising the Legacy DFU service UUID or a DFU-style name, so discovery - still succeeds if the MAC+1 guess is off. Falls back to the original address - after 10 s in case this particular bootloader keeps the same address. + advertising the Legacy DFU service UUID, so discovery still succeeds if the + MAC+1 guess is off. Selection never matches on device name — a DFU-ish name + would pick the wrong tag when two are in bootloader mode (finding MJ-23). + Falls back to the original address after 10 s in case this particular + bootloader keeps the same address. Works in both plain bleak environments and HA's cached scanner — in HA, BleakScanner.discover() returns the passive-scan cache, so repeated calls @@ -227,17 +230,16 @@ async def find_nrf_dfu_device(original_address: str) -> BLEDevice | None: if attempt >= 5: candidates.append(original_address.upper()) - # return_adv=True gives the advertisement data (service UUIDs + local - # name) needed for the DFU service-UUID/name fallback below. + # return_adv=True gives the advertisement data (service UUIDs) needed for + # the DFU service-UUID fallback below. discovered = await BleakScanner.discover(timeout=0.1, return_adv=True) for device, adv in discovered.values(): if device.address.upper() in candidates: return device - # Fallback: a genuine DFU advertisement is safe to return even before - # the 10 s mark — the stale app-mode entry carries neither the DFU - # service UUID nor a DFU name, so it can't be matched here by mistake. - name = adv.local_name or device.name or "" - if _is_dfu_advertisement(name, adv.service_uuids or []): + # Fallback: a device advertising the Legacy DFU service is safe to + # return even before the 10 s mark — the stale app-mode entry doesn't + # expose that service, so it can't be matched here by mistake. + if _advertises_dfu_service(adv.service_uuids or []): return device return None diff --git a/tests/unit/test_ota.py b/tests/unit/test_ota.py index 8030f42..0918170 100644 --- a/tests/unit/test_ota.py +++ b/tests/unit/test_ota.py @@ -129,19 +129,24 @@ async def test_find_nrf_dfu_device_service_uuid_fallback() -> None: @pytest.mark.asyncio -async def test_find_nrf_dfu_device_name_fallback() -> None: - """A device with a DFU-style advertised name is found even if its MAC doesn't match.""" - dfu_dev = _make_scanner_device("11:22:33:44:55:66") - adv = _make_adv(local_name="DfuTarg") +async def test_find_nrf_dfu_device_dfu_name_alone_not_selected() -> None: + """A DFU-ish name with no matching UUID/MAC must NOT be selected (finding MJ-23). + + Selecting on a name that merely contains "DFU"/"DfuTarg" picks the wrong tag + when two devices are in bootloader mode at once, so name is never a + selection discriminator — only MAC+1 or the DFU service UUID are. + """ + dfu_named = _make_scanner_device("11:22:33:44:55:66", name="DfuTarg") + adv = _make_adv(local_name="DfuTarg") # no DFU service UUID advertised with ( patch("opendisplay.ota.asyncio.sleep", new=AsyncMock()), patch("bleak.BleakScanner") as scanner_cls, ): - scanner_cls.discover = AsyncMock(return_value=_discovered((dfu_dev, adv))) + scanner_cls.discover = AsyncMock(return_value=_discovered((dfu_named, adv))) result = await find_nrf_dfu_device("AA:BB:CC:DD:EE:01") - assert result is dfu_dev + assert result is None # --------------------------------------------------------------------------- From 028e7fa149ac88df538259724fb83572a828ed16 Mon Sep 17 00:00:00 2001 From: Paulus Schoutsen Date: Mon, 6 Jul 2026 03:05:50 -0400 Subject: [PATCH 3/3] style: fix lint Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_012g2e8mr132vcizx92WsgiR --- src/opendisplay/ota.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/opendisplay/ota.py b/src/opendisplay/ota.py index 8c497d7..54f0dfd 100644 --- a/src/opendisplay/ota.py +++ b/src/opendisplay/ota.py @@ -77,7 +77,7 @@ async def perform_nrf_dfu( """ try: from bleak_retry_connector import BleakClientWithServiceCache, establish_connection - from nrf_ota._const import DEFAULT_PRN, LEGACY_DFU_SERVICE_UUID, TYPE_APPLICATION + from nrf_ota._const import DEFAULT_PRN, TYPE_APPLICATION from nrf_ota._zip import _parse_zip_bytes from nrf_ota.dfu import LegacyDFU except ImportError as exc: