From fc29b0befee50fff82b831bf9b45a9d106c74315 Mon Sep 17 00:00:00 2001
From: Michael Sauer <michael.sauer@qconnex.com>
Date: Wed, 16 Sep 2026 17:40:21 +0200
Subject: [PATCH] fix: Katalog-SSID-Cache threadsicher, Sicherheitshinweis zu
 mDNS/MAC
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

scan() konnte den SSID-Cache filtern, während contains_ssid() aus einem
anderen Executor-Thread gleichzeitig hineinschrieb ("dictionary changed
size during iteration"). Ein Lock schützt jetzt beide Zugriffe.

README ergänzt: mDNS-Updates auf eine bekannte MAC laufen ungeprüft: Ein
Gerät, das dieselbe MAC ankündigt, übernimmt den Host-Eintrag und könnte
so an Anfragen mit Token gelangen (nur in einem bereits kompromittierten
Netz relevant).
---
 README.md                              |  3 ++
 custom_components/ninlight/firmware.py | 48 +++++++++++++++++++++-----
 2 files changed, 43 insertions(+), 8 deletions(-)

diff --git a/README.md b/README.md
index d05c1cf..b6bae16 100644
--- a/README.md
+++ b/README.md
@@ -29,6 +29,9 @@ HA zeigt dafür einen Reparaturhinweis. Eigenes Token auf der Lampe setzen
 *Neu konfigurieren* eintragen.
 
 Wechselt eine Lampe ihre IP, aktualisiert mDNS den Eintrag selbst (Identität ist die MAC).
+Das gilt ungeprüft: Ein Gerät im selben Netz, das dieselbe MAC per mDNS ankündigt, übernimmt
+damit den Host-Eintrag und kann so an Anfragen mit Token gelangen, die eigentlich an die echte
+Lampe gehen (relevant nur in einem bereits kompromittierten Netz).
 
 ## Entitäten
 
diff --git a/custom_components/ninlight/firmware.py b/custom_components/ninlight/firmware.py
index c45552a..0c41a5d 100644
--- a/custom_components/ninlight/firmware.py
+++ b/custom_components/ninlight/firmware.py
@@ -11,6 +11,7 @@ import hashlib
 import logging
 from pathlib import Path
 import re
+import threading
 import zlib
 
 _LOGGER = logging.getLogger(__name__)
@@ -132,6 +133,9 @@ class FirmwareCatalog:
         self.images: tuple[FirmwareImage, ...] = ()
         self._parsed: dict[tuple[str, int, int], FirmwareImage | None] = {}
         self._ssid_hits: dict[tuple[str, int, int, str], bool] = {}
+        # scan() runs in one executor thread while contains_ssid() may run in another (e.g. an
+        # install's Wi-Fi re-check racing a periodic rescan): both read and mutate _ssid_hits.
+        self._ssid_hits_lock = threading.Lock()
 
     def scan(self) -> tuple[FirmwareImage, ...]:
         """Blocking. Invalid files are logged once per (path, mtime, size)."""
@@ -161,7 +165,8 @@ class FirmwareCatalog:
             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}
+        with self._ssid_hits_lock:
+            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
 
@@ -184,13 +189,40 @@ class FirmwareCatalog:
         except OSError, InvalidImageError:
             return False
 
+    def verify_and_read(self, image: FirmwareImage) -> bytes | None:
+        """Blocking. Like `verify`, but returns the exact bytes that were checked (or None).
+
+        Callers that need to act on the verified content (e.g. serve it, or scan it for an SSID)
+        must use these bytes rather than reading the file again afterwards - the file on disk can
+        change between the two reads.
+        """
+        try:
+            stat = image.path.stat()
+            if (stat.st_mtime_ns, stat.st_size) != (image.mtime_ns, image.size):
+                return None
+            data = image.path.read_bytes()
+        except OSError:
+            return None
+        try:
+            if parse_trailer(data) != (image.device_type, image.version):
+                return None
+        except InvalidImageError:
+            return None
+        return data
+
     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]
+        with self._ssid_hits_lock:
+            cached = key in self._ssid_hits
+            hit = self._ssid_hits.get(key)
+        if cached:
+            return bool(hit)
+        try:
+            data = image.path.read_bytes()
+        except OSError:
+            return False
+        hit = contains_ssid(data, ssid)
+        with self._ssid_hits_lock:
+            self._ssid_hits[key] = hit
+        return hit