Sync ETag sidecar mtime with cache file; treat mismatched sidecars as stale

The ETag sidecar describes a specific snapshot of the cache file. If the
cache file is replaced or edited out-of-band (manual edit, restore from
backup, another tool overwriting it), the sidecar's recorded ETag no
longer matches the bytes on disk -- using it would cause the server to
return 304 and we'd serve the wrong content from cache.

- _write_etag now os.utime()'s the sidecar to share the cache file's
  st_mtime_ns immediately after writing it.
- _read_etag compares the two mtimes; if they don't match it returns
  None and removes the sidecar so subsequent calls don't re-check it.

esphome/external_files.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import contextlib
 from datetime import UTC, datetime
 import logging
+import os
 from pathlib import Path
 
 import requests
@@ -31,8 +32,33 @@ def _etag_sidecar_path(local_file_path: Path) -> Path:
 
 
 def _read_etag(local_file_path: Path) -> str | None:
+    """Return the cached ETag if it still describes the current cached file.
+
+    The sidecar's mtime is kept in sync with the cache file's mtime by
+    `_write_etag`. If they disagree, the cache file was modified out-of-band
+    (manual edit, restored from backup, replaced by another tool) and the
+    ETag no longer corresponds to its contents, so treat the sidecar as
+    stale and delete it.
+    """
+    etag_path = _etag_sidecar_path(local_file_path)
     try:
-        etag = _etag_sidecar_path(local_file_path).read_text().strip()
+        etag_mtime_ns = etag_path.stat().st_mtime_ns
+        file_mtime_ns = local_file_path.stat().st_mtime_ns
+    except OSError:
+        return None
+    if etag_mtime_ns != file_mtime_ns:
+        _LOGGER.debug(
+            "ETag sidecar mtime (%d) does not match cached file mtime (%d) at %s; "
+            "treating ETag as stale",
+            etag_mtime_ns,
+            file_mtime_ns,
+            local_file_path,
+        )
+        with contextlib.suppress(OSError):
+            etag_path.unlink()
+        return None
+    try:
+        etag = etag_path.read_text().strip()
     except OSError:
         return None
     return etag or None
@@ -48,6 +74,16 @@ def _write_etag(local_file_path: Path, etag: str | None) -> None:
         write_file(etag_path, etag)
     except EsphomeError as e:
         _LOGGER.debug("Could not save ETag for %s: %s", local_file_path, e)
+        return
+    # Pin the sidecar's mtime to the cache file's mtime. _read_etag relies on
+    # this match to detect out-of-band edits to the cache file.
+    try:
+        file_mtime_ns = local_file_path.stat().st_mtime_ns
+        os.utime(etag_path, ns=(file_mtime_ns, file_mtime_ns))
+    except OSError as e:
+        _LOGGER.debug(
+            "Could not sync ETag sidecar mtime for %s: %s", local_file_path, e
+        )
 
 
 def has_remote_file_changed(url: str, local_file_path: Path) -> bool:

tests/unit_tests/test_external_files.py

@@ -1,5 +1,6 @@
 """Tests for external_files.py functions."""
 
+import os
 from pathlib import Path
 import time
 from unittest.mock import MagicMock, patch
@@ -12,6 +13,17 @@ from esphome.config_validation import Invalid
 from esphome.core import CORE, TimePeriod
 
 
+def _seed_etag(cache_file: Path, etag: str) -> Path:
+    """Write an ETag sidecar with its mtime synced to the cache file's mtime,
+    matching the invariant that `_write_etag` enforces in production.
+    """
+    sidecar = external_files._etag_sidecar_path(cache_file)
+    sidecar.write_text(etag)
+    file_mtime_ns = cache_file.stat().st_mtime_ns
+    os.utime(sidecar, ns=(file_mtime_ns, file_mtime_ns))
+    return sidecar
+
+
 def test_compute_local_file_dir(setup_core: Path) -> None:
     """Test compute_local_file_dir creates and returns correct path."""
     domain = "font"
@@ -185,7 +197,7 @@ def test_has_remote_file_changed_uses_etag(
     """Test has_remote_file_changed sends If-None-Match when ETag is cached."""
     test_file = setup_core / "cached.txt"
     test_file.write_text("cached content")
-    external_files._etag_sidecar_path(test_file).write_text('"abc123"')
+    _seed_etag(test_file, '"abc123"')
 
     mock_response = MagicMock()
     mock_response.status_code = 304
@@ -227,7 +239,7 @@ def test_has_remote_file_changed_refreshes_etag_on_304(
     """Test has_remote_file_changed updates the cached ETag when the 304 sends a new one."""
     test_file = setup_core / "cached.txt"
     test_file.write_text("cached content")
-    external_files._etag_sidecar_path(test_file).write_text('"old"')
+    _seed_etag(test_file, '"old"')
 
     mock_response = MagicMock()
     mock_response.status_code = 304
@@ -240,6 +252,63 @@ def test_has_remote_file_changed_refreshes_etag_on_304(
     assert external_files._etag_sidecar_path(test_file).read_text() == '"new"'
 
 
+@patch("esphome.external_files.requests.head")
+def test_has_remote_file_changed_ignores_etag_when_mtime_diverges(
+    mock_head: MagicMock, setup_core: Path
+) -> None:
+    """If the cache file was edited out-of-band (mtime no longer matches the
+    sidecar's), the cached ETag must not be used -- it no longer describes the
+    bytes on disk.
+    """
+    test_file = setup_core / "cached.txt"
+    test_file.write_text("cached content")
+    sidecar = _seed_etag(test_file, '"abc123"')
+
+    # Simulate an out-of-band edit to the cache file -- mtime advances but the
+    # sidecar is left untouched, so the recorded ETag is now stale.
+    file_stat = test_file.stat()
+    os.utime(
+        test_file, ns=(file_stat.st_atime_ns, file_stat.st_mtime_ns + 1_000_000_000)
+    )
+
+    mock_response = MagicMock()
+    mock_response.status_code = 304
+    mock_response.headers = {}
+    mock_head.return_value = mock_response
+
+    external_files.has_remote_file_changed("https://example.com/file.txt", test_file)
+
+    headers = mock_head.call_args[1]["headers"]
+    assert external_files.IF_NONE_MATCH not in headers
+    # Stale sidecar should be removed so future calls don't keep paying the
+    # mtime-comparison cost on a known-bad sidecar.
+    assert not sidecar.exists()
+
+
+@patch("esphome.external_files.requests.get")
+@patch("esphome.external_files.has_remote_file_changed")
+def test_download_content_pins_etag_mtime_to_file_mtime(
+    mock_has_changed: MagicMock,
+    mock_get: MagicMock,
+    setup_core: Path,
+) -> None:
+    """After a successful download, the sidecar's mtime must equal the cache
+    file's mtime so `_read_etag` accepts it on the next call.
+    """
+    test_file = setup_core / "fresh.txt"
+    mock_has_changed.return_value = True
+    mock_response = MagicMock()
+    mock_response.content = b"fresh content"
+    mock_response.headers = {external_files.ETAG: '"deadbeef"'}
+    mock_response.raise_for_status = MagicMock()
+    mock_get.return_value = mock_response
+
+    external_files.download_content("https://example.com/file.txt", test_file)
+
+    sidecar = external_files._etag_sidecar_path(test_file)
+    assert sidecar.stat().st_mtime_ns == test_file.stat().st_mtime_ns
+
+
 def test_compute_local_file_dir_creates_parent_dirs(setup_core: Path) -> None:
     """Test compute_local_file_dir creates parent directories."""
     domain = "level1/level2/level3/level4"