feat(firmware): Trailer, Versionsordnung, WLAN-Pruefung und Firmware-Katalog

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<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)"
+    r"(?:-rc(?P<rc>\d+))?"
+    r"(?:-(?P<commits>\d+)(?P<hash>-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]

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

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)

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")

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"
+    )