From 53e7d781560d4245cb2024d22ec3fe68727e9370 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Mon, 12 Oct 2026 13:36:02 -1000 Subject: [PATCH] Simplify store_yaml, trim comments, dedup tests, add DOMAIN --- esphome/components/api/api.proto | 6 +- esphome/components/api/api_connection.cpp | 5 +- esphome/components/store_yaml/__init__.py | 83 +-- script/analyze_component_buses.py | 2 +- tests/integration/conftest.py | 3 +- tests/integration/raw_api_client.py | 25 +- tests/integration/test_store_yaml_recovery.py | 14 +- .../unit_tests/components/test_store_yaml.py | 630 +++++++----------- 8 files changed, 297 insertions(+), 471 deletions(-) diff --git a/esphome/components/api/api.proto b/esphome/components/api/api.proto index 06d33697b9..1959a6997b 100644 --- a/esphome/components/api/api.proto +++ b/esphome/components/api/api.proto @@ -397,8 +397,7 @@ message ZWaveProxyCapabilities { } message StoreYamlCapabilities { - // Whether this firmware embeds its YAML configuration for recovery via - // `get_yaml`, so clients can skip the request instead of waiting for a timeout + // Firmware embeds its YAML, so clients can request it with get_yaml bool supported = 1; } @@ -3148,8 +3147,7 @@ message GetYamlResponse { // The blob is in flash, and the device sends it from there bytes data = 1 [(force) = true, (pointer_to_buffer) = true, (progmem) = true]; bool done = 2 [(force) = true]; - // Sent on the first chunk only — the client is expected to cache it. - // (firmware bandwidth/flash is expensive; the client has gigabytes.) + // Set on the first chunk only uint32 total_size = 3; string encoding = 4 [(max_data_length) = 8]; } diff --git a/esphome/components/api/api_connection.cpp b/esphome/components/api/api_connection.cpp index 6988ffe6d0..28fcb7fa49 100644 --- a/esphome/components/api/api_connection.cpp +++ b/esphome/components/api/api_connection.cpp @@ -2406,9 +2406,8 @@ bool APIConnection::try_to_clear_buffer_slow_(bool log_out_of_space) { bool APIConnection::send_message_(uint32_t payload_size, uint16_t message_type, MessageEncodeFn encode_fn, const void *msg) { #ifdef HAS_PROTO_MESSAGE_DUMP - // Skip dump for log messages (recursive logging risk), camera frames (high-frequency noise), - // and YAML recovery payloads (every chunk would log the embedded config, including any - // secrets the user opted into). + // Skip dump for log messages (recursive logging risk), camera frames (high-frequency noise) + // and YAML chunks (may contain secrets) if (message_type != SubscribeLogsResponse::MESSAGE_TYPE #ifdef USE_CAMERA && message_type != CameraImageResponse::MESSAGE_TYPE diff --git a/esphome/components/store_yaml/__init__.py b/esphome/components/store_yaml/__init__.py index 827c01fcaa..4360cc5f37 100644 --- a/esphome/components/store_yaml/__init__.py +++ b/esphome/components/store_yaml/__init__.py @@ -22,6 +22,7 @@ _LOGGER = logging.getLogger(__name__) CODEOWNERS = ["@bdraco"] DEPENDENCIES = ["api"] +DOMAIN = "store_yaml" CONF_INCLUDE_SECRETS = "include_secrets" CONF_ALLOW_UNENCRYPTED = "allow_unencrypted" @@ -107,23 +108,17 @@ def _gather_files( ", ".join(discovered.unresolved), ) - # Resolved (not just absolute) because discovery returns resolved paths - # and relative_to needs both sides on the same footing when symlinks are - # in play; CORE.config_dir is only absolute. + # Resolved to match discovery's resolved paths; CORE.config_dir is only absolute. root = CORE.config_path.resolve().parent entries: list[tuple[str, Path]] = [] secret_rels: set[str] = set() for path in discovered.files: - # Files outside the project root (e.g. ../common.yaml or a secrets file - # in $HOME) keep their ".." components so the include graph is preserved - # and files from different directories with the same basename don't - # collide. + # Keep ".." for files outside the root so same-named files don't collide. try: rel_str = path.relative_to(root, walk_up=True).as_posix() except ValueError as err: - # Different anchors (a Windows file on another drive) cannot be - # expressed relative to the config root. + # No common anchor, e.g. a Windows file on another drive. raise EsphomeError( f"store_yaml: cannot place {path} in the recovery envelope; " f"it does not share a root with {root}: {err}" @@ -143,8 +138,6 @@ def _read_files_verbatim(entries: list[tuple[str, Path]]) -> list[tuple[str, byt 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 @@ -201,37 +194,33 @@ def _check_sensitive_usage( for rel, tree in trees.items(): for path, node, is_key in _iter_nodes(tree): text = str(node) - info = sensitive.get(text) - if is_key: - if info is not None: + if (info := sensitive.get(text)) is not None: + if is_key: key_hits.append( f"{info.config_path} (as the mapping key at " f"{'.'.join((*path, text))} in {rel})" ) - continue - if info is not None: - if path and ( - path[0] == "substitutions" or path[-1] in info.sensitive_keys + elif not path or ( + path[0] != "substitutions" and path[-1] not in info.sensitive_keys ): - continue - _LOGGER.warning( - "store_yaml: the sensitive value at %s also matches the " - "scalar at %s in %s; the recovered config will reference " - "!secret %s there too", - info.config_path, - ".".join(path), - rel, - info.secret_name, - ) + _LOGGER.warning( + "store_yaml: the sensitive value at %s also matches the " + "scalar at %s in %s; the recovered config will reference " + "!secret %s there too", + info.config_path, + ".".join(path), + rel, + info.secret_name, + ) + continue + if is_key: continue for value, other in sensitive.items(): if value not in text: continue if other.from_secret: - # The value lives in a real secrets.yaml, so the - # move-into-!secret remedy does not apply; overlaps like - # an SSID inside an entity name are common and benign. - # Warn naming the location, never the value. + # Already a real !secret and overlaps (an SSID in a name) + # are common, so warn without the value instead of failing. _LOGGER.warning( "store_yaml: the value of !secret %s (sensitive at %s) " "appears inside the value at %s in %s; substring " @@ -292,10 +281,7 @@ def _collect_sensitive_values() -> dict[str, _SensitiveValue]: def _uncaptured_note( unresolved: list[str], remote_packages: list[str] ) -> tuple[str, bytes] | None: - """Comment-only YAML entry listing content that could not be captured, so - a recovered config never silently appears complete. Emitted for both the - redacted and verbatim paths; user files are never modified to carry it. - Returns None when there is nothing to record.""" + """Return a comment-only entry listing uncaptured content, or None if there is none.""" if not unresolved and not remote_packages: return None parts = ["# store_yaml: the following content could not be captured.\n"] @@ -330,10 +316,9 @@ def _build_secrets_skeleton(keys: set[str]) -> bytes: def _generate_redacted_files( entries: list[tuple[str, Path]], secret_rels: set[str], remote_packages: list[str] ) -> list[tuple[str, bytes]]: - """Re-dump each captured file with cv.sensitive values as `!secret` references - and secrets files as a fill-in skeleton, so the recovered config flashes once - secrets.yaml is filled in. The dumper does the swap; comments and formatting - are not preserved. + """Re-dump captured files with sensitive values as `!secret` and secrets files as a skeleton. + + Comments and formatting are not preserved. """ sensitive = _collect_sensitive_values() @@ -349,10 +334,8 @@ def _generate_redacted_files( with yaml_util.secret_values_registered(registered) as skeleton_keys: texts = {rel: yaml_util.dump(tree) for rel, tree in trees.items()} - # skeleton_keys now holds exactly the `!secret` names the dumper emitted. - # A registered inline value whose name was never emitted was not found as - # a whole scalar in any captured file, so it would ship nowhere and the - # redaction promise cannot be verified — fail the build. + # An inline value whose name was never emitted was not found as a whole + # scalar, so its redaction cannot be verified. leaked = [ info.config_path for info in sensitive.values() @@ -390,8 +373,7 @@ def _pack_envelope(files: list[tuple[str, bytes]]) -> bytes: seen: set[str] = set() for path, content in files: if path in seen: - # unpack_envelope builds a dict, so a duplicate would silently - # replace the earlier entry — fail the build instead. + # unpack_envelope would silently keep only the last duplicate. raise EsphomeError(f"store_yaml: duplicate envelope path: {path}") seen.add(path) path_bytes = path.encode("utf-8") @@ -407,10 +389,10 @@ def _pack_envelope(files: list[tuple[str, bytes]]) -> bytes: def unpack_envelope(blob: bytes) -> dict[str, bytes]: - """Reference decoder for the EHY1 envelope, for tests and recovery tools. + """Reference decoder for the EHY1 envelope. - Absolute paths are rejected. ``..`` paths are valid (files outside the config - root), so callers writing to disk must confine paths to their target dir.""" + Absolute paths are rejected but ``..`` is allowed; callers writing to disk must confine paths. + """ if blob[:4] != ENVELOPE_MAGIC: raise EsphomeError("envelope must start with EHY1 magic") pos = 4 @@ -445,8 +427,7 @@ def unpack_envelope(blob: bytes) -> dict[str, bytes]: async def to_code(config: ConfigType) -> None: cg.add_define("USE_STORE_YAML") - # Re-parse at codegen time to capture only user files; this also refills the - # secret registry that save_compiled_config cleared, which redaction relies on. + # Re-parse to capture only user files; also refills the secret registry redaction needs. discovered = yaml_util.discover_user_yaml_files(CORE.config_path) entries, secret_rels = _gather_files(discovered) remote_packages = _remote_package_descriptions() diff --git a/script/analyze_component_buses.py b/script/analyze_component_buses.py index 92b58c12c9..c259371b85 100755 --- a/script/analyze_component_buses.py +++ b/script/analyze_component_buses.py @@ -96,7 +96,7 @@ ISOLATED_COMPONENTS = { "modbus_controller": "Defines multiple modbus buses for testing client/server functionality - conflicts with package modbus bus", "neopixelbus": "RMT type conflict with ESP32 Arduino/ESP-IDF headers (enum vs struct rmt_channel_t)", "packages": "cannot merge packages", - "store_yaml": "Embeds every captured user YAML file in firmware; grouping would make the blob and its secret redaction depend on every grouped component's test config", + "store_yaml": "Embeds the test's own YAML; grouping would change the blob and its redaction", "tinyusb": "Conflicts with usb_host component - cannot be used together", "uart_mux": "Depends on tinyusb which conflicts with usb_host", "usb_cdc_acm": "Depends on tinyusb which conflicts with usb_host", diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 5090ad2003..3527171276 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -212,8 +212,7 @@ async def yaml_config(request: pytest.FixtureRequest, unused_tcp_port: int) -> s loop = asyncio.get_running_loop() content = await loop.run_in_executor(None, read_file, fixture_path) - # Replace the port in the config if it contains an api section. Anchored to - # the start of a line so keys that merely end in "api:" are left alone. + # Add the port to a top-level api section; keys ending in "api:" are left alone. content = re.sub(r"(?m)^api:", f"api:\n port: {unused_tcp_port}", content) # Add debug build flags for integration tests to enable assertions diff --git a/tests/integration/raw_api_client.py b/tests/integration/raw_api_client.py index 96b6af890f..82257874df 100644 --- a/tests/integration/raw_api_client.py +++ b/tests/integration/raw_api_client.py @@ -73,9 +73,6 @@ def decode_fields(payload: bytes) -> dict[int, list[int | bytes]]: length, pos = varint(pos) assert pos + length <= len(buf), "truncated length-delimited field" value, pos = bytes(buf[pos : pos + length]), pos + length - elif wire_type == 5: - assert pos + 4 <= len(buf), "truncated fixed32 field" - value, pos = int.from_bytes(buf[pos : pos + 4], "little"), pos + 4 else: raise AssertionError(f"unexpected wire type {wire_type}") fields.setdefault(tag >> 3, []).append(value) @@ -92,15 +89,13 @@ class FrameParser: def __init__(self) -> None: self._buf = bytearray() - self.frames: deque[tuple[int, bytes]] = deque() - def feed(self, data: bytes) -> list[int]: + def feed(self, data: bytes) -> list[tuple[int, bytes]]: self._buf.extend(data) - types: list[int] = [] + frames: list[tuple[int, bytes]] = [] while (frame := self._try_parse()) is not None: - self.frames.append(frame) - types.append(frame[0]) - return types + frames.append(frame) + return frames def _try_parse(self) -> tuple[int, bytes] | None: buf = self._buf @@ -126,6 +121,7 @@ class RawApiClient: def __init__(self, port: int, recv_buffer_size: int | None = None) -> None: self._port = port self._parser = FrameParser() + self._frames: deque[tuple[int, bytes]] = deque() self.bytes_received = 0 self.frame_counts: Counter[int] = Counter() sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) @@ -172,7 +168,9 @@ class RawApiClient: data = await asyncio.get_running_loop().sock_recv(self._sock, _READ_CHUNK) assert data, "server closed the connection unexpectedly" self.bytes_received += len(data) - self.frame_counts.update(self._parser.feed(data)) + frames = self._parser.feed(data) + self.frame_counts.update(msg_type for msg_type, _ in frames) + self._frames.extend(frames) async def read_until_frame(self, msg_type: int, timeout: float = 10.0) -> None: """Read until at least one frame of msg_type has been received.""" @@ -184,11 +182,8 @@ class RawApiClient: await asyncio.wait_for(_read_loop(), timeout) async def read_frame(self, msg_type: int, timeout: float = 10.0) -> bytes: - """Return the payload of the oldest unconsumed frame of msg_type, dropping other frames before it. - - Frames received earlier, also through read_until_frame, count as unconsumed. - """ - frames = self._parser.frames + """Return the oldest unconsumed payload of msg_type, dropping earlier frames of other types.""" + frames = self._frames async def _read_loop() -> bytes: while True: diff --git a/tests/integration/test_store_yaml_recovery.py b/tests/integration/test_store_yaml_recovery.py index 042799e736..e361ed63c1 100644 --- a/tests/integration/test_store_yaml_recovery.py +++ b/tests/integration/test_store_yaml_recovery.py @@ -21,6 +21,8 @@ GET_YAML_RESPONSE = 162 # DeviceCapabilitiesResponse.store_yaml and StoreYamlCapabilities.supported CAPABILITIES_STORE_YAML_FIELD = 7 STORE_YAML_SUPPORTED_FIELD = 1 +# GetYamlResponse fields +DATA_FIELD, DONE_FIELD, TOTAL_SIZE_FIELD, ENCODING_FIELD = 1, 2, 3, 4 @pytest.mark.asyncio @@ -50,12 +52,12 @@ async def test_store_yaml_recovery( done = False while not done: fields = decode_fields(await client.read_frame(GET_YAML_RESPONSE)) - if data := fields.get(1): + if data := fields.get(DATA_FIELD): chunks.append(data[0]) - done = fields.get(2) == [1] - if total := fields.get(3): + done = fields.get(DONE_FIELD) == [1] + if total := fields.get(TOTAL_SIZE_FIELD): advertised_total = total[0] - if encoding := fields.get(4): + if encoding := fields.get(ENCODING_FIELD): advertised_encoding = encoding[0].decode("utf-8") compressed = b"".join(chunks) @@ -78,9 +80,7 @@ async def test_store_yaml_recovery( "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. + # The inline OTA password must come back as a !secret listed in the skeleton. assert b"recoverme123" not in envelope, ( "inline sensitive value leaked into the recovery blob" ) diff --git a/tests/unit_tests/components/test_store_yaml.py b/tests/unit_tests/components/test_store_yaml.py index 004de8a909..1872fff184 100644 --- a/tests/unit_tests/components/test_store_yaml.py +++ b/tests/unit_tests/components/test_store_yaml.py @@ -1,9 +1,9 @@ -"""Tests for the store_yaml component's file gathering, secret redaction, and -envelope packing.""" +"""Tests for store_yaml file gathering, redaction and envelope packing.""" from __future__ import annotations from pathlib import Path +import struct import pytest @@ -11,6 +11,7 @@ from esphome import yaml_util from esphome.components import packages from esphome.components.store_yaml import ( CONF_ALLOW_UNENCRYPTED, + ENVELOPE_MAGIC, SECRETS_SKELETON_HEADER, UNCAPTURED_NOTE_PATH, _final_validate, @@ -30,7 +31,7 @@ from esphome.yaml_util import DiscoveredYamlFiles, SensitiveStr @pytest.fixture def project(tmp_path: Path) -> Path: - """Lay out a tiny ESPHome-like project: entry yaml, an include, and a secrets file.""" + """Lay out an entry yaml, an include and a secrets file.""" project_dir = tmp_path / "project" project_dir.mkdir() (project_dir / "entry.yaml").write_text( @@ -41,18 +42,20 @@ def project(tmp_path: Path) -> Path: return project_dir -@pytest.fixture(autouse=True) -def _clear_config() -> None: - CORE.config = {} - - def _sources( - project_dir: Path, *names: str, secrets: tuple[str, ...] = () + project_dir: Path, + *names: str, + secrets: tuple[str, ...] = (), + unresolved: list[str] | None = None, + load_errors: list[str] | None = None, ) -> DiscoveredYamlFiles: CORE.config_path = project_dir / "entry.yaml" - files = [project_dir / name for name in names] - secret_paths = {(project_dir / name).resolve() for name in secrets} - return DiscoveredYamlFiles(files, secret_paths) + return DiscoveredYamlFiles( + [project_dir / name for name in names], + {(project_dir / name).resolve() for name in secrets}, + unresolved or [], + load_errors or [], + ) def _gather_redacted(discovered: DiscoveredYamlFiles) -> dict[str, bytes]: @@ -62,9 +65,12 @@ def _gather_redacted(discovered: DiscoveredYamlFiles) -> dict[str, bytes]: ) -# --------------------------------------------------------------------------- -# _gather_files -# --------------------------------------------------------------------------- +def _run_final_validate(full_config: dict, config: dict) -> dict: + token = fv.full_config.set(full_config) + try: + return _final_validate(config) + finally: + fv.full_config.reset(token) def test_gather_maps_rel_paths_and_flags_secrets(project: Path) -> None: @@ -90,51 +96,26 @@ def test_read_files_verbatim_returns_exact_bytes(project: Path) -> None: assert contents["entry.yaml"] == (project / "entry.yaml").read_bytes() -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 flagged - because the un-resolved basename was captured upstream.""" - target = tmp_path / "actual_creds.yaml" - target.write_text("api_key: FROM_SYMLINK\n") - link = project / "secrets.yaml" - link.unlink() # remove the regular file laid down by the fixture - link.symlink_to(target) - # Discovery records the un-resolved listener fname under SECRETS_FILES - # but stores the resolved path; mimic that here. - resolved = link.resolve() - CORE.config_path = project / "entry.yaml" - files = _gather_redacted(DiscoveredYamlFiles([resolved], {resolved})) - assert b"FROM_SYMLINK" not in b"".join(files.values()) - - def test_gather_uses_relative_path_for_external_files( project: Path, tmp_path: Path ) -> None: - """Files outside the project root use a ``..``-style relative path so they don't collide.""" + """Files outside the project root keep a ``..`` path.""" sibling = tmp_path / "outside.yaml" sibling.write_text("foo: bar\n") - CORE.config_path = project / "entry.yaml" - discovered = DiscoveredYamlFiles([project / "entry.yaml", sibling], set()) + discovered = _sources(project, "entry.yaml") + discovered.files.append(sibling) 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 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()) + _gather_files(_sources(project)) 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"] - ) + """A failed include load fails the build.""" + discovered = _sources(project, "entry.yaml", load_errors=["oops.yaml: boom"]) with pytest.raises(EsphomeError, match="oops.yaml"): _gather_files(discovered) @@ -142,10 +123,8 @@ def test_gather_raises_on_load_errors(project: Path) -> None: 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) + """An unreadable file fails the build.""" + entries, _ = _gather_files(_sources(project, "entry.yaml", "wifi.yaml")) orig_read_bytes = Path.read_bytes def fake_read_bytes(self: Path) -> bytes: @@ -161,10 +140,8 @@ def test_read_files_verbatim_raises_on_unreadable_file( def test_gather_warns_on_unresolved_includes( project: Path, caplog: pytest.LogCaptureFixture ) -> None: - """Substitution-pathed includes that discovery could not capture produce a - warning naming them, so the user knows the blob is incomplete.""" - CORE.config_path = project / "entry.yaml" - discovered = DiscoveredYamlFiles([project / "entry.yaml"], set(), ["${board}.yaml"]) + """Substitution-pathed includes are named in a warning.""" + discovered = _sources(project, "entry.yaml", unresolved=["${board}.yaml"]) with caplog.at_level("WARNING", logger="esphome.components.store_yaml"): files, _ = _gather_files(discovered) assert len(files) == 1 @@ -174,14 +151,21 @@ def test_gather_warns_on_unresolved_includes( ) -# --------------------------------------------------------------------------- -# _generate_redacted_files -# --------------------------------------------------------------------------- +def test_gather_raises_esphome_error_on_cross_anchor_path( + project: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A path with no common anchor raises EsphomeError, not ValueError.""" + + def fake_relative_to(self: Path, other: Path, walk_up: bool = False) -> Path: + raise ValueError("paths have different anchors") + + monkeypatch.setattr(Path, "relative_to", fake_relative_to) + with pytest.raises(EsphomeError, match="does not share a root"): + _gather_files(_sources(project, "entry.yaml")) 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.""" + """The secrets file becomes a skeleton of the referenced keys.""" discovered = _sources( project, "entry.yaml", "secrets.yaml", secrets=("secrets.yaml",) ) @@ -190,63 +174,60 @@ def test_redacted_secrets_file_becomes_skeleton(project: Path) -> None: 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",) +def test_redacted_symlinked_secrets_outside_root(project: Path, tmp_path: Path) -> None: + """A secrets.yaml linked to another name outside the root is flagged and gets a root skeleton.""" + target = tmp_path / "actual_creds.yaml" + target.write_text("api_key: SUPER_SECRET\n") + link = project / "secrets.yaml" + link.unlink() + link.symlink_to(target) + # Discovery flags the unresolved name but records the resolved path. + resolved = link.resolve() + CORE.config_path = project / "entry.yaml" + files = _gather_redacted( + DiscoveredYamlFiles([project / "entry.yaml", resolved], {resolved}) ) - files = _gather_redacted(discovered) - text = files["wifi.yaml"].decode() - assert "my_password" not in text - assert "password: !secret 'wifi_password'" in text + assert 'api_key: ""' in files["secrets.yaml"].decode() + assert b"SUPER_SECRET" not in b"".join(files.values()) + + +@pytest.mark.parametrize( + ("source", "value"), + [ + ("password: my_password\n", "my_password"), + ('password: "my_password"\n', "my_password"), + ("password: 'my_password'\n", "my_password"), + # A value that is a substring of its key name. + ("password: word\n", "word"), + ], +) +def test_redacted_inline_value_becomes_secret_ref( + project: Path, source: str, value: str +) -> None: + """An inline sensitive value becomes a path-named !secret listed in the skeleton.""" + (project / "wifi.yaml").write_text(source) + CORE.config = {"wifi": [{"password": SensitiveStr(value)}]} + files = _gather_redacted(_sources(project, "wifi.yaml")) + assert files["wifi.yaml"] == b"password: !secret 'wifi_password'\n" 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). The recovered config stays semantically identical - once the secret is filled.""" - (project / "wifi.yaml").write_text("platform: esp32\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 - - def test_redacted_include_reference_round_trips(project: Path) -> None: - """A nested `!include` stays a reference in the generated file.""" + """A nested `!include` stays a reference.""" (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) + files = _gather_redacted(_sources(project, "entry.yaml", "wifi.yaml")) 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.""" + """An inline value equal to a !secret value reuses that 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" @@ -258,18 +239,15 @@ def test_redacted_reuses_existing_secret_name_for_duplicated_value( def test_redacted_raises_when_value_not_locatable(project: Path) -> None: - """A sensitive value that never appears as a whole scalar (e.g. composed - via substitutions) would ship verbatim — fail the build, naming the config - path but never the value.""" + """A value not found as a whole scalar fails the build without echoing it.""" CORE.config = {"wifi": [{"password": SensitiveStr("not_in_any_file")}]} - discovered = _sources(project, "wifi.yaml") with pytest.raises(EsphomeError, match="wifi.password") as err: - _gather_redacted(discovered) + _gather_redacted(_sources(project, "wifi.yaml")) assert "not_in_any_file" not in str(err.value) def test_redacted_accepts_secret_only_values(project: Path) -> None: - """A value that only exists via `!secret` legitimately never appears inline.""" + """A value only set through `!secret` never needs to appear inline.""" CORE.config = {"api": {"encryption": {"key": SensitiveStr("SUPER_SECRET")}}} yaml_util._SECRET_VALUES["SUPER_SECRET"] = "api_key" discovered = _sources( @@ -279,34 +257,142 @@ def test_redacted_accepts_secret_only_values(project: Path) -> None: assert 'api_key: ""' in files["secrets.yaml"].decode() -def test_uncaptured_note_lists_missing_includes() -> None: - """Substitution-pathed includes that can't be captured are recorded in a - dedicated envelope entry (both modes), not just a compile-time log line.""" - rel, content = _uncaptured_note(["${board}.yaml"], []) +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("")}} + files = _gather_redacted(_sources(project, "wifi.yaml")) + assert files["wifi.yaml"] == b"ssid: my_ssid\n" + + +def test_redacted_generates_unique_names_on_collision(project: Path) -> None: + """Two inline values with colliding paths 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")}, + ] + } + files = _gather_redacted(_sources(project, "wifi.yaml", "wifi2.yaml")) + assert files["wifi.yaml"] == b"password: !secret 'wifi_password'\n" + assert files["wifi2.yaml"] == b"password: !secret 'wifi_password_2'\n" + + +@pytest.mark.parametrize( + ("files", "sources"), + [ + ( + {"wifi.yaml": "password: my_password\nurl: http://user:my_password@host\n"}, + ("wifi.yaml",), + ), + ( + { + "wifi.yaml": "password: my_password\nx: !lambda 'return \"my_password\";'\n" + }, + ("wifi.yaml",), + ), + ( + { + "entry.yaml": "password: my_password\n" + "pkg: !include {file: wifi.yaml, vars: {url: 'http://user:my_password@host'}}\n" + }, + ("entry.yaml", "wifi.yaml"), + ), + ], + ids=["url", "lambda", "include_vars"], +) +def test_redacted_embedded_sensitive_value_fails_build( + project: Path, files: dict[str, str], sources: tuple[str, ...] +) -> None: + """A sensitive value inside a larger scalar fails the build.""" + for name, text in files.items(): + (project / name).write_text(text) + CORE.config = {"wifi": [{"password": SensitiveStr("my_password")}]} + with pytest.raises(EsphomeError, match="embedded"): + _gather_redacted(_sources(project, *sources)) + + +def test_redacted_warns_on_value_collision( + project: Path, caplog: pytest.LogCaptureFixture +) -> None: + """An unrelated equal scalar is swapped with a warning; the value's own key is not warned about.""" + (project / "wifi.yaml").write_text("password: esp32\nplatform: esp32\n") + CORE.config = {"wifi": [{"password": SensitiveStr("esp32")}]} + files = _gather_redacted(_sources(project, "wifi.yaml")) + assert files["wifi.yaml"] == ( + b"password: !secret 'wifi_password'\nplatform: !secret 'wifi_password'\n" + ) + assert "also matches the scalar at platform in wifi.yaml" in caplog.text + assert "at password in wifi.yaml" not in caplog.text + + +def test_redacted_no_warning_for_substitution_definition( + project: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A swapped substitutions definition does not warn.""" + (project / "wifi.yaml").write_text( + "substitutions:\n wifi_password: hunter2\nwifi:\n password: ${wifi_password}\n" + ) + CORE.config = {"wifi": [{"password": SensitiveStr("hunter2")}]} + _gather_redacted(_sources(project, "wifi.yaml")) + assert "also matches" not in caplog.text + + +def test_redacted_sensitive_value_as_mapping_key_fails_build(project: Path) -> None: + """A sensitive value used as a mapping key fails the build.""" + (project / "wifi.yaml").write_text("password: password\n") + CORE.config = {"ota": [{"password": SensitiveStr("password")}]} + with pytest.raises(EsphomeError, match="mapping key"): + _gather_redacted(_sources(project, "wifi.yaml")) + + +def test_redacted_secret_sourced_overlap_warns_not_fails( + project: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A !secret value inside another scalar warns instead of failing.""" + yaml_util._SECRET_VALUES["Kitchen"] = "wifi_ssid" + (project / "wifi.yaml").write_text("ssid: Kitchen\nname: Kitchen Temperature\n") + CORE.config = {"wifi": [{"ssid": SensitiveStr("Kitchen")}]} + files = _gather_redacted(_sources(project, "wifi.yaml")) + assert b"!secret 'wifi_ssid'" in files["wifi.yaml"] + assert "appears inside the value at name in wifi.yaml" in caplog.text + + +@pytest.mark.parametrize( + ("unresolved", "remote", "expected"), + [ + (["${board}.yaml"], [], ["# ${board}.yaml"]), + ( + [], + ["https://github.com/org/repo@main", "https://github.com/org/other"], + [ + "# https://github.com/org/repo@main", + "# https://github.com/org/other", + ], + ), + ], + ids=["includes", "remote_packages"], +) +def test_uncaptured_note_lists_content( + unresolved: list[str], remote: list[str], expected: list[str] +) -> None: + """Uncaptured includes and remote packages are listed in the note entry.""" + rel, content = _uncaptured_note(unresolved, remote) assert rel == UNCAPTURED_NOTE_PATH text = content.decode() assert text.startswith("# store_yaml:") - assert "# ${board}.yaml" in text + for line in expected: + assert line in text -def test_uncaptured_note_lists_remote_packages() -> None: - """Remote packages that can't be captured are recorded with their source - so the user knows to re-fetch them.""" - rel, content = _uncaptured_note( - [], ["https://github.com/org/repo@main", "https://github.com/org/other"] - ) - assert rel == UNCAPTURED_NOTE_PATH - text = content.decode() - assert "# https://github.com/org/repo@main" in text - assert "# https://github.com/org/other" in text +def test_uncaptured_note_none_when_complete() -> None: + assert _uncaptured_note([], []) is None -def test_remote_package_descriptions_read_packages_record( - monkeypatch: pytest.MonkeyPatch, -) -> None: - """Remote sources recorded by the packages component during config - processing are formatted as url@ref (url alone when ref is absent).""" - monkeypatch.delitem(CORE.data, packages.DOMAIN, raising=False) +def test_remote_package_descriptions_read_packages_record() -> None: + """Remote sources format as url@ref, or url alone.""" data = packages._get_data() data.remote_sources[ packages.RemotePackageSource("https://github.com/org/repo", "main") @@ -321,70 +407,25 @@ def test_remote_package_descriptions_read_packages_record( def test_remote_package_descriptions_empty_without_packages() -> None: - CORE.data.pop(packages.DOMAIN, None) assert _remote_package_descriptions() == [] -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"), - ("wifi.yaml", b"ssid: a\n"), + ("dossiers/élan.yaml", b"ssid: a\n"), ] - blob = _pack_envelope(files) - assert unpack_envelope(blob) == dict(files) - - -def test_pack_envelope_handles_utf8_paths() -> None: - files = [("dossiers/maison.yaml", b"foo: bar\n")] - blob = _pack_envelope(files) - assert unpack_envelope(blob) == dict(files) + assert unpack_envelope(_pack_envelope(files)) == dict(files) def test_pack_envelope_rejects_overlong_path() -> None: - long_path = "a" * (0xFFFF + 1) with pytest.raises(EsphomeError): - _pack_envelope([(long_path, b"")]) + _pack_envelope([("a" * (0xFFFF + 1), b"")]) + + +def test_pack_envelope_rejects_duplicate_paths() -> None: + with pytest.raises(EsphomeError, match="duplicate"): + _pack_envelope([("entry.yaml", b"a: 1\n"), ("entry.yaml", b"b: 2\n")]) def test_unpack_envelope_rejects_bad_magic() -> None: @@ -406,127 +447,23 @@ def test_unpack_envelope_rejects_trailing_bytes() -> None: def test_unpack_envelope_rejects_invalid_utf8_path() -> None: - """A corrupted envelope raises EsphomeError, never a bare UnicodeDecodeError.""" - import struct - - blob = ( - b"EHY1" - + struct.pack(" None: - """A tampered envelope with duplicate paths must not silently drop data.""" - import struct - entry = struct.pack(" None: - """A duplicate path would silently clobber the earlier entry on unpack.""" - with pytest.raises(EsphomeError, match="duplicate"): - _pack_envelope([("entry.yaml", b"a: 1\n"), ("entry.yaml", b"b: 2\n")]) + unpack_envelope(ENVELOPE_MAGIC + struct.pack(" None: - """The packer never emits absolute or drive-qualified paths, so their - presence means a malformed or hostile envelope.""" - blob = _pack_envelope([(path, b"boom\n")]) + """Absolute or drive-qualified paths are rejected.""" with pytest.raises(EsphomeError, match="non-relative"): - unpack_envelope(blob) - - -# --------------------------------------------------------------------------- -# embedded-value leak scan and collision warning -# --------------------------------------------------------------------------- - - -def test_redacted_embedded_sensitive_value_fails_build(project: Path) -> None: - """A sensitive value inside a larger scalar (URL, lambda body) is not - swapped by the whole-scalar redaction; the substring scan fails closed.""" - (project / "wifi.yaml").write_text( - "password: my_password\nurl: http://user:my_password@host\n" - ) - CORE.config = {"wifi": [{"password": SensitiveStr("my_password")}]} - discovered = _sources(project, "wifi.yaml") - with pytest.raises(EsphomeError, match="embedded"): - _gather_redacted(discovered) - - -def test_redacted_warns_on_value_collision( - project: Path, caplog: pytest.LogCaptureFixture -) -> None: - """An unrelated scalar equal to a sensitive value gets rewritten by the - value-keyed swap; a warning documents the trap. The value's own - occurrence (under its sensitive key) does not warn.""" - (project / "wifi.yaml").write_text("password: esp32\nplatform: esp32\n") - CORE.config = {"wifi": [{"password": SensitiveStr("esp32")}]} - discovered = _sources(project, "wifi.yaml") - files = _gather_redacted(discovered) - assert files["wifi.yaml"] == ( - b"password: !secret 'wifi_password'\nplatform: !secret 'wifi_password'\n" - ) - assert "also matches the scalar at platform in wifi.yaml" in caplog.text - assert "at password in wifi.yaml" not in caplog.text - - -def test_redacted_no_warning_for_substitution_definition( - project: Path, caplog: pytest.LogCaptureFixture -) -> None: - """A swapped `substitutions:` definition keeps `${...}` working in the - recovered config, so it is expected and does not warn.""" - (project / "wifi.yaml").write_text( - "substitutions:\n wifi_password: hunter2\nwifi:\n password: ${wifi_password}\n" - ) - CORE.config = {"wifi": [{"password": SensitiveStr("hunter2")}]} - discovered = _sources(project, "wifi.yaml") - _gather_redacted(discovered) - assert "also matches" not in caplog.text - - -def test_redacted_sensitive_value_as_mapping_key_fails_build(project: Path) -> None: - """A sensitive value equal to a mapping key would be swapped in key - position and corrupt the recovered structure; the build fails instead.""" - (project / "wifi.yaml").write_text("password: password\n") - CORE.config = {"ota": [{"password": SensitiveStr("password")}]} - discovered = _sources(project, "wifi.yaml") - with pytest.raises(EsphomeError, match="mapping key"): - _gather_redacted(discovered) - - -def test_redacted_key_names_do_not_false_positive_embedded_scan( - project: Path, -) -> None: - """The embedded scan runs on tree scalars, not serialized text, so a - sensitive value that is a substring of a key name (or of the generated - `!secret` reference text) does not fail the build.""" - (project / "wifi.yaml").write_text("password: word\n") - CORE.config = {"wifi": [{"password": SensitiveStr("word")}]} - discovered = _sources(project, "wifi.yaml") - files = _gather_redacted(discovered) - assert files["wifi.yaml"] == b"password: !secret 'wifi_password'\n" - - -# --------------------------------------------------------------------------- -# _final_validate encryption gate -# --------------------------------------------------------------------------- - - -def _run_final_validate(full_config: dict, config: dict) -> dict: - token = fv.full_config.set(full_config) - try: - return _final_validate(config) - finally: - fv.full_config.reset(token) + unpack_envelope(_pack_envelope([(path, b"boom\n")])) def test_final_validate_accepts_encrypted_api() -> None: @@ -537,108 +474,25 @@ def test_final_validate_accepts_encrypted_api() -> None: assert _run_final_validate(full, config) is config -def test_final_validate_rejects_unencrypted_api() -> None: +@pytest.mark.parametrize( + "api", [{}, {"encryption": {}}], ids=["no_encryption", "keyless_encryption"] +) +def test_final_validate_rejects_unencrypted_api(api: dict) -> None: + """Keyless encryption accepts the all-zeros key, so it does not count.""" with pytest.raises(cv.Invalid, match="requires API encryption"): - _run_final_validate({"api": {}}, {CONF_ALLOW_UNENCRYPTED: False}) + _run_final_validate({"api": api}, {CONF_ALLOW_UNENCRYPTED: False}) +@pytest.mark.parametrize( + ("include_secrets", "expected"), + [ + (False, "without API encryption"), + (True, "verbatim contents of secrets.yaml"), + ], +) def test_final_validate_allows_unencrypted_with_escape_hatch( - caplog: pytest.LogCaptureFixture, + caplog: pytest.LogCaptureFixture, include_secrets: bool, expected: str ) -> None: - config = {CONF_ALLOW_UNENCRYPTED: True} + config = {CONF_ALLOW_UNENCRYPTED: True, "include_secrets": include_secrets} assert _run_final_validate({"api": {}}, config) is config - assert "without API encryption" in caplog.text - - -def test_redacted_lambda_body_leak_fails_build(project: Path) -> None: - """A sensitive value inside a !lambda body would be emitted verbatim by - the dumper; the scan must catch it.""" - (project / "wifi.yaml").write_text( - "password: my_password\nx: !lambda 'return \"my_password\";'\n" - ) - CORE.config = {"wifi": [{"password": SensitiveStr("my_password")}]} - discovered = _sources(project, "wifi.yaml") - with pytest.raises(EsphomeError, match="embedded"): - _gather_redacted(discovered) - - -def test_redacted_include_vars_leak_fails_build(project: Path) -> None: - """A sensitive value inside an !include vars: mapping is emitted by the - dumper; the scan must catch it.""" - (project / "entry.yaml").write_text( - "password: my_password\n" - "pkg: !include {file: wifi.yaml, vars: {url: 'http://user:my_password@host'}}\n" - ) - CORE.config = {"wifi": [{"password": SensitiveStr("my_password")}]} - discovered = _sources(project, "entry.yaml", "wifi.yaml") - with pytest.raises(EsphomeError, match="embedded"): - _gather_redacted(discovered) - - -def test_redacted_secret_sourced_overlap_warns_not_fails( - project: Path, caplog: pytest.LogCaptureFixture -) -> None: - """A value from a real !secret appearing inside another scalar (SSID in an - entity name) warns instead of failing; the remedy text of the hard error - does not apply to it.""" - yaml_util._SECRET_VALUES["Kitchen"] = "wifi_ssid" - (project / "wifi.yaml").write_text("ssid: Kitchen\nname: Kitchen Temperature\n") - CORE.config = {"wifi": [{"ssid": SensitiveStr("Kitchen")}]} - discovered = _sources(project, "wifi.yaml") - files = _gather_redacted(discovered) - assert b"!secret 'wifi_ssid'" in files["wifi.yaml"] - assert "appears inside the value at name in wifi.yaml" in caplog.text - - -def test_final_validate_rejects_keyless_encryption() -> None: - """A keyless provisionable `api: encryption:` accepts the all-zeros PSK - until provisioned, so it must not satisfy the gate.""" - with pytest.raises(cv.Invalid, match="requires API encryption"): - _run_final_validate( - {"api": {"encryption": {}}}, {CONF_ALLOW_UNENCRYPTED: False} - ) - - -def test_redacted_outside_root_secrets_gets_root_skeleton( - project: Path, tmp_path: Path -) -> None: - """A secrets file resolving outside the config root still yields a root - secrets.yaml skeleton, so the recovered config can resolve !secret.""" - target = tmp_path / "actual_creds.yaml" - target.write_text("api_key: SUPER_SECRET\n") - link = project / "secrets.yaml" - link.unlink() - link.symlink_to(target) - resolved = link.resolve() - CORE.config_path = project / "entry.yaml" - files = _gather_redacted( - DiscoveredYamlFiles([project / "entry.yaml", resolved], {resolved}) - ) - assert "secrets.yaml" in files - assert 'api_key: ""' in files["secrets.yaml"].decode() - assert b"SUPER_SECRET" not in b"".join(files.values()) - - -def test_gather_raises_esphome_error_on_cross_anchor_path( - project: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - """A file that cannot be made relative to the config root (a Windows - cross-drive path) surfaces as EsphomeError, not a raw ValueError.""" - - def fake_relative_to(self: Path, other: Path, walk_up: bool = False) -> Path: - raise ValueError("paths have different anchors") - - monkeypatch.setattr(Path, "relative_to", fake_relative_to) - discovered = _sources(project, "entry.yaml") - with pytest.raises(EsphomeError, match="does not share a root"): - _gather_files(discovered) - - -def test_final_validate_unencrypted_with_secrets_names_secrets_yaml( - caplog: pytest.LogCaptureFixture, -) -> None: - """allow_unencrypted combined with include_secrets warns about the - verbatim secrets.yaml specifically.""" - config = {CONF_ALLOW_UNENCRYPTED: True, "include_secrets": True} - assert _run_final_validate({"api": {}}, config) is config - assert "verbatim contents of secrets.yaml" in caplog.text + assert expected in caplog.text