From 3d42a510535a30a330c9249eeb9fcc808ee1ace6 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Wed, 7 Oct 2026 22:40:09 -1000 Subject: [PATCH] [core] Add YAML secret reference and registration helpers --- esphome/bundle.py | 20 +++--- esphome/yaml_util.py | 72 +++++++++++++++++++++- tests/unit_tests/conftest.py | 11 ++++ tests/unit_tests/test_yaml_util.py | 97 +++++++++++++++++++++++++++--- 4 files changed, 177 insertions(+), 23 deletions(-) diff --git a/esphome/bundle.py b/esphome/bundle.py index b633c5ca4f0..45d9a74b49a 100644 --- a/esphome/bundle.py +++ b/esphome/bundle.py @@ -102,25 +102,21 @@ _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() for fpath in yaml_files: try: text = fpath.read_text(encoding="utf-8") - except (OSError, UnicodeDecodeError): + except (OSError, UnicodeDecodeError) as err: + _LOGGER.warning( + "Could not scan %s for !secret references (%s); the bundled " + "secret set may be incomplete", + fpath, + err, + ) 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/yaml_util.py b/esphome/yaml_util.py index e892664724c..2ab3b56ec5e 100644 --- a/esphome/yaml_util.py +++ b/esphome/yaml_util.py @@ -11,6 +11,7 @@ import logging import math import os from pathlib import Path +import re from typing import Any import uuid @@ -46,6 +47,9 @@ _LOGGER = logging.getLogger(__name__) SECRET_YAML = "secrets.yaml" _SECRET_CACHE = {} _SECRET_VALUES = {} +# Set while a secret_values_registered context is active; the dumper records +# every emitted `!secret` name into it. +_EMITTED_SECRET_NAMES: set[str] | None = None # Not thread-safe — config processing is single-threaded today. _load_listeners: list[Callable[[Path], None]] = [] @@ -440,6 +444,20 @@ def _load_include_candidates( ) +# 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)} + + def force_load_include_files( obj: Any, *, @@ -1119,6 +1137,44 @@ 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[set[str]]: + """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. + + Yields a set that collects the name of every ``!secret`` reference the + dumper emits while the context is active, so callers can tell exactly + which registered values were actually swapped. + + A wrapper (``!extend``, ``!remove``, scalar ``!include``) whose value is + registered dumps as a bare ``!secret``, dropping its tag rather than the + value leaking. + """ + global _EMITTED_SECRET_NAMES # noqa: PLW0603 + added = {v: n for v, n in values.items() if v not in _SECRET_VALUES} + _SECRET_VALUES.update(added) + outer = _EMITTED_SECRET_NAMES + emitted: set[str] = set() + _EMITTED_SECRET_NAMES = emitted + try: + yield emitted + finally: + _EMITTED_SECRET_NAMES = outer + # Only drop mappings this context still owns; a real !secret loaded + # meanwhile may have registered the same value under its own name. + for value, name in added.items(): + if _SECRET_VALUES.get(value) == name: + del _SECRET_VALUES[value] + + def dump( dict_, show_secrets=False, @@ -1333,7 +1389,10 @@ class ESPHomeDumper(yaml.SafeDumper): return node def represent_secret(self, value): - return self.represent_scalar(tag="!secret", value=_SECRET_VALUES[str(value)]) + name = _SECRET_VALUES[str(value)] + if _EMITTED_SECRET_NAMES is not None: + _EMITTED_SECRET_NAMES.add(name) + return self.represent_scalar(tag="!secret", value=name) def represent_stringify(self, value): if is_secret(value): @@ -1414,13 +1473,22 @@ class ESPHomeDumper(yaml.SafeDumper): return self.represent_scalar(tag="!lambda", value=value.value, style="|") def represent_extend(self, value): + # Consult is_secret like the other scalar representers so a payload + # equal to a registered secret is never written out in cleartext; the + # tag is lost then, which beats leaking the value. + if is_secret(value.value): + return self.represent_secret(value.value) return self.represent_scalar(tag="!extend", value=value.value) def represent_remove(self, value): + if is_secret(value.value): + return self.represent_secret(value.value) return self.represent_scalar(tag="!remove", value=value.value) def represent_include_file(self, value): if value.vars or value.condition is not None: + # The mapping values route through the regular representers, + # which already consult is_secret. mapping = {"file": value.file} if value.vars: mapping["vars"] = value.vars @@ -1429,6 +1497,8 @@ class ESPHomeDumper(yaml.SafeDumper): return self.represent_mapping( tag="!include", mapping=mapping, flow_style=False ) + if is_secret(value.file): + return self.represent_secret(value.file) return self.represent_scalar(tag="!include", value=value.file) def represent_id(self, value): diff --git a/tests/unit_tests/conftest.py b/tests/unit_tests/conftest.py index ff5564e8150..aba47c9237e 100644 --- a/tests/unit_tests/conftest.py +++ b/tests/unit_tests/conftest.py @@ -17,6 +17,7 @@ from unittest.mock import Mock, patch import pytest +from esphome import yaml_util from esphome.core import CORE here = Path(__file__).parent @@ -33,6 +34,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 e1c3080e291..6ee0d76ff28 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.""" @@ -1941,6 +1931,66 @@ 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_collects_emitted_names() -> None: + """The context yields a set recording exactly the `!secret` names the + dumper emitted, including real secrets, and not names never swapped.""" + yaml_util._SECRET_VALUES["real_value"] = "real_name" + with yaml_util.secret_values_registered({"hunter2": "wifi_password"}) as emitted: + yaml_util.dump( + { + "password": make_data_base("hunter2"), + "key": make_data_base("real_value"), + "plain": make_data_base("nothing"), + } + ) + assert emitted == {"wifi_password", "real_name"} + + +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" + 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_secret_values_registered_keeps_a_real_secret_loaded_meanwhile() -> None: + """A real `!secret` registered inside the context survives its exit.""" + with yaml_util.secret_values_registered({"hunter2": "generated_name"}): + yaml_util._SECRET_VALUES["hunter2"] = "real_name" + assert yaml_util.is_secret("hunter2") == "real_name" + + +def test_secret_values_registered_nested_keeps_the_outer_collector() -> None: + """An inner context does not stop the outer one from collecting names.""" + with yaml_util.secret_values_registered({"outer_value": "outer"}) as outer: + with yaml_util.secret_values_registered({"inner_value": "inner"}) as inner: + yaml_util.dump({"a": make_data_base("inner_value")}) + yaml_util.dump({"b": make_data_base("outer_value")}) + assert inner == {"inner"} + assert outer == {"outer"} + + +def test_registered_secret_names() -> None: + yaml_util._SECRET_VALUES["value_a"] = "name_a" + assert "name_a" in yaml_util.registered_secret_names() + + @pytest.fixture(autouse=True) def clear_dropped_merge_keys() -> None: """Reset the dropped-merge-key queue between tests.""" @@ -1986,6 +2036,33 @@ def test_merge_include_no_overlap_records_nothing(tmp_path: Path) -> None: assert yaml_util.take_dropped_merge_keys() == [] +def test_wrapper_representers_consult_is_secret() -> None: + """!extend / !remove payloads and scalar !include paths equal to a + registered secret are swapped, never written in cleartext.""" + from esphome.config_helpers import Extend, Remove + + with yaml_util.secret_values_registered({"hunter2": "the_secret"}): + out = yaml_util.dump( + { + "a": Extend("hunter2"), + "b": Remove("hunter2"), + "c": Extend("plain_id"), + } + ) + assert out.count("!secret 'the_secret'") == 2 + assert "hunter2" not in out + assert "!extend 'plain_id'" in out + + +def test_scalar_include_path_equal_to_secret_is_swapped() -> None: + """A scalar !include whose path equals a registered secret is swapped, + matching the other wrapper representers.""" + include = yaml_util.IncludeFile(Path("/fake/main.yaml"), "hunter2", lambda _: {}) + with yaml_util.secret_values_registered({"hunter2": "the_secret"}): + out = yaml_util.dump({"key": include}) + assert out == "key: !secret 'the_secret'\n" + + # --------------------------------------------------------------------------- # track_document_range=False (validated-config-cache fast path) # ---------------------------------------------------------------------------