"""Liefert OTA-Abbilder an die Geraete aus, unter einmaligen Adressen.""" 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_DEVICE_BASE_URL, DOMAIN, FIRMWARE_URL_PREFIX class OtaRegistry: """Zufaellige Schluessel, jeder an genau die geprueften Bytes eines Abbildes gebunden und nur gueltig, solange dessen Aktualisierung laeuft. Bytes statt eines Pfades: die Datei koennte zwischen Pruefung und Abruf ersetzt werden - die WLAN-Sicherung wuerde dann fuer eine Datei gelten, die das Geraet nie bekommt. """ 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 NinstripFirmwareView(HomeAssistantView): """Ohne HA-Anmeldung: das Geraet kann sich nicht anmelden. Der unratbare Schluessel ist der Schutz.""" url = FIRMWARE_URL_PREFIX + "{key}" name = "api:ninstrip: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 device_base_url(hass: HomeAssistant, options: Mapping[str, Any]) -> str: """Die http://-Adresse, unter der Geraete Home Assistant erreichen.""" base = options.get(CONF_DEVICE_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)