python/setprotocol/firmware_catalog.py

@@ -0,0 +1,75 @@
+"""Shared firmware release catalog contract from ``update.json``."""
+
+from __future__ import annotations
+
+import json
+import re
+from dataclasses import dataclass
+from pathlib import Path
+from urllib.parse import urljoin, urlparse
+
+MAX_MANIFEST_BYTES = 128 * 1024
+SUPPORTED_TRANSPORTS = frozenset({"rs485", "can", "stm32"})
+
+
+@dataclass(frozen=True)
+class FirmwareRelease:
+    product: str
+    version: str
+    version_code: int
+    image_url: str
+    sha256: str
+    file_name: str
+    notes: str = ""
+    transport: str = "rs485"
+    base_address: int | None = None
+
+
+def parse_firmware_catalog(data: bytes, manifest_url: str) -> list[FirmwareRelease]:
+    """Parse and validate the optional ``firmware.releases`` section."""
+    if len(data) > MAX_MANIFEST_BYTES:
+        raise ValueError("Файл каталога прошивок слишком большой")
+    try:
+        root = json.loads(data.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as error:
+        raise ValueError("Некорректный JSON каталога прошивок") from error
+    firmware = root.get("firmware") if isinstance(root, dict) else None
+    rows = firmware.get("releases") if isinstance(firmware, dict) else firmware
+    if not isinstance(rows, list):
+        raise ValueError("В update.json не опубликован каталог прошивок")
+    releases: list[FirmwareRelease] = []
+    for index, row in enumerate(rows, 1):
+        if not isinstance(row, dict):
+            raise ValueError(f"Некорректная запись прошивки №{index}")
+        product = str(row.get("product", row.get("device", ""))).strip()
+        version = str(row.get("versionName", row.get("version", ""))).strip()
+        raw_url = str(row.get("imageUrl", row.get("url", ""))).strip()
+        sha256 = str(row.get("sha256", "")).strip().lower()
+        try:
+            version_code = int(row.get("versionCode", 0))
+        except (TypeError, ValueError) as error:
+            raise ValueError(f"Некорректная версия прошивки №{index}") from error
+        if not product or not version or version_code < 0 or not raw_url:
+            raise ValueError(f"Не заполнены обязательные поля прошивки №{index}")
+        if len(sha256) != 64 or any(char not in "0123456789abcdef" for char in sha256):
+            raise ValueError(f"Некорректный SHA-256 прошивки №{index}")
+        image_url = urljoin(manifest_url, raw_url)
+        parsed = urlparse(image_url)
+        if parsed.scheme.lower() != "https" or not parsed.hostname:
+            raise ValueError("Обновления разрешены только по HTTPS")
+        file_name = str(row.get("fileName", Path(parsed.path).name)).strip()
+        if not re.fullmatch(r"[A-Za-zА-Яа-яЁё0-9._ -]+\.(?:bin|hex)", file_name, re.I):
+            raise ValueError(f"Некорректное имя файла прошивки №{index}")
+        base_raw = row.get("baseAddress")
+        try:
+            base_address = int(str(base_raw), 0) if base_raw is not None else None
+        except ValueError as error:
+            raise ValueError(f"Некорректный адрес прошивки №{index}") from error
+        transport = str(row.get("transport", "rs485")).strip().lower()
+        if transport not in SUPPORTED_TRANSPORTS:
+            raise ValueError(f"Некорректный канал прошивки №{index}")
+        releases.append(FirmwareRelease(
+            product, version, version_code, image_url, sha256, file_name,
+            str(row.get("notes", "")).strip(), transport, base_address,
+        ))
+    return sorted(releases, key=lambda item: (item.product.casefold(), -item.version_code))