Add repository brand icon endpoint

custom_components/hacs/__init__.py

@@ -23,6 +23,7 @@
 from homeassistant.loader import async_get_integration
 
 from .base import HacsBase
+from .brands import async_register_icon_view
 from .const import DOMAIN, HACS_SYSTEM_ID, MINIMUM_HA_VERSION
 from .data_client import HacsDataClient
 from .enums import HacsDisabledReason, HacsStage, LovelaceMode
@@ -143,6 +144,7 @@ async def async_startup():
 
         async_register_websocket_commands(hass)
         await async_register_frontend(hass, hacs)
+        async_register_icon_view(hass)
 
         await hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS)
 

custom_components/hacs/brands.py

@@ -0,0 +1,280 @@
+"""Serve brand icons for repositories managed by HACS.
+
+Custom integrations ship their brand images in a local brand folder
+(custom_components/<domain>/brand/) since Home Assistant 2026.3, and
+home-assistant/brands no longer accepts images for custom integrations.
+
+This view serves those icons to the HACS frontend:
+- Downloaded integrations are served straight from the local brand folder.
+- Integrations that are not downloaded are fetched from the repository
+  content on GitHub and cached on disk. The cache is keyed by the version
+  the icon was fetched for, so a new release invalidates it automatically.
+- Repositories without a brand icon are redirected to the brands CDN,
+  which still hosts previously accepted images.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from http import HTTPStatus
+from pathlib import Path
+import re
+import time
+from typing import TYPE_CHECKING
+from urllib.parse import quote
+
+from aiohttp import ClientError, ClientTimeout, hdrs, web
+
+from homeassistant.components.http import KEY_AUTHENTICATED, HomeAssistantView
+from homeassistant.core import HomeAssistant, callback
+
+from .const import DOMAIN
+from .enums import HacsCategory
+
+if TYPE_CHECKING:
+    from .base import HacsBase
+    from .repositories.base import HacsRepository
+
+URL_BASE = "/api/hacs/repository"
+BRANDS_CDN_URL = "https://brands.home-assistant.io/_"
+CACHE_DIR = ".storage/hacs.icons"
+CACHE_CONTROL = f"public, max-age={60 * 60 * 24}"
+NO_CACHE = "no-store"
+NEGATIVE_CACHE_TTL = 60 * 60 * 24 * 7
+MAX_ICON_SIZE = 1024 * 1024
+DOWNLOAD_CHUNK_SIZE = 64 * 1024
+PNG_MAGIC = b"\x89PNG\r\n\x1a\n"
+VALID_FILENAMES = ("icon.png", "dark_icon.png")
+BRANDS_DOMAIN = "brands"
+
+_DOMAIN_RE = re.compile(r"[a-z0-9_]+")
+
+_VIEW_REGISTERED = "hacs_repository_icon_view"
+
+
+def _read_file(path: Path) -> bytes | None:
+    """Read a file, returning None if it can not be read."""
+    try:
+        return path.read_bytes()
+    except OSError:
+        return None
+
+
+def _validate_icon(content: bytes | None) -> bytes | None:
+    """Return the content if it is a PNG image within the size limit."""
+    if content is None or len(content) > MAX_ICON_SIZE or not content.startswith(PNG_MAGIC):
+        return None
+    return content
+
+
+def _cache_lookup(cache_file: Path, marker_file: Path) -> tuple[bytes | None, bool]:
+    """Return cached icon content and whether a fresh negative marker exists."""
+    if (content := _validate_icon(_read_file(cache_file))) is not None:
+        return content, False
+    try:
+        fresh = (time.time() - marker_file.stat().st_mtime) < NEGATIVE_CACHE_TTL
+    except OSError:
+        fresh = False
+    return None, fresh
+
+
+def _cache_write(
+    cache_dir: Path,
+    prefix: str,
+    filename: str,
+    target: Path,
+    content: bytes | None,
+) -> None:
+    """Write icon content (or a negative marker) and drop entries for old versions."""
+    cache_dir.mkdir(parents=True, exist_ok=True)
+    for stale in (
+        *cache_dir.glob(f"{prefix}*-{filename}"),
+        *cache_dir.glob(f"{prefix}*-{filename}.missing"),
+    ):
+        if stale != target:
+            stale.unlink(missing_ok=True)
+    if content is None:
+        target.touch()
+    else:
+        target.write_bytes(content)
+
+
+class HacsRepositoryIconView(HomeAssistantView):
+    """Serve brand icons for HACS repositories."""
+
+    url = f"{URL_BASE}/{{repository_id}}/{{filename}}"
+    name = "api:hacs:repository:icon"
+    requires_auth = False
+
+    def __init__(self, hass: HomeAssistant) -> None:
+        """Initialize the view."""
+        self._hass = hass
+        self._cache_dir = Path(hass.config.path(CACHE_DIR))
+        self._locks: dict[str, asyncio.Lock] = {}
+
+    def _authenticate(self, request: web.Request) -> None:
+        """Authenticate with Home Assistant or a brands access token."""
+        access_tokens = self._hass.data.get(BRANDS_DOMAIN, ())
+        authenticated = request.get(KEY_AUTHENTICATED, False) or (
+            request.query.get("token") in access_tokens
+        )
+        if authenticated:
+            return
+        if hdrs.AUTHORIZATION in request.headers:
+            raise web.HTTPUnauthorized
+        raise web.HTTPForbidden
+
+    async def get(
+        self,
+        request: web.Request,
+        repository_id: str,
+        filename: str,
+    ) -> web.StreamResponse:
+        """Serve the brand icon for a repository."""
+        self._authenticate(request)
+        if (hacs := self._hass.data.get(DOMAIN)) is None:
+            raise web.HTTPServiceUnavailable
+
+        repository = hacs.repositories.get_by_id(repository_id)
+        if (
+            filename not in VALID_FILENAMES
+            or repository is None
+            or repository.data.category != HacsCategory.INTEGRATION
+            or not repository.data.domain
+            or not _DOMAIN_RE.fullmatch(repository.data.domain)
+        ):
+            raise web.HTTPNotFound
+
+        token = request.query.get("token")
+        if repository.data.installed:
+            return await self._async_serve_local(repository, filename, token)
+        return await self._async_serve_remote(hacs, repository, filename, token)
+
+    async def _async_serve_local(
+        self, repository: HacsRepository, filename: str, token: str | None
+    ) -> web.StreamResponse:
+        """Serve the icon from the downloaded integration."""
+        brand_path = Path(
+            self._hass.config.path("custom_components", repository.data.domain, "brand", filename)
+        )
+        content = _validate_icon(await self._hass.async_add_executor_job(_read_file, brand_path))
+        if content is not None:
+            return self._icon_response(content)
+        return self._fallback_response(repository, filename, token=token)
+
+    async def _async_serve_remote(
+        self,
+        hacs: HacsBase,
+        repository: HacsRepository,
+        filename: str,
+        token: str | None,
+    ) -> web.StreamResponse:
+        """Serve the icon from the GitHub repository content, with caching."""
+        ref = repository.data.last_version or repository.data.default_branch
+        cache_ref = repository.data.last_version or repository.data.last_commit or ref
+        if ref is None or cache_ref is None:
+            return self._fallback_response(repository, filename, token=token)
+
+        prefix = f"{repository.data.id}-"
+        cache_file = self._cache_dir / f"{prefix}{quote(cache_ref, safe='')}-{filename}"
+        marker_file = cache_file.parent / f"{cache_file.name}.missing"
+        retry = False
+
+        lock = self._locks.setdefault(str(repository.data.id), asyncio.Lock())
+        async with lock:
+            content, missing = await self._hass.async_add_executor_job(
+                _cache_lookup, cache_file, marker_file
+            )
+            if content is None and not missing:
+                content, cache_missing = await self._async_download_icon(
+                    hacs, repository, ref, filename
+                )
+                if content is not None or cache_missing:
+                    await self._hass.async_add_executor_job(
+                        _cache_write,
+                        self._cache_dir,
+                        prefix,
+                        filename,
+                        cache_file if content is not None else marker_file,
+                        content,
+                    )
+                else:
+                    retry = True
+
+        if content is not None:
+            return self._icon_response(content)
+        return self._fallback_response(repository, filename, cache=not retry, token=token)
+
+    async def _async_download_icon(
+        self, hacs: HacsBase, repository: HacsRepository, ref: str, filename: str
+    ) -> tuple[bytes | None, bool]:
+        """Download the icon and report whether a missing result can be cached."""
+        brand_path = (
+            "brand"
+            if repository.repository_manifest.content_in_root
+            else f"custom_components/{repository.data.domain}/brand"
+        )
+        url = (
+            f"https://raw.githubusercontent.com/{repository.data.full_name}/"
+            f"{quote(ref, safe='/')}/{brand_path}/{filename}"
+        )
+        try:
+            response = await hacs.session.get(url, timeout=ClientTimeout(total=60))
+        except (ClientError, TimeoutError):
+            return None, False
+
+        try:
+            if response.status == HTTPStatus.NOT_FOUND:
+                return None, True
+            if response.status != HTTPStatus.OK:
+                return None, False
+
+            content = bytearray()
+            async for chunk in response.content.iter_chunked(DOWNLOAD_CHUNK_SIZE):
+                content.extend(chunk)
+                if len(content) > MAX_ICON_SIZE:
+                    return None, True
+            validated = _validate_icon(bytes(content))
+            return validated, validated is None
+        except (ClientError, TimeoutError):
+            return None, False
+        finally:
+            response.release()
+
+    def _icon_response(self, content: bytes) -> web.Response:
+        """Return the icon content."""
+        return web.Response(
+            body=content,
+            content_type="image/png",
+            headers={"Cache-Control": CACHE_CONTROL},
+        )
+
+    def _fallback_response(
+        self,
+        repository: HacsRepository,
+        filename: str,
+        *,
+        cache: bool = True,
+        token: str | None = None,
+    ) -> web.StreamResponse:
+        """Redirect to the icon variant, or to the brands CDN."""
+        if filename != "icon.png":
+            location = f"{URL_BASE}/{repository.data.id}/icon.png"
+            if token is not None:
+                location = f"{location}?token={quote(token, safe='')}"
+        else:
+            location = f"{BRANDS_CDN_URL}/{repository.data.domain}/{filename}"
+        return web.HTTPFound(
+            location,
+            headers={"Cache-Control": CACHE_CONTROL if cache else NO_CACHE},
+        )
+
+
+@callback
+def async_register_icon_view(hass: HomeAssistant) -> None:
+    """Register the repository icon view."""
+    if hass.data.get(_VIEW_REGISTERED):
+        return
+    view = HacsRepositoryIconView(hass)
+    hass.data[_VIEW_REGISTERED] = view
+    hass.http.register_view(view)

tests/common.py

@@ -402,6 +402,10 @@ def url(self):
     def headers(self):
         return self.kwargs.get("headers", {})
 
+    @property
+    def content(self):
+        return MockedResponseContent(self)
+
     async def read(self, **kwargs):
         if (content := self.kwargs.get("content")) is not None:
             return content
@@ -421,6 +425,19 @@ def raise_for_status(self) -> None:
         if self.status >= 300:
             raise ClientError(self.status)
 
+    def release(self) -> None:
+        """Release the mocked response."""
+
+
+class MockedResponseContent:
+    def __init__(self, response: MockedResponse) -> None:
+        self.response = response
+
+    async def iter_chunked(self, size: int):
+        content = await self.response.read()
+        for offset in range(0, len(content), size):
+            yield content[offset : offset + size]
+
 
 class ResponseMocker:
     calls: list[dict[str, Any]] = []

tests/snapshots/api-usage/tests/test_brandstest-icon-view-downloaded-icon-invalid-content.json

@@ -0,0 +1,9 @@
+{
+    "tests/test_brands.py::test_icon_view_downloaded_icon_invalid_content": {
+        "https://api.github.com/repos/hacs/integration": 1,
+        "https://api.github.com/repos/hacs/integration/contents/custom_components/hacs/manifest.json": 1,
+        "https://api.github.com/repos/hacs/integration/contents/hacs.json": 1,
+        "https://api.github.com/repos/hacs/integration/git/trees/main": 1,
+        "https://api.github.com/repos/hacs/integration/releases": 1
+    }
+}

tests/snapshots/api-usage/tests/test_brandstest-icon-view-downloaded-icon-missing.json

@@ -0,0 +1,9 @@
+{
+    "tests/test_brands.py::test_icon_view_downloaded_icon_missing": {
+        "https://api.github.com/repos/hacs/integration": 1,
+        "https://api.github.com/repos/hacs/integration/contents/custom_components/hacs/manifest.json": 1,
+        "https://api.github.com/repos/hacs/integration/contents/hacs.json": 1,
+        "https://api.github.com/repos/hacs/integration/git/trees/main": 1,
+        "https://api.github.com/repos/hacs/integration/releases": 1
+    }
+}

tests/snapshots/api-usage/tests/test_brandstest-icon-view-downloaded-icon.json

@@ -0,0 +1,9 @@
+{
+    "tests/test_brands.py::test_icon_view_downloaded_icon": {
+        "https://api.github.com/repos/hacs/integration": 1,
+        "https://api.github.com/repos/hacs/integration/contents/custom_components/hacs/manifest.json": 1,
+        "https://api.github.com/repos/hacs/integration/contents/hacs.json": 1,
+        "https://api.github.com/repos/hacs/integration/git/trees/main": 1,
+        "https://api.github.com/repos/hacs/integration/releases": 1
+    }
+}

tests/snapshots/api-usage/tests/test_brandstest-icon-view-not-found-0000000-icon-png.json

@@ -0,0 +1,9 @@
+{
+    "tests/test_brands.py::test_icon_view_not_found[0000000-icon.png]": {
+        "https://api.github.com/repos/hacs/integration": 1,
+        "https://api.github.com/repos/hacs/integration/contents/custom_components/hacs/manifest.json": 1,
+        "https://api.github.com/repos/hacs/integration/contents/hacs.json": 1,
+        "https://api.github.com/repos/hacs/integration/git/trees/main": 1,
+        "https://api.github.com/repos/hacs/integration/releases": 1
+    }
+}

tests/snapshots/api-usage/tests/test_brandstest-icon-view-not-found-1296269-icon-png.json

@@ -0,0 +1,9 @@
+{
+    "tests/test_brands.py::test_icon_view_not_found[1296269-../icon.png]": {
+        "https://api.github.com/repos/hacs/integration": 1,
+        "https://api.github.com/repos/hacs/integration/contents/custom_components/hacs/manifest.json": 1,
+        "https://api.github.com/repos/hacs/integration/contents/hacs.json": 1,
+        "https://api.github.com/repos/hacs/integration/git/trees/main": 1,
+        "https://api.github.com/repos/hacs/integration/releases": 1
+    }
+}

tests/snapshots/api-usage/tests/test_brandstest-icon-view-not-found-1296269-not-an-icon-png.json

@@ -0,0 +1,9 @@
+{
+    "tests/test_brands.py::test_icon_view_not_found[1296269-not_an_icon.png]": {
+        "https://api.github.com/repos/hacs/integration": 1,
+        "https://api.github.com/repos/hacs/integration/contents/custom_components/hacs/manifest.json": 1,
+        "https://api.github.com/repos/hacs/integration/contents/hacs.json": 1,
+        "https://api.github.com/repos/hacs/integration/git/trees/main": 1,
+        "https://api.github.com/repos/hacs/integration/releases": 1
+    }
+}