76 lines
3.7 KiB
Python
76 lines
3.7 KiB
Python
"""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))
|