fix(assets): bound resolve integrity work

Issue: #432
User-Visible: yes

custom_components/houseplan/asset_integrity.py

@@ -0,0 +1,139 @@
+"""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

custom_components/houseplan/decor_assets.py

@@ -372,20 +372,42 @@ def asset_meta_path(root: Path, asset_id: str) -> Path:
     return root / f"{asset_id}.json"
 
 
+def _read_catalog_row(root: Path, path: Path) -> dict[str, Any] | None:
+    """Read one sidecar through the validation shared by list and resolve."""
+    try:
+        row = json.loads(path.read_text(encoding="utf-8"))
+        if not isinstance(row, dict):
+            return None
+        aid = str(row.get("asset_id") or "")
+        ext = row.get("ext")
+        blob = root / f"{aid}{ext}"
+        if (
+            path.stem != aid
+            or not ASSET_ID_RE.fullmatch(aid)
+            or ext not in ASSET_EXTENSIONS
+            or not blob.is_file()
+        ):
+            return None
+        return row
+    except (OSError, ValueError, TypeError):
+        return None
+
+
+def read_asset(root: Path, asset_id: str) -> dict[str, Any] | None:
+    """Read one exact catalog row without scanning unrelated sidecars."""
+    if not ASSET_ID_RE.fullmatch(asset_id):
+        return None
+    return _read_catalog_row(root, asset_meta_path(root, asset_id))
+
+
 def read_catalog(root: Path) -> list[dict[str, Any]]:
     rows: list[dict[str, Any]] = []
     if not root.is_dir():
         return rows
     for path in root.glob("*.json"):
-        try:
-            row = json.loads(path.read_text(encoding="utf-8"))
-            aid = str(row.get("asset_id") or "")
-            ext = row.get("ext")
-            blob = root / f"{aid}{ext}"
-            if ASSET_ID_RE.fullmatch(aid) and ext in ASSET_EXTENSIONS and blob.is_file():
-                rows.append(row)
-        except (OSError, ValueError, TypeError):
-            continue
+        row = _read_catalog_row(root, path)
+        if row is not None:
+            rows.append(row)
     return sorted(
         rows,
         key=lambda row: (str(row.get("created_at", "")), str(row["asset_id"])),

custom_components/houseplan/http_api.py

@@ -24,6 +24,7 @@ try:  # KEY_HASS — the modern way to access hass from the aiohttp application
 except ImportError:  # older HA versions
     KEY_HASS = "hass"  # type: ignore[assignment]
 
+from .asset_integrity import get_asset_integrity_verifier
 from .auth import may_write
 from .const import (
     ASSETS_DIR,
@@ -176,18 +177,13 @@ class HouseplanContentView(HomeAssistantView):
         if not str(path).startswith(str(base)):
             return web.Response(status=404)
 
-        if not await hass.async_add_executor_job(path.is_file):
-            return web.Response(status=404)
         suffix = path.suffix.lower()
         if kind == "assets":
-            try:
-                digest = await hass.async_add_executor_job(
-                    lambda: hashlib.sha256(path.read_bytes()).hexdigest(),
-                )
-            except OSError:
-                return web.Response(status=404)
-            if digest != path.stem:
+            verifier = get_asset_integrity_verifier(hass)
+            if not await hass.async_add_executor_job(verifier.verify, path, path.stem):
                 return web.Response(status=404)
+        elif not await hass.async_add_executor_job(path.is_file):
+            return web.Response(status=404)
         headers = {
             "Cache-Control": "private, max-age=31536000, immutable"
                 if kind == "assets" else "private, max-age=3600",

custom_components/houseplan/websocket_api.py

@@ -21,6 +21,7 @@ from homeassistant.const import __version__ as HA_VERSION
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers import issue_registry as ir
 
+from .asset_integrity import get_asset_integrity_verifier
 from .auth import may_write
 from .const import (
     ASSETS_DIR,
@@ -54,6 +55,7 @@ from .decor_assets import (
     asset_meta_path,
     asset_refs,
     public_asset,
+    read_asset,
     read_catalog,
 )
 from .import_export import (
@@ -1135,22 +1137,29 @@ async def ws_assets_list(hass: HomeAssistant, connection, msg: dict[str, Any]) -
 @websocket_api.async_response
 async def ws_assets_resolve(hass: HomeAssistant, connection, msg: dict[str, Any]) -> None:
     """Resolve each unique id once; absent/corrupt content is reported missing."""
-    root = Path(hass.config.path(ASSETS_DIR))
+    rt = _runtime(hass, connection, msg["id"])
+    if rt is None:
+        return
     requested = set(msg["asset_ids"])
+    allowed = requested
+    if not _check_write(hass, connection):
+        async with rt.write_lock:
+            stored = await rt.config_store.async_load() or {}
+            referenced = set(asset_refs(stored.get("config") or {}))
+        allowed = requested & referenced
+
+    root = Path(hass.config.path(ASSETS_DIR))
+    verifier = get_asset_integrity_verifier(hass)
 
     def _resolve() -> tuple[list[dict], list[str]]:
         rows: list[dict] = []
         found: set[str] = set()
-        for row in read_catalog(root):
-            aid = row["asset_id"]
-            if aid not in requested:
+        for aid in sorted(allowed):
+            row = read_asset(root, aid)
+            if row is None:
                 continue
             path = root / f"{aid}{row['ext']}"
-            try:
-                import hashlib
-                if hashlib.sha256(path.read_bytes()).hexdigest() != aid:
-                    continue
-            except OSError:
+            if not verifier.verify(path, aid):
                 continue
             rows.append(public_asset(row))
             found.add(aid)

docs/ARCHITECTURE.md

@@ -380,10 +380,23 @@ Custom Background images use a separate content-addressed store at
 `<config>/houseplan/assets/`. Raster input is fully decoded and SVG is parsed
 through a strict allowlist before promotion; the SHA-256 of canonical bytes is
 the persisted `asset_id`. Config never carries file bytes or a signed URL.
-`houseplan/assets/resolve` maps unique ids to authenticated content paths,
-while the shared `ContentSigner` batches signatures for `<image>` elements.
-Catalog deletion rechecks references across every space under the config write
-lock. Missing or corrupt assets are never painted in View.
+`houseplan/assets/resolve` maps unique ids to authenticated content paths.
+Writers may resolve any catalog id; a read-only household member may resolve
+only ids referenced by the current saved config, with forbidden ids reported as
+ordinary `missing` entries. The reference snapshot is taken under the config
+write lock, but file I/O happens after releasing it. The resolve path reads only
+the requested sidecars rather than scanning the catalog. The HTTP content view
+keeps its authenticated/signed exact-URL contract.
+
+Resolve and HTTP GET share one HA-instance memory-only integrity verifier. It
+streams SHA-256 in bounded chunks and caches at most 256 actual digests by
+canonical path plus size/mtime/ctime signature. Per-file-version single-flight
+deduplicates concurrent reads without serialising different files; a second
+`stat` prevents a digest for bytes changed mid-read from entering the cache.
+Missing, changed and corrupt files fail dark. The shared `ContentSigner` batches
+signatures for `<image>` elements. Catalog deletion rechecks references across
+every space under the config write lock. Missing or corrupt assets are never
+painted in View.
 
 `removed:true` is a binding tombstone, not a renderable marker. It claims an
 HA binding against automatic discovery while intentionally exposing that same
@@ -918,7 +931,7 @@ transmit light is the separate `zero_wall_style` policy.
 | `houseplan/files/migrate` | `from_id`, `to_id` | `{mapping}` — COPY, never move |
 | `houseplan/files/cleanup` | `marker_id`, `keep?` | replacement-only collection |
 | `houseplan/assets/list` | — | reusable image metadata plus authoritative `used_by` references |
-| `houseplan/assets/resolve` | `asset_ids[]` (max 200) | verified metadata/content paths plus missing ids |
+| `houseplan/assets/resolve` | `asset_ids[]` (max 200) | verified metadata/content paths plus missing ids; writer: catalog, read-only: saved references only |
 | `houseplan/assets/delete` | `asset_id` | explicit deletion only when no decor record refers to it |
 | `houseplan/content/sign` | `paths[]` | `{urls}` — authSig for `<image>`/`<a>` fetches |
 | `houseplan/export/create` | `kind`, `space_id?`, `plan_only?`, `card_version` | consistent versioned JSON document + safe filename; plan-only is valid only for one space |

docs/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## Unreleased
 
+- Saved custom images remain visible to read-only household members while
+  arbitrary asset lookup is blocked, and repeated card/HTTP loads now reuse one
+  bounded streaming integrity check instead of re-reading every image
+  ([#432](https://github.com/Matysh/houseplan-card/issues/432)).
 - Custom decor images now pass through the same stable coordinate-write barrier
   as furniture and shapes, so repeated saves and **Optimize Plans** no longer
   retain image-only floating-point noise

docs/CHANGELOG.ru.md

@@ -8,6 +8,11 @@
 
 ## Не выпущено
 
+- Сохранённые пользовательские картинки по-прежнему видны домочадцам без права
+  редактирования, но произвольный поиск файлов теперь закрыт; повторные загрузки
+  карточки и HTTP используют одну ограниченную потоковую проверку вместо нового
+  чтения каждой картинки
+  ([#432](https://github.com/Matysh/houseplan-card/issues/432)).
 - Пользовательские изображения декора теперь проходят тот же стабильный барьер
   записи координат, что мебель и фигуры, поэтому повторные сохранения и
   «Оптимизировать планы» больше не сохраняют float-шум только у изображений

docs/CONFIG-COMPATIBILITY.md

@@ -263,6 +263,15 @@ identity/hash remain exact, and every supplied non-null MIME must be supported.
 Before a permanent downgrade, remove all image objects with a current card and
 then explicitly delete their now-unused files from the palette.
 
+The #432 backend hardening does not change that schema, URL shape, export format
+or `decor_assets_api:1` capability. A read-only user still resolves images used
+by the saved config; only arbitrary unreferenced ids are now returned as
+`missing`. Writers keep the full catalog contract. Authenticated and signed
+exact content URLs remain valid, while integrity results are shared in a bounded
+memory-only cache. Old and new cards therefore remain rolling-compatible with
+the hardened integration; the cache is discarded on restart and needs no data
+migration or downgrade step.
+
 ## Independent-wall opening host (#132)
 
 `space.openings[].host` is an optional discriminated object

scripts/mutation-gate.mjs

@@ -4797,6 +4797,58 @@ const MUTANT_DEFINITIONS = [
       replace: '    if (row.url !== expectedUrl',
     }],
   },
+  {
+    id: 'asset-resolve-readonly-membership-removed',
+    guard: 'node scripts/backend-test-guard.mjs '
+      + 'decor_asset_resolve_readonly_is_limited_to_referenced_ids '
+      + 'tests_backend/test_ha_websocket.py',
+    because: 'a read-only household member needs referenced images for View but must not use '
+      + 'assets/resolve to probe or hash arbitrary catalog ids (#432 AC2)',
+    patches: [{
+      file: 'custom_components/houseplan/websocket_api.py',
+      find: '        allowed = requested & referenced\n',
+      replace: '        allowed = requested\n',
+    }],
+  },
+  {
+    id: 'asset-integrity-cache-hit-disabled',
+    guard: 'node scripts/backend-test-guard.mjs '
+      + 'integrity_cache_reuses_digest_and_caches_corrupt_signature '
+      + 'tests_backend/test_decor_assets.py',
+    because: 'unchanged valid and corrupt files must reuse the actual digest instead of '
+      + 're-reading the blob for every WS resolve or HTTP GET (#432 AC5)',
+    patches: [{
+      file: 'custom_components/houseplan/asset_integrity.py',
+      find: '            if cached is not None and cached.signature == before:\n',
+      replace: '            if False and cached is not None and cached.signature == before:\n',
+    }],
+  },
+  {
+    id: 'asset-integrity-single-flight-disabled',
+    guard: 'node scripts/backend-test-guard.mjs '
+      + 'integrity_cache_single_flights_same_path_and_releases_after_error '
+      + 'tests_backend/test_decor_assets.py',
+    because: 'parallel requests for one file version must share one streaming hash and wake '
+      + 'all waiters after success or failure (#432 AC6)',
+    patches: [{
+      file: 'custom_components/houseplan/asset_integrity.py',
+      find: '            flight = self._inflight.get(key)\n',
+      replace: '            flight = None\n',
+    }],
+  },
+  {
+    id: 'asset-integrity-post-read-signature-ignored',
+    guard: 'node scripts/backend-test-guard.mjs '
+      + 'integrity_cache_invalidates_changed_signature_and_rejects_mid_read_change '
+      + 'tests_backend/test_decor_assets.py',
+    because: 'a digest computed while the blob changes must fail dark and never become a '
+      + 'trusted cache entry for either transport (#432 AC7)',
+    patches: [{
+      file: 'custom_components/houseplan/asset_integrity.py',
+      find: '            stable = _signature(path) == before\n',
+      replace: '            stable = True\n',
+    }],
+  },
   {
     id: 'pure-backend-test-pulls-home-assistant',
     guard: 'python3 -m pytest tests_backend/test_backend_quality.py -q -p no:cacheprovider',

tests_backend/test_decor_assets.py

@@ -2,19 +2,28 @@
 from __future__ import annotations
 
 import base64
+import hashlib
 import importlib
 import json
 import struct
+import threading
 import zlib
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
 
 import pytest
 
+from custom_components.houseplan.asset_integrity import (
+    ASSET_INTEGRITY_CACHE_ENTRIES,
+    AssetIntegrityVerifier,
+)
 from custom_components.houseplan.const import MAX_DECOR_ASSET_BYTES
 from custom_components.houseplan.decor_assets import (
     DecorAssetError,
     asset_meta_path,
     asset_refs,
     public_asset,
+    read_asset,
     read_catalog,
     validate_asset,
 )
@@ -323,6 +332,192 @@ def test_catalog_empty_directory_and_metadata_path(tmp_path) -> None:
     assert asset_meta_path(tmp_path, aid) == tmp_path / f"{aid}.json"
 
 
+def test_direct_asset_lookup_never_scans_or_accepts_mismatched_sidecars(
+    tmp_path, monkeypatch,
+) -> None:
+    payload = b"one"
+    aid = hashlib.sha256(payload).hexdigest()
+    other = "d" * 64
+    (tmp_path / f"{aid}.png").write_bytes(payload)
+    (tmp_path / f"{aid}.json").write_text(json.dumps({
+        "asset_id": aid, "ext": ".png", "mime": "image/png",
+    }), encoding="utf-8")
+    (tmp_path / f"{other}.json").write_text(json.dumps({
+        "asset_id": aid, "ext": ".png", "mime": "image/png",
+    }), encoding="utf-8")
+    assert [row["asset_id"] for row in read_catalog(tmp_path)] == [aid]
+
+    def no_scan(_self, _pattern):
+        raise AssertionError("direct lookup must not scan the catalog")
+
+    monkeypatch.setattr(Path, "glob", no_scan)
+    assert read_asset(tmp_path, aid)["asset_id"] == aid
+    assert read_asset(tmp_path, other) is None
+
+
+def test_integrity_cache_reuses_digest_and_caches_corrupt_signature(tmp_path) -> None:
+    payload = b"stable-content"
+    path = tmp_path / "asset.bin"
+    path.write_bytes(payload)
+    expected = hashlib.sha256(payload).hexdigest()
+    calls = 0
+
+    def counted(candidate: Path) -> str:
+        nonlocal calls
+        calls += 1
+        return hashlib.sha256(candidate.read_bytes()).hexdigest()
+
+    verifier = AssetIntegrityVerifier(hasher=counted)
+    assert verifier.verify(path, expected)
+    assert verifier.verify(path, expected)
+    assert calls == 1
+
+    wrong = "0" * 64
+    assert not verifier.verify(path, wrong)
+    assert not verifier.verify(path, wrong)
+    assert calls == 1, "the actual digest also caches a negative comparison"
+
+
+def test_integrity_cache_invalidates_changed_signature_and_rejects_mid_read_change(
+    tmp_path,
+) -> None:
+    path = tmp_path / "asset.bin"
+    first = b"first"
+    second = b"second-version"
+    path.write_bytes(first)
+    calls = 0
+
+    def counted(candidate: Path) -> str:
+        nonlocal calls
+        calls += 1
+        return hashlib.sha256(candidate.read_bytes()).hexdigest()
+
+    verifier = AssetIntegrityVerifier(hasher=counted)
+    assert verifier.verify(path, hashlib.sha256(first).hexdigest())
+    path.write_bytes(second)
+    assert verifier.verify(path, hashlib.sha256(second).hexdigest())
+    assert calls == 2
+
+    replacement = b"changed-during-read"
+
+    def mutating(candidate: Path) -> str:
+        original = candidate.read_bytes()
+        candidate.write_bytes(replacement)
+        return hashlib.sha256(original).hexdigest()
+
+    unstable = AssetIntegrityVerifier(hasher=mutating)
+    path.write_bytes(first)
+    assert not unstable.verify(path, hashlib.sha256(first).hexdigest())
+    assert not unstable._cache, "an unstable digest must not become a cache hit"
+
+
+def test_integrity_cache_single_flights_same_path_and_releases_after_error(tmp_path) -> None:
+    path = tmp_path / "asset.bin"
+    payload = b"concurrent"
+    path.write_bytes(payload)
+    expected = hashlib.sha256(payload).hexdigest()
+    waiter_joined = threading.Event()
+    real_event = threading.Event
+
+    class ObservedEvent:
+        def __init__(self) -> None:
+            self._event = real_event()
+
+        def set(self) -> None:
+            self._event.set()
+
+        def wait(self, timeout=None) -> bool:
+            waiter_joined.set()
+            return self._event.wait(timeout)
+
+    calls = 0
+
+    def coordinated(candidate: Path) -> str:
+        nonlocal calls
+        calls += 1
+        assert waiter_joined.wait(2), "the concurrent caller never joined the flight"
+        return hashlib.sha256(candidate.read_bytes()).hexdigest()
+
+    verifier = AssetIntegrityVerifier(hasher=coordinated, event_factory=ObservedEvent)
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        first = pool.submit(verifier.verify, path, expected)
+        second = pool.submit(verifier.verify, path, expected)
+        assert first.result(timeout=3) and second.result(timeout=3)
+    assert calls == 1
+
+    attempts = 0
+
+    def once_broken(candidate: Path) -> str:
+        nonlocal attempts
+        attempts += 1
+        if attempts == 1:
+            raise OSError("injected read failure")
+        return hashlib.sha256(candidate.read_bytes()).hexdigest()
+
+    recovered = AssetIntegrityVerifier(hasher=once_broken)
+    assert not recovered.verify(path, expected)
+    assert recovered.verify(path, expected)
+    assert attempts == 2 and not recovered._inflight
+
+
+def test_integrity_checks_for_different_paths_do_not_share_a_hash_lock(tmp_path) -> None:
+    first_path = tmp_path / "first.bin"
+    second_path = tmp_path / "second.bin"
+    first_path.write_bytes(b"first")
+    second_path.write_bytes(b"second")
+    first_started = threading.Event()
+    release_first = threading.Event()
+
+    def coordinated(candidate: Path) -> str:
+        if candidate == first_path:
+            first_started.set()
+            assert release_first.wait(2)
+        return hashlib.sha256(candidate.read_bytes()).hexdigest()
+
+    verifier = AssetIntegrityVerifier(hasher=coordinated)
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        first = pool.submit(
+            verifier.verify, first_path, hashlib.sha256(b"first").hexdigest(),
+        )
+        assert first_started.wait(1)
+        independent = pool.submit(
+            verifier.verify, second_path, hashlib.sha256(b"second").hexdigest(),
+        )
+        assert independent.result(timeout=1)
+        release_first.set()
+        assert first.result(timeout=2)
+
+
+def test_integrity_cache_is_bounded_lru_and_stream_reader_avoids_read_bytes(
+    tmp_path, monkeypatch,
+) -> None:
+    assert ASSET_INTEGRITY_CACHE_ENTRIES == 256
+    paths = []
+    for index in range(ASSET_INTEGRITY_CACHE_ENTRIES + 1):
+        path = tmp_path / f"{index}.bin"
+        path.write_bytes(str(index).encode())
+        paths.append(path)
+
+    verifier = AssetIntegrityVerifier()
+
+    def forbidden_read_bytes(_self):
+        raise AssertionError("integrity verification must stream chunks")
+
+    monkeypatch.setattr(Path, "read_bytes", forbidden_read_bytes)
+    for index in range(ASSET_INTEGRITY_CACHE_ENTRIES):
+        assert verifier.verify(
+            paths[index], hashlib.sha256(str(index).encode()).hexdigest(),
+        )
+    # Refresh zero, then the 257th insert must evict one rather than zero.
+    assert verifier.verify(paths[0], hashlib.sha256(b"0").hexdigest())
+    last = ASSET_INTEGRITY_CACHE_ENTRIES
+    assert verifier.verify(paths[last], hashlib.sha256(str(last).encode()).hexdigest())
+    assert len(verifier._cache) == ASSET_INTEGRITY_CACHE_ENTRIES
+    assert str(paths[0].resolve()) in verifier._cache
+    assert str(paths[1].resolve()) not in verifier._cache
+    assert str(paths[last].resolve()) in verifier._cache
+
+
 def test_reference_scan_is_cross_space_and_image_only() -> None:
     aid = "b" * 64
     refs = asset_refs({"spaces": [

tests_backend/test_ha_websocket.py

@@ -1173,6 +1173,26 @@ async def test_not_ready_without_entry(hass: HomeAssistant, hass_ws_client: WebS
     assert not resp["success"] and resp["error"]["code"] == "not_ready"
 
 
+async def test_decor_asset_resolve_requires_runtime_before_io(
+    hass: HomeAssistant, hass_ws_client: WebSocketGenerator, monkeypatch,
+) -> None:
+    """#432 AC3: lifecycle refusal precedes even construction of a store path."""
+    from custom_components.houseplan import websocket_api as hp_ws
+
+    hp_ws.async_register(hass)
+
+    def forbidden_path(*_args, **_kwargs):
+        raise AssertionError("asset filesystem touched before the runtime gate")
+
+    monkeypatch.setattr(hp_ws, "Path", forbidden_path)
+    client = await hass_ws_client(hass)
+    await client.send_json_auto_id({
+        "type": "houseplan/assets/resolve", "asset_ids": ["a" * 64],
+    })
+    resp = await client.receive_json()
+    assert not resp["success"] and resp["error"]["code"] == "not_ready"
+
+
 async def test_plan_set_validates(hass: HomeAssistant, hass_ws_client: WebSocketGenerator) -> None:
     await _setup(hass)
     client = await hass_ws_client(hass)
