Merge remote-tracking branch 'origin/dev' into store-yaml-firmware

# Conflicts:
#	tests/integration/raw_api_client.py

CODEOWNERS

@@ -634,6 +634,7 @@ esphome/components/st7789v/* @kbx81
 esphome/components/st7920/* @marsjan155
 esphome/components/statsd/* @Links2004
 esphome/components/stcc4/* @j9brown
+esphome/components/store_yaml/* @bdraco
 esphome/components/stts22h/* @B48D81EFCC
 esphome/components/substitutions/* @esphome/core
 esphome/components/sun/* @OttoWinter

esphome/components/api/api.proto

@@ -83,6 +83,7 @@ service APIConnection {
   rpc serial_proxy_set_mode(SerialProxySetModeRequest) returns (void) {}
 
   rpc sendspin_pairing_token(SendspinPairingTokenRequest) returns (SendspinPairingTokenResponse) {}
+  rpc get_yaml(GetYamlRequest) returns (void) {}
 }
 
 
@@ -401,6 +402,11 @@ message ZWaveProxyCapabilities {
   uint32 home_id = 2;
 }
 
+message StoreYamlCapabilities {
+  // Firmware embeds its YAML, so clients can request it with get_yaml
+  bool supported = 1;
+}
+
 message WizardCapabilities {
   // True when the device has a wizard configured, so DeviceWizardRequest will be answered
   bool configured = 1;
@@ -425,6 +431,7 @@ message DeviceCapabilitiesResponse {
       [(field_ifdef) = "USE_SERIAL_PROXY", (fixed_array_size_define) = "SERIAL_PROXY_COUNT"];
   WizardCapabilities wizard = 5 [(field_ifdef) = "USE_API_WIZARD"];
   SendspinCapabilities sendspin = 6 [(field_ifdef) = "USE_SENDSPIN"];
+  StoreYamlCapabilities store_yaml = 7 [(field_ifdef) = "USE_STORE_YAML"];
 }
 
 // ==================== DEVICE WIZARD ====================
@@ -3143,3 +3150,25 @@ message SendspinPairingTokenResponse {
   // The pairing token (Sendspin pairing.md "Pairing Token"). Set only when status is OK.
   string token = 2 [(max_data_length) = 127];
 }
+
+// ==================== STORE YAML ====================
+// Streams the compressed user YAML embedded in firmware; the client decompresses.
+// A GetYamlRequest during an active transfer is ignored; reconnect to restart.
+message GetYamlRequest {
+  option (id) = 161;
+  option (source) = SOURCE_CLIENT;
+  option (ifdef) = "USE_STORE_YAML";
+  option (no_delay) = true;
+}
+
+message GetYamlResponse {
+  option (id) = 162;
+  option (source) = SOURCE_SERVER;
+  option (ifdef) = "USE_STORE_YAML";
+
+  // A zstd frame of the EHY1 file envelope, sent from flash
+  bytes data = 1 [(force) = true, (pointer_to_buffer) = true, (progmem) = true];
+  bool done = 2 [(force) = true];
+  // Set on the first chunk only
+  uint32 total_size = 3;
+}

esphome/components/api/api_connection.cpp

@@ -366,6 +366,13 @@ void APIConnection::loop() {
   // (missing a frame is fine, missing a state update is not)
   this->try_send_camera_image_();
 #endif
+
+#ifdef USE_STORE_YAML
+  // Guard inline so the idle hot path pays a compare, not a function call.
+  if (this->store_yaml_pos_ != STORE_YAML_IDLE) {
+    this->try_send_store_yaml_();
+  }
+#endif
 }
 
 void APIConnection::check_keepalive_(uint32_t now) {
@@ -1222,6 +1229,36 @@ void APIConnection::on_camera_image_request(const CameraImageRequest &msg) {
 }
 #endif
 
+#ifdef USE_STORE_YAML
+void APIConnection::on_get_yaml_request() {
+  // A re-request while a transfer is in flight is ignored (see api.proto)
+  if (this->store_yaml_pos_ != STORE_YAML_IDLE)
+    return;
+  this->store_yaml_pos_ = 0;
+  this->try_send_store_yaml_();
+}
+
+void APIConnection::try_send_store_yaml_() {
+  // Advance only after a successful send, so a full TX buffer retries the same chunk from loop()
+  while (this->helper_->can_write_without_blocking()) {
+    const size_t to_send = std::min(STORE_YAML_DATA_SIZE - this->store_yaml_pos_, MAX_BATCH_PACKET_SIZE);
+    GetYamlResponse resp;
+    resp.data = STORE_YAML_DATA + this->store_yaml_pos_;
+    resp.data_len = to_send;
+    if (this->store_yaml_pos_ == 0)
+      resp.total_size = STORE_YAML_DATA_SIZE;
+    resp.done = this->store_yaml_pos_ + to_send == STORE_YAML_DATA_SIZE;
+    if (!this->send_message(resp))
+      return;
+    if (resp.done) {
+      this->store_yaml_pos_ = STORE_YAML_IDLE;
+      return;
+    }
+    this->store_yaml_pos_ += to_send;
+  }
+}
+#endif
+
 #ifdef USE_HOMEASSISTANT_TIME
 void APIConnection::on_get_time_response(const GetTimeResponse &value) {
   if (homeassistant::global_homeassistant_time != nullptr) {
@@ -2122,6 +2159,9 @@ bool APIConnection::send_device_capabilities_response_() {
 #if defined(USE_SENDSPIN) && defined(USE_ESP_IDF)
   resp.sendspin.feature_flags = sendspin_::global_sendspin_hub->get_feature_flags();
   resp.sendspin.client_id = sendspin_::global_sendspin_hub->get_client_id();
+#endif
+#ifdef USE_STORE_YAML
+  resp.store_yaml.supported = true;
 #endif
   return this->send_message(resp);
 }
@@ -2364,10 +2404,14 @@ 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) and camera frames (high-frequency noise)
+  // 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
+#endif
+#ifdef USE_STORE_YAML
+      && message_type != GetYamlResponse::MESSAGE_TYPE
 #endif
   ) {
     auto *proto_msg = static_cast<const ProtoMessage *>(msg);

esphome/components/api/api_connection.h

@@ -32,6 +32,11 @@ namespace esphome::api {
 // Forward-declared to break the api_server.h cycle; full-type inlines are in api_connection_buffer.h.
 class APIServer;
 
+#ifdef USE_STORE_YAML
+/// The zstd YAML envelope generated by store_yaml, STORE_YAML_DATA_SIZE bytes in flash.
+extern const uint8_t STORE_YAML_DATA[] PROGMEM;
+#endif
+
 // One shared flash string for every refused-frame warning: send_message()
 // fails as soon as the TCP buffer is full, and each caller only pays for its
 // short name. The guard drops the helper and its arguments below WARN.
@@ -123,6 +128,9 @@ class APIConnection final : public APIServerConnectionBase {
   void set_camera_state(std::shared_ptr<camera::CameraImage> image);
   void on_camera_image_request(const CameraImageRequest &msg);
 #endif
+#ifdef USE_STORE_YAML
+  void on_get_yaml_request();
+#endif
 #ifdef USE_CLIMATE
   bool send_climate_state(climate::Climate *climate);
   void on_climate_command_request(const ClimateCommandRequest &msg);
@@ -421,6 +429,13 @@ class APIConnection final : public APIServerConnectionBase {
   void try_send_camera_image_();
 #endif
 
+#ifdef USE_STORE_YAML
+  void try_send_store_yaml_();
+  static constexpr size_t STORE_YAML_IDLE = std::numeric_limits<size_t>::max();
+  // Streaming offset into STORE_YAML_DATA, or STORE_YAML_IDLE
+  size_t store_yaml_pos_{STORE_YAML_IDLE};
+#endif
+
 #ifdef USE_API_HOMEASSISTANT_STATES
   void process_state_subscriptions_();
 #endif

esphome/components/api/api_pb2.cpp

@@ -299,6 +299,19 @@ uint32_t ZWaveProxyCapabilities::calc_size_msg(const void *self) {
   return size;
 }
 #endif
+#ifdef USE_STORE_YAML
+uint8_t *StoreYamlCapabilities::encode_msg(const void *self, uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM) {
+  const auto &msg = *static_cast<const StoreYamlCapabilities *>(self);
+  pos = ProtoEncode::encode_bool(pos PROTO_ENCODE_DEBUG_ARG, 1, msg.supported);
+  return pos;
+}
+uint32_t StoreYamlCapabilities::calc_size_msg(const void *self) {
+  const auto &msg = *static_cast<const StoreYamlCapabilities *>(self);
+  uint32_t size = 0;
+  size += ProtoSize::calc_bool(1, msg.supported);
+  return size;
+}
+#endif
 #ifdef USE_API_WIZARD
 uint8_t *WizardCapabilities::encode_msg(const void *self, uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM) {
   const auto &msg = *static_cast<const WizardCapabilities *>(self);
@@ -348,6 +361,9 @@ uint8_t *DeviceCapabilitiesResponse::encode_msg(const void *self, uint8_t *__res
 #endif
 #ifdef USE_SENDSPIN
   pos = ProtoEncode::encode_optional_sub_message(pos PROTO_ENCODE_DEBUG_ARG, 6, msg.sendspin);
+#endif
+#ifdef USE_STORE_YAML
+  pos = ProtoEncode::encode_optional_sub_message(pos PROTO_ENCODE_DEBUG_ARG, 7, msg.store_yaml);
 #endif
   return pos;
 }
@@ -373,6 +389,9 @@ uint32_t DeviceCapabilitiesResponse::calc_size_msg(const void *self) {
 #endif
 #ifdef USE_SENDSPIN
   size += ProtoSize::calc_message(1, msg.sendspin.calculate_size());
+#endif
+#ifdef USE_STORE_YAML
+  size += ProtoSize::calc_message(1, msg.store_yaml.calculate_size());
 #endif
   return size;
 }
@@ -4407,6 +4426,24 @@ uint32_t SendspinPairingTokenResponse::calc_size_msg(const void *self) {
   return size;
 }
 #endif
+#ifdef USE_STORE_YAML
+uint8_t *GetYamlResponse::encode_msg(const void *self, uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM) {
+  const auto &msg = *static_cast<const GetYamlResponse *>(self);
+  pos = ProtoEncode::encode_progmem_bytes_force(pos PROTO_ENCODE_DEBUG_ARG, 1, msg.data, msg.data_len);
+  pos = ProtoEncode::write_raw_byte(pos PROTO_ENCODE_DEBUG_ARG, 16);
+  pos = ProtoEncode::write_raw_byte(pos PROTO_ENCODE_DEBUG_ARG, msg.done ? 0x01 : 0x00);
+  pos = ProtoEncode::encode_uint32(pos PROTO_ENCODE_DEBUG_ARG, 3, msg.total_size);
+  return pos;
+}
+uint32_t GetYamlResponse::calc_size_msg(const void *self) {
+  const auto &msg = *static_cast<const GetYamlResponse *>(self);
+  uint32_t size = 0;
+  size += ProtoSize::calc_length_force(1, msg.data_len);
+  size += ProtoSize::calc_bool_force(1);
+  size += ProtoSize::calc_uint32(1, msg.total_size);
+  return size;
+}
+#endif
 #ifndef HAS_PROTO_MESSAGE_DUMP
 static_assert(!std::is_polymorphic_v<HelloRequest>, "decodable messages carry no vtable");
 static_assert(!std::is_polymorphic_v<DisconnectRequest>, "decodable messages carry no vtable");

esphome/components/api/api_pb2.h

@@ -734,6 +734,23 @@ class ZWaveProxyCapabilities final : public ProtoMessage {
  protected:
 };
 #endif
+#ifdef USE_STORE_YAML
+class StoreYamlCapabilities final : public ProtoMessage {
+ public:
+  bool supported{false};
+  static uint8_t *encode_msg(const void *self, uint8_t *pos PROTO_ENCODE_DEBUG_PARAM);
+  uint8_t *encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const {
+    return encode_msg(this, buffer.get_pos() PROTO_ENCODE_DEBUG_ARG);
+  }
+  static uint32_t calc_size_msg(const void *self);
+  uint32_t calculate_size() const { return calc_size_msg(this); }
+#ifdef HAS_PROTO_MESSAGE_DUMP
+  const char *dump_to(DumpBuffer &out) const override;
+#endif
+
+ protected:
+};
+#endif
 #ifdef USE_API_WIZARD
 class WizardCapabilities final : public ProtoMessage {
  public:
@@ -772,7 +789,7 @@ class SendspinCapabilities final : public ProtoMessage {
 class DeviceCapabilitiesResponse final : public ProtoMessage {
  public:
   static constexpr uint16_t MESSAGE_TYPE = 150;
-  static constexpr uint8_t ESTIMATED_SIZE = 136;
+  static constexpr uint8_t ESTIMATED_SIZE = 153;
 #ifdef HAS_PROTO_MESSAGE_DUMP
   const LogString *message_name() const override { return LOG_STR("device_capabilities_response"); }
 #endif
@@ -793,6 +810,9 @@ class DeviceCapabilitiesResponse final : public ProtoMessage {
 #endif
 #ifdef USE_SENDSPIN
   SendspinCapabilities sendspin{};
+#endif
+#ifdef USE_STORE_YAML
+  StoreYamlCapabilities store_yaml{};
 #endif
   static uint8_t *encode_msg(const void *self, uint8_t *pos PROTO_ENCODE_DEBUG_PARAM);
   uint8_t *encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const {
@@ -4209,5 +4229,30 @@ class SendspinPairingTokenResponse final : public ProtoMessage {
  protected:
 };
 #endif
+#ifdef USE_STORE_YAML
+class GetYamlResponse final : public ProtoMessage {
+ public:
+  static constexpr uint16_t MESSAGE_TYPE = 162;
+  static constexpr uint8_t ESTIMATED_SIZE = 25;
+#ifdef HAS_PROTO_MESSAGE_DUMP
+  const LogString *message_name() const override { return LOG_STR("get_yaml_response"); }
+#endif
+  const uint8_t *data{nullptr};
+  uint16_t data_len{0};
+  bool done{false};
+  uint32_t total_size{0};
+  static uint8_t *encode_msg(const void *self, uint8_t *pos PROTO_ENCODE_DEBUG_PARAM);
+  uint8_t *encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const {
+    return encode_msg(this, buffer.get_pos() PROTO_ENCODE_DEBUG_ARG);
+  }
+  static uint32_t calc_size_msg(const void *self);
+  uint32_t calculate_size() const { return calc_size_msg(this); }
+#ifdef HAS_PROTO_MESSAGE_DUMP
+  const char *dump_to(DumpBuffer &out) const override;
+#endif
+
+ protected:
+};
+#endif
 
 }  // namespace esphome::api

esphome/components/api/api_pb2_dump.cpp

@@ -1127,6 +1127,13 @@ const char *ZWaveProxyCapabilities::dump_to(DumpBuffer &out) const {
   return out.c_str();
 }
 #endif
+#ifdef USE_STORE_YAML
+const char *StoreYamlCapabilities::dump_to(DumpBuffer &out) const {
+  MessageDumpHelper helper(out, ESPHOME_PSTR("StoreYamlCapabilities"));
+  dump_field(out, ESPHOME_PSTR("supported"), this->supported);
+  return out.c_str();
+}
+#endif
 #ifdef USE_API_WIZARD
 const char *WizardCapabilities::dump_to(DumpBuffer &out) const {
   MessageDumpHelper helper(out, ESPHOME_PSTR("WizardCapabilities"));
@@ -1175,6 +1182,11 @@ const char *DeviceCapabilitiesResponse::dump_to(DumpBuffer &out) const {
   out.append(2, ' ').append_p(ESPHOME_PSTR("sendspin")).append(": ");
   this->sendspin.dump_to(out);
   out.append("\n");
+#endif
+#ifdef USE_STORE_YAML
+  out.append(2, ' ').append_p(ESPHOME_PSTR("store_yaml")).append(": ");
+  this->store_yaml.dump_to(out);
+  out.append("\n");
 #endif
   return out.c_str();
 }
@@ -3016,6 +3028,15 @@ const char *SendspinPairingTokenResponse::dump_to(DumpBuffer &out) const {
   return out.c_str();
 }
 #endif
+#ifdef USE_STORE_YAML
+const char *GetYamlResponse::dump_to(DumpBuffer &out) const {
+  MessageDumpHelper helper(out, ESPHOME_PSTR("GetYamlResponse"));
+  dump_progmem_bytes_field(out, ESPHOME_PSTR("data"), this->data, this->data_len);
+  dump_field(out, ESPHOME_PSTR("done"), this->done);
+  dump_field(out, ESPHOME_PSTR("total_size"), this->total_size);
+  return out.c_str();
+}
+#endif
 
 }  // namespace esphome::api
 

esphome/components/api/api_pb2_service.cpp

@@ -760,6 +760,15 @@ void APIConnection::read_message_(uint32_t msg_size, uint32_t msg_type, const ui
       this->on_sendspin_pairing_token_request();
       break;
     }
+#endif
+#ifdef USE_STORE_YAML
+    case 161 /* GetYamlRequest is empty */: {
+#ifdef HAS_PROTO_MESSAGE_DUMP
+      this->log_receive_message_(LOG_STR("on_get_yaml_request"));
+#endif
+      this->on_get_yaml_request();
+      break;
+    }
 #endif
     default:
       break;

