From a25b7a806b43f00f14ff5c108d4f6def058bbb05 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 3 Jul 2026 20:31:32 -0500 Subject: [PATCH] [store_yaml] Recover secrets as !secret references using cv.sensitive --- esphome/bundle.py | 12 +- esphome/components/api/api_connection.cpp | 7 +- esphome/components/store_yaml/__init__.py | 193 ++++++++++--- esphome/yaml_util.py | 81 +++++- .../fixtures/store_yaml_recovery.yaml | 4 + tests/integration/test_store_yaml_recovery.py | 13 + .../unit_tests/components/test_store_yaml.py | 263 +++++++++++++++--- tests/unit_tests/test_yaml_util.py | 59 +++- 8 files changed, 526 insertions(+), 106 deletions(-) diff --git a/esphome/bundle.py b/esphome/bundle.py index d38f68ebfd0..fbeeb1815f6 100644 --- a/esphome/bundle.py +++ b/esphome/bundle.py @@ -98,15 +98,6 @@ _KNOWN_FILE_EXTENSIONS = frozenset( ) -# Matches !secret references in YAML text. An optional surrounding -# quote pair around the key is allowed and ignored: YAML treats -# ``!secret 'foo'`` and ``!secret foo`` as the same key. This is -# intentionally a simple regex scan rather than a YAML parse — it may -# match inside comments or multi-line strings, which is the conservative -# direction (include more secrets rather than fewer). -_SECRET_RE = re.compile(r"""!secret\s+['"]?([^\s'"]+)""") - - def _find_used_secret_keys(yaml_files: list[Path]) -> set[str]: """Scan YAML files for ``!secret `` references.""" keys: set[str] = set() @@ -115,8 +106,7 @@ def _find_used_secret_keys(yaml_files: list[Path]) -> set[str]: text = fpath.read_text(encoding="utf-8") except (OSError, UnicodeDecodeError): continue - for match in _SECRET_RE.finditer(text): - keys.add(match.group(1)) + keys |= yaml_util.find_secret_references(text) return keys diff --git a/esphome/components/api/api_connection.cpp b/esphome/components/api/api_connection.cpp index 2741ff9c904..5af7bfffad0 100644 --- a/esphome/components/api/api_connection.cpp +++ b/esphome/components/api/api_connection.cpp @@ -1180,9 +1180,10 @@ static uint8_t store_yaml_chunk_buf[STORE_YAML_CHUNK_SIZE]; #endif void APIConnection::on_get_yaml_request() { - if (store_yaml::global_store_yaml == nullptr) { - // Request arrived before the component's setup() ran — send a single - // done=true response so the client doesn't hang. + auto *comp = store_yaml::global_store_yaml; + if (comp == nullptr || comp->get_size() == 0) { + // No component yet (request before setup()) or empty blob — send a single + // done=true response so the client always gets a terminal frame. GetYamlResponse resp; resp.done = true; this->send_message(resp); diff --git a/esphome/components/store_yaml/__init__.py b/esphome/components/store_yaml/__init__.py index c71ccb7669f..6b3910bfa37 100644 --- a/esphome/components/store_yaml/__init__.py +++ b/esphome/components/store_yaml/__init__.py @@ -1,5 +1,6 @@ from __future__ import annotations +from dataclasses import dataclass import logging import os from pathlib import Path @@ -37,8 +38,12 @@ StoreYamlComponent = store_yaml_ns.class_("StoreYamlComponent", cg.Component) ZSTD_LEVEL = 22 # Envelope magic: "EHY1" = ESPHome YAML, version 1. ENVELOPE_MAGIC = b"EHY1" -# Replacement content when secrets are not included. -REDACTED_PLACEHOLDER = b"# redacted\n" +# Replacement content for secrets files: a fill-in skeleton listing every +# `!secret` key the recovered config needs. +SECRETS_SKELETON_HEADER = ( + "# Redacted by store_yaml. Fill in these values and the recovered\n" + "# config is ready to flash.\n" +) CONFIG_SCHEMA = cv.Schema( { @@ -78,15 +83,28 @@ FINAL_VALIDATE_SCHEMA = _final_validate def _gather_files( - discovered: yaml_util.DiscoveredYamlFiles, include_secrets: bool -) -> list[tuple[str, bytes]]: - """Read each discovered YAML file, return (relative_path, content) pairs.""" + discovered: yaml_util.DiscoveredYamlFiles, +) -> tuple[list[tuple[str, bytes]], set[str]]: + """Read each discovered YAML file verbatim. + + 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). + """ if not discovered.files: raise EsphomeError( "store_yaml could not discover any YAML files for " f"{CORE.config_path}; nothing to embed." ) + if discovered.load_errors: + # A silently partial recovery blob defeats the feature; fail the build + # instead of embedding an incomplete file set. + raise EsphomeError( + "store_yaml: could not load all configuration files: " + + "; ".join(discovered.load_errors) + ) + if discovered.unresolved: _LOGGER.warning( "store_yaml: %d !include path(s) use substitutions and cannot be " @@ -97,24 +115,16 @@ def _gather_files( config_path = Path(CORE.config_path).resolve() root = config_path.parent - secret_paths = discovered.secrets files: list[tuple[str, bytes]] = [] + secret_rels: set[str] = set() for path in discovered.files: - # `secret_paths` was collected from the *un-resolved* basename, so a - # `secrets.yaml` symlinked to a differently-named target is still - # treated as secrets here. - if path in secret_paths and not include_secrets: - content = REDACTED_PLACEHOLDER - else: - try: - content = 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 + 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() @@ -125,9 +135,119 @@ def _gather_files( # different directories with the same basename don't collide. rel_str = os.path.relpath(path, root).replace(os.sep, "/") + if path in discovered.secrets: + secret_rels.add(rel_str) files.append((rel_str, content)) - return files + return files, secret_rels + + +def _iter_sensitive_values(node: object, path: tuple[str, ...] = ()): + """Yield (config_path, value) for every cv.sensitive value in a config tree.""" + if isinstance(node, yaml_util.SensitiveStr): + yield path, str(node) + elif isinstance(node, dict): + for key, value in node.items(): + yield from _iter_sensitive_values(value, (*path, str(key))) + elif isinstance(node, (list, tuple)): + for item in node: + yield from _iter_sensitive_values(item, path) + + +@dataclass +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]: + """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`. + """ + used = set(reserved_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: + base = "_".join(path) or "secret" + name = base + counter = 2 + while name in used: + name = f"{base}_{counter}" + counter += 1 + used.add(name) + result[value] = _SensitiveValue(name, ".".join(path), bool(existing)) + return result + + +def _build_secrets_skeleton(keys: set[str]) -> bytes: + parts = [SECRETS_SKELETON_HEADER] + parts.extend(f'{key}: ""\n' for key in sorted(keys)) + return "".join(parts).encode("utf-8") + + +def _generate_redacted_files( + files: list[tuple[str, bytes]], 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 + with a fill-in skeleton — the recovered config is flashable once the user + restores their secrets.yaml values. + + The swap happens inside the YAML dumper (`represent_stringify` consults + the registered secret values), not by mutating text afterwards. Nested + `!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 + } + + 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: + if rel in secret_rels: + continue + tree = yaml_util.load_yaml(root / rel, 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: + _LOGGER.warning( + "store_yaml: could not locate the sensitive value of '%s' in the " + "source YAML (built via substitutions?); it may still be " + "embedded verbatim", + info.config_path, + ) + + skeleton = _build_secrets_skeleton(skeleton_keys) + result = [ + (rel, skeleton if rel in secret_rels else texts[rel].encode("utf-8")) + for rel, _ in files + ] + 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)) + return result def _pack_envelope(files: list[tuple[str, bytes]]) -> bytes: @@ -156,18 +276,25 @@ def unpack_envelope(blob: bytes) -> dict[str, bytes]: if blob[:4] != ENVELOPE_MAGIC: raise EsphomeError("envelope must start with EHY1 magic") pos = 4 - (count,) = struct.unpack_from(" len(blob): + raise EsphomeError("truncated envelope") + path = blob[pos : pos + path_len].decode("utf-8") + pos += path_len + (content_len,) = struct.unpack_from(" len(blob): + raise EsphomeError("truncated envelope") + files[path] = blob[pos : pos + content_len] + pos += content_len + except struct.error as err: + raise EsphomeError(f"truncated envelope: {err}") from err if pos != len(blob): raise EsphomeError("envelope has trailing bytes") return files @@ -182,7 +309,9 @@ 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 = _gather_files(discovered, config[CONF_INCLUDE_SECRETS]) + files, secret_rels = _gather_files(discovered) + if not config[CONF_INCLUDE_SECRETS]: + files = _generate_redacted_files(files, secret_rels) envelope = _pack_envelope(files) compressed = zstd.compress(envelope, level=ZSTD_LEVEL) diff --git a/esphome/yaml_util.py b/esphome/yaml_util.py index 038d1d07065..d20c6302577 100644 --- a/esphome/yaml_util.py +++ b/esphome/yaml_util.py @@ -11,7 +11,8 @@ import logging import math import os from pathlib import Path -from typing import Any +import re +from typing import Any, NamedTuple import uuid import yaml @@ -267,11 +268,37 @@ class IncludeFile: return has_substitution_or_expression(str(self.file)) +# Matches !secret references in YAML text. An optional surrounding +# quote pair around the key is allowed and ignored: YAML treats +# ``!secret 'foo'`` and ``!secret foo`` as the same key. This is +# intentionally a simple regex scan rather than a YAML parse — it may +# match inside comments or multi-line strings, which is the conservative +# direction (include more secrets rather than fewer). +_SECRET_REFERENCE_RE = re.compile(r"""!secret\s+['"]?([^\s'"]+)""") + + +def find_secret_references(text: str) -> set[str]: + """Return the ``!secret `` names referenced in a YAML document text.""" + return {match.group(1) for match in _SECRET_REFERENCE_RE.finditer(text)} + + +class ForceLoadResult(NamedTuple): + """Outcome of :func:`force_load_include_files`. + + ``unresolved`` lists ``!include`` path strings that contain substitution + variables and therefore could not be loaded; ``errors`` lists includes + that failed to load. Either being non-empty means the walk was incomplete. + """ + + unresolved: list[str] + errors: list[str] + + def force_load_include_files( obj: Any, *, warn_on_unresolved: bool = True, -) -> list[str]: +) -> ForceLoadResult: """Recursively resolve any deferred ``IncludeFile`` instances in a YAML tree. Nested ``!include`` returns a deferred ``IncludeFile`` that is only resolved @@ -284,11 +311,11 @@ def force_load_include_files( variables cannot be loaded. By default a warning is logged for each one; pass ``warn_on_unresolved=False`` (used by discovery paths that run on a fresh re-parse where substitutions haven't been applied yet) to demote it - to a debug log. Returns the path strings of those unloadable includes so - callers can tell the walk was incomplete. + to a debug log. """ seen: set[int] = set() unresolved: list[str] = [] + errors: list[str] = [] def walk(node: Any) -> None: if not isinstance(node, (IncludeFile, dict, list, tuple)) or id(node) in seen: @@ -313,6 +340,7 @@ def force_load_include_files( node.parent_file, err, ) + errors.append(f"{node.file}: {err}") return walk(loaded) elif isinstance(node, dict): @@ -323,7 +351,7 @@ def force_load_include_files( walk(item) walk(obj) - return unresolved + return ForceLoadResult(unresolved, errors) @dataclass(slots=True) @@ -335,13 +363,15 @@ class DiscoveredYamlFiles: *un-resolved* filename matched :data:`esphome.const.SECRETS_FILES` (so a ``secrets.yaml`` symlinked to a differently-named target is still flagged as secrets). ``unresolved`` lists ``!include`` path strings that - contain substitution variables and therefore could not be loaded — - consumers should treat ``files`` as incomplete when it is non-empty. + contain substitution variables and therefore could not be loaded, and + ``load_errors`` lists files that failed to parse or load — consumers + should treat ``files`` as incomplete when either is non-empty. """ files: list[Path] = field(default_factory=list) secrets: set[Path] = field(default_factory=set) unresolved: list[str] = field(default_factory=list) + load_errors: list[str] = field(default_factory=list) def discover_user_yaml_files(config_path: Path) -> DiscoveredYamlFiles: @@ -371,9 +401,16 @@ def discover_user_yaml_files(config_path: Path) -> DiscoveredYamlFiles: try: try: data = load_yaml(config_path) - except EsphomeError: - return DiscoveredYamlFiles(list(loaded), secrets) - unresolved = force_load_include_files(data, warn_on_unresolved=False) + except EsphomeError as err: + _LOGGER.warning( + "YAML discovery failed to parse %s: %s", config_path, err + ) + return DiscoveredYamlFiles( + list(loaded), secrets, load_errors=[f"{config_path}: {err}"] + ) + unresolved, load_errors = force_load_include_files( + data, warn_on_unresolved=False + ) finally: _load_listeners.remove(_capture_secret) @@ -384,7 +421,7 @@ def discover_user_yaml_files(config_path: Path) -> DiscoveredYamlFiles: if path not in seen: seen.add(path) unique.append(path) - return DiscoveredYamlFiles(unique, secrets, unresolved) + return DiscoveredYamlFiles(unique, secrets, unresolved, load_errors) def _add_data_ref(fn): @@ -836,6 +873,28 @@ def _load_yaml_internal_with_type( loader.dispose() +def registered_secret_names() -> set[str]: + """Names of all ``!secret`` keys the loader has seen since the last clear.""" + return set(_SECRET_VALUES.values()) + + +@contextmanager +def secret_values_registered(values: dict[str, str]) -> Generator[None]: + """Temporarily register value→name mappings so :func:`dump` renders those + scalars as ``!secret ``. + + Mappings already present in ``_SECRET_VALUES`` (values loaded through a + real ``!secret``) win over the supplied ones and are left untouched. + """ + added = {v: n for v, n in values.items() if v not in _SECRET_VALUES} + _SECRET_VALUES.update(added) + try: + yield + finally: + for value in added: + _SECRET_VALUES.pop(value, None) + + def dump(dict_, show_secrets=False, sort_keys=False): """Dump YAML to a string and remove null.""" if show_secrets: diff --git a/tests/integration/fixtures/store_yaml_recovery.yaml b/tests/integration/fixtures/store_yaml_recovery.yaml index d55100c17d6..c416e41f50f 100644 --- a/tests/integration/fixtures/store_yaml_recovery.yaml +++ b/tests/integration/fixtures/store_yaml_recovery.yaml @@ -10,5 +10,9 @@ logger: api: +ota: + - platform: esphome + password: recoverme123 + store_yaml: allow_unencrypted: true diff --git a/tests/integration/test_store_yaml_recovery.py b/tests/integration/test_store_yaml_recovery.py index f8248779606..98e08882b93 100644 --- a/tests/integration/test_store_yaml_recovery.py +++ b/tests/integration/test_store_yaml_recovery.py @@ -205,3 +205,16 @@ async def test_store_yaml_recovery( assert b"store_yaml:" in combined, ( "expected the store_yaml config line to be in the recovery blob" ) + + # The inline cv.sensitive OTA password must be recovered as a `!secret` + # reference, never as its raw value, and the synthetic secrets.yaml + # skeleton must list the key so the recovered config is flashable. + assert b"recoverme123" not in envelope, ( + "inline sensitive value leaked into the recovery blob" + ) + assert b"!secret 'ota_password'" in combined, ( + "expected the inline OTA password to be recovered as a !secret reference" + ) + assert b'ota_password: ""' in files["secrets.yaml"], ( + "expected the secrets.yaml skeleton to list the ota_password key" + ) diff --git a/tests/unit_tests/components/test_store_yaml.py b/tests/unit_tests/components/test_store_yaml.py index 40b6ca7da80..73372ead45a 100644 --- a/tests/unit_tests/components/test_store_yaml.py +++ b/tests/unit_tests/components/test_store_yaml.py @@ -1,4 +1,5 @@ -"""Tests for the store_yaml component's file gathering and envelope packing.""" +"""Tests for the store_yaml component's file gathering, secret redaction, and +envelope packing.""" from __future__ import annotations @@ -6,14 +7,16 @@ from pathlib import Path import pytest +from esphome import yaml_util from esphome.components.store_yaml import ( - REDACTED_PLACEHOLDER, + SECRETS_SKELETON_HEADER, _gather_files, + _generate_redacted_files, _pack_envelope, unpack_envelope, ) from esphome.core import CORE, EsphomeError -from esphome.yaml_util import DiscoveredYamlFiles +from esphome.yaml_util import DiscoveredYamlFiles, SensitiveStr @pytest.fixture @@ -21,12 +24,21 @@ def project(tmp_path: Path) -> Path: """Lay out a tiny ESPHome-like project: entry yaml, an include, and a secrets file.""" project_dir = tmp_path / "project" project_dir.mkdir() - (project_dir / "entry.yaml").write_text("esphome:\n name: test\n") + (project_dir / "entry.yaml").write_text( + "esphome:\n name: test\napi:\n encryption:\n key: !secret api_key\n" + ) (project_dir / "wifi.yaml").write_text("ssid: my_ssid\npassword: my_password\n") (project_dir / "secrets.yaml").write_text("api_key: SUPER_SECRET\n") return project_dir +@pytest.fixture(autouse=True) +def _clear_config() -> None: + CORE.config = {} + yield + yaml_util._SECRET_VALUES.clear() + + def _sources( project_dir: Path, *names: str, secrets: tuple[str, ...] = () ) -> DiscoveredYamlFiles: @@ -36,34 +48,30 @@ def _sources( return DiscoveredYamlFiles(files, secret_paths) -def test_gather_redacts_secrets_by_default(project: Path) -> None: +def _gather_redacted(discovered: DiscoveredYamlFiles) -> dict[str, bytes]: + files, secret_rels = _gather_files(discovered) + return dict(_generate_redacted_files(files, secret_rels)) + + +# --------------------------------------------------------------------------- +# _gather_files +# --------------------------------------------------------------------------- + + +def test_gather_returns_verbatim_content_and_flags_secrets(project: Path) -> None: discovered = _sources( - project, - "entry.yaml", - "wifi.yaml", - "secrets.yaml", - secrets=("secrets.yaml",), + project, "entry.yaml", "secrets.yaml", secrets=("secrets.yaml",) ) - files = dict(_gather_files(discovered, include_secrets=False)) - assert files["secrets.yaml"] == REDACTED_PLACEHOLDER - assert b"SUPER_SECRET" not in files["secrets.yaml"] - assert files["wifi.yaml"] == (project / "wifi.yaml").read_bytes() + files, secret_rels = _gather_files(discovered) + contents = dict(files) + assert contents["secrets.yaml"] == b"api_key: SUPER_SECRET\n" + assert secret_rels == {"secrets.yaml"} -def test_gather_redacts_yml_extension(project: Path) -> None: - yml = project / "secrets.yml" - yml.write_text("api_key: OTHER_SECRET\n") - discovered = _sources( - project, "entry.yaml", "secrets.yml", secrets=("secrets.yml",) - ) - files = dict(_gather_files(discovered, include_secrets=False)) - assert files["secrets.yml"] == REDACTED_PLACEHOLDER - - -def test_gather_redacts_secret_symlinked_to_other_name( +def test_gather_flags_secret_symlinked_to_other_name( project: Path, tmp_path: Path ) -> None: - """A `secrets.yaml` symlinked to a non-secrets-named target is still redacted + """A `secrets.yaml` symlinked to a non-secrets-named target is still flagged because the un-resolved basename was captured upstream.""" target = tmp_path / "actual_creds.yaml" target.write_text("api_key: FROM_SYMLINK\n") @@ -74,20 +82,10 @@ def test_gather_redacts_secret_symlinked_to_other_name( # but stores the resolved path; mimic that here. resolved = link.resolve() CORE.config_path = project / "entry.yaml" - discovered = DiscoveredYamlFiles([resolved], {resolved}) - files = dict(_gather_files(discovered, include_secrets=False)) - assert REDACTED_PLACEHOLDER in files.values() + files = _gather_redacted(DiscoveredYamlFiles([resolved], {resolved})) assert b"FROM_SYMLINK" not in b"".join(files.values()) -def test_gather_embeds_secrets_when_opted_in(project: Path) -> None: - discovered = _sources( - project, "entry.yaml", "secrets.yaml", secrets=("secrets.yaml",) - ) - files = dict(_gather_files(discovered, include_secrets=True)) - assert b"SUPER_SECRET" in files["secrets.yaml"] - - def test_gather_uses_relative_path_for_external_files( project: Path, tmp_path: Path ) -> None: @@ -96,16 +94,27 @@ def test_gather_uses_relative_path_for_external_files( sibling.write_text("foo: bar\n") CORE.config_path = project / "entry.yaml" discovered = DiscoveredYamlFiles([project / "entry.yaml", sibling], set()) - files = dict(_gather_files(discovered, include_secrets=False)) + files, _ = _gather_files(discovered) # project root is `tmp_path/project`, sibling is in `tmp_path` so it # resolves to `../outside.yaml`. - assert "../outside.yaml" in files + assert "../outside.yaml" in dict(files) def test_gather_raises_when_no_sources(project: Path) -> None: CORE.config_path = project / "entry.yaml" with pytest.raises(EsphomeError): - _gather_files(DiscoveredYamlFiles(), include_secrets=False) + _gather_files(DiscoveredYamlFiles()) + + +def test_gather_raises_on_load_errors(project: Path) -> None: + """A failed include load during discovery fails the build instead of + embedding an incomplete recovery bundle.""" + CORE.config_path = project / "entry.yaml" + discovered = DiscoveredYamlFiles( + [project / "entry.yaml"], set(), load_errors=["oops.yaml: boom"] + ) + with pytest.raises(EsphomeError, match="oops.yaml"): + _gather_files(discovered) def test_gather_raises_on_unreadable_file( @@ -123,7 +132,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, include_secrets=False) + _gather_files(discovered) def test_gather_warns_on_unresolved_includes( @@ -134,7 +143,7 @@ def test_gather_warns_on_unresolved_includes( CORE.config_path = project / "entry.yaml" discovered = DiscoveredYamlFiles([project / "entry.yaml"], set(), ["${board}.yaml"]) with caplog.at_level("WARNING", logger="esphome.components.store_yaml"): - files = _gather_files(discovered, include_secrets=False) + files, _ = _gather_files(discovered) assert len(files) == 1 assert any( "${board}.yaml" in r.message and "not contain" in r.message @@ -142,6 +151,163 @@ def test_gather_warns_on_unresolved_includes( ) +# --------------------------------------------------------------------------- +# _generate_redacted_files +# --------------------------------------------------------------------------- + + +def test_redacted_secrets_file_becomes_skeleton(project: Path) -> None: + """The secrets file is replaced by a fill-in skeleton listing every + referenced `!secret` key, so the recovered config is flashable.""" + discovered = _sources( + project, "entry.yaml", "secrets.yaml", secrets=("secrets.yaml",) + ) + files = _gather_redacted(discovered) + skeleton = files["secrets.yaml"].decode() + assert skeleton.startswith(SECRETS_SKELETON_HEADER) + assert 'api_key: ""' in skeleton + assert b"SUPER_SECRET" not in files["secrets.yaml"] + # The entry's own `!secret` reference is re-emitted as a reference. + assert "key: !secret 'api_key'" in files["entry.yaml"].decode() + + +def test_redacted_inline_sensitive_value_becomes_secret_ref(project: Path) -> None: + """An inline cv.sensitive value is generated as `!secret ` and lands in the skeleton.""" + CORE.config = {"wifi": [{"password": SensitiveStr("my_password")}]} + discovered = _sources( + project, "wifi.yaml", "secrets.yaml", secrets=("secrets.yaml",) + ) + files = _gather_redacted(discovered) + text = files["wifi.yaml"].decode() + assert "my_password" not in text + assert "password: !secret 'wifi_password'" in text + assert 'wifi_password: ""' in files["secrets.yaml"].decode() + + +@pytest.mark.parametrize("quote", ['"', "'"]) +def test_redacted_quoted_inline_value(project: Path, quote: str) -> None: + """Quoting in the source doesn't matter — the swap happens on the parsed + scalar, not the text.""" + (project / "wifi.yaml").write_text(f"password: {quote}my_password{quote}\n") + CORE.config = {"wifi": [{"password": SensitiveStr("my_password")}]} + discovered = _sources(project, "wifi.yaml") + files = _gather_redacted(discovered) + assert files["wifi.yaml"] == b"password: !secret 'wifi_password'\n" + + +def test_redacted_swap_is_whole_scalar_and_value_keyed(project: Path) -> None: + """Every whole scalar equal to the sensitive value is swapped (value-keyed, + like `!secret` itself); substrings inside other scalars are never touched. + The recovered config stays semantically identical once the secret is filled.""" + (project / "wifi.yaml").write_text( + "platform: esp32\nnote: esp32 is great\npassword: esp32\n" + ) + CORE.config = {"wifi": [{"password": SensitiveStr("esp32")}]} + discovered = _sources(project, "wifi.yaml") + files = _gather_redacted(discovered) + text = files["wifi.yaml"].decode() + assert "password: !secret 'wifi_password'" in text + assert "platform: !secret 'wifi_password'" in text + assert "note: esp32 is great" in text + + +def test_redacted_include_reference_round_trips(project: Path) -> None: + """A nested `!include` stays a reference in the generated file.""" + (project / "entry.yaml").write_text( + "esphome:\n name: test\nwifi: !include wifi.yaml\n" + ) + discovered = _sources(project, "entry.yaml", "wifi.yaml") + files = _gather_redacted(discovered) + assert "wifi: !include 'wifi.yaml'" in files["entry.yaml"].decode() + + +def test_redacted_reuses_existing_secret_name_for_duplicated_value( + project: Path, +) -> None: + """A value that comes from `!secret` somewhere but is ALSO written inline + elsewhere is generated with the existing secret name.""" + (project / "wifi.yaml").write_text("password: SUPER_SECRET\n") + CORE.config = {"wifi": [{"password": SensitiveStr("SUPER_SECRET")}]} + yaml_util._SECRET_VALUES["SUPER_SECRET"] = "api_key" + discovered = _sources( + project, "wifi.yaml", "secrets.yaml", secrets=("secrets.yaml",) + ) + files = _gather_redacted(discovered) + assert files["wifi.yaml"] == b"password: !secret 'api_key'\n" + + +def test_redacted_warns_when_value_not_locatable( + project: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A sensitive value that never appears as a whole scalar (e.g. composed + via substitutions) produces a warning naming the config path, not the value.""" + CORE.config = {"wifi": [{"password": SensitiveStr("not_in_any_file")}]} + discovered = _sources(project, "wifi.yaml") + with caplog.at_level("WARNING", logger="esphome.components.store_yaml"): + _gather_redacted(discovered) + assert any( + "wifi.password" in r.message and "could not locate" in r.message + for r in caplog.records + ) + assert not any("not_in_any_file" in r.message for r in caplog.records) + + +def test_redacted_does_not_warn_for_secret_only_values( + project: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A value that only exists via `!secret` legitimately never appears inline.""" + CORE.config = {"api": {"encryption": {"key": SensitiveStr("SUPER_SECRET")}}} + yaml_util._SECRET_VALUES["SUPER_SECRET"] = "api_key" + discovered = _sources( + project, "entry.yaml", "secrets.yaml", secrets=("secrets.yaml",) + ) + with caplog.at_level("WARNING", logger="esphome.components.store_yaml"): + _gather_redacted(discovered) + assert not any("could not locate" in r.message for r in caplog.records) + + +def test_redacted_skips_empty_sensitive_values(project: Path) -> None: + """Empty defaults (e.g. mqtt password) are never swapped.""" + (project / "wifi.yaml").write_text("ssid: my_ssid\n") + CORE.config = {"mqtt": {"password": SensitiveStr("")}} + discovered = _sources(project, "wifi.yaml") + files = _gather_redacted(discovered) + assert files["wifi.yaml"] == b"ssid: my_ssid\n" + + +def test_redacted_adds_synthetic_secrets_file_when_none_captured( + project: Path, +) -> None: + """Inline secrets in a project without a secrets.yaml still produce a + skeleton so the recovered config is complete.""" + CORE.config = {"wifi": [{"password": SensitiveStr("my_password")}]} + discovered = _sources(project, "wifi.yaml") + files = _gather_redacted(discovered) + assert 'wifi_password: ""' in files["secrets.yaml"].decode() + + +def test_redacted_generates_unique_names_on_collision(project: Path) -> None: + """Two different inline values whose paths collide get distinct names.""" + (project / "wifi.yaml").write_text("password: first_pw\n") + (project / "wifi2.yaml").write_text("password: second_pw\n") + CORE.config = { + "wifi": [ + {"password": SensitiveStr("first_pw")}, + {"password": SensitiveStr("second_pw")}, + ] + } + discovered = _sources(project, "wifi.yaml", "wifi2.yaml") + files = _gather_redacted(discovered) + assert files["wifi.yaml"] == b"password: !secret 'wifi_password'\n" + assert files["wifi2.yaml"] == b"password: !secret 'wifi_password_2'\n" + + +# --------------------------------------------------------------------------- +# envelope pack/unpack +# --------------------------------------------------------------------------- + + def test_pack_envelope_roundtrip() -> None: files = [ ("entry.yaml", b"esphome:\n name: test\n"), @@ -166,3 +332,16 @@ def test_pack_envelope_rejects_overlong_path() -> None: def test_unpack_envelope_rejects_bad_magic() -> None: with pytest.raises(EsphomeError): unpack_envelope(b"NOPE" + b"\x00" * 4) + + +@pytest.mark.parametrize("cut", [5, 9, 12, -1]) +def test_unpack_envelope_rejects_truncated_input(cut: int) -> None: + blob = _pack_envelope([("entry.yaml", b"esphome:\n")]) + with pytest.raises(EsphomeError, match="truncated"): + unpack_envelope(blob[:cut]) + + +def test_unpack_envelope_rejects_trailing_bytes() -> None: + blob = _pack_envelope([("entry.yaml", b"esphome:\n")]) + with pytest.raises(EsphomeError, match="trailing"): + unpack_envelope(blob + b"\x00") diff --git a/tests/unit_tests/test_yaml_util.py b/tests/unit_tests/test_yaml_util.py index 04e0a84a03f..749e2b9710c 100644 --- a/tests/unit_tests/test_yaml_util.py +++ b/tests/unit_tests/test_yaml_util.py @@ -1091,8 +1091,9 @@ def test_force_load_include_files_returns_unresolved_paths( caller; resolvable ones are not.""" templated = _StubInclude("${var}.yaml", unresolved=True) plain = _StubInclude("ok.yaml") - unresolved = force_load_include_files({"a": templated, "b": plain}) - assert unresolved == [str(templated.file)] + result = force_load_include_files({"a": templated, "b": plain}) + assert result.unresolved == [str(templated.file)] + assert result.errors == [] assert plain.load_calls == 1 @@ -1100,14 +1101,17 @@ def test_force_load_include_files_warns_on_load_failure( patch_include_file: None, caplog: pytest.LogCaptureFixture, ) -> None: - """An `EsphomeError` raised by `load()` is caught and logged, not propagated.""" + """An `EsphomeError` raised by `load()` is caught, logged, and reported to + the caller — not propagated.""" stub = _StubInclude("missing.yaml", raise_on_load=EsphomeError("boom")) with caplog.at_level("WARNING", logger="esphome.yaml_util"): - force_load_include_files({"k": stub}) + result = force_load_include_files({"k": stub}) assert any( "Failed to load !include" in r.message and "missing.yaml" in r.message for r in caplog.records ) + assert result.errors == [f"{stub.file}: boom"] + assert result.unresolved == [] def test_discovered_yaml_files_holds_files_and_secrets() -> None: @@ -1174,11 +1178,18 @@ def test_discover_user_yaml_files_flags_secrets_symlink(tmp_path: Path) -> None: assert target.resolve() in discovered.secrets -def test_discover_user_yaml_files_swallows_parse_errors(tmp_path: Path) -> None: - """A YAML parse failure returns whatever was tracked so far without raising.""" +def test_discover_user_yaml_files_reports_parse_errors( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A YAML parse failure is logged and surfaced in `.load_errors` (not + raised), so consumers can tell the file set is incomplete.""" entry = _write(tmp_path, "entry.yaml", "esphome: [unterminated\n") - discovered = discover_user_yaml_files(entry) + with caplog.at_level("WARNING", logger="esphome.yaml_util"): + discovered = discover_user_yaml_files(entry) assert isinstance(discovered, DiscoveredYamlFiles) + assert len(discovered.load_errors) == 1 + assert "entry.yaml" in discovered.load_errors[0] + assert any("discovery failed to parse" in r.message for r in caplog.records) def test_discover_user_yaml_files_reports_unresolved_includes( @@ -1419,6 +1430,40 @@ def test_dump__redaction_flag_does_not_leak_between_calls() -> None: assert "\\033[8m" in redacted_again +def test_secret_values_registered_swaps_scalars_in_dump() -> None: + """Registered value→name mappings make dump() emit `!secret ` for + matching scalars, and are removed again on exit.""" + with yaml_util.secret_values_registered({"hunter2": "wifi_password"}): + out = yaml_util.dump({"password": make_data_base("hunter2")}) + assert "password: !secret 'wifi_password'" in out + assert "hunter2" not in out + out_after = yaml_util.dump({"password": make_data_base("hunter2")}) + assert "hunter2" in out_after + assert "!secret" not in out_after + assert yaml_util.is_secret("hunter2") is 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() + + +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() + + @pytest.fixture(autouse=True) def clear_dropped_merge_keys() -> None: """Reset the dropped-merge-key queue between tests."""