@@ -2122,6 +2142,104 @@ async def test_decor_asset_upload_deduplicates_and_rejects_mime_spoofing(
     assert json.loads(spoofed.text)["error"] == "invalid_format"
 
 
+async def test_decor_asset_resolve_readonly_is_limited_to_referenced_ids(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    hass_read_only_access_token: str,
+    monkeypatch,
+) -> None:
+    """#432 AC1/AC2: View keeps its images without exposing the catalog."""
+    import hashlib
+
+    from custom_components.houseplan import websocket_api as wsapi
+    from custom_components.houseplan.const import ASSETS_DIR
+
+    await _setup(hass)
+    admin = await hass_ws_client(hass)
+    root = Path(hass.config.path(ASSETS_DIR))
+    root.mkdir(parents=True, exist_ok=True)
+    payloads = (b"referenced", b"not-referenced")
+    asset_ids = []
+    for index, payload in enumerate(payloads):
+        aid = hashlib.sha256(payload).hexdigest()
+        asset_ids.append(aid)
+        (root / f"{aid}.png").write_bytes(payload)
+        (root / f"{aid}.json").write_text(json.dumps({
+            "asset_id": aid, "name": f"{index}.png", "mime": "image/png",
+            "ext": ".png", "width": 1, "height": 1, "bytes": len(payload),
+            "created_at": f"2026-01-0{index + 1}T00:00:00Z",
+        }), encoding="utf-8")
+
+    cfg = await _cfg([{"id": "one", "plan_url": None}])
+    cfg["spaces"][0]["decor"] = [{
+        "id": "picture", "kind": "image", "asset_id": asset_ids[0],
+        "x": 0.1, "y": 0.2, "w": 0.3, "h": 0.4,
+    }]
+    assert (await _save(admin, cfg, 0))["success"]
+
+    looked_up = []
+    real_read_asset = wsapi.read_asset
+
+    def observed_read_asset(asset_root: Path, asset_id: str):
+        looked_up.append(asset_id)
+        return real_read_asset(asset_root, asset_id)
+
+    monkeypatch.setattr(wsapi, "read_asset", observed_read_asset)
+    readonly = await hass_ws_client(hass, access_token=hass_read_only_access_token)
+    await readonly.send_json_auto_id({
+        "type": "houseplan/assets/resolve", "asset_ids": asset_ids,
+    })
+    response = await readonly.receive_json()
+    assert response["success"]
+    assert [row["asset_id"] for row in response["result"]["assets"]] == [asset_ids[0]]
+    assert response["result"]["missing"] == [asset_ids[1]]
+    assert looked_up == [asset_ids[0]], "forbidden metadata/blob must not be touched"
+
+    looked_up.clear()
+    await admin.send_json_auto_id({
+        "type": "houseplan/assets/resolve", "asset_ids": asset_ids,
+    })
+    response = await admin.receive_json()
+    assert response["success"] and len(response["result"]["assets"]) == 2
+    assert set(looked_up) == set(asset_ids)
+
+
+async def test_decor_asset_resolve_non_admin_is_writer_when_admin_only_is_off(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    hass_read_only_access_token: str,
+) -> None:
+    """#432 AC2: resolve follows may_write instead of hard-coding admin."""
+    import hashlib
+
+    from custom_components.houseplan.const import ASSETS_DIR
+
+    entry = MockConfigEntry(
+        domain=DOMAIN, title="House Plan", data={}, options={CONF_ADMIN_ONLY: False},
+    )
+    entry.add_to_hass(hass)
+    assert await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+    payload = b"writer-by-option"
+    aid = hashlib.sha256(payload).hexdigest()
+    root = Path(hass.config.path(ASSETS_DIR))
+    root.mkdir(parents=True, exist_ok=True)
+    (root / f"{aid}.png").write_bytes(payload)
+    (root / f"{aid}.json").write_text(json.dumps({
+        "asset_id": aid, "name": "writer.png", "mime": "image/png", "ext": ".png",
+        "width": 1, "height": 1, "bytes": len(payload),
+        "created_at": "2026-01-01T00:00:00Z",
+    }), encoding="utf-8")
+
+    client = await hass_ws_client(hass, access_token=hass_read_only_access_token)
+    await client.send_json_auto_id({
+        "type": "houseplan/assets/resolve", "asset_ids": [aid],
+    })
+    response = await client.receive_json()
+    assert response["success"]
+    assert response["result"]["assets"][0]["asset_id"] == aid
+
+
 async def test_decor_asset_list_resolve_delete_and_signed_content(
     hass: HomeAssistant, hass_ws_client: WebSocketGenerator, hass_client_no_auth,
 ) -> None:
@@ -2130,6 +2248,7 @@ async def test_decor_asset_list_resolve_delete_and_signed_content(
     import hashlib
 
     from custom_components.houseplan.const import ASSETS_DIR, CONTENT_URL
+    from custom_components.houseplan.asset_integrity import AssetIntegrityVerifier
 
     await _setup(hass)
     client = await hass_ws_client(hass)
@@ -2144,6 +2263,16 @@ async def test_decor_asset_list_resolve_delete_and_signed_content(
         "asset_id": aid, "name": "pixel.png", "mime": "image/png", "ext": ".png",
         "width": 1, "height": 1, "bytes": len(png), "created_at": "2026-01-01T00:00:00Z",
     }), encoding="utf-8")