esphome/components/api/api_pb2_service.h

@@ -256,6 +256,10 @@ class APIServerConnectionBase {
 #ifdef USE_SENDSPIN
   void on_sendspin_pairing_token_request() {};
 #endif
+
+#ifdef USE_STORE_YAML
+  void on_get_yaml_request() {};
+#endif
 };
 
 }  // namespace esphome::api

esphome/components/store_yaml/__init__.py

@@ -0,0 +1,723 @@
+from __future__ import annotations
+
+from collections.abc import Generator
+from dataclasses import dataclass
+import logging
+import os
+from pathlib import Path
+import re
+import struct
+
+import yaml
+
+from esphome import yaml_util
+import esphome.codegen as cg
+from esphome.components import noise, packages
+from esphome.config_helpers import Extend, Remove
+import esphome.config_validation as cv
+from esphome.const import (
+    CONF_API,
+    CONF_DEFAULTS,
+    CONF_ID,
+    CONF_PACKAGES,
+    CONF_SSID,
+    CONF_SUBSTITUTIONS,
+)
+from esphome.core import CORE, EsphomeError, Lambda
+from esphome.expression import SUBSTITUTION_VARIABLE_PROG
+import esphome.final_validate as fv
+from esphome.git import URL_USERINFO_RE
+from esphome.helpers import ensure_unique_string, list_starts_with, zstd_module
+from esphome.types import ConfigType
+
+_LOGGER = logging.getLogger(__name__)
+
+CODEOWNERS = ["@bdraco"]
+DEPENDENCIES = ["api"]
+DOMAIN = "store_yaml"
+
+CONF_INCLUDE_SECRETS = "include_secrets"
+CONF_ALLOW_UNENCRYPTED = "allow_unencrypted"
+
+# 22 is the max and gives ~70-90% reduction on YAML.
+ZSTD_LEVEL = 22
+# "EHY1" = ESPHome YAML, version 1.
+ENVELOPE_MAGIC = b"EHY1"
+SECRETS_SKELETON_HEADER = (
+    "# Redacted by store_yaml. Fill in these values and the recovered\n"
+    "# config is ready to flash.\n"
+)
+UNCAPTURED_NOTE_PATH = "store_yaml_uncaptured.yaml"
+_FIX = (
+    "Reference them with `!secret`, or set `include_secrets: true` to embed "
+    "secrets deliberately."
+)
+# Substitution names to every value they may take in one file.
+_Scope = dict[str, list[object]]
+# An `!include`, its vars before any folding, and the config path of its site.
+_Include = tuple[yaml_util.IncludeFile, dict[str, object], tuple[str, ...]]
+# Stands in for an integer credential at its keys and definitions.
+_INT_MARK = "\x00"
+# A removed branch is never validated, so its credentials are known only by name.
+_CREDENTIAL_WORDS = (
+    "password",
+    "key",
+    "secret",
+    "token",
+    "psk",
+    "tlv",
+    "pairing_code",
+    "pin_code",
+)
+# Every key validated with cv.sensitive (a test keeps this complete) and the
+# packet_transport `encryption: <key>` shorthand: credentials wherever they appear,
+# also when a merge or override drops them before validation. SSIDs are broadcast,
+# and a bare `key` also names keypad keys; both are checked where validation kept them.
+_SENSITIVE_KEYS = frozenset(
+    {"password", "new_password", "private_key", "peer_preshared_key"}
+    | {"client_certificate_key", "network_key", "old_key", "pin_code", "pskc", "tlv"}
+    | {"static_pairing_code", "encryption", "bindkey", "decryption_key", "auth_key"}
+)
+# Keys under these name declared arguments, not values (`variables: {password: string}`).
+_DECLARATIONS = frozenset({"variables", "parameters"})
+# cv.bind_key's normalised output: 32 uppercase hex digits.
+_BIND_KEY_RE = re.compile(r"[0-9A-F]{32}")
+# A quoted string or a number: a lambda without one cannot spell a credential.
+_LITERAL_RE = re.compile(r"""["']|(?<![\w.])\d""")
+
+CONFIG_SCHEMA = cv.Schema(
+    {
+        cv.Optional(CONF_INCLUDE_SECRETS, default=False): cv.boolean,
+        cv.Optional(CONF_ALLOW_UNENCRYPTED, default=False): cv.boolean,
+    }
+)
+
+
+def _final_validate(config: ConfigType) -> ConfigType:
+    """Require an API encryption key unless ``allow_unencrypted`` is set."""
+    full = fv.full_config.get()
+    if noise.static_encryption_key(full.get(CONF_API, {})) is not None:
+        return config
+    if config.get(CONF_ALLOW_UNENCRYPTED):
+        _LOGGER.warning(
+            "store_yaml is enabled without API encryption; any client that can "
+            "reach the device on the network can pull the embedded YAML%s.",
+            " including the verbatim contents of secrets.yaml"
+            if config.get(CONF_INCLUDE_SECRETS)
+            else "",
+        )
+        return config
+    raise cv.Invalid(
+        "store_yaml requires API encryption (`api.encryption.key`), or any client "
+        "that reaches the device can read the embedded YAML. Set "
+        "`store_yaml.allow_unencrypted: true` to accept that."
+    )
+
+
+FINAL_VALIDATE_SCHEMA = _final_validate
+
+
+def _gather_files(
+    discovered: yaml_util.DiscoveredYamlFiles,
+) -> tuple[list[tuple[str, Path]], set[str]]:
+    """(envelope path, named path) per name a file was reached by, and the envelope
+    paths of secrets files."""
+    if discovered.load_errors:
+        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 "
+            "captured (%s); the embedded recovery data will not contain them",
+            len(discovered.unresolved),
+            ", ".join(discovered.unresolved),
+        )
+
+    root = yaml_util.named_path(CORE.config_path).parent
+    entries: list[tuple[str, Path]] = []
+    secret_rels: set[str] = set()
+    for path in discovered.files:
+        for alias in discovered.aliases.get(path, [path]):
+            # Keep ".." for files outside the root so same-named files don't collide.
+            try:
+                rel_str = alias.relative_to(root, walk_up=True).as_posix()
+            except ValueError as err:
+                raise EsphomeError(
+                    f"store_yaml: cannot place {alias} in the recovery envelope; "
+                    f"it does not share a root with {root}: {err}"
+                ) from err
+            if path in discovered.secrets:
+                secret_rels.add(rel_str)
+            entries.append((rel_str, alias))
+    return entries, secret_rels
+
+
+def _read_files_verbatim(entries: list[tuple[str, Path]]) -> list[tuple[str, bytes]]:
+    """Read each file's exact on-disk bytes (the `include_secrets: true` path)."""
+    files: list[tuple[str, bytes]] = []
+    for rel, path in entries:
+        try:
+            files.append((rel, path.read_bytes()))
+        except OSError as err:
+            raise EsphomeError(
+                f"store_yaml: cannot read tracked YAML file {path}: {err}"
+            ) from err
+    return files
+
+
+class _IncludeName(str):
+    """An `!include` file name: scanned for secrets, never taken for the value it loads."""
+
+    __slots__ = ()
+
+
+def _iter_nodes(
+    node: object, path: tuple[str, ...] = ()
+) -> Generator[tuple[tuple[str, ...], object, bool]]:
+    """Yield (config_path, value, is_key) for every key and scalar, unwrapping the
+    tagged wrappers the dumper writes as text."""
+    if isinstance(node, dict):
+        for key, value in node.items():
+            yield path, str(key), True
+            yield from _iter_nodes(value, (*path, str(key)))
+    elif isinstance(node, (list, tuple)):
+        for item in node:
+            yield from _iter_nodes(item, path)
+    elif isinstance(node, Lambda):
+        yield path, node, False
+    elif isinstance(node, (Extend, Remove)):
+        yield path, node.value, False
+    elif isinstance(node, yaml_util.IncludeFile):
+        yield path, _IncludeName(node.file), False
+        if isinstance(node.condition, str):
+            yield path, node.condition, False
+        if node.vars:
+            yield from _iter_nodes(node.vars, path)
+    elif isinstance(node, (str, int, float)) and not isinstance(node, bool):
+        yield path, node, False
+
+
+def _str_keys(value: object) -> dict[str, object]:
+    """`value` as a mapping with string keys, or empty."""
+    return {str(k): v for k, v in value.items()} if isinstance(value, dict) else {}
+
+
+def _ends_with(full: tuple[str, ...], part: tuple[str, ...]) -> bool:
+    """`part` ends `full`, skipping segments validation inserted (`wifi.password`
+    to `wifi.networks.password`)."""
+    if full[-1] != part[-1]:
+        return False
+    rest = iter(reversed(full[:-1]))
+    return all(seg in rest for seg in reversed(part[:-1]))
+
+
+def _merged(path: tuple[str, ...]) -> set[tuple[str, ...]]:
+    """`path` as validation sees it: every `packages[.<name>]` prefix merges away."""
+    if path and path[0] == CONF_PACKAGES:
+        return _merged(path[1:]) | _merged(path[2:])
+    return {path}
+
+
+def _defines_substitution(path: tuple[str, ...]) -> bool:
+    """`path` is a `substitutions:` entry, also inside an inline package."""
+    return any(len(p) > 1 and p[0] == CONF_SUBSTITUTIONS for p in _merged(path))
+
+
+@dataclass
+class _SensitiveValue:
+    secret_name: str
+    config_path: str  # dotted path, for messages (never the value itself)
+    from_secret: bool  # loaded through a real `!secret`
+    # Keys the value sits at, to tell its own occurrences from collisions.
+    sensitive_keys: set[str]
+    # Integers skip the substring scan; short PINs would match unrelated text.
+    is_int: bool = True
+
+
+def _include_scopes(
+    includes: dict[str, list[_Include]],
+    trees: dict[str, object],
+    sources: dict[str, Path],
+) -> tuple[dict[str, _Scope], dict[str, list[tuple[str, ...]]]]:
+    """Per file: every value a substitution name may take (per include: its vars,
+    else the file's `defaults:`, else the caller's chain, else substitutions) and
+    the config paths its content lands under."""
+    substitutions: _Scope = {}
+    for tree in trees.values():
+        for path, node, is_key in _iter_nodes(tree):
+            if not is_key and _defines_substitution(path):
+                substitutions.setdefault(path[-1], []).append(node)
+    defaults = {
+        rel: _str_keys(_str_keys(tree).get(CONF_DEFAULTS))
+        for rel, tree in trees.items()
+    }
+    rels_by_path: dict[Path, list[str]] = {}
+    for rel, path in sources.items():
+        rels_by_path.setdefault(path.resolve(), []).append(rel)
+    edges: dict[str, list[tuple[str, dict[str, object], tuple[str, ...]]]] = {}
+    for src, items in includes.items():
+        for include, vars, site in items:
+            if include.has_unresolved_file() or not yaml_util.include_enabled(include):
+                continue
+            target = (include.parent_file.parent / include.file).resolve()
+            edges.setdefault(src, []).extend(
+                (rel, vars, site) for rel in rels_by_path.get(target, ())
+            )
+    # Each way a file is reached: its vars chain, built as the loader does, and
+    # the include-site prefix; a caller's chain flows into nested includes.
+    targeted = {rel for targets in edges.values() for rel, _, _ in targets}
+    invocations: dict[str, list[tuple[dict[str, object], tuple[str, ...]]]] = {}
+    pending = [(rel, defaults[rel], (), ()) for rel in trees if rel not in targeted]
+    while pending:
+        rel, chain, prefix, lineage = pending.pop()
+        if (chain, prefix) in invocations.setdefault(rel, []):
+            continue
+        invocations[rel].append((chain, prefix))
+        pending.extend(
+            (dst, {**chain, **defaults[dst], **vars}, prefix + site, (*lineage, rel))
+            for dst, vars, site in edges.get(rel, ())
+            if dst not in lineage  # an include cycle never ends otherwise
+        )
+    scopes: dict[str, _Scope] = {}
+    prefixes: dict[str, list[tuple[str, ...]]] = {}
+    for rel in trees:
+        chains = invocations.get(rel) or [(defaults[rel], ())]
+        scope: _Scope = {}
+        for chain, _ in chains:
+            for name in {*substitutions, *chain}:
+                scope.setdefault(name, []).extend(
+                    [chain[name]] if name in chain else substitutions[name]
+                )
+        scopes[rel] = scope
+        prefixes[rel] = [prefix for _, prefix in chains]
+    return scopes, prefixes
+
+
+def _check_sensitive_usage(
+    sensitive: dict[str, _SensitiveValue],
+    sensitive_paths: set[tuple[str, ...]],
+    trees: dict[str, object],
+    scopes: dict[str, _Scope],
+    prefixes: dict[str, list[tuple[str, ...]]],
+    removed: set[tuple[str, ...]],
+) -> None:
+    """A sensitive value is a whole scalar (swapped) or fails, except a secret SSID
+    inside a name (warned); unverifiable credential locations and URL userinfo fail."""
+    credential_keys = {sp[-1] for sp in sensitive_paths if sp}
+
+    def verified(text: str) -> bool:
+        return not text or text in sensitive or yaml_util.is_secret(text) is not None
+
+    def references_verified(
+        text: str, scope: _Scope, seen: frozenset = frozenset()
+    ) -> bool:
+        """Every reference names definitions that are verified or alias verified ones."""
+        names = [m.strip("{}") for m in SUBSTITUTION_VARIABLE_PROG.findall(text)]
+        return bool(names) and all(
+            name in scope
+            and name not in seen
+            and all(
+                verified(str(v))
+                or SUBSTITUTION_VARIABLE_PROG.fullmatch(str(v)) is not None
+                and references_verified(str(v), scope, seen | {name})
+                for v in scope[name]
+            )
+            for name in names
+        )
+
+    def credential_path(path: tuple[str, ...]) -> bool:
+        """Whether `path` lands where the validated config holds a credential."""
+        if not path:
+            return False
+        merged = _merged(path)
+        if path[-1] in _SENSITIVE_KEYS and not _DECLARATIONS.intersection(path[-2:-1]):
+            return True
+        if any(p and any(list_starts_with(p, r) for r in removed) for p in merged):
+            return path[-1] in credential_keys or any(
+                word in path[-1] for word in _CREDENTIAL_WORDS
+            )
+        return path[-1] in credential_keys and any(
+            # One segment says nothing about where an `!include_dir_*` file lands
+            len(p) > 1 and any(_ends_with(sp, p) for sp in sensitive_paths)
+            for p in merged
+        )
+
+    def shown(path: tuple[str, ...]) -> str:
+        """`path` for messages, credential segments masked."""
+        return ".".join(
+            f"<!secret {sensitive[seg].secret_name}>" if seg in sensitive else seg
+            for seg in path
+        )
+
+    failures: dict[str, list[str]] = {
+        "appear embedded inside larger values or as mapping keys, which redaction "
+        "does not replace": [],
+        "cannot be confirmed redacted: they sit at a credential key or location but "
+        "are not a known secret (for example a literal password, or one a merge, "
+        "override or `!remove` dropped before validation)": [],
+    }
+    embedded, unverified = failures.values()
+    for rel, tree in trees.items():
+        scope = scopes[rel]
+        for path, node, is_key in _iter_nodes(tree):
+            text = str(node)
+            where = (
+                f"a mapping key under {shown(path) or 'the document root'}"
+                if is_key
+                else shown(path)
+            )
+            if verified(text):
+                # A whole scalar is swapped for its `!secret`; a key never is.
+                if is_key and (info := sensitive.get(text)) is not None:
+                    embedded.append(f"{info.config_path} (as {where} in {rel})")
+                continue
+            if not is_key and any(
+                SUBSTITUTION_VARIABLE_PROG.sub("", userinfo) not in ("", ":")
+                or not references_verified(userinfo, scope)
+                for userinfo in URL_USERINFO_RE.findall(text)
+            ):
+                unverified.append(f"{where} in {rel} (a URL with credentials)")
+            elif (
+                not is_key
+                and not isinstance(node, _IncludeName)
+                and any(credential_path(prefix + path) for prefix in prefixes[rel])
+                and not (
+                    _LITERAL_RE.search(text) is None
+                    if isinstance(node, Lambda)
+                    else SUBSTITUTION_VARIABLE_PROG.fullmatch(text) is not None
+                    and references_verified(text, scope)
+                )
+            ):
+                unverified.append(f"{where} in {rel}")
+                continue
+            for value, other in sensitive.items():
+                # A key the value is sensitive at (`password` for `word`) is structure.
+                if (
+                    other.is_int
+                    or value not in text
+                    or (is_key and text in other.sensitive_keys)
+                ):
+                    continue
+                if other.from_secret and other.sensitive_keys <= {CONF_SSID}:
+                    # A broadcast SSID inside a name is common; warn without the value.
+                    _LOGGER.warning(
+                        "store_yaml: !secret %s is not redacted inside %s in %s",
+                        other.secret_name,
+                        where,
+                        rel,
+                    )
+                else:
+                    embedded.append(f"{other.config_path} (inside {where} in {rel})")
+    for why, hits in failures.items():
+        if hits:
+            raise EsphomeError(
+                f"store_yaml: sensitive value(s) {why}: {', '.join(hits)}. {_FIX}"
+            )
+
+
+def _collect_sensitive_values() -> tuple[
+    dict[str, _SensitiveValue], set[tuple[str, ...]]
+]:
+    """Map each cv.sensitive value to its `!secret` name (inline values get one from
+    their path) and collect every sensitive path, empty values included."""
+    used = yaml_util.registered_secret_names()
+    result: dict[str, _SensitiveValue] = {}
+    paths: set[tuple[str, ...]] = set()
+    for path, node, is_key in _iter_nodes(CORE.config):
+        if is_key or not isinstance(node, yaml_util.Sensitive):
+            continue
+        paths.add(path)
+        if isinstance(node, Lambda):
+            if _LITERAL_RE.search(node.value) is not None:
+                raise EsphomeError(
+                    f"store_yaml: the credential at {'.'.join(path)} is set by a "
+                    f"lambda holding a literal, which cannot be redacted. {_FIX}"
+                )
+            continue
+        if isinstance(node, str) and not node:
+            continue
+        # Decimal for ints, to match the raw tree's parsed scalars, not HexInt's hex.
+        value = str(int(node)) if isinstance(node, int) else str(node)
+        if (entry := result.get(value)) is None:
+            name = yaml_util.is_secret(value)
+            if name is None and _BIND_KEY_RE.fullmatch(value):
+                # cv.bind_key uppercases the key; its secret keeps the user's spelling
+                secrets = yaml_util.registered_secret_values().items()
+                name = next((n for v, n in secrets if v.upper() == value), None)
+            from_secret = name is not None
+            if not from_secret:
+                name = ensure_unique_string("_".join(path) or "secret", used)
+                used.add(name)
+            entry = result[value] = _SensitiveValue(
+                name, ".".join(path), from_secret, set()
+            )
+        entry.is_int = entry.is_int and isinstance(node, int)
+        entry.sensitive_keys.update(path[-1:])
+    return result, paths
+
+
+def _symlink_descriptions(discovered: yaml_util.DiscoveredYamlFiles) -> list[str]:
+    """Each directory symlink the config went through, as `link -> target`."""
+    root = Path(CORE.config_path).absolute().parent
+    return [
+        f"{link.relative_to(root, walk_up=True).as_posix()} -> "
+        f"{Path(os.path.relpath(target, link.parent.resolve())).as_posix()}"
+        for link, target in discovered.links.items()
+    ]
+
+
+def _uncaptured_note(
+    unresolved: list[str], remote_packages: list[str], symlinks: list[str]
+) -> tuple[str, bytes] | None:
+    """A comment-only entry listing uncaptured content, or None if there is none."""
+    sections = (
+        (
+            "These !include paths use substitutions; restore the files manually:",
+            unresolved,
+        ),
+        (
+            "These packages come from remote sources; re-fetch them to complete this config:",
+            remote_packages,
+        ),
+        ("These directories were symlinks; recreate them as link -> target:", symlinks),
+    )
+    body = "".join(
+        f"# {title}\n" + "".join(f"#   {item}\n" for item in items)
+        for title, items in sections
+        if items
+    )
+    if not body:
+        return None
+    header = "# store_yaml: the following content could not be captured.\n"
+    return UNCAPTURED_NOTE_PATH, (header + body).encode("utf-8")
+
+
+def _remote_package_descriptions() -> list[str]:
+    """Each remote package source as its secret name or masked URL, plus ref."""
+
+    def describe(text: str) -> str:
+        if (name := yaml_util.is_secret(text)) is not None:
+            return f"!secret {name}"
+        # A secret substituted into the URL or ref is shown by name, longest first so
+        # a secret containing another is replaced whole.
+        secrets = yaml_util.registered_secret_values().items()
+        for value, name in sorted(secrets, key=lambda item: -len(item[0])):
+            if value and value in text:
+                text = text.replace(value, f"<!secret {name}>")
+        return URL_USERINFO_RE.sub("://***@", text)
+
+    return [
+        describe(source.url) + (f"@{describe(source.ref)}" if source.ref else "")
+        for source in packages.get_remote_package_sources()
+    ]
+
+
+def _build_secrets_skeleton(keys: set[str]) -> bytes:
+    # Plain PyYAML quotes a key such as "true" and does no secret substitution.
+    body = yaml.safe_dump(dict.fromkeys(sorted(keys), ""), sort_keys=False)
+    return (SECRETS_SKELETON_HEADER + (body if keys else "")).encode("utf-8")
+
+
+def _generate_redacted_files(
+    entries: list[tuple[str, Path]], secret_rels: set[str], remote_packages: list[str]
+) -> list[tuple[str, bytes]]:
+    """Re-dump captured files with sensitive values as `!secret`, secrets files as a
+    skeleton; comments and formatting are lost."""
+    sensitive, sensitive_paths = _collect_sensitive_values()
+    includes: dict[str, list[_Include]] = {}
+    removed: dict[str, set[tuple[str, ...]]] = {}
+
+    def prepare(
+        node: object,
+        source: Path,
+        rel: str,
+        path: tuple[str, ...],
+        in_vars: bool = False,
+    ) -> object:
+        """Fold container secrets to their registered text (a plain str would skip
+        the secret check), swap integer credentials at their own keys, point each
+        `!include` at its target's captured name, collect includes and removals."""
+        if isinstance(node, (dict, list)) and yaml_util.is_secret(node) is not None:
+            return yaml_util.make_data_base(str(node))
+        if isinstance(node, dict):
+            for key, value in node.items():
+                node[key] = prepare(value, source, rel, (*path, str(key)), in_vars)
+        elif isinstance(node, list):
+            for i, item in enumerate(node):
+                node[i] = prepare(item, source, rel, path, in_vars)
+        elif isinstance(node, Remove):
+            # `id: !remove x` drops the whole list item, not just its id.
+            removed.setdefault(rel, set()).add(
+                path[:-1] if path[-1:] == (CONF_ID,) else path
+            )
+        elif isinstance(node, (int, str)) and not isinstance(node, bool):
+            # An integer credential, quoted or not, at its own keys or as a variable
+            # definition a credential may reference.
+            info = sensitive.get(str(node))
+            at_key = info is not None and info.sensitive_keys.intersection(path[-1:])
+            definition = (
+                in_vars or path[:1] == (CONF_DEFAULTS,) or _defines_substitution(path)
+            )
+            if info and info.is_int and (at_key or definition):
+                return yaml_util.make_data_base(_INT_MARK + info.secret_name)
+            # cv.bind_key uppercases the key, so match an inline one the same way
+            upper = node.upper() if isinstance(node, str) else ""
+            bind = sensitive.get(upper) if _BIND_KEY_RE.fullmatch(upper) else None
+            if bind and bind.sensitive_keys.intersection(path[-1:]):
+                return yaml_util.make_data_base(upper)
+        elif isinstance(node, yaml_util.IncludeFile):
+            vars = _str_keys(node.vars)
+            if yaml_util.is_secret(node.vars) is not None:
+                vars = dict.fromkeys(vars, node.vars)
+            includes.setdefault(rel, []).append((node, vars, path))
+            if node.vars:
+                node.vars = prepare(node.vars, source, rel, path, in_vars=True)
+            if not node.has_unresolved_file():
+                # Where the loader read it, through directory symlinks and `!include_dir_*` moves.
+                target = yaml_util.named_path(Path(node.parent_file).parent / node.file)
+                node = node.with_file(
+                    Path(os.path.relpath(target, source.parent)).as_posix()
+                )
+        return node
+
+    for rel, path in entries:
+        # A secrets file becomes a key skeleton; what it includes would ship verbatim.
+        if rel in secret_rels and any(
+            isinstance(token, yaml.TagToken) and token.value[1].startswith("include")
+            for token in yaml.scan(path.read_text(encoding="utf-8"), yaml.SafeLoader)
+        ):
+            raise EsphomeError(
+                f"store_yaml: {rel} uses !include, which redaction cannot follow. "
+                f"Inline the files into {rel}, or set `include_secrets: true`."
+            )
+    captured = {rel: path for rel, path in entries if rel not in secret_rels}
+    trees: dict[str, object] = {}
+    for rel, path in captured.items():
+        # A document that is one `!include` keeps its vars and condition.
+        with path.open(encoding="utf-8") as handle:
+            tree = yaml_util.parse_yaml(path, handle, resolve_document_include=False)
+        trees[rel] = prepare(tree, yaml_util.named_path(path), rel, ())
+    scopes, prefixes = _include_scopes(includes, trees, captured)
+    removed_at = {
+        merged
+        for rel, paths in removed.items()
+        for path in paths
+        for prefix in prefixes[rel]
+        for merged in _merged(prefix + path)
+        if merged
+    }
+
+    # An integer credential is swapped by prepare(), not by the dumper, so an
+    # unrelated short number (`accuracy_decimals: 0`) is left alone.
+    registered = {
+        (_INT_MARK + info.secret_name if info.is_int else value): info.secret_name
+        for value, info in sensitive.items()
+    }
+    with yaml_util.secret_values_registered(registered) as skeleton_keys:
+        _check_sensitive_usage(
+            sensitive, sensitive_paths, trees, scopes, prefixes, removed_at
+        )
+        texts = {rel: yaml_util.dump(tree) for rel, tree in trees.items()}
+
+    # An inline value whose name was never emitted was not found as a whole scalar.
+    leaked = [
+        info.config_path
+        for info in sensitive.values()
+        if not info.from_secret and info.secret_name not in skeleton_keys
+    ]
+    if leaked:
+        raise EsphomeError(
+            f"store_yaml: could not redact the sensitive value(s) of {', '.join(leaked)}; "
+            "not found in any captured file (composed via substitutions, set with -s, "
+            f"or defined in a remote package: {', '.join(remote_packages) or 'none'}). "
+            + _FIX
+        )
+
+    skeleton = _build_secrets_skeleton(skeleton_keys)
+    result = [
+        (rel, skeleton if rel in secret_rels else texts[rel].encode("utf-8"))
+        for rel, _ in entries
+    ]
+    if skeleton_keys and yaml_util.SECRET_YAML not in secret_rels:
+        # `!secret` looks for secrets.yaml beside the config; ship a skeleton there.
+        result.append((yaml_util.SECRET_YAML, skeleton))
+    return result
+
+
+def _pack_envelope(files: list[tuple[str, bytes]]) -> bytes:
+    """EHY1 | u32 count | {u16 path_len, path, u32 content_len, content}..., little-endian."""
+    paths = [path for path, _ in files]
+    if len(set(paths)) != len(paths):
+        raise EsphomeError(f"store_yaml: duplicate envelope path in {paths}")
+    return (
+        ENVELOPE_MAGIC
+        + struct.pack("<I", len(files))
+        + b"".join(
+            struct.pack("<H", len(encoded := path.encode("utf-8")))
+            + encoded
+            + struct.pack("<I", len(content))
+            + content
+            for path, content in files
+        )
+    )
+
+
+def unpack_envelope(blob: bytes) -> dict[str, bytes]:
+    """Reference decoder; rejects absolute paths, allows `..`, so disk writers confine paths."""
+    if blob[:4] != ENVELOPE_MAGIC:
+        raise EsphomeError("envelope must start with EHY1 magic")
+    files: dict[str, bytes] = {}
+    try:
+        (count,) = struct.unpack_from("<I", blob, 4)
+        pos = 8
+        for _ in range(count):
+            (path_len,) = struct.unpack_from("<H", blob, pos)
+            path = blob[pos + 2 : pos + 2 + path_len].decode("utf-8")
+            (size,) = struct.unpack_from("<I", blob, pos + 2 + path_len)
+            pos += 6 + path_len
+            if path in files or path.startswith(("/", "\\")) or path[1:2] == ":":
+                raise EsphomeError(f"envelope path rejected: {path}")
+            files[path], pos = blob[pos : pos + size], pos + size
+    except (struct.error, UnicodeDecodeError) as err:
+        raise EsphomeError(f"corrupt envelope: {err}") from err
+    if pos != len(blob):
+        raise EsphomeError("corrupt envelope: size mismatch")
+    return files
+
+
+async def to_code(config: ConfigType) -> None:
+    cg.add_define("USE_STORE_YAML")
+
+    # A fresh parse captures only user files and refills the secret registry.
+    discovered = yaml_util.discover_user_yaml_files(CORE.config_path)
+    entries, secret_rels = _gather_files(discovered)
+    remote_packages = _remote_package_descriptions()
+    if config[CONF_INCLUDE_SECRETS]:
+        files = _read_files_verbatim(entries)
+    else:
+        files = _generate_redacted_files(entries, secret_rels, remote_packages)
+    if remote_packages:
+        _LOGGER.warning(
+            "store_yaml: remote packages are not embedded, only recorded: %s",
+            ", ".join(remote_packages),
+        )
+    note = _uncaptured_note(
+        discovered.unresolved, remote_packages, _symlink_descriptions(discovered)
+    )
+    if note is not None:
+        files.append(note)
+    envelope = _pack_envelope(files)
+    compressed = zstd_module().compress(envelope, level=ZSTD_LEVEL)
+
+    _LOGGER.info(
+        "store_yaml: embedding %d file(s) as %d bytes (%d uncompressed, %.1f%% ratio)",
+        len(files),
+        len(compressed),
+        len(envelope),
+        100.0 * len(compressed) / len(envelope),
+    )
+
+    cg.extern_progmem_array("esphome::api::STORE_YAML_DATA", cg.uint8, list(compressed))
+    cg.add_define("STORE_YAML_DATA_SIZE", len(compressed))

