"""Bounded, shared integrity verification for content-addressed decor assets.""" from __future__ import annotations import hashlib import logging import stat 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 ASSET_INTEGRITY_FOLLOWER_TIMEOUT_SECONDS = 30.0 _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: current = path.stat() if not stat.S_ISREG(current.st_mode): raise OSError("asset is not a regular file") return FileSignature( size=current.st_size, mtime_ns=current.st_mtime_ns, ctime_ns=current.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: if not flight.event.wait(ASSET_INTEGRITY_FOLLOWER_TIMEOUT_SECONDS): return False 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