Michael Sauer
b8f3c94278
Die Registry für Einmal-URLs speicherte bisher nur einen Pfad; View und verify() lasen die Datei getrennt und zu unterschiedlichen Zeitpunkten erneut ein. Wurde die Datei zwischen Prüfung und Abruf ersetzt (z. B. gleicher Dateiname in build/ und build-nocreds/), erreichte ein nie geprüftes Image die Lampe und hebelte damit die WLAN-Sicherung aus. Jetzt werden die Bytes einmal im Install-Pfad gelesen, Gerätetyp/Version und WLAN-SSID auf genau diesen Bytes geprüft (FirmwareCatalog.verify_and_read, firmware.contains_ssid) und exakt diese Bytes in der Registry hinterlegt; die View serviert nur noch das, was tatsächlich geprüft wurde. Außerdem: Bewertungen (Katalog-Rescan-Signal) werden nicht mehr angestoßen, während ein Update läuft - das würde eine gerade flashende Lampe zusätzlich anfragen und Angebot/Issue unter dem laufenden Update verändern; nach jedem Update wird stattdessen genau einmal neu bewertet. Kommentar zum Install-Lock korrigiert (der eigentliche Lücke liegt vor dem Setzen der Flags, nicht in HAs eigener "bereits läuft"-Prüfung). Testlücke geschlossen: die View liefert jetzt nachweislich nur die zu ihrem Schlüssel gehörenden Bytes. Timeout in einem WLAN-Regressionstest verkürzt, damit er bei einem Rückfall nicht 12 Minuten hängt.
260 lines
11 KiB
Python
260 lines
11 KiB
Python
"""Firmware update entity for ninlight lamps, including the Wi-Fi guard (spec §7)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import time
|
|
from typing import Any
|
|
|
|
from homeassistant.components.update import (
|
|
UpdateDeviceClass,
|
|
UpdateEntity,
|
|
UpdateEntityFeature,
|
|
)
|
|
from homeassistant.core import HomeAssistant, callback
|
|
from homeassistant.exceptions import HomeAssistantError
|
|
from homeassistant.helpers import issue_registry as ir
|
|
from homeassistant.helpers.dispatcher import async_dispatcher_connect
|
|
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
|
|
|
from .api import NinlightError
|
|
from .const import (
|
|
DOMAIN,
|
|
MAX_LAMP_URL_LENGTH,
|
|
OTA_POLL_INTERVAL_S,
|
|
OTA_TIMEOUT_S,
|
|
SIGNAL_CATALOG_UPDATED,
|
|
)
|
|
from .coordinator import NinlightConfigEntry, NinlightCoordinator
|
|
from .entity import NinlightEntity
|
|
from .firmware import FirmwareImage
|
|
from .firmware import contains_ssid as _firmware_contains_ssid
|
|
from .firmware import is_newer as _firmware_is_newer
|
|
from .http import lamp_base_url
|
|
from .shared import DATA_SHARED
|
|
from .urls import firmware_url
|
|
|
|
PARALLEL_UPDATES = 0 # several lamps may update at once; each lamp serializes via its lock
|
|
|
|
|
|
def depends_on_compiled_wifi(settings: dict[str, Any]) -> bool:
|
|
"""Only an explicit 'stored' makes a lamp independent of compiled-in credentials."""
|
|
wifi = settings.get("wifi")
|
|
return not (isinstance(wifi, dict) and wifi.get("source") == "stored")
|
|
|
|
|
|
def _ssid(info: dict[str, Any]) -> str:
|
|
wifi = info.get("wifi")
|
|
return str(wifi.get("ssid") or "") if isinstance(wifi, dict) else ""
|
|
|
|
|
|
def _error(key: str, **placeholders: str) -> HomeAssistantError:
|
|
return HomeAssistantError(
|
|
translation_domain=DOMAIN,
|
|
translation_key=key,
|
|
translation_placeholders=placeholders or None,
|
|
)
|
|
|
|
|
|
async def async_setup_entry(
|
|
hass: HomeAssistant,
|
|
entry: NinlightConfigEntry,
|
|
async_add_entities: AddConfigEntryEntitiesCallback,
|
|
) -> None:
|
|
async_add_entities([NinlightUpdateEntity(entry.runtime_data)])
|
|
|
|
|
|
class NinlightUpdateEntity(NinlightEntity, UpdateEntity):
|
|
"""Offers the newest suitable image from the firmware folder."""
|
|
|
|
_attr_device_class = UpdateDeviceClass.FIRMWARE
|
|
_attr_supported_features = UpdateEntityFeature.INSTALL | UpdateEntityFeature.PROGRESS
|
|
|
|
def __init__(self, coordinator: NinlightCoordinator) -> None:
|
|
super().__init__(coordinator, "firmware")
|
|
self._offer: FirmwareImage | None = None
|
|
self._evaluated_for: tuple[str, str, str] | None = None
|
|
self._eval_lock = asyncio.Lock()
|
|
self._install_lock = asyncio.Lock()
|
|
self._attr_in_progress = False
|
|
|
|
@property
|
|
def installed_version(self) -> str | None:
|
|
return self.coordinator.data.info.get("fw_version")
|
|
|
|
@property
|
|
def latest_version(self) -> str | None:
|
|
return self._offer.version if self._offer is not None else self.installed_version
|
|
|
|
def version_is_newer(self, latest_version: str, installed_version: str) -> bool:
|
|
"""Use our own `git describe` comparison (spec §2): HA's AwesomeVersion misparses these.
|
|
|
|
HA 2026.9's default comparator treats "0.3.8-47-g71a57" and "0.3.8-36-gca9bb" as equal
|
|
version cores and falls back to a string compare, which returns False here even though
|
|
commit 47 is newer than commit 36 - the STATE_ON promise this entity makes would silently
|
|
break. `firmware.is_newer` already implements the correct `git describe` ordering used to
|
|
build `self._offer` in the first place, so state and offer never disagree.
|
|
"""
|
|
return _firmware_is_newer(latest_version, installed_version)
|
|
|
|
async def async_added_to_hass(self) -> None:
|
|
await super().async_added_to_hass()
|
|
self.async_on_remove(
|
|
async_dispatcher_connect(self.hass, SIGNAL_CATALOG_UPDATED, self._schedule_evaluation)
|
|
)
|
|
self._schedule_evaluation()
|
|
|
|
@callback
|
|
def _schedule_evaluation(self) -> None:
|
|
# Never evaluate while an install is running: it would poll a flashing lamp for settings
|
|
# and could churn the offer/issue underneath the install. `_async_install_locked` always
|
|
# schedules exactly one evaluation itself once the flags are cleared.
|
|
if self._attr_in_progress:
|
|
return
|
|
self.coordinator.config_entry.async_create_task(
|
|
self.hass, self._async_evaluate(), name="ninlight firmware evaluation"
|
|
)
|
|
|
|
@callback
|
|
def _handle_coordinator_update(self) -> None:
|
|
if not self._attr_in_progress and self._evaluation_key() != self._evaluated_for:
|
|
self._schedule_evaluation()
|
|
super()._handle_coordinator_update()
|
|
|
|
def _evaluation_key(self) -> tuple[str, str, str]:
|
|
info = self.coordinator.data.info
|
|
return (str(info.get("device_type")), str(info.get("fw_version")), _ssid(info))
|
|
|
|
async def _async_evaluate(self) -> None:
|
|
async with self._eval_lock:
|
|
catalog = self.hass.data[DATA_SHARED].catalog
|
|
key = self._evaluation_key()
|
|
device_type, installed, ssid = key
|
|
offer: FirmwareImage | None = None
|
|
blocked: FirmwareImage | None = None
|
|
newer = catalog.newer_images(device_type, installed)
|
|
if newer:
|
|
try:
|
|
settings = await self.coordinator.async_call(lambda c: c.get_settings())
|
|
except HomeAssistantError:
|
|
return # keep the previous offer; the next catalog scan retries
|
|
dependent = depends_on_compiled_wifi(settings)
|
|
for image in newer: # newest first
|
|
if not dependent or await self.hass.async_add_executor_job(
|
|
catalog.contains_ssid, image, ssid
|
|
):
|
|
offer = image
|
|
break
|
|
if blocked is None:
|
|
blocked = image
|
|
self._offer = offer
|
|
self._evaluated_for = key
|
|
self._update_wifi_issue(blocked, ssid)
|
|
self.async_write_ha_state()
|
|
|
|
@callback
|
|
def _update_wifi_issue(self, blocked: FirmwareImage | None, ssid: str) -> None:
|
|
entry = self.coordinator.config_entry
|
|
issue_id = f"wifi_guard_{entry.entry_id}"
|
|
if blocked is None:
|
|
ir.async_delete_issue(self.hass, DOMAIN, issue_id)
|
|
return
|
|
ir.async_create_issue(
|
|
self.hass,
|
|
DOMAIN,
|
|
issue_id,
|
|
is_fixable=False,
|
|
severity=ir.IssueSeverity.WARNING,
|
|
translation_key="wifi_guard",
|
|
translation_placeholders={
|
|
"name": entry.title,
|
|
"version": blocked.version,
|
|
"device_type": blocked.device_type,
|
|
"ssid": ssid or "?",
|
|
},
|
|
)
|
|
|
|
async def async_install(self, version: str | None, backup: bool, **kwargs: Any) -> None:
|
|
# Only the call that owns this lock may touch in_progress/ota_in_progress below: without
|
|
# this check, a second concurrent call could reach the image verification and lamp
|
|
# re-checks before those flags are set - HA's own "already in progress" guard only helps
|
|
# once _attr_in_progress is already True - and its `finally` would clear the flags out
|
|
# from under the first call while that one is still flashing the lamp.
|
|
if self._install_lock.locked():
|
|
raise _error("update_busy")
|
|
async with self._install_lock:
|
|
await self._async_install_locked()
|
|
|
|
async def _async_install_locked(self) -> None:
|
|
shared = self.hass.data[DATA_SHARED]
|
|
image = self._offer
|
|
if image is None:
|
|
raise _error("no_update")
|
|
base = lamp_base_url(self.hass, self.coordinator.config_entry.options)
|
|
# Read the bytes ONCE here and verify exactly those bytes: the view below serves this
|
|
# same in-memory copy, so nothing that happens to the file afterwards (e.g. another build
|
|
# overwriting it mid-update) can change what actually reaches the lamp.
|
|
data = await self.hass.async_add_executor_job(shared.catalog.verify_and_read, image)
|
|
if data is None:
|
|
raise _error("image_invalid")
|
|
|
|
# Re-check right before sending: the lamp's firmware may have changed since the offer was
|
|
# computed (e.g. a stale offer surviving a failed evaluation - see _async_evaluate) and
|
|
# must never be flashed if it is no longer actually newer than what is running now.
|
|
info = await self.coordinator.async_call(lambda c: c.get_info())
|
|
if not _firmware_is_newer(image.version, str(info.get("fw_version"))):
|
|
raise _error("no_update")
|
|
settings = await self.coordinator.async_call(lambda c: c.get_settings())
|
|
if info.get("device_type") != image.device_type:
|
|
raise _error("image_invalid")
|
|
ssid = _ssid(info)
|
|
if depends_on_compiled_wifi(settings) and not await self.hass.async_add_executor_job(
|
|
_firmware_contains_ssid, data, ssid
|
|
):
|
|
raise _error("wifi_guard", ssid=ssid or "?")
|
|
|
|
key = shared.registry.create(data)
|
|
url = firmware_url(base, key)
|
|
if len(url) > MAX_LAMP_URL_LENGTH:
|
|
shared.registry.revoke(key)
|
|
raise _error("lamp_url_too_long")
|
|
|
|
self._attr_in_progress = True
|
|
self.coordinator.ota_in_progress = True
|
|
self.async_write_ha_state()
|
|
try:
|
|
await self.coordinator.async_call(lambda c: c.fw_update(url))
|
|
await self._async_wait_for(image.version, info.get("uptime"))
|
|
finally:
|
|
shared.registry.revoke(key)
|
|
self._attr_in_progress = False
|
|
self.coordinator.ota_in_progress = False
|
|
await self.coordinator.async_refresh()
|
|
self.async_write_ha_state()
|
|
self._schedule_evaluation()
|
|
|
|
async def _async_wait_for(self, target: str, start_uptime: Any) -> None:
|
|
"""Wait for the lamp to report the target version, an error, or an old-version reboot.
|
|
|
|
The firmware clears last_ota_error when an update starts (nl_ota.c), so any value seen
|
|
here belongs to this attempt.
|
|
"""
|
|
deadline = time.monotonic() + OTA_TIMEOUT_S
|
|
while True:
|
|
await asyncio.sleep(OTA_POLL_INTERVAL_S)
|
|
if time.monotonic() >= deadline:
|
|
raise _error("update_timeout")
|
|
try:
|
|
async with self.coordinator.lock:
|
|
info = await self.coordinator.client.get_info()
|
|
except NinlightError:
|
|
continue # flashing or rebooting
|
|
version = str(info.get("fw_version"))
|
|
if version == target:
|
|
return
|
|
if error := info.get("last_ota_error"):
|
|
raise _error("update_failed", reason=str(error))
|
|
uptime = info.get("uptime")
|
|
if isinstance(uptime, int) and isinstance(start_uptime, int) and uptime < start_uptime:
|
|
raise _error("update_failed", reason=f"rebooted with {version}")
|