"""Serve firmware images to lamps under one-time URLs.""" from __future__ import annotations from collections.abc import Mapping import secrets from typing import Any from aiohttp import web from homeassistant.components.http import HomeAssistantView from homeassistant.core import HomeAssistant from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers.network import NoURLAvailableError, get_url from .const import CONF_LAMP_BASE_URL, DOMAIN, FIRMWARE_URL_PREFIX class OtaRegistry: """Random keys, each bound to exactly the verified bytes of one image, valid only while its update runs. Bytes rather than a path: the file on disk could be replaced between verification and download (spec ยง7 guard would otherwise apply to a file the lamp never actually receives). """ def __init__(self) -> None: self._images: dict[str, bytes] = {} def create(self, data: bytes) -> str: key = secrets.token_urlsafe(32) self._images[key] = data return key def revoke(self, key: str) -> None: self._images.pop(key, None) def get(self, key: str) -> bytes | None: return self._images.get(key) def __len__(self) -> int: return len(self._images) class NinlightFirmwareView(HomeAssistantView): """No HA login: the lamp cannot authenticate. The unguessable key is the protection.""" url = FIRMWARE_URL_PREFIX + "{key}" name = "api:ninlight:firmware" requires_auth = False def __init__(self, hass: HomeAssistant, registry: OtaRegistry) -> None: self._hass = hass self._registry = registry async def get(self, request: web.Request, key: str) -> web.Response: body = self._registry.get(key) if body is None: return web.Response(status=404) return web.Response(body=body, content_type="application/octet-stream") def lamp_base_url(hass: HomeAssistant, options: Mapping[str, Any]) -> str: """The http:// address lamps use to reach Home Assistant.""" base = options.get(CONF_LAMP_BASE_URL) if not base: try: base = get_url(hass, allow_external=False, allow_cloud=False, prefer_external=False) except NoURLAvailableError as err: raise HomeAssistantError( translation_domain=DOMAIN, translation_key="no_http_url" ) from err if not str(base).startswith("http://"): raise HomeAssistantError(translation_domain=DOMAIN, translation_key="no_http_url") return str(base)