From 75dc2b7dd66bfdaaada727f7bedfccd35a0562e1 Mon Sep 17 00:00:00 2001 From: Michael Sauer Date: Wed, 16 Sep 2026 15:11:28 +0200 Subject: [PATCH] feat(firmware): Trailer, Versionsordnung, WLAN-Pruefung und Firmware-Katalog --- custom_components/ninlight/firmware.py | 196 +++++++++++++++++++++ custom_components/ninlight/urls.py | 34 ++++ tests_unit/images.py | 37 ++++ tests_unit/test_firmware.py | 231 +++++++++++++++++++++++++ tests_unit/test_urls.py | 46 +++++ 5 files changed, 544 insertions(+) create mode 100644 custom_components/ninlight/firmware.py create mode 100644 custom_components/ninlight/urls.py create mode 100644 tests_unit/images.py create mode 100644 tests_unit/test_firmware.py create mode 100644 tests_unit/test_urls.py diff --git a/custom_components/ninlight/firmware.py b/custom_components/ninlight/firmware.py new file mode 100644 index 0000000..c45552a --- /dev/null +++ b/custom_components/ninlight/firmware.py @@ -0,0 +1,196 @@ +"""Firmware images for ninlight lamps: trailer, version order, Wi-Fi check, catalog. + +Deliberately free of Home Assistant imports. Blocking functions are marked; callers run them in an +executor. Facts behind the constants: spec §2 (layout checked against a real image). +""" + +from __future__ import annotations + +from dataclasses import dataclass +import hashlib +import logging +from pathlib import Path +import re +import zlib + +_LOGGER = logging.getLogger(__name__) + +TRAILER_SIZE = 512 +TRAILER_MAGIC = b"NINL" +TRAILER_FORMAT = 1 +W600_HEADER_SIZE = 56 +GZIP_MAGIC = b"\x1f\x8b\x08" +# cmake/write_version.cmake cuts the version hard to 15 characters. +VERSION_FIELD_MAX = 15 + +_VERSION_RE = re.compile( + r"^(?P\d+)\.(?P\d+)\.(?P\d+)" + r"(?:-rc(?P\d+))?" + r"(?:-(?P\d+)(?P-g[0-9a-f]*)?)?" + r"(?:-d(?:i(?:r(?:t(?:y)?)?)?)?)?$" +) + + +class InvalidImageError(ValueError): + """The file is not a valid ninlight OTA image.""" + + +@dataclass(frozen=True, slots=True) +class FirmwareVersion: + """A parsed `git describe` version. The hash is ignored on purpose (usually truncated).""" + + base: tuple[int, int, int] + rc: int | None + commits: int + + @property + def sort_key(self) -> tuple[int, int, int, int, int, int]: + # A release candidate sorts before the release of the same base. + rc_rank, rc_number = (0, self.rc) if self.rc is not None else (1, 0) + return (*self.base, rc_rank, rc_number, self.commits) + + +def parse_version(text: str) -> FirmwareVersion | None: + match = _VERSION_RE.match(text) + if match is None: + return None + if len(text) == VERSION_FIELD_MAX and text[-1].isdigit() and match["hash"] is None: + # The cut may have landed inside the commit (or rc) number: not comparable. + return None + return FirmwareVersion( + (int(match["major"]), int(match["minor"]), int(match["patch"])), + int(match["rc"]) if match["rc"] is not None else None, + int(match["commits"]) if match["commits"] is not None else 0, + ) + + +def is_newer(candidate: str, installed: str) -> bool: + new, old = parse_version(candidate), parse_version(installed) + if new is None or old is None: + return False + return new.sort_key > old.sort_key + + +def parse_trailer(data: bytes) -> tuple[str, str]: + """Return (device_type, version) of a valid image or raise InvalidImageError.""" + if len(data) <= TRAILER_SIZE: + raise InvalidImageError("too short for a trailer") + payload, trailer = data[:-TRAILER_SIZE], data[-TRAILER_SIZE:] + if trailer[0:4] != TRAILER_MAGIC: + raise InvalidImageError("no NINL trailer") + if trailer[4] != TRAILER_FORMAT: + raise InvalidImageError(f"unsupported trailer format {trailer[4]}") + if trailer[37:69] != hashlib.sha256(payload).digest(): + raise InvalidImageError("sha256 mismatch") + device_type = _text_field(trailer[5:21]) + version = _text_field(trailer[21:37]) + if not device_type or not version: + raise InvalidImageError("empty device type or version") + return device_type, version + + +def _text_field(raw: bytes) -> str: + return raw.split(b"\x00", 1)[0].decode("ascii", errors="replace") + + +def contains_ssid(data: bytes, ssid: str) -> bool: + """True if the decompressed firmware carries `ssid` as a NUL-framed C string (spec §7).""" + if not ssid or len(data) <= TRAILER_SIZE: + return False + payload = data[:-TRAILER_SIZE] + start = payload.find(GZIP_MAGIC, W600_HEADER_SIZE) + if start < 0: + return False + try: + raw = zlib.decompressobj(31).decompress(payload[start:]) + except zlib.error: + return False + return b"\x00" + ssid.encode() + b"\x00" in raw + + +@dataclass(frozen=True, slots=True) +class FirmwareImage: + """One valid image in the firmware folder.""" + + path: Path + device_type: str + version: str + size: int + mtime_ns: int + + +def _sort_key(image: FirmwareImage) -> tuple[int, ...]: + version = parse_version(image.version) + return version.sort_key if version is not None else () + + +class FirmwareCatalog: + """All valid images in one folder. Not thread-safe: callers serialize scans.""" + + def __init__(self, directory: Path) -> None: + self.directory = directory + self.images: tuple[FirmwareImage, ...] = () + self._parsed: dict[tuple[str, int, int], FirmwareImage | None] = {} + self._ssid_hits: dict[tuple[str, int, int, str], bool] = {} + + def scan(self) -> tuple[FirmwareImage, ...]: + """Blocking. Invalid files are logged once per (path, mtime, size).""" + self.directory.mkdir(parents=True, exist_ok=True) + found: list[FirmwareImage] = [] + seen: set[tuple[str, int, int]] = set() + for path in sorted(self.directory.glob("*.img")): + try: + if not path.is_file(): + continue + stat = path.stat() + except OSError: + continue + key = (str(path), stat.st_mtime_ns, stat.st_size) + seen.add(key) + if key not in self._parsed: + try: + device_type, version = parse_trailer(path.read_bytes()) + except (OSError, InvalidImageError) as err: + _LOGGER.warning("Ignoring firmware file %s: %s", path.name, err) + self._parsed[key] = None + else: + self._parsed[key] = FirmwareImage( + path, device_type, version, stat.st_size, stat.st_mtime_ns + ) + image = self._parsed[key] + if image is not None: + found.append(image) + self._parsed = {k: v for k, v in self._parsed.items() if k in seen} + self._ssid_hits = {k: v for k, v in self._ssid_hits.items() if k[:3] in seen} + self.images = tuple(found) + return self.images + + def newer_images(self, device_type: str, installed: str) -> list[FirmwareImage]: + """Images for exactly this device type that are newer than `installed`, newest first.""" + candidates = [ + image + for image in self.images + if image.device_type == device_type and is_newer(image.version, installed) + ] + return sorted(candidates, key=_sort_key, reverse=True) + + def verify(self, image: FirmwareImage) -> bool: + """Blocking. The file is still there, unchanged and valid for the same type/version.""" + try: + stat = image.path.stat() + if (stat.st_mtime_ns, stat.st_size) != (image.mtime_ns, image.size): + return False + return parse_trailer(image.path.read_bytes()) == (image.device_type, image.version) + except OSError, InvalidImageError: + return False + + def contains_ssid(self, image: FirmwareImage, ssid: str) -> bool: + """Blocking, cached per (image file, ssid).""" + key = (str(image.path), image.mtime_ns, image.size, ssid) + if key not in self._ssid_hits: + try: + data = image.path.read_bytes() + except OSError: + return False + self._ssid_hits[key] = contains_ssid(data, ssid) + return self._ssid_hits[key] diff --git a/custom_components/ninlight/urls.py b/custom_components/ninlight/urls.py new file mode 100644 index 0000000..12fff61 --- /dev/null +++ b/custom_components/ninlight/urls.py @@ -0,0 +1,34 @@ +"""The address lamps use to reach Home Assistant. Deliberately free of Home Assistant imports.""" + +from __future__ import annotations + +from urllib.parse import urlsplit + +from .const import FIRMWARE_KEY_LENGTH, FIRMWARE_URL_PREFIX, MAX_LAMP_URL_LENGTH + + +def firmware_url(base: str, key: str) -> str: + return f"{base.rstrip('/')}{FIRMWARE_URL_PREFIX}{key}" + + +def validate_lamp_base_url(url: str) -> str | None: + """Return None if usable, else the error key for the options form.""" + try: + parts = urlsplit(url) + port_ok = parts.port is None or parts.port > 0 + except ValueError: + return "invalid_base_url" + if ( + parts.scheme != "http" + or not parts.hostname + or not port_ok + or parts.username + or parts.password + or parts.query + or parts.fragment + or parts.path not in ("", "/") + ): + return "invalid_base_url" + if len(firmware_url(url, "k" * FIRMWARE_KEY_LENGTH)) > MAX_LAMP_URL_LENGTH: + return "base_url_too_long" + return None diff --git a/tests_unit/images.py b/tests_unit/images.py new file mode 100644 index 0000000..3503b94 --- /dev/null +++ b/tests_unit/images.py @@ -0,0 +1,37 @@ +"""Synthetic ninlight OTA images: 56-byte W600 header + gzip stream + 512-byte NINL trailer. + +The layout mirrors ninlight-firmware/tools/mkimage/nl_mkimage.py and was checked against a real +image (spec §2). Tests build their own images instead of copying firmware binaries. +""" + +from __future__ import annotations + +import gzip +import hashlib + +CONTENT_WITH_SSID = b"\x00boot\x00ninIOT\x00ninlight-%s\x00" +CONTENT_WITHOUT_SSID = b"\x00boot\x00ninlight-%s\x00" + + +def build_payload(content: bytes) -> bytes: + return b"\xab" * 56 + gzip.compress(content, mtime=0) + + +def build_image( + device_type: str = "gu10_rgb", + version: str = "0.3.8-47-g71a57", + *, + content: bytes = CONTENT_WITH_SSID, + payload: bytes | None = None, + magic: bytes = b"NINL", + fmt: int = 1, + corrupt_hash: bool = False, +) -> bytes: + body = payload if payload is not None else build_payload(content) + trailer = bytearray(b"\xff" * 512) + trailer[0:4] = magic + trailer[4] = fmt + trailer[5:21] = device_type.encode().ljust(16, b"\x00") + trailer[21:37] = version.encode().ljust(16, b"\x00") + trailer[37:69] = bytes(32) if corrupt_hash else hashlib.sha256(body).digest() + return body + bytes(trailer) diff --git a/tests_unit/test_firmware.py b/tests_unit/test_firmware.py new file mode 100644 index 0000000..f75c3ea --- /dev/null +++ b/tests_unit/test_firmware.py @@ -0,0 +1,231 @@ +"""Trailer, version order, Wi-Fi credential check and the firmware catalog.""" + +from __future__ import annotations + +import gzip +from pathlib import Path + +import pytest + +from custom_components.ninlight import firmware +from custom_components.ninlight.firmware import ( + FirmwareCatalog, + FirmwareVersion, + InvalidImageError, + contains_ssid, + is_newer, + parse_trailer, + parse_version, +) +from tests_unit.images import CONTENT_WITH_SSID, CONTENT_WITHOUT_SSID, build_image + +# Strictly ascending. Every neighbour pair and every pair across the list is checked. +ORDER = [ + "0.3.8", + "0.3.8-36-gca9bb", + "0.3.8-47-g71a57", + "0.3.9-rc1", + "0.3.9-rc1-4-gab12", + "0.3.9-rc2", + "0.3.9", + "0.4.0-rc1", + "0.10.0", +] + + +@pytest.mark.parametrize(("older", "newer"), list(zip(ORDER, ORDER[1:], strict=False))) # noqa: RUF007 +def test_version_neighbours(older: str, newer: str) -> None: + assert is_newer(newer, older) + assert not is_newer(older, newer) + + +def test_version_order_holds_across_the_whole_list() -> None: + for i, older in enumerate(ORDER): + for newer in ORDER[i + 1 :]: + assert is_newer(newer, older), (newer, older) + + +@pytest.mark.parametrize( + ("text", "expected"), + [ + ("0.3.8-36-gca9bb", FirmwareVersion((0, 3, 8), None, 36)), + ("0.3.8-38-gd3cc9", FirmwareVersion((0, 3, 8), None, 38)), # exactly 15 chars, hash present + ("0.3.8-36-g", FirmwareVersion((0, 3, 8), None, 36)), # hash cut to nothing + ("0.3.8-36", FirmwareVersion((0, 3, 8), None, 36)), + ("0.3.8-3-gab-dir", FirmwareVersion((0, 3, 8), None, 3)), # "-dirty" cut to "-dir" + ("0.3.8-dirty", FirmwareVersion((0, 3, 8), None, 0)), + ("0.4.0-rc1-12-ga", FirmwareVersion((0, 4, 0), 1, 12)), # 15 chars, hash cut to one char + ("0.3.9-rc2", FirmwareVersion((0, 3, 9), 2, 0)), + ], +) +def test_parse_version(text: str, expected: FirmwareVersion) -> None: + assert parse_version(text) == expected + + +@pytest.mark.parametrize( + "text", + [ + "", + "gca9bb", # git describe --always without a tag + "v0.3.8", # the firmware strips the "v" + "0.3", + "0.3.8-", + "0.10.10-rc10-12", # 15 chars ending in digits without "-g": commit count may be cut + "0.3.8-36-gXYZ", + "0.3.8 ", + ], +) +def test_unparseable_versions(text: str) -> None: + assert parse_version(text) is None + + +def test_same_commit_count_with_other_hash_is_not_newer() -> None: + assert not is_newer("0.3.8-36-gffff0", "0.3.8-36-gca9bb") + assert not is_newer("0.3.8-36-gca9bb", "0.3.8-36-gffff0") + + +def test_unparseable_side_is_never_newer() -> None: + assert not is_newer("gca9bb", "0.3.8") + assert not is_newer("0.3.9", "gca9bb") + + +def test_parse_trailer_reads_nul_padded_fields() -> None: + assert parse_trailer(build_image("g95_cw", "0.3.8-38-gd3cc9")) == ("g95_cw", "0.3.8-38-gd3cc9") + + +@pytest.mark.parametrize( + ("data", "reason"), + [ + (b"\xff" * 512, "short"), + (build_image(magic=b"NINX"), "NINL"), + (build_image(fmt=2), "format"), + (build_image(corrupt_hash=True), "sha256"), + ], +) +def test_invalid_images(data: bytes, reason: str) -> None: + with pytest.raises(InvalidImageError, match=reason): + parse_trailer(data) + + +def test_ssid_found_when_nul_framed() -> None: + assert contains_ssid(build_image(content=CONTENT_WITH_SSID), "ninIOT") + + +def test_ssid_missing_in_image_without_credentials() -> None: + assert not contains_ssid(build_image(content=CONTENT_WITHOUT_SSID), "ninIOT") + + +@pytest.mark.parametrize("content", [b"\x00xninIOT\x00", b"\x00ninIOTx\x00", b"ninIOT"]) +def test_ssid_must_be_nul_framed(content: bytes) -> None: + assert not contains_ssid(build_image(content=content), "ninIOT") + + +def test_other_ssid_does_not_match() -> None: + assert not contains_ssid(build_image(content=CONTENT_WITH_SSID), "ninIOT-gast") + + +def test_empty_ssid_never_matches() -> None: + assert not contains_ssid(build_image(content=b"\x00\x00\x00"), "") + + +def test_broken_gzip_is_not_a_match() -> None: + payload = b"\xab" * 56 + b"\x1f\x8b\x08" + bytes(7) + b"\xff" * 64 + assert not contains_ssid(build_image(payload=payload), "ninIOT") + + +def test_gzip_stream_inside_the_header_is_ignored() -> None: + payload = gzip.compress(CONTENT_WITH_SSID, mtime=0).ljust(120, b"\x00") + assert not contains_ssid(build_image(payload=payload), "ninIOT") + + +def _write(directory: Path, name: str, data: bytes) -> Path: + path = directory / name + path.write_bytes(data) + return path + + +def test_scan_creates_directory_and_accepts_valid_images_only(tmp_path: Path) -> None: + directory = tmp_path / "ninlight_firmware" + catalog = FirmwareCatalog(directory) + assert catalog.scan() == () + assert directory.is_dir() + _write(directory, "a.img", build_image("gu10_rgb", "0.3.8-47-g71a57")) + _write(directory, "broken.img", build_image(corrupt_hash=True)) + _write(directory, "notes.txt", build_image()) + images = catalog.scan() + assert [(i.path.name, i.device_type, i.version) for i in images] == [ + ("a.img", "gu10_rgb", "0.3.8-47-g71a57") + ] + assert catalog.images == images + + +def test_invalid_file_is_logged_once(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None: + catalog = FirmwareCatalog(tmp_path) + _write(tmp_path, "broken.img", build_image(corrupt_hash=True)) + catalog.scan() + catalog.scan() + assert sum("broken.img" in r.getMessage() for r in caplog.records) == 1 + + +def test_rescan_sees_new_and_removed_files(tmp_path: Path) -> None: + catalog = FirmwareCatalog(tmp_path) + old = _write(tmp_path, "a.img", build_image(version="0.3.8-47-g71a57")) + catalog.scan() + _write(tmp_path, "b.img", build_image(version="0.3.9-rc1")) + old.unlink() + assert [i.version for i in catalog.scan()] == ["0.3.9-rc1"] + + +def test_newer_images_filters_exact_device_type_and_sorts_descending(tmp_path: Path) -> None: + catalog = FirmwareCatalog(tmp_path) + _write(tmp_path, "1.img", build_image("gu10_rgb", "0.3.8-47-g71a57")) + _write(tmp_path, "2.img", build_image("gu10_rgb", "0.3.9-rc1")) + _write(tmp_path, "3.img", build_image("gu10_cw", "0.4.0")) # newest, but other type + _write(tmp_path, "4.img", build_image("gu10", "0.5.0")) # prefix of the type + _write(tmp_path, "5.img", build_image("gu10_rgb", "0.3.8-36-gffff0")) # equal to installed + _write(tmp_path, "6.img", build_image("gu10_rgb", "0.3.8")) # older + catalog.scan() + newer = catalog.newer_images("gu10_rgb", "0.3.8-36-gca9bb") + assert [(i.device_type, i.version) for i in newer] == [ + ("gu10_rgb", "0.3.9-rc1"), + ("gu10_rgb", "0.3.8-47-g71a57"), + ] + + +def test_verify_detects_changed_and_removed_files(tmp_path: Path) -> None: + catalog = FirmwareCatalog(tmp_path) + path = _write(tmp_path, "a.img", build_image(version="0.3.8-47-g71a57")) + (image,) = catalog.scan() + assert catalog.verify(image) + path.write_bytes(build_image(version="0.3.9-rc1")) + assert not catalog.verify(image) + path.unlink() + assert not catalog.verify(image) + + +def test_catalog_ssid_check_is_cached_per_image_and_ssid( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + catalog = FirmwareCatalog(tmp_path) + _write(tmp_path, "a.img", build_image(content=CONTENT_WITH_SSID)) + (image,) = catalog.scan() + calls: list[str] = [] + real = firmware.contains_ssid + + def spy(data: bytes, ssid: str) -> bool: + calls.append(ssid) + return real(data, ssid) + + monkeypatch.setattr(firmware, "contains_ssid", spy) + assert catalog.contains_ssid(image, "ninIOT") + assert catalog.contains_ssid(image, "ninIOT") + assert not catalog.contains_ssid(image, "andere") + assert calls == ["ninIOT", "andere"] + + +def test_catalog_ssid_check_on_removed_file_is_false(tmp_path: Path) -> None: + catalog = FirmwareCatalog(tmp_path) + path = _write(tmp_path, "a.img", build_image(content=CONTENT_WITH_SSID)) + (image,) = catalog.scan() + path.unlink() + assert not catalog.contains_ssid(image, "ninIOT") diff --git a/tests_unit/test_urls.py b/tests_unit/test_urls.py new file mode 100644 index 0000000..55ec58d --- /dev/null +++ b/tests_unit/test_urls.py @@ -0,0 +1,46 @@ +"""Validation of the address lamps use to reach Home Assistant.""" + +from __future__ import annotations + +import pytest + +from custom_components.ninlight.urls import firmware_url, validate_lamp_base_url + + +@pytest.mark.parametrize( + "url", ["http://192.168.1.50:8123", "http://ha.local:8123/", "http://192.168.1.50"] +) +def test_valid_base_urls(url: str) -> None: + assert validate_lamp_base_url(url) is None + + +@pytest.mark.parametrize( + "url", + [ + "https://192.168.1.50:8123", + "http://", + "ftp://ha.local", + "http://user:pass@ha.local:8123", + "http://ha.local:8123/ha", + "http://ha.local:8123?x=1", + "http://ha.local:8123#frag", + "http://ha.local:99999", + "192.168.1.50:8123", + ], +) +def test_invalid_base_urls(url: str) -> None: + assert validate_lamp_base_url(url) == "invalid_base_url" + + +def test_length_boundary_is_the_lamps_255_characters() -> None: + longest_ok = "http://" + "a" * 182 + assert len(firmware_url(longest_ok, "k" * 43)) == 255 + assert validate_lamp_base_url(longest_ok) is None + assert validate_lamp_base_url(longest_ok + "a") == "base_url_too_long" + + +def test_firmware_url_joins_without_double_slash() -> None: + assert ( + firmware_url("http://ha.local:8123/", "abc") + == "http://ha.local:8123/api/ninlight/firmware/abc" + )