From e691dfc3378572b97bab1e24b61f07abddb03165 Mon Sep 17 00:00:00 2001 From: Andrey Kruchinkin Date: Sat, 5 Sep 2026 03:05:17 +0300 Subject: [PATCH] =?UTF-8?q?=D0=94=D0=BE=D0=B1=D0=B0=D0=B2=D0=B8=D1=82?= =?UTF-8?q?=D1=8C=20=D0=BF=D1=83=D0=B1=D0=BB=D0=B8=D0=BA=D0=B0=D1=86=D0=B8?= =?UTF-8?q?=D1=8E=20=D0=BA=D0=B0=D1=82=D0=B0=D0=BB=D0=BE=D0=B3=D0=B0=20?= =?UTF-8?q?=D0=BF=D1=80=D0=BE=D1=88=D0=B8=D0=B2=D0=BE=D0=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- python/setprotocol/firmware_publish.py | 145 +++++++++++++++++++++++++ python/tests/test_firmware_publish.py | 93 ++++++++++++++++ 2 files changed, 238 insertions(+) create mode 100644 python/setprotocol/firmware_publish.py create mode 100644 python/tests/test_firmware_publish.py diff --git a/python/setprotocol/firmware_publish.py b/python/setprotocol/firmware_publish.py new file mode 100644 index 0000000..c1f86a2 --- /dev/null +++ b/python/setprotocol/firmware_publish.py @@ -0,0 +1,145 @@ +"""Reusable helpers for publishing the shared firmware release catalog.""" + +from __future__ import annotations + +import hashlib +import json +import re +from dataclasses import dataclass +from pathlib import Path + +from .firmware_catalog import ( + MAX_MANIFEST_BYTES, + SUPPORTED_TRANSPORTS, + parse_firmware_catalog, +) + +MAX_FIRMWARE_BYTES = 128 * 1024 * 1024 + + +@dataclass(frozen=True) +class FirmwarePublication: + """Metadata required to publish one firmware image.""" + + path: Path + product: str + version_name: str + version_code: int + transport: str + base_address: int | None = None + notes: str = "" + + def validate(self) -> None: + if not self.path.is_file(): + raise ValueError(f"Firmware file is missing: {self.path}") + if self.path.suffix.lower() not in {".bin", ".hex"}: + raise ValueError("Firmware file must have a .bin or .hex extension") + size = self.path.stat().st_size + if size <= 0: + raise ValueError("Firmware file is empty") + if size > MAX_FIRMWARE_BYTES: + raise ValueError("Firmware file exceeds the maximum size") + if not self.product.strip(): + raise ValueError("Firmware product is empty") + if not self.version_name.strip(): + raise ValueError("Firmware version name is empty") + if not 0 <= self.version_code <= 0x7FFFFFFF: + raise ValueError( + "Firmware version code must be between 0 and 2147483647" + ) + if self.transport not in SUPPORTED_TRANSPORTS: + raise ValueError("Unsupported firmware transport") + if ( + self.base_address is not None + and not 0 <= self.base_address <= 0xFFFFFFFF + ): + raise ValueError("Firmware base address is outside the uint32 range") + + +def sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as source: + for chunk in iter(lambda: source.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def safe_release_tag_part(value: str) -> str: + source = value.strip() + part = re.sub(r"[^A-Za-z0-9._-]+", "-", source).strip("-.") + digest = hashlib.sha256(source.encode("utf-8")).hexdigest()[:8] + if not part: + return digest + return part if part == source else f"{part}-{digest}" + + +def firmware_release_tag(publication: FirmwarePublication) -> str: + return "firmware-%s-v%s" % ( + safe_release_tag_part(publication.product), + safe_release_tag_part(publication.version_name), + ) + + +def firmware_release_entry( + publication: FirmwarePublication, image_url: str, sha256: str +) -> dict: + result = { + "product": publication.product.strip(), + "versionCode": publication.version_code, + "versionName": publication.version_name.strip(), + "imageUrl": image_url, + "fileName": publication.path.name, + "sha256": sha256, + "transport": publication.transport, + "notes": publication.notes.strip(), + } + if publication.base_address is not None: + result["baseAddress"] = f"0x{publication.base_address:08X}" + return result + + +def firmware_entry_identity(entry: dict) -> tuple[str, int, str]: + return ( + str(entry.get("product", entry.get("device", ""))).strip().casefold(), + int(entry.get("versionCode", 0)), + str(entry.get("transport", "rs485")).strip().lower(), + ) + + +def update_firmware_manifest(manifest: dict, entry: dict) -> dict: + """Insert or replace one release without disturbing other manifest data.""" + result = dict(manifest) + existing = manifest.get("firmware") + firmware = dict(existing) if isinstance(existing, dict) else {} + rows = firmware.get("releases") if isinstance(existing, dict) else existing + releases = ( + [dict(row) for row in rows if isinstance(row, dict)] + if isinstance(rows, list) + else [] + ) + identity = firmware_entry_identity(entry) + releases = [ + row for row in releases if firmware_entry_identity(row) != identity + ] + releases.append(dict(entry)) + releases.sort( + key=lambda row: ( + str(row.get("product", row.get("device", ""))).casefold(), + -int(row.get("versionCode", 0)), + str(row.get("transport", "rs485")), + ) + ) + previous_rows = firmware.get("releases") + changed = releases != previous_rows + firmware["catalogVersion"] = ( + int(firmware.get("catalogVersion", 0)) + int(changed) + ) + firmware["releases"] = releases + result["firmware"] = firmware + + encoded = (json.dumps(result, ensure_ascii=False) + "\n").encode("utf-8") + if len(encoded) > MAX_MANIFEST_BYTES: + raise ValueError("Updated update.json exceeds the maximum size") + parse_firmware_catalog(encoded, "https://catalog.invalid/update.json") + return result + diff --git a/python/tests/test_firmware_publish.py b/python/tests/test_firmware_publish.py new file mode 100644 index 0000000..a50b354 --- /dev/null +++ b/python/tests/test_firmware_publish.py @@ -0,0 +1,93 @@ +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +from setprotocol.firmware_publish import ( + FirmwarePublication, + firmware_release_entry, + firmware_release_tag, + update_firmware_manifest, +) + + +class FirmwarePublishTests(unittest.TestCase): + def publication(self, path: Path, **overrides) -> FirmwarePublication: + fields = { + "path": path, + "product": "F103DS18", + "version_name": "1.1.0", + "version_code": 0x00010100, + "transport": "can", + "base_address": 0x08003000, + "notes": "Verified release", + } + fields.update(overrides) + return FirmwarePublication(**fields) + + def test_publication_validates_file_and_metadata(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + image = Path(temporary) / "image.hex" + image.write_text(":00000001FF\n", encoding="ascii") + self.publication(image).validate() + with self.assertRaisesRegex(ValueError, "Unsupported"): + self.publication(image, transport="unknown").validate() + + def test_entry_and_tag_are_deterministic(self) -> None: + publication = self.publication(Path("image.hex")) + entry = firmware_release_entry( + publication, "https://example.test/image.hex", "ab" * 32 + ) + self.assertEqual( + firmware_release_tag(publication), "firmware-F103DS18-v1.1.0" + ) + self.assertEqual(entry["baseAddress"], "0x08003000") + + def test_update_preserves_sections_and_replaces_same_release(self) -> None: + first = { + "product": "Device", + "versionCode": 7, + "versionName": "1.2.3", + "imageUrl": "https://example.test/old.bin", + "fileName": "old.bin", + "sha256": "11" * 32, + "transport": "rs485", + } + manifest = update_firmware_manifest( + {"windows": {"versionCode": 8}}, first + ) + replacement = { + **first, + "imageUrl": "https://example.test/new.bin", + "fileName": "new.bin", + "sha256": "22" * 32, + } + updated = update_firmware_manifest(manifest, replacement) + self.assertEqual(updated["windows"], {"versionCode": 8}) + self.assertEqual(len(updated["firmware"]["releases"]), 1) + self.assertEqual( + updated["firmware"]["releases"][0]["sha256"], "22" * 32 + ) + self.assertEqual(updated["firmware"]["catalogVersion"], 2) + + def test_legacy_array_is_migrated_without_data_loss(self) -> None: + legacy = { + "product": "Legacy", + "versionCode": 1, + "versionName": "1.0", + "imageUrl": "https://example.test/legacy.bin", + "fileName": "legacy.bin", + "sha256": "33" * 32, + "transport": "rs485", + } + current = {**legacy, "product": "Current", "versionCode": 2} + updated = update_firmware_manifest({"firmware": [legacy]}, current) + self.assertCountEqual( + [row["product"] for row in updated["firmware"]["releases"]], + ["Legacy", "Current"], + ) + + +if __name__ == "__main__": + unittest.main()