diff --git a/python/setprotocol/firmware_catalog.py b/python/setprotocol/firmware_catalog.py new file mode 100644 index 0000000..2b1c2e9 --- /dev/null +++ b/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))