esphome/core/defines.h

@@ -377,8 +377,10 @@
 #define USE_API_USER_DEFINED_ACTION_OPTIONAL_ARGS
 #define USE_API_USER_DEFINED_ACTION_RESPONSES
 #define USE_API_USER_DEFINED_ACTION_RESPONSES_JSON
+#define USE_STORE_YAML
 #define API_MAX_SEND_QUEUE 8
 #define API_WIZARD_DATA_SIZE 1
+#define STORE_YAML_DATA_SIZE 1
 #define API_WIZARD_INPUT_COUNT 1
 #define MAX_API_CONNECTIONS 6
 // The Improv library is not in the Zephyr tidy environment

script/analyze_component_buses.py

@@ -101,6 +101,7 @@ ISOLATED_COMPONENTS = {
     "modem": "Cellular-only node with api - grouping with wifi-bearing components hides that case",
     "neopixelbus": "RMT type conflict with ESP32 Arduino/ESP-IDF headers (enum vs struct rmt_channel_t)",
     "packages": "cannot merge packages",
+    "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",
     "tinyusb_keyboard": "Depends on tinyusb which conflicts with usb_host",
     "uart_mux": "Depends on tinyusb which conflicts with usb_host",

tests/components/store_yaml/common.yaml

@@ -0,0 +1,4 @@
+api:
+
+store_yaml:
+  allow_unencrypted: true

tests/components/store_yaml/test.bk72xx-ard.yaml

@@ -0,0 +1,6 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+packages:
+  store_yaml: !include common.yaml

tests/components/store_yaml/test.esp32-idf.yaml

@@ -0,0 +1,6 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+packages:
+  store_yaml: !include common.yaml

tests/components/store_yaml/test.esp8266-ard.yaml

@@ -0,0 +1,6 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+packages:
+  store_yaml: !include common.yaml

tests/components/store_yaml/test.host.yaml

@@ -0,0 +1,4 @@
+network:
+
+packages:
+  store_yaml: !include common.yaml

tests/components/store_yaml/test.ln882x-ard.yaml

@@ -0,0 +1,6 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+packages:
+  store_yaml: !include common.yaml

tests/components/store_yaml/test.rp2040-ard.yaml

@@ -0,0 +1,6 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+packages:
+  store_yaml: !include common.yaml

tests/components/store_yaml/test.rtl87xx-ard.yaml

@@ -0,0 +1,6 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+packages:
+  store_yaml: !include common.yaml

tests/components/store_yaml/validate.esp32-idf.yaml

@@ -0,0 +1,9 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+api:
+  encryption:
+    key: AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=
+
+store_yaml:

tests/integration/fixtures/store_yaml_recovery.yaml

@@ -0,0 +1,18 @@
+esphome:
+  name: store-yaml-test
+  areas:
+    - id: living_room
+      name: "Living Room"
+
+host:
+
+logger:
+
+api:
+
+ota:
+  - platform: esphome
+    password: recoverme123
+
+store_yaml:
+  allow_unencrypted: true

tests/integration/fixtures/store_yaml_recovery_backpressure.yaml

@@ -0,0 +1,24 @@
+esphome:
+  name: store-yaml-backpressure
+
+host:
+
+logger:
+
+api:
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+    components: [sndbuf_pin_component]
+
+# Pins the device's socket send buffers for deterministic TCP backpressure
+sndbuf_pin_component:
+  buffer_size: SERVER_SNDBUF
+
+store_yaml:
+  allow_unencrypted: true
+
+substitutions:
+# GENERATED_SUBSTITUTIONS

tests/integration/raw_api_client.py

@@ -1,7 +1,8 @@
 """Minimal plaintext native-api client over a raw socket.
 
 Reads only when told to, so tests control when the TCP pipe backs up toward
-the device; message types are counted and each frame is kept for inspection.
+the device; message types are counted and each frame is kept for inspection,
+and read_frame() hands them out in arrival order.
 """
 
 from __future__ import annotations
@@ -137,6 +138,7 @@ class RawApiClient:
     def __init__(self, port: int, recv_buffer_size: int | None = None) -> None:
         self._port = port
         self._parser = FrameParser()
+        self._read_pos = 0  # frames before this index were consumed by read_frame()
         self.bytes_received = 0
         self.frame_counts: Counter[int] = Counter()
         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@@ -179,21 +181,38 @@ class RawApiClient:
         loop = asyncio.get_running_loop()
         await loop.sock_sendall(self._sock, encode_frame(msg_type, payload))
 
+    async def _recv_once(self) -> None:
+        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))
+
     async def read_until_frame(
         self, msg_type: int, timeout: float = 10.0, count: int = 1
     ) -> None:
         """Read until at least count frames of msg_type have been received."""