+    hash_calls = 0
+
+    def counted_hash(path: Path) -> str:
+        nonlocal hash_calls
+        hash_calls += 1
+        return hashlib.sha256(path.read_bytes()).hexdigest()
+
+    hass.data[DOMAIN]["asset_integrity_verifier"] = AssetIntegrityVerifier(
+        hasher=counted_hash,
+    )
 
     cfg = await _cfg([{"id": "one", "plan_url": None}])
     cfg["spaces"][0]["decor"] = [{
@@ -2174,11 +2303,21 @@ async def test_decor_asset_list_resolve_delete_and_signed_content(
     assert response.status == 200 and await response.read() == png
     assert response.headers["Content-Type"].startswith("image/png")
     assert response.headers["X-Content-Type-Options"] == "nosniff"
+    assert hash_calls == 1, "WS and HTTP must share one unchanged-file digest"
 
     (root / f"{aid}.png").write_bytes(b"tampered")
     assert (await http.get(signed)).status == 404
+    assert hash_calls == 2
 
     (root / f"{aid}.png").write_bytes(png)
+    assert (await http.get(signed)).status == 200
+    assert hash_calls == 3
+    await client.send_json_auto_id({
+        "type": "houseplan/assets/resolve", "asset_ids": [aid],
+    })
+    resolved_after_http = await client.receive_json()
+    assert resolved_after_http["success"]
+    assert hash_calls == 3, "HTTP and WS must share one unchanged-file digest"
     cfg["spaces"][0]["decor"] = []
     saved = await _save(client, cfg, saved["result"]["rev"])
     assert saved["success"]