From 8f2c09e3c86b0deb5b8542c61a53db97b60667d5 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 3 Jul 2026 20:40:14 -0500 Subject: [PATCH] [store_yaml] Simplify redaction pipeline, fix Windows newline test failure --- esphome/components/store_yaml/__init__.py | 93 ++++++++++--------- tests/integration/test_store_yaml_recovery.py | 3 +- .../unit_tests/components/test_store_yaml.py | 29 ++++-- tests/unit_tests/conftest.py | 11 +++ tests/unit_tests/test_yaml_util.py | 28 ++---- 5 files changed, 89 insertions(+), 75 deletions(-) diff --git a/esphome/components/store_yaml/__init__.py b/esphome/components/store_yaml/__init__.py index 6b3910bfa3..f90b8807ef 100644 --- a/esphome/components/store_yaml/__init__.py +++ b/esphome/components/store_yaml/__init__.py @@ -85,11 +85,12 @@ FINAL_VALIDATE_SCHEMA = _final_validate def _gather_files( discovered: yaml_util.DiscoveredYamlFiles, ) -> tuple[list[tuple[str, bytes]], set[str]]: - """Read each discovered YAML file verbatim. + """Map each discovered YAML file to its envelope path. - Returns (relative_path, content) pairs plus the subset of relative paths - that are secrets files (matched upstream on the *un-resolved* basename, so - a `secrets.yaml` symlinked to a differently-named target is still flagged). + Returns (relative_path, source_path) pairs plus the subset of relative + paths that are secrets files (matched upstream on the *un-resolved* + basename, so a `secrets.yaml` symlinked to a differently-named target is + still flagged). """ if not discovered.files: raise EsphomeError( @@ -113,19 +114,11 @@ def _gather_files( ", ".join(discovered.unresolved), ) - config_path = Path(CORE.config_path).resolve() - root = config_path.parent + root = Path(CORE.config_path).resolve().parent - files: list[tuple[str, bytes]] = [] + entries: list[tuple[str, Path]] = [] secret_rels: set[str] = set() for path in discovered.files: - try: - content = path.read_bytes() - except OSError as err: - raise EsphomeError( - f"store_yaml: cannot read tracked YAML file {path}: {err}" - ) from err - try: rel_str = path.relative_to(root).as_posix() except ValueError: @@ -137,9 +130,24 @@ def _gather_files( if path in discovered.secrets: secret_rels.add(rel_str) - files.append((rel_str, content)) + entries.append((rel_str, path)) - return files, secret_rels + return entries, secret_rels + + +def _read_files_verbatim(entries: list[tuple[str, Path]]) -> list[tuple[str, bytes]]: + """Read each file's exact on-disk bytes (the `include_secrets: true` path).""" + files: list[tuple[str, bytes]] = [] + for rel, path in entries: + try: + files.append((rel, path.read_bytes())) + except OSError as err: + # A silently partial recovery blob defeats the feature; fail the + # build instead of embedding an incomplete file set. + raise EsphomeError( + f"store_yaml: cannot read tracked YAML file {path}: {err}" + ) from err + return files def _iter_sensitive_values(node: object, path: tuple[str, ...] = ()): @@ -158,25 +166,23 @@ def _iter_sensitive_values(node: object, path: tuple[str, ...] = ()): class _SensitiveValue: secret_name: str config_path: str # dotted path, for warnings (never log the value itself) - from_secret: bool # already loaded via !secret somewhere -def _collect_sensitive_values(reserved_names: set[str]) -> dict[str, _SensitiveValue]: +def _collect_sensitive_values() -> dict[str, _SensitiveValue]: """Map each cv.sensitive value in the validated config to the `!secret` name it should be recovered as. Values that already come from `!secret` keep their existing name; inline - values get a name generated from their config path, avoiding - `reserved_names`. + values get a name generated from their config path, avoiding names already + taken by real secrets. """ - used = set(reserved_names) + used = yaml_util.registered_secret_names() result: dict[str, _SensitiveValue] = {} for path, value in _iter_sensitive_values(CORE.config): if not value or value in result: continue - if existing := yaml_util.is_secret(value): - name = existing - else: + name = yaml_util.is_secret(value) + if name is None: base = "_".join(path) or "secret" name = base counter = 2 @@ -184,7 +190,7 @@ def _collect_sensitive_values(reserved_names: set[str]) -> dict[str, _SensitiveV name = f"{base}_{counter}" counter += 1 used.add(name) - result[value] = _SensitiveValue(name, ".".join(path), bool(existing)) + result[value] = _SensitiveValue(name, ".".join(path)) return result @@ -195,7 +201,7 @@ def _build_secrets_skeleton(keys: set[str]) -> bytes: def _generate_redacted_files( - files: list[tuple[str, bytes]], secret_rels: set[str] + entries: list[tuple[str, Path]], secret_rels: set[str] ) -> list[tuple[str, bytes]]: """Re-generate each captured file from its parse tree with cv.sensitive values emitted as `!secret ` references, and replace secrets files @@ -207,29 +213,26 @@ def _generate_redacted_files( `!include` references round-trip via the dumper's IncludeFile support; comments and formatting of the originals are not preserved. """ - sensitive = _collect_sensitive_values(yaml_util.registered_secret_names()) - inline = { - value: info.secret_name - for value, info in sensitive.items() - if not info.from_secret - } + sensitive = _collect_sensitive_values() - config_path = Path(CORE.config_path).resolve() - root = config_path.parent texts: dict[str, str] = {} - with yaml_util.secret_values_registered(inline): - for rel, _ in files: + registered = {value: info.secret_name for value, info in sensitive.items()} + with yaml_util.secret_values_registered(registered): + for rel, path in entries: if rel in secret_rels: continue - tree = yaml_util.load_yaml(root / rel, clear_secrets=False) + tree = yaml_util.load_yaml(path, clear_secrets=False) texts[rel] = yaml_util.dump(tree) skeleton_keys: set[str] = set() for text in texts.values(): skeleton_keys |= yaml_util.find_secret_references(text) - for info in sensitive.values(): - if not info.from_secret and info.secret_name not in skeleton_keys: + for value, info in sensitive.items(): + # After the context manager exits, only values loaded through a real + # `!secret` are still registered — those legitimately never appear + # inline, so only warn for inline values that were never swapped. + if yaml_util.is_secret(value) is None and info.secret_name not in skeleton_keys: _LOGGER.warning( "store_yaml: could not locate the sensitive value of '%s' in the " "source YAML (built via substitutions?); it may still be " @@ -240,13 +243,13 @@ def _generate_redacted_files( skeleton = _build_secrets_skeleton(skeleton_keys) result = [ (rel, skeleton if rel in secret_rels else texts[rel].encode("utf-8")) - for rel, _ in files + for rel, _ in entries ] if skeleton_keys and not secret_rels: # The generated files reference `!secret` keys but the project has no # secrets file (all secrets were inline) — ship a synthetic one so the # recovered config is complete. - result.append(("secrets.yaml", skeleton)) + result.append((yaml_util.SECRET_YAML, skeleton)) return result @@ -309,9 +312,11 @@ async def to_code(config: ConfigType) -> None: # that components load internally (e.g. LVGL's `hello_world.yaml`), and # costs nothing on validate-only runs or configs without this component. discovered = yaml_util.discover_user_yaml_files(CORE.config_path) - files, secret_rels = _gather_files(discovered) - if not config[CONF_INCLUDE_SECRETS]: - files = _generate_redacted_files(files, secret_rels) + entries, secret_rels = _gather_files(discovered) + if config[CONF_INCLUDE_SECRETS]: + files = _read_files_verbatim(entries) + else: + files = _generate_redacted_files(entries, secret_rels) envelope = _pack_envelope(files) compressed = zstd.compress(envelope, level=ZSTD_LEVEL) diff --git a/tests/integration/test_store_yaml_recovery.py b/tests/integration/test_store_yaml_recovery.py index 98e08882b9..8470dbcdde 100644 --- a/tests/integration/test_store_yaml_recovery.py +++ b/tests/integration/test_store_yaml_recovery.py @@ -25,6 +25,7 @@ except ImportError: from backports import zstd # type: ignore[import-not-found, no-redef] from esphome.components.store_yaml import unpack_envelope +from esphome.yaml_util import find_secret_references from .types import RunCompiledFunction @@ -212,7 +213,7 @@ async def test_store_yaml_recovery( assert b"recoverme123" not in envelope, ( "inline sensitive value leaked into the recovery blob" ) - assert b"!secret 'ota_password'" in combined, ( + assert "ota_password" in find_secret_references(combined.decode()), ( "expected the inline OTA password to be recovered as a !secret reference" ) assert b'ota_password: ""' in files["secrets.yaml"], ( diff --git a/tests/unit_tests/components/test_store_yaml.py b/tests/unit_tests/components/test_store_yaml.py index 73372ead45..2e911b2d22 100644 --- a/tests/unit_tests/components/test_store_yaml.py +++ b/tests/unit_tests/components/test_store_yaml.py @@ -13,6 +13,7 @@ from esphome.components.store_yaml import ( _gather_files, _generate_redacted_files, _pack_envelope, + _read_files_verbatim, unpack_envelope, ) from esphome.core import CORE, EsphomeError @@ -35,8 +36,6 @@ def project(tmp_path: Path) -> Path: @pytest.fixture(autouse=True) def _clear_config() -> None: CORE.config = {} - yield - yaml_util._SECRET_VALUES.clear() def _sources( @@ -58,16 +57,29 @@ def _gather_redacted(discovered: DiscoveredYamlFiles) -> dict[str, bytes]: # --------------------------------------------------------------------------- -def test_gather_returns_verbatim_content_and_flags_secrets(project: Path) -> None: +def test_gather_maps_rel_paths_and_flags_secrets(project: Path) -> None: discovered = _sources( project, "entry.yaml", "secrets.yaml", secrets=("secrets.yaml",) ) - files, secret_rels = _gather_files(discovered) - contents = dict(files) - assert contents["secrets.yaml"] == b"api_key: SUPER_SECRET\n" + entries, secret_rels = _gather_files(discovered) + assert dict(entries) == { + "entry.yaml": project / "entry.yaml", + "secrets.yaml": project / "secrets.yaml", + } assert secret_rels == {"secrets.yaml"} +def test_read_files_verbatim_returns_exact_bytes(project: Path) -> None: + """`include_secrets: true` embeds the on-disk bytes untouched.""" + discovered = _sources( + project, "entry.yaml", "secrets.yaml", secrets=("secrets.yaml",) + ) + entries, _ = _gather_files(discovered) + contents = dict(_read_files_verbatim(entries)) + assert contents["secrets.yaml"] == (project / "secrets.yaml").read_bytes() + assert contents["entry.yaml"] == (project / "entry.yaml").read_bytes() + + def test_gather_flags_secret_symlinked_to_other_name( project: Path, tmp_path: Path ) -> None: @@ -117,12 +129,13 @@ def test_gather_raises_on_load_errors(project: Path) -> None: _gather_files(discovered) -def test_gather_raises_on_unreadable_file( +def test_read_files_verbatim_raises_on_unreadable_file( project: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """An unreadable tracked file fails the build instead of producing a silently partial recovery blob.""" discovered = _sources(project, "entry.yaml", "wifi.yaml") + entries, _ = _gather_files(discovered) orig_read_bytes = Path.read_bytes def fake_read_bytes(self: Path) -> bytes: @@ -132,7 +145,7 @@ def test_gather_raises_on_unreadable_file( monkeypatch.setattr(Path, "read_bytes", fake_read_bytes) with pytest.raises(EsphomeError, match="wifi.yaml"): - _gather_files(discovered) + _read_files_verbatim(entries) def test_gather_warns_on_unresolved_includes( diff --git a/tests/unit_tests/conftest.py b/tests/unit_tests/conftest.py index 13450b10f0..aec23aaf6d 100644 --- a/tests/unit_tests/conftest.py +++ b/tests/unit_tests/conftest.py @@ -16,6 +16,7 @@ from unittest.mock import Mock, patch import pytest +from esphome import yaml_util from esphome.core import CORE here = Path(__file__).parent @@ -32,6 +33,16 @@ def reset_core(): CORE.reset() +@pytest.fixture(autouse=True) +def clear_yaml_secrets(): + """Isolate the yaml_util secrets registry between tests.""" + yaml_util._SECRET_VALUES.clear() + yaml_util._SECRET_CACHE.clear() + yield + yaml_util._SECRET_VALUES.clear() + yaml_util._SECRET_CACHE.clear() + + @pytest.fixture def fixture_path() -> Path: """ diff --git a/tests/unit_tests/test_yaml_util.py b/tests/unit_tests/test_yaml_util.py index 749e2b9710..00f67e7e5e 100644 --- a/tests/unit_tests/test_yaml_util.py +++ b/tests/unit_tests/test_yaml_util.py @@ -25,16 +25,6 @@ from esphome.yaml_util import ( ) -@pytest.fixture(autouse=True) -def clear_secrets_cache() -> None: - """Clear the secrets cache before each test.""" - yaml_util._SECRET_VALUES.clear() - yaml_util._SECRET_CACHE.clear() - yield - yaml_util._SECRET_VALUES.clear() - yaml_util._SECRET_CACHE.clear() - - @pytest.fixture(autouse=True) def clear_core_frontmatter() -> None: """Reset CORE.frontmatter between tests.""" @@ -1446,22 +1436,16 @@ def test_secret_values_registered_swaps_scalars_in_dump() -> None: def test_secret_values_registered_does_not_clobber_real_secrets() -> None: """A value already mapped by a real `!secret` keeps its original name.""" yaml_util._SECRET_VALUES["hunter2"] = "original_name" - try: - with yaml_util.secret_values_registered({"hunter2": "generated_name"}): - out = yaml_util.dump({"password": make_data_base("hunter2")}) - assert "!secret 'original_name'" in out - # The pre-existing mapping survives the context exit. - assert yaml_util.is_secret("hunter2") == "original_name" - finally: - yaml_util._SECRET_VALUES.clear() + with yaml_util.secret_values_registered({"hunter2": "generated_name"}): + out = yaml_util.dump({"password": make_data_base("hunter2")}) + assert "!secret 'original_name'" in out + # The pre-existing mapping survives the context exit. + assert yaml_util.is_secret("hunter2") == "original_name" def test_registered_secret_names() -> None: yaml_util._SECRET_VALUES["value_a"] = "name_a" - try: - assert "name_a" in yaml_util.registered_secret_names() - finally: - yaml_util._SECRET_VALUES.clear() + assert "name_a" in yaml_util.registered_secret_names() @pytest.fixture(autouse=True)