-        loop = asyncio.get_running_loop()
 
         async def _read_loop() -> None:
             while self.frame_counts[msg_type] < count:
-                data = await 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))
+                await self._recv_once()
 
         await asyncio.wait_for(_read_loop(), timeout)
 
+    async def read_frame(self, msg_type: int, timeout: float = 10.0) -> bytes:
+        """Return the oldest unconsumed payload of msg_type, skipping earlier frames of other types."""
+        frames = self._parser.frames
+
+        async def _read_loop() -> bytes:
+            while True:
+                while self._read_pos < len(frames):
+                    frame_type, payload = frames[self._read_pos]
+                    self._read_pos += 1
+                    if frame_type == msg_type:
+                        return payload
+                await self._recv_once()
+
+        return await asyncio.wait_for(_read_loop(), timeout)
+
     def frames(self, msg_type: int) -> list[bytes]:
         """Payloads of every frame of msg_type received so far."""
         return [payload for kind, payload in self._parser.frames if kind == msg_type]

tests/integration/test_store_yaml_recovery.py

@@ -0,0 +1,109 @@
+"""End-to-end test of `store_yaml` recovery over the native API.
+
+Uses the raw client so a test controls when it reads; aioesphomeapi has no get_yaml helper.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import random
+
+from aioesphomeapi import api_pb2
+import pytest
+
+from esphome.components.store_yaml import unpack_envelope
+from esphome.helpers import zstd_module
+from esphome.yaml_util import find_secret_references
+
+from .log_utils import LineWaiter
+from .raw_api_client import MESSAGE_TYPE_OF, RawApiClient
+from .types import RunCompiledFunction
+
+GET_YAML_RESPONSE = MESSAGE_TYPE_OF[api_pb2.GetYamlResponse]
+
+
+async def _read_yaml_stream(client: RawApiClient) -> list[api_pb2.GetYamlResponse]:
+    """Read GetYamlResponse chunks up to and including the one marked done."""
+    responses: list[api_pb2.GetYamlResponse] = []
+    while not responses or not responses[-1].done:
+        payload = await client.read_frame(GET_YAML_RESPONSE)
+        responses.append(api_pb2.GetYamlResponse.FromString(payload))
+    return responses
+
+
+@pytest.mark.asyncio
+async def test_store_yaml_recovery(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    unused_tcp_port: int,
+) -> None:
+    """Stream the embedded YAML back from a host build and check the recovered files."""
+    async with run_compiled(yaml_config), RawApiClient(unused_tcp_port) as client:
+        await client.connect("store_yaml integration test")
+
+        # Clients learn the device can answer get_yaml from its capabilities.
+        await client.send_message(api_pb2.DeviceCapabilitiesRequest())
+        capabilities = api_pb2.DeviceCapabilitiesResponse.FromString(
+            await client.read_frame(MESSAGE_TYPE_OF[api_pb2.DeviceCapabilitiesResponse])
+        )
+        assert capabilities.store_yaml.supported
+
+        await client.send_message(api_pb2.GetYamlRequest())
+        responses = await _read_yaml_stream(client)
+
+    compressed = b"".join(r.data for r in responses)
+    assert responses[0].total_size == len(compressed)
+
+    envelope = zstd_module().decompress(compressed)
+    files = unpack_envelope(envelope)
+    combined = b"\n".join(files.values())
+    assert b"store-yaml-test" in combined
+    assert b"store_yaml:" in combined
+    # The inline OTA password comes back as a !secret listed in the skeleton.
+    assert b"recoverme123" not in envelope
+    assert "ota_password" in find_secret_references(combined.decode())
+    assert b"ota_password: ''" in files["secrets.yaml"]
+
+
+# Random hex barely compresses, so the blob spans many chunks and outgrows both buffers
+SERVER_SNDBUF = 8192
+RECV_BUFFER_SIZE = 4096
+CHUNK_LIMIT = 1390
+
+
+@pytest.mark.asyncio
+async def test_store_yaml_recovery_backpressure(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    unused_tcp_port: int,
+) -> None:
+    """A client that stops reading mid-stream still recovers every byte once it drains."""
+    rng = random.Random(16445)
+    values = {f"blob_{i}": rng.randbytes(60).hex() for i in range(1000)}
+    config = yaml_config.replace("SERVER_SNDBUF", str(SERVER_SNDBUF)).replace(
+        "# GENERATED_SUBSTITUTIONS",
+        "\n".join(f"  {name}: {value}" for name, value in values.items()),
+    )
+    log = LineWaiter()
+    async with (
+        run_compiled(config, line_callback=log.callback),
+        RawApiClient(unused_tcp_port, recv_buffer_size=RECV_BUFFER_SIZE) as client,
+    ):
+        # Without the pin the kernel buffers could swallow the stall
+        await log.wait_for("SO_SNDBUF pinned to", timeout=10.0)
+        await client.connect("store_yaml backpressure test")
+        await client.send_message(api_pb2.GetYamlRequest())
+        # Stop reading so the device fills both buffers and has to resume later
+        await asyncio.sleep(0.5)
+        responses = await _read_yaml_stream(client)
+
+    chunks = [r.data for r in responses]
+    compressed = b"".join(chunks)
+    assert len(compressed) > 4 * (SERVER_SNDBUF + RECV_BUFFER_SIZE)
+    assert len(responses) > 1 and all(len(c) <= CHUNK_LIMIT for c in chunks)
+    assert not any(r.done for r in responses[:-1])
+    assert responses[0].total_size == len(compressed)
+    combined = b"\n".join(
+        unpack_envelope(zstd_module().decompress(compressed)).values()
+    )
+    assert all(value.encode() in combined for value in values.values())