75 lines
2.4 KiB
Python
75 lines
2.4 KiB
Python
"""Serve firmware images to lamps under one-time URLs."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
from pathlib import Path
|
|
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 one file, valid only while its update runs."""
|
|
|
|
def __init__(self) -> None:
|
|
self._paths: dict[str, Path] = {}
|
|
|
|
def create(self, path: Path) -> str:
|
|
key = secrets.token_urlsafe(32)
|
|
self._paths[key] = path
|
|
return key
|
|
|
|
def revoke(self, key: str) -> None:
|
|
self._paths.pop(key, None)
|
|
|
|
def get(self, key: str) -> Path | None:
|
|
return self._paths.get(key)
|
|
|
|
def __len__(self) -> int:
|
|
return len(self._paths)
|
|
|
|
|
|
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:
|
|
path = self._registry.get(key)
|
|
if path is None:
|
|
return web.Response(status=404)
|
|
try:
|
|
body = await self._hass.async_add_executor_job(path.read_bytes)
|
|
except OSError:
|
|
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)
|