Implement a generic OTA provider using the zigpy OTA format

zigpy/config/__init__.py

@@ -64,6 +64,9 @@
 CONF_OTA_SONOFF = "sonoff_provider"
 CONF_OTA_SONOFF_URL = "sonoff_update_url"
 CONF_OTA_THIRDREALITY = "thirdreality_provider"
+CONF_OTA_REMOTE_PROVIDERS = "remote_providers"
+CONF_OTA_PROVIDER_URL = "url"
+CONF_OTA_PROVIDER_MANUF_IDS = "manufacturer_ids"
 CONF_SOURCE_ROUTING = "source_routing"
 CONF_TOPO_SCAN_PERIOD = "topology_scan_period"
 CONF_TOPO_SCAN_ENABLED = "topology_scan_enabled"
@@ -101,6 +104,14 @@
         ),
     }
 )
+
+SCHEMA_OTA_PROVIDER = vol.Schema(
+    {
+        vol.Required(CONF_OTA_PROVIDER_URL): str,
+        vol.Optional(CONF_OTA_PROVIDER_MANUF_IDS, default=[]): [cv_hex],
+    }
+)
+
 SCHEMA_OTA = {
     vol.Optional(CONF_OTA_DIR, default=CONF_OTA_OTAU_DIR_DEFAULT): vol.Any(None, str),
     vol.Optional(CONF_OTA_IKEA, default=CONF_OTA_IKEA_DEFAULT): cv_boolean,
@@ -112,6 +123,7 @@
     vol.Optional(
         CONF_OTA_THIRDREALITY, default=CONF_OTA_THIRDREALITY_DEFAULT
     ): cv_boolean,
+    vol.Optional(CONF_OTA_REMOTE_PROVIDERS, default=[]): [SCHEMA_OTA_PROVIDER],
     # Deprecated keys
     vol.Optional(CONF_OTA_IKEA_URL): vol.All(
         cv_deprecated("The `ikea_update_url` key is deprecated and should be removed"),

zigpy/ota/__init__.py

@@ -11,6 +11,9 @@
     CONF_OTA_IKEA,
     CONF_OTA_INOVELLI,
     CONF_OTA_LEDVANCE,
+    CONF_OTA_PROVIDER_MANUF_IDS,
+    CONF_OTA_PROVIDER_URL,
+    CONF_OTA_REMOTE_PROVIDERS,
     CONF_OTA_SALUS,
     CONF_OTA_SONOFF,
     CONF_OTA_THIRDREALITY,
@@ -122,6 +125,14 @@ def __init__(self, app: ControllerApplicationType, *args, **kwargs):
         if ota_config[CONF_OTA_THIRDREALITY]:
             self.add_listener(zigpy.ota.provider.ThirdReality())
 
+        for provider_config in ota_config[CONF_OTA_REMOTE_PROVIDERS]:
+            self.add_listener(
+                zigpy.ota.provider.RemoteProvider(
+                    url=provider_config[CONF_OTA_PROVIDER_URL],
+                    manufacturer_ids=provider_config[CONF_OTA_PROVIDER_MANUF_IDS],
+                )
+            )
+
     async def initialize(self) -> None:
         await self.async_event("initialize_provider", self._app.config[CONF_OTA])
         self._not_initialized = False

zigpy/ota/provider.py

@@ -5,6 +5,7 @@
 import asyncio
 from collections import defaultdict
 import datetime
+import hashlib
 import io
 import logging
 import os
@@ -796,3 +797,114 @@ async def refresh_firmware_list(self) -> None:
 
     async def filter_get_image(self, key: ImageKey) -> bool:
         return key.manufacturer_id not in self.MANUFACTURER_IDS
+
+
+@attr.s
+class RemoteImage:
+    binary_url = attr.ib()
+    file_version = attr.ib()
+    image_type = attr.ib()
+    manufacturer_id = attr.ib()
+    changelog = attr.ib()
+    checksum = attr.ib()
+
+    # Optional
+    min_hardware_version = attr.ib()
+    max_hardware_version = attr.ib()
+
+    @classmethod
+    def from_json(cls, obj: dict[str, typing.Any]) -> RemoteImage:
+        return cls(
+            binary_url=obj["binary_url"],
+            file_version=obj["file_version"],
+            image_type=obj["image_type"],
+            manufacturer_id=obj["manufacturer_id"],
+            changelog=obj["changelog"],
+            checksum=obj["checksum"],
+            min_hardware_version=obj.get("min_hardware_version"),
+            max_hardware_version=obj.get("max_hardware_version"),
+        )
+
+    @property
+    def key(self) -> ImageKey:
+        return ImageKey(self.manufacturer_id, self.image_type)
+
+    async def fetch_image(self) -> BaseOTAImage:
+        async with aiohttp.ClientSession() as req:
+            LOGGER.debug("Downloading %s for %s", self.binary_url, self.key)
+            async with req.get(self.binary_url) as rsp:
+                data = await rsp.read()
+
+        algorithm, checksum = self.checksum.split(":")
+        hasher = hashlib.new(algorithm)
+        await asyncio.get_running_loop().run_in_executor(None, hasher.update, data)
+
+        if hasher.hexdigest() != checksum:
+            raise ValueError(
+                f"Image checksum is invalid: expected {self.checksum},"
+                f" got {hasher.hexdigest()}"
+            )
+
+        ota_image, _ = parse_ota_image(data)
+
+        if ota_image.header.key != self.key:
+            raise ValueError(
+                f"Image key does not match metadata: {ota_image.header.key}"
+                f" != {self.key}"
+            )
+
+        LOGGER.debug(
+            "Finished downloading from %s for %s ver %s",
+            self.binary_url,
+            self.key,
+            self.version,
+        )
+        return ota_image
+
+
+class RemoteProvider(Basic):
+    """Generic zigpy OTA URL provider."""
+
+    HEADERS = {"accept": "application/json"}
+
+    def __init__(self, url: str, manufacturer_ids: list[int] | None) -> None:
+        super().__init__()
+
+        self.url = url
+        self.manufacturer_ids = manufacturer_ids
+
+    async def initialize_provider(self, ota_config: dict) -> None:
+        self.info("OTA provider enabled")
+        await self.refresh_firmware_list()
+        self.enable()
+
+    async def refresh_firmware_list(self) -> None:
+        if self._locks[LOCK_REFRESH].locked():
+            return
+
+        async with self._locks[LOCK_REFRESH]:
+            async with aiohttp.ClientSession(headers=self.HEADERS) as req:
+                async with req.get(self.url) as rsp:
+                    if not (200 <= rsp.status <= 299):
+                        self.warning(
+                            "Couldn't download '%s': %s/%s",
+                            rsp.url,
+                            rsp.status,
+                            rsp.reason,
+                        )
+                        return
+                    fw_lst = await rsp.json()
+
+        self.debug("Finished downloading firmware update list")
+        self._cache.clear()
+        for obj in fw_lst:
+            img = RemoteImage.from_json(obj)
+            self._cache[img.key] = img
+
+        self.update_expiration()
+
+    async def filter_get_image(self, key: ImageKey) -> bool:
+        if not self.manufacturer_ids:
+            return False
+
+        return key.manufacturer_id not in self.manufacturer_ids