houseplan-card/custom_components/houseplan/asset_integrity.py

"""Bounded, shared integrity verification for content-addressed decor assets."""
from __future__ import annotations

import hashlib
import logging
import threading
from collections import OrderedDict
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable

from .const import DOMAIN

_LOGGER = logging.getLogger(__name__)

ASSET_INTEGRITY_CACHE_ENTRIES = 256
ASSET_HASH_CHUNK_BYTES = 64 * 1024
_HASS_DATA_KEY = "asset_integrity_verifier"


@dataclass(frozen=True)
class FileSignature:
    """File version facts available without reading its content."""

    size: int
    mtime_ns: int
    ctime_ns: int


@dataclass(frozen=True)
class _CacheEntry:
    signature: FileSignature
    digest: str


@dataclass
class _Flight:
    event: threading.Event
    digest: str | None = None


def _signature(path: Path) -> FileSignature:
    stat = path.stat()
    return FileSignature(
        size=stat.st_size,
        mtime_ns=stat.st_mtime_ns,
        ctime_ns=stat.st_ctime_ns,
    )


def _stream_sha256(path: Path) -> str:
    """Hash a blob without retaining its bytes in memory."""
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        while chunk := stream.read(ASSET_HASH_CHUNK_BYTES):
            digest.update(chunk)
    return digest.hexdigest()


class AssetIntegrityVerifier:
    """Thread-safe LRU digest cache with per-file-version single-flight."""

    def __init__(
        self,
        max_entries: int = ASSET_INTEGRITY_CACHE_ENTRIES,
        *,
        hasher: Callable[[Path], str] | None = None,
        event_factory: Callable[[], threading.Event] | None = None,
    ) -> None:
        if max_entries < 1:
            raise ValueError("max_entries must be positive")
        self._max_entries = max_entries
        self._hasher = hasher or _stream_sha256
        self._event_factory = event_factory or threading.Event
        self._lock = threading.Lock()
        self._cache: OrderedDict[str, _CacheEntry] = OrderedDict()
        self._inflight: dict[tuple[str, FileSignature], _Flight] = {}

    def verify(self, path: Path, expected_digest: str) -> bool:
        """Return whether one stable file version has the expected digest."""
        try:
            canonical = str(path.resolve())
        except (OSError, RuntimeError):
            return False
        try:
            before = _signature(path)
        except OSError:
            with self._lock:
                self._cache.pop(canonical, None)
            return False

        key = (canonical, before)
        with self._lock:
            cached = self._cache.get(canonical)
            if cached is not None and cached.signature == before:
                self._cache.move_to_end(canonical)
                return cached.digest == expected_digest
            if cached is not None:
                self._cache.pop(canonical, None)
            flight = self._inflight.get(key)
            owner = flight is None
            if owner:
                flight = _Flight(self._event_factory())
                self._inflight[key] = flight

        assert flight is not None
        if not owner:
            flight.event.wait()
            return flight.digest == expected_digest

        digest: str | None = None
        stable = False
        try:
            digest = self._hasher(path)
            # Never publish a digest for bytes that changed while they were read.
            stable = _signature(path) == before
        except Exception as err:  # noqa: BLE001 - filesystem/hash seam fails dark
            _LOGGER.debug("House Plan asset integrity check failed: %s", err)
        finally:
            with self._lock:
                if stable and digest is not None:
                    self._cache[canonical] = _CacheEntry(before, digest)
                    self._cache.move_to_end(canonical)
                    while len(self._cache) > self._max_entries:
                        self._cache.popitem(last=False)
                    flight.digest = digest
                self._inflight.pop(key, None)
                flight.event.set()
        return stable and digest == expected_digest


def get_asset_integrity_verifier(hass: Any) -> AssetIntegrityVerifier:
    """Return the single verifier shared by HTTP and WS on this HA instance."""
    domain_data = hass.data.setdefault(DOMAIN, {})
    verifier = domain_data.get(_HASS_DATA_KEY)
    if not isinstance(verifier, AssetIntegrityVerifier):
        verifier = AssetIntegrityVerifier()
        domain_data[_HASS_DATA_KEY] = verifier
    return verifier