Offer with a provisioned api key and let the CLI use the api key with a plaintext fallback until 2027.3.0

THREAT_MODEL.md

@@ -126,16 +126,20 @@ design is optimal or that it will not change.
 
 The `esphome` OTA platform optionally encrypts updates with the same Noise
 `NNpsk0` pattern the native API uses; one key protects the device. A device
-whose `api:` block has a static encryption key compiles in the transport and
-offers it on every OTA connection, so an uploader presenting the key gets the
-guarantees below even without an `ota: encryption:` block; only that block
-makes the device require encryption. The guarantees are: the firmware image is
-confidential in transit, the uploader is authenticated by the pre-shared key,
-and the plaintext negotiation preceding the handshake is bound into the
-handshake prologue, so stripping or tampering with it fails the first MAC.
-With `ota: encryption:` configured both ends fail closed with no override: the
-device refuses plaintext uploads, and the CLI refuses to send plaintext when a
-key is configured.
+whose `api:` block has an encryption key, static in the YAML or provisioned at
+runtime, compiles in the transport and offers it on every OTA connection once
+it holds a key, so an uploader presenting that key gets the guarantees below
+even without an `ota: encryption:` block; only that block makes the device
+require encryption. The guarantees are: the firmware image is confidential in
+transit, the uploader is authenticated by the pre-shared key, and the plaintext
+negotiation preceding the handshake is bound into the handshake prologue, so
+stripping or tampering with it fails the first MAC. With `ota: encryption:`
+configured both ends fail closed with no override: the device refuses
+plaintext uploads, and the CLI refuses to send plaintext when a key is
+configured. Without that block the CLI tries a static api key when the device
+offers and, until 2027.3.0, falls back to plaintext with a warning when the
+offer is missing or the handshake fails; a runtime provisioned key never
+reaches the CLI, so those uploads stay plaintext.
 
 Defeating any of that without the key is in scope: a device that requires
 encryption accepting a plaintext or downgraded upload, getting past the MAC,
@@ -144,9 +148,13 @@ or recovering image contents from captured traffic.
 The following are **not** vulnerabilities, by design:
 
 - Plaintext OTA on a device with no `ota: encryption:` block, including one
-  that offers encryption because it has a static api key. That is the
-  documented default, authenticated (if at all) by the OTA password. An
-  uploader that takes the offer skips the password; the key authenticates it.
+  that offers encryption because it has an api key. That is the documented
+  default, authenticated (if at all) by the OTA password. An uploader that
+  takes the offer skips the password; the key authenticates it.
+- The CLI plaintext fallback until 2027.3.0: without `ota: encryption:` an
+  active attacker who strips the offer or breaks the handshake can make a
+  keyed CLI upload plaintext, with the pre-existing plaintext exposure. A
+  device that requires encryption still refuses that upload.
 - The enablement window: firmware built with a static api key already offers
   encryption, so turning on `ota: encryption:` is itself an encrypted upload.
   Older firmware needs one last plaintext upload of an offering build, with

esphome/__main__.py

@@ -1325,6 +1325,19 @@ def _choose_ota_platform(config: ConfigType, requested: str | None) -> str:
     return CONF_WEB_SERVER
 
 
+def _static_api_encryption_key(config: ConfigType) -> str | None:
+    """The api encryption key when it is fixed in the YAML; None when there is
+    none, when it is provisioned at runtime, or when it is the reserved all
+    zeros key."""
+    from esphome.components.noise import is_reserved_key
+
+    api_conf = config.get(CONF_API) or {}
+    key = (api_conf.get(CONF_ENCRYPTION) or {}).get(CONF_KEY)
+    if not key or is_reserved_key(key):
+        return None
+    return str(key)
+
+
 def _upload_via_native_api(
     config: ConfigType, network_devices: list[str], args: ArgsProtocol
 ) -> tuple[int, str | None]:
@@ -1341,6 +1354,7 @@ def _upload_via_native_api(
     # Fail closed: an encryption block whose key did not resolve must never
     # fall back to a plaintext upload
     noise_psk = None
+    plaintext_fallback = False
     if (encryption_conf := ota_conf.get(CONF_ENCRYPTION)) is not None:
         noise_psk = encryption_conf.get(CONF_KEY)
         if not noise_psk:
@@ -1351,6 +1365,11 @@ def _upload_via_native_api(
         # Ensure the key is a string, as required by the underlying OTA implementation.
         # It arrives here as a SensitiveStr which aioesphomeapi rejects.
         noise_psk = str(noise_psk)
+    elif api_key := _static_api_encryption_key(config):
+        # Remove before 2027.3.0: without an ota block the api key is tried
+        # when the device offers encryption, falling back to plaintext
+        noise_psk = api_key
+        plaintext_fallback = True
 
     def check_partition_access(option_string: str) -> None:
         if not ota_conf.get("allow_partition_access"):
@@ -1382,7 +1401,13 @@ def _upload_via_native_api(
         _validate_bootloader_binary(binary)
 
     return espota2.run_ota(
-        network_devices, remote_port, password, binary, ota_type, noise_psk
+        network_devices,
+        remote_port,
+        password,
+        binary,
+        ota_type,
+        noise_psk,
+        plaintext_fallback=plaintext_fallback,
     )
 
 

esphome/components/esphome/ota/__init__.py

@@ -129,7 +129,9 @@ def ota_esphome_final_validate(config: ConfigType) -> None:
         _validate_no_password_with_encryption(ota_conf)
         if (encryption_conf := ota_conf.get(CONF_ENCRYPTION)) is not None:
             _resolve_encryption_key(encryption_conf, api_conf)
-        elif CONF_PASSWORD in ota_conf and _api_static_key(api_conf) is not None:
+        elif CONF_PASSWORD in ota_conf and (
+            _api_static_key(api_conf) is not None or _api_runtime_key(api_conf)
+        ):
             _LOGGER.warning(
                 "'%s' %s wastes significant flash and RAM (about 3.5 KB and 60 "
                 "bytes plus the password on the heap): the device already offers "
@@ -187,6 +189,12 @@ def _warn_web_server_ota() -> None:
     )
 
 
+def _api_runtime_key(api_conf: ConfigType) -> bool:
+    """True when the api key is provisioned at runtime: an encryption block
+    with no key at all."""
+    return CONF_ENCRYPTION in api_conf and not api_conf[CONF_ENCRYPTION].get(CONF_KEY)
+
+
 def _api_static_key(api_conf: ConfigType) -> str | None:
     """The api key when fixed at build time; None for a runtime provisioned
     or all-zeros key, neither can seed the encryption offer."""
@@ -313,18 +321,26 @@ async def to_code(config: ConfigType) -> None:
     if config.get(CONF_ALLOW_PARTITION_ACCESS):
         cg.add_define("USE_OTA_PARTITIONS")
 
+    api_conf = CORE.config.get(CONF_API) or {}
+    from_api = False
     if (encryption_conf := config.get(CONF_ENCRYPTION)) is not None:
         # A missing key was resolved from the api component in final validate.
         key = encryption_conf[CONF_KEY]
     else:
         # An api key alone makes the device offer encryption while still
-        # accepting plaintext, so the upload that adds the block is encrypted
-        key = _api_static_key(CORE.config.get(CONF_API) or {})
-    if key is not None:
+        # accepting plaintext, so the upload that adds the block is encrypted.
+        # A key provisioned at runtime lives in the api server; the offer then
+        # uses whatever key it holds, so it follows provisioning and rotation.
+        key = _api_static_key(api_conf)
+        from_api = key is None and _api_runtime_key(api_conf)
+    if key is not None or from_api:
         cg.add_define("USE_OTA_ENCRYPTION")
         if encryption_conf is not None:
             cg.add_define("USE_OTA_ENCRYPTION_REQUIRED")
-        cg.add(var.set_noise_psk(new_psk_progmem(config[CONF_ID], key)))
+        if from_api:
+            cg.add_define("USE_OTA_ENCRYPTION_FROM_API")
+        else:
+            cg.add(var.set_noise_psk(new_psk_progmem(config[CONF_ID], key)))
 
     # Build flag so lwip_fast_select.c (a .c file that can't include defines.h) sees it.
     cg.add_build_flag("-DUSE_OTA_PLATFORM_ESPHOME")

esphome/components/esphome/ota/ota_esphome.cpp

@@ -100,11 +100,16 @@ void ESPHomeOTAComponent::dump_config() {
                 "  Version: %d"
 #ifdef USE_OTA_ENCRYPTION_REQUIRED
                 "\n  Encryption: required"
-#elif defined(USE_OTA_ENCRYPTION)
+#elif defined(USE_OTA_ENCRYPTION) && !defined(USE_OTA_ENCRYPTION_FROM_API)
                 "\n  Encryption: offered, plaintext accepted"
 #endif
                 ,
                 network::get_use_address_to(addr_buf), this->port_, USE_OTA_VERSION);
+#ifdef USE_OTA_ENCRYPTION_FROM_API
+  ESP_LOGCONFIG(TAG, "  Encryption: offered %s, plaintext accepted",
+                this->noise_context_().has_psk() ? LOG_STR_LITERAL("with the api key")
+                                                 : LOG_STR_LITERAL("once the api key is provisioned"));
+#endif
 #ifdef USE_OTA_PASSWORD
   if (!this->password_.empty()) {
     ESP_LOGCONFIG(TAG, "  Password configured");
@@ -264,7 +269,10 @@ void ESPHomeOTAComponent::handle_handshake_() {
         this->handshake_buf_[1] |= SERVER_FEATURE_SUPPORTS_PARTITION_ACCESS;
 #endif
 #ifdef USE_OTA_ENCRYPTION
-        this->handshake_buf_[1] |= SERVER_FEATURE_SUPPORTS_NOISE;
+        // A runtime provisioned key may not exist yet; offer only with a key
+        if (this->noise_context_().has_psk()) {
+          this->handshake_buf_[1] |= SERVER_FEATURE_SUPPORTS_NOISE;
+        }
 #endif
       } else {
         this->handshake_buf_[0] =
@@ -284,7 +292,7 @@ void ESPHomeOTAComponent::handle_handshake_() {
       // The client took the encryption offer: the rest of the session runs
       // inside the noise transport, which also authenticates it. Nothing to
       // do until its first handshake frame arrives.
-      if ((this->ota_features_ & CLIENT_NOISE_FEATURES) == CLIENT_NOISE_FEATURES) {
+      if (this->noise_context_().has_psk() && (this->ota_features_ & CLIENT_NOISE_FEATURES) == CLIENT_NOISE_FEATURES) {
         // handshake_buf_ still holds the feature ack composed above; a
         // would-block re-entry lands here without rebuilding it
         if (!this->noise_start_session_(this->handshake_buf_[1])) {

esphome/components/esphome/ota/ota_esphome.h

@@ -7,6 +7,9 @@
 #ifdef USE_OTA_ENCRYPTION
 #include "esphome/components/noise/noise_handshake.h"
 #endif
+#ifdef USE_OTA_ENCRYPTION_FROM_API
+#include "esphome/components/api/api_server.h"
+#endif
 #include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
 #include "esphome/core/preferences.h"
@@ -44,7 +47,7 @@ class ESPHomeOTAComponent final : public ota::OTAComponent {
   }
 #endif  // USE_OTA_PASSWORD
 
-#ifdef USE_OTA_ENCRYPTION
+#if defined(USE_OTA_ENCRYPTION) && !defined(USE_OTA_ENCRYPTION_FROM_API)
   /// psk points at 32 bytes that live in flash for the life of the program
   void set_noise_psk(const uint8_t *psk) { this->noise_ctx_.set_psk(psk); }
 #endif
@@ -86,6 +89,15 @@ class ESPHomeOTAComponent final : public ota::OTAComponent {
     bool writing{false};    // a produced handshake frame is still being flushed
     uint8_t frame_buf[noise::FRAME_HEADER_SIZE + 1 + noise::MAX_HANDSHAKE_SIZE];
   };
+  // The key the offer and the handshake use: the api server's live context
+  // for a runtime provisioned key, otherwise the component's own
+  inline const noise::NoiseContext &noise_context_() const {
+#ifdef USE_OTA_ENCRYPTION_FROM_API
+    return api::global_api_server->get_noise_ctx();
+#else
+    return this->noise_ctx_;
+#endif
+  }
   bool noise_start_session_(uint8_t server_feature_flags);
   bool handle_noise_handshake_();
   bool noise_try_read_frame_();
@@ -146,7 +158,9 @@ class ESPHomeOTAComponent final : public ota::OTAComponent {
   std::unique_ptr<uint8_t[]> auth_buf_;
 #endif  // USE_OTA_PASSWORD
 #ifdef USE_OTA_ENCRYPTION
+#ifndef USE_OTA_ENCRYPTION_FROM_API
   noise::NoiseContext noise_ctx_;
+#endif
   std::unique_ptr<NoiseSession> noise_;
 #endif  // USE_OTA_ENCRYPTION
 

esphome/components/esphome/ota/ota_esphome_noise.cpp

@@ -64,11 +64,12 @@ bool ESPHomeOTAComponent::noise_start_session_(uint8_t server_feature_flags) {
   *p++ = ota::OTA_RESPONSE_FEATURE_FLAGS;
   *p++ = server_feature_flags;
 
-  // Codegen always pairs USE_OTA_ENCRYPTION with a real key; never fall back
-  // to the all-zeros provisioning key here
-  int err = this->noise_ == nullptr       ? NOISE_ERROR_NO_MEMORY
-            : !this->noise_ctx_.has_psk() ? NOISE_ERROR_PSK_REQUIRED
-                                          : this->noise_->handshake.init(this->noise_ctx_, prologue, sizeof(prologue));
+  // Never run the handshake with the all-zeros provisioning key: a static key
+  // is always present, a runtime provisioned one may not be yet
+  const noise::NoiseContext &ctx = this->noise_context_();
+  int err = this->noise_ == nullptr ? NOISE_ERROR_NO_MEMORY
+            : !ctx.has_psk()        ? NOISE_ERROR_PSK_REQUIRED
+                                    : this->noise_->handshake.init(ctx, prologue, sizeof(prologue));
   if (err != 0) {
     ESP_LOGW(TAG, "Session init: %d", err);
     this->cleanup_connection_();

esphome/core/defines.h

@@ -244,6 +244,7 @@
 #define USE_RUNTIME_STATS
 #define USE_OTA
 #define USE_OTA_ENCRYPTION
+#define USE_OTA_ENCRYPTION_FROM_API
 #define USE_OTA_ENCRYPTION_REQUIRED
 #define USE_OTA_PASSWORD
 #define USE_OTA_VERSION 2

esphome/espota2.py

@@ -202,6 +202,11 @@ class OTANetworkError(OTAError):
     """Network-level OTA failure (timeout, reset, closed connection); retrying may succeed."""
 
 
+# Remove before 2027.3.0
+class OTAEncryptionFallback(OTAError):
+    """The encrypted attempt failed and the caller may retry in plaintext."""
+
+
 def _committed_error(err: OTANetworkError) -> OTAError:
     """Wrap a network failure that happened once the device had the full image.
 
@@ -464,6 +469,7 @@ def perform_ota(
     filename: Path,
     ota_type: int = OTA_TYPE_UPDATE_APP,
     noise_psk: str | None = None,
+    plaintext_fallback: bool = False,
 ) -> None:
     # Validate up front; an out-of-range value would only surface as a
     # ValueError deep inside send_check, bypassing OTAError handling
@@ -528,18 +534,24 @@ def perform_ota(
     else:
         features = 0
 
-    if not noise_psk and extended_proto and features & SERVER_FEATURE_SUPPORTS_NOISE:
-        _LOGGER.warning(
-            "The device offers OTA encryption but this upload is plaintext; "
-            "add 'encryption:' under 'ota: platform: esphome' to use it"
-        )
-
-    if noise_psk:
-        # Fail closed: never fall back to a plaintext upload when an
-        # encryption key is configured, an active attacker could otherwise
-        # strip the feature flag and capture the image (it contains the wifi
-        # credentials and the api encryption key).
-        if not (extended_proto and features & SERVER_FEATURE_SUPPORTS_NOISE):
+    if noise_psk and not (extended_proto and features & SERVER_FEATURE_SUPPORTS_NOISE):
+        if plaintext_fallback:
+            # Remove before 2027.3.0: the api key is tried opportunistically
+            # without an 'ota: encryption:' block, so an older firmware that
+            # cannot encrypt still gets its update
+            _LOGGER.warning(
+                "The device did not offer OTA encryption, so this upload "
+                "continues in plaintext. After this install a device with an "
+                "api encryption key offers encryption; add 'encryption:' under "
+                "'ota: platform: esphome' to require it. This plaintext "
+                "fallback is removed in 2027.3.0."
+            )
+            noise_psk = None
+        else:
+            # Fail closed: never fall back to a plaintext upload when an
+            # encryption key is configured, an active attacker could otherwise
+            # strip the feature flag and capture the image (it contains the wifi
+            # credentials and the api encryption key).
             raise OTAError(
                 "An OTA encryption key is configured but the device did not "
                 "offer encryption; refusing to send the image in plaintext. "
@@ -551,6 +563,7 @@ def perform_ota(
                 "again. Otherwise flash by serial or the web_server OTA "
                 "platform."
             )
+    if noise_psk:
         # The prologue binds every negotiation byte both sides saw, so any
         # tampering with the plaintext preamble breaks the handshake.
         prologue = (
@@ -559,8 +572,14 @@ def perform_ota(
             + bytes([RESPONSE_OK, version, features_to_send])
             + bytes([RESPONSE_FEATURE_FLAGS, features])
         )
-        sock = NoiseSocketWrapper(sock, noise_psk, prologue)
-        sock.do_handshake()
+        try:
+            sock = NoiseSocketWrapper(sock, noise_psk, prologue)
+            sock.do_handshake()
+        except OTAError as err:
+            # Remove before 2027.3.0
+            if plaintext_fallback:
+                raise OTAEncryptionFallback(str(err)) from err
+            raise
         _LOGGER.info("Encrypted connection established")
 
     if ota_type != OTA_TYPE_UPDATE_APP:
@@ -767,6 +786,7 @@ def run_ota_impl_(
     filename: Path,
     ota_type: int = OTA_TYPE_UPDATE_APP,
     noise_psk: str | None = None,
+    plaintext_fallback: bool = False,
 ) -> tuple[int, str | None]:
     from esphome.core import CORE
 
@@ -805,8 +825,10 @@ def run_ota_impl_(
     total_attempts = len(res) + EXTRA_UPLOAD_ATTEMPTS
     last_error = ""
     reached_device = False
-    for attempt in range(total_attempts):
-        af, socktype, _, _, sa = res[attempt % len(res)]
+    attempt = 0
+    addr_index = 0
+    while attempt < total_attempts:
+        af, socktype, _, _, sa = res[addr_index % len(res)]
         if reached_device or attempt >= len(res):
             _LOGGER.info(
                 "Retrying in %.0f seconds (attempt %d of %d)...",
@@ -825,17 +847,43 @@ def run_ota_impl_(
             sock.close()
             _LOGGER.warning("Connecting to %s port %s failed: %s", sa[0], sa[1], err)
             last_error = f"connecting to {sa[0]} failed: {err}"
+            attempt += 1
+            addr_index += 1
             continue
 
         _LOGGER.info("Connected to %s", sa[0])
         reached_device = True
         with contextlib.closing(sock), Path(filename).open("rb") as file_handle:
             try:
-                perform_ota(sock, password, file_handle, filename, ota_type, noise_psk)
+                perform_ota(
+                    sock,
+                    password,
+                    file_handle,
+                    filename,
+                    ota_type,
+                    noise_psk,
+                    plaintext_fallback,
+                )
+            except OTAEncryptionFallback as err:
+                # Remove before 2027.3.0: retry this address in plaintext
+                # without spending one of the network retries
+                _LOGGER.warning(
+                    "%s. The upload is retried in plaintext; a device that "
+                    "requires encryption will refuse it. This plaintext "
+                    "fallback is removed in 2027.3.0.",
+                    err,
+                )
+                noise_psk = None
+                plaintext_fallback = False
+                total_attempts += 1
+                attempt += 1
+                continue
             except OTANetworkError as err:
                 # Transient network failure; retry
                 last_error = str(err)
                 _LOGGER.warning("%s", last_error)
+                attempt += 1
+                addr_index += 1
                 continue
             except OTAError as err:
                 # Device-reported error (wrong password, wrong flash size, ...);
@@ -857,10 +905,17 @@ def run_ota(
     filename: Path,
     ota_type: int = OTA_TYPE_UPDATE_APP,
     noise_psk: str | None = None,
+    plaintext_fallback: bool = False,
 ) -> tuple[int, str | None]:
     try:
         return run_ota_impl_(
-            remote_host, remote_port, password, filename, ota_type, noise_psk
+            remote_host,
+            remote_port,
+            password,
+            filename,
+            ota_type,
+            noise_psk,
+            plaintext_fallback,
         )
     except OTAError as err:
         _LOGGER.error(err)

tests/component_tests/ota/test_esphome_ota.py

@@ -12,6 +12,7 @@ from esphome import config_validation as cv
 from esphome.components.esphome.ota import (
     AUTO_LOAD,
     FILTER_SOURCE_FILES,
+    _api_runtime_key,
     _api_static_key,
     _validate_no_password_with_encryption,
     ota_esphome_final_validate,
@@ -349,11 +350,19 @@ def test_encryption_with_captive_portal_does_not_warn(
         fv.full_config.reset(token)
 
 
-def test_password_with_api_key_warns(caplog: pytest.LogCaptureFixture) -> None:
-    """An api key makes the device offer encryption, which authenticates an
-    uploader without the password; the config validates with a warning."""
+@pytest.mark.parametrize(
+    "api_conf",
+    [{CONF_ENCRYPTION: {CONF_KEY: API_KEY}}, {CONF_ENCRYPTION: {}}],
+    ids=["static_key", "runtime_key"],
+)
+def test_password_with_api_key_warns(
+    caplog: pytest.LogCaptureFixture, api_conf: dict[str, Any]
+) -> None:
+    """An api key, static or provisioned, makes the device offer encryption,
+    which authenticates an uploader without the password; the config
+    validates with a warning."""
     full_conf = {
-        CONF_API: {CONF_ENCRYPTION: {CONF_KEY: API_KEY}},
+        CONF_API: api_conf,
         CONF_OTA: [_make_ota_config(port=3232, **{CONF_PASSWORD: "pw"})],
     }
     token = fv.full_config.set(full_conf)
@@ -367,8 +376,8 @@ def test_password_with_api_key_warns(caplog: pytest.LogCaptureFixture) -> None:
 
 @pytest.mark.parametrize(
     "api_conf",
-    [{}, {CONF_ENCRYPTION: {}}, {CONF_ENCRYPTION: {CONF_KEY: ZEROS_KEY}}],
-    ids=["no_api", "runtime_key", "zeros_key"],
+    [{}, {CONF_ENCRYPTION: {CONF_KEY: ZEROS_KEY}}],
+    ids=["no_api", "zeros_key"],
 )
 def test_password_without_static_api_key_no_warning(
     caplog: pytest.LogCaptureFixture, api_conf: dict[str, Any]
@@ -454,6 +463,14 @@ def test_auto_load_pulls_noise_only_for_encryption() -> None:
     assert "noise" in AUTO_LOAD({})
 
 
+def test_api_runtime_key() -> None:
+    """Only an encryption block with no key at all is provisioned at runtime."""
+    assert _api_runtime_key({}) is False
+    assert _api_runtime_key({CONF_ENCRYPTION: {}}) is True
+    assert _api_runtime_key({CONF_ENCRYPTION: {CONF_KEY: ZEROS_KEY}}) is False
+    assert _api_runtime_key({CONF_ENCRYPTION: {CONF_KEY: API_KEY}}) is False
+
+
 def test_api_static_key() -> None:
     """Only a real build-time api key can seed the encryption offer."""
     assert _api_static_key({}) is None
@@ -466,21 +483,30 @@ def test_api_static_key() -> None:
     ("yaml_name", "defines_present", "defines_absent"),
     [
         # An api key alone compiles the transport in without requiring it
-        ("api_key_offer", {"USE_OTA_ENCRYPTION"}, {"USE_OTA_ENCRYPTION_REQUIRED"}),
+        (
+            "api_key_offer",
+            {"USE_OTA_ENCRYPTION"},
+            {"USE_OTA_ENCRYPTION_REQUIRED", "USE_OTA_ENCRYPTION_FROM_API"},
+        ),
         # A password still guards plaintext uploads on an offering device
         (
             "api_key_offer_password",
             {"USE_OTA_ENCRYPTION", "USE_OTA_PASSWORD"},
-            {"USE_OTA_ENCRYPTION_REQUIRED"},
+            {"USE_OTA_ENCRYPTION_REQUIRED", "USE_OTA_ENCRYPTION_FROM_API"},
         ),
         # The ota encryption block is what makes the device refuse plaintext
         (
             "encryption_required",
             {"USE_OTA_ENCRYPTION", "USE_OTA_ENCRYPTION_REQUIRED"},
-            set(),
+            {"USE_OTA_ENCRYPTION_FROM_API"},
+        ),
+        # A key provisioned at runtime lives in the api server; the device
+        # offers with it once provisioned and never requires it
+        (
+            "runtime_api_key",
+            {"USE_OTA_ENCRYPTION", "USE_OTA_ENCRYPTION_FROM_API"},
+            {"USE_OTA_ENCRYPTION_REQUIRED"},
         ),
-        # A key provisioned at runtime is unknown at build time, so no offer
-        ("runtime_api_key", set(), {"USE_OTA_ENCRYPTION"}),
     ],
 )
 def test_encryption_offer_codegen(
@@ -496,7 +522,8 @@ def test_encryption_offer_codegen(
     assert defines_present <= defines
     assert not (defines_absent & defines)
     encrypted = "USE_OTA_ENCRYPTION" in defines_present
-    assert ("esphome_esphomeotacomponent_id->set_noise_psk(" in main_cpp) is encrypted
+    own_key = encrypted and "USE_OTA_ENCRYPTION_FROM_API" not in defines_present
+    assert ("esphome_esphomeotacomponent_id->set_noise_psk(" in main_cpp) is own_key
     assert ("set_auth_password(" in main_cpp) is ("USE_OTA_PASSWORD" in defines_present)
     # The noise transport source compiles only when the define is set
     assert FILTER_SOURCE_FILES() == ([] if encrypted else ["ota_esphome_noise.cpp"])

tests/components/ota/api_runtime_key.yaml

@@ -0,0 +1,10 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+api:
+  encryption:
+
+ota:
+  - platform: esphome
+    port: 3291

tests/components/ota/test-api_runtime_key.esp32-idf.yaml

@@ -0,0 +1,2 @@
+packages:
+  ota: !include api_runtime_key.yaml

tests/components/ota/test-api_runtime_key.esp8266-ard.yaml

@@ -0,0 +1,2 @@
+packages:
+  ota: !include api_runtime_key.yaml

tests/integration/fixtures/host_ota_api_key_fallback.yaml

@@ -0,0 +1,12 @@
+esphome:
+  name: host-ota-test
+host:
+api:
+  encryption:
+    key: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+ota:
+  - platform: esphome
+    port: __OTA_PORT__
+    password: "hunter2"
+logger:
+  level: DEBUG

tests/integration/fixtures/host_ota_provisioned_api_key.yaml

@@ -0,0 +1,10 @@
+esphome:
+  name: host-ota-test
+host:
+api:
+  encryption:
+ota:
+  - platform: esphome
+    port: __OTA_PORT__
+logger:
+  level: DEBUG

tests/integration/test_host_ota.py

@@ -8,9 +8,11 @@ instance covers the FD_CLOEXEC path.
 from __future__ import annotations
 
 import asyncio
+import base64
 from collections.abc import Generator
 from contextlib import contextmanager
 import functools
+import logging
 from pathlib import Path
 import socket
 
@@ -20,7 +22,7 @@ from esphome import espota2
 
 from .conftest import run_binary, wait_and_connect_api_client
 from .const import LOCALHOST, PORT_POLL_INTERVAL, PORT_WAIT_TIMEOUT
-from .types import CompileFunction, ConfigWriter
+from .types import APIClientConnectedFactory, CompileFunction, ConfigWriter
 
 DEVICE_NAME = "host-ota-test"
 API_KEY = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
@@ -177,7 +179,11 @@ class _RebootCounter:
 
 
 async def _run_ota(
-    ota_port: int, password: str | None, binary_path: Path, noise_psk: str | None
+    ota_port: int,
+    password: str | None,
+    binary_path: Path,
+    noise_psk: str | None,
+    plaintext_fallback: bool = False,
 ) -> int:
     """espota2 is blocking; run it in the executor and return its exit code."""
     rc, _ = await asyncio.get_running_loop().run_in_executor(
@@ -189,6 +195,7 @@ async def _run_ota(
             password,
             binary_path,
             noise_psk=noise_psk,
+            plaintext_fallback=plaintext_fallback,
         ),
     )
     return rc
@@ -242,6 +249,127 @@ async def test_host_ota_api_key_offer_with_password(
             assert any("Encryption: offered" in line for line in lines)
 
 
+# The well-known provisioning PSK and a key to provision, as in the api
+# provisioning tests
+ZERO_PSK = base64.b64encode(bytes(32)).decode()
+PROVISIONED_KEY = base64.b64encode(b"p" * 32)
+KEY_ACTIVATION_DELAY = 0.5
+
+
+@pytest.mark.asyncio
+async def test_host_ota_provisioned_api_key(
+    yaml_config: str,
+    write_yaml_config: ConfigWriter,
+    compile_esphome: CompileFunction,
+    reserved_tcp_port: tuple[int, socket.socket],
+    api_client_connected: APIClientConnectedFactory,
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    """A key provisioned over the api feeds the OTA offer: plaintext works
+    while unprovisioned, the provisioned key encrypts, and the key loaded from
+    preferences on the next boot keeps encrypting."""
+    pytest.importorskip("aioesphomeapi.noise")
+    # Host preferences persist per device name; keep this run unprovisioned
+    monkeypatch.setenv("ESPHOME_PREFDIR", str(tmp_path / "prefs"))
+    api_port, api_socket = reserved_tcp_port
+    with _reserve_port() as (ota_port, ota_socket):
+        yaml_config = yaml_config.replace("__OTA_PORT__", str(ota_port))
+        config_path = await write_yaml_config(yaml_config)
+        binary_path = await compile_esphome(config_path)
+        api_socket.close()
+        ota_socket.close()
+
+        reboots = _RebootCounter()
+        async with run_binary(binary_path, line_callback=reboots.on_log) as (
+            proc,
+            lines,
+        ):
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            pid_before = proc.pid
+            assert any("once the api key is provisioned" in line for line in lines)
+
+            rc = await _run_ota(ota_port, None, binary_path, None)
+            assert rc == 0, "plaintext upload to an unprovisioned device must succeed"
+            await reboots.wait(1)
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            assert proc.pid == pid_before
+
+            async with api_client_connected(
+                port=api_port, noise_psk=ZERO_PSK
+            ) as client:
+                assert await client.noise_encryption_set_key(PROVISIONED_KEY) is True
+            await asyncio.sleep(KEY_ACTIVATION_DELAY)
+
+            rc = await _run_ota(ota_port, None, binary_path, PROVISIONED_KEY.decode())
+            assert rc == 0, "encrypted upload with the provisioned key must succeed"
+            await reboots.wait(2)
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            assert proc.pid == pid_before
+
+            # After the re-exec the key came from preferences at boot
+            rc = await _run_ota(ota_port, None, binary_path, PROVISIONED_KEY.decode())
+            assert rc == 0, "the key loaded at boot must feed the OTA offer"
+            await reboots.wait(3)
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            assert proc.pid == pid_before
+
+            # The offer never becomes a requirement without ota: encryption:
+            rc = await _run_ota(ota_port, None, binary_path, None)
+            assert rc == 0, "plaintext must stay accepted on an offering device"
+            await reboots.wait(4)
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            assert proc.pid == pid_before
+
+
+# Remove before 2027.3.0
+@pytest.mark.asyncio
+async def test_host_ota_api_key_fallback(
+    yaml_config: str,
+    write_yaml_config: ConfigWriter,
+    compile_esphome: CompileFunction,
+    reserved_tcp_port: tuple[int, socket.socket],
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Without an ota encryption block the api key is tried and a failed
+    handshake falls back to plaintext, which the password still guards."""
+    pytest.importorskip("aioesphomeapi.noise")
+    wrong_key = base64.b64encode(b"w" * 32).decode()
+    api_port, api_socket = reserved_tcp_port
+    with _reserve_port() as (ota_port, ota_socket):
+        yaml_config = yaml_config.replace("__OTA_PORT__", str(ota_port))
+        config_path = await write_yaml_config(yaml_config)
+        binary_path = await compile_esphome(config_path)
+        api_socket.close()
+        ota_socket.close()
+
+        reboots = _RebootCounter()
+        async with run_binary(binary_path, line_callback=reboots.on_log) as (
+            proc,
+            _lines,
+        ):
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            pid_before = proc.pid
+
+            with caplog.at_level(logging.WARNING, logger="esphome.espota2"):
+                rc = await _run_ota(
+                    ota_port, "hunter2", binary_path, wrong_key, plaintext_fallback=True
+                )
+            assert rc == 0, "the plaintext retry with the password must succeed"
+            assert any("retried in plaintext" in r.message for r in caplog.records)
+            await reboots.wait(1)
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            assert proc.pid == pid_before
+
+            rc = await _run_ota(
+                ota_port, None, binary_path, API_KEY, plaintext_fallback=True
+            )
+            assert rc == 0, "the right api key encrypts without touching the fallback"
+            await reboots.wait(2)
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            assert proc.pid == pid_before
+
+
 @pytest.mark.asyncio
 async def test_host_ota_rejects_garbage(
     yaml_config: str,

tests/unit_tests/test_espota2_noise.py

@@ -67,8 +67,10 @@ class FakeEncryptedDevice(threading.Thread):
         offer_noise: bool = True,
         require_noise: bool = True,
         prologue_features_override: int | None = None,
+        connections: int = 1,
     ) -> None:
         super().__init__(daemon=True)
+        self.connections = connections
         self.psk = psk
         self.version = version
         self.offer_noise = offer_noise
@@ -83,10 +85,11 @@ class FakeEncryptedDevice(threading.Thread):
 
     def run(self) -> None:
         try:
-            sock, _ = self.listener.accept()
-            sock.settimeout(10)
-            with sock:
-                self._serve(sock)
+            for _ in range(self.connections):
+                sock, _ = self.listener.accept()
+                sock.settimeout(10)
+                with sock:
+                    self._serve(sock)
         except Exception as err:  # noqa: BLE001 - surfaced via join_and_check
             self.error = err
         finally:
@@ -111,19 +114,23 @@ class FakeEncryptedDevice(threading.Thread):
             return
         server_flags = espota2.SERVER_FEATURE_SUPPORTS_NOISE if self.offer_noise else 0
         sock.sendall(bytes([espota2.RESPONSE_FEATURE_FLAGS, server_flags]))
-        if not noise_negotiated:
-            # A device that does not require encryption lets a plaintext
-            # client through
-            self._transfer(
-                lambda byte: sock.sendall(bytes([byte])),
-                lambda length: _recv_exact(sock, length),
-                lambda remaining: _recv_exact(
-                    sock, min(remaining, espota2.UPLOAD_BLOCK_SIZE)
-                ),
-            )
+        if not (noise_negotiated and self.offer_noise):
+            # A device that does not require encryption continues in
+            # plaintext whatever the client asked for, like older firmware
+            try:
+                self._transfer(
+                    lambda byte: sock.sendall(bytes([byte])),
+                    lambda length: _recv_exact(sock, length),
+                    lambda remaining: _recv_exact(
+                        sock, min(remaining, espota2.UPLOAD_BLOCK_SIZE)
+                    ),
+                )
+            except ConnectionError:
+                # A keyed client without fallback fails closed and hangs up
+                if noise_negotiated and not self.offer_noise:
+                    return
+                raise
             return
-        if not self.offer_noise:
-            return  # the client fails closed; nothing further arrives
 
         from cryptography.exceptions import InvalidTag
         from noise.connection import NoiseConnection
@@ -201,7 +208,10 @@ class FakeEncryptedDevice(threading.Thread):
 
 
 def _upload(
-    device: FakeEncryptedDevice, firmware: bytes, noise_psk: str | None
+    device: FakeEncryptedDevice,
+    firmware: bytes,
+    noise_psk: str | None,
+    plaintext_fallback: bool = False,
 ) -> None:
     device.start()
     sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@@ -209,12 +219,35 @@ def _upload(
     sock.connect(("127.0.0.1", device.port))
     try:
         espota2.perform_ota(
-            sock, None, io.BytesIO(firmware), Path("firmware.bin"), noise_psk=noise_psk
+            sock,
+            None,
+            io.BytesIO(firmware),
+            Path("firmware.bin"),
+            noise_psk=noise_psk,
+            plaintext_fallback=plaintext_fallback,
         )
     finally:
         sock.close()
 
 
+def _run_ota(
+    device: FakeEncryptedDevice, firmware: bytes, tmp_path: Path, noise_psk: str
+) -> int:
+    """Drive the retry loop, which is where the plaintext fallback reconnects."""
+    path = tmp_path / "firmware.bin"
+    path.write_bytes(firmware)
+    device.start()
+    rc, _ = espota2.run_ota(
+        "127.0.0.1",
+        device.port,
+        None,
+        path,
+        noise_psk=noise_psk,
+        plaintext_fallback=True,
+    )
+    return rc
+
+
 def test_encrypted_upload_success() -> None:
     """A full encrypted v2 upload spanning several 8192-byte blocks."""
     pytest.importorskip("aioesphomeapi.noise")
@@ -265,15 +298,47 @@ def test_client_fails_closed_when_device_lacks_encryption() -> None:
     device.join_and_check()
 
 
-def test_plaintext_upload_to_offering_device_warns(
-    caplog: pytest.LogCaptureFixture,
-) -> None:
-    """A keyless client is told the device could have encrypted the upload."""
-    device = FakeEncryptedDevice(offer_noise=True, require_noise=False)
+# Remove before 2027.3.0
+def test_fallback_when_device_does_not_offer(caplog: pytest.LogCaptureFixture) -> None:
+    """The api key is tried opportunistically; an older device that cannot
+    encrypt still gets its update, with a warning."""
+    firmware = b"firmware"
+    device = FakeEncryptedDevice(offer_noise=False, require_noise=False)
     with patch("time.sleep"), caplog.at_level(logging.WARNING):
-        _upload(device, b"firmware", None)
+        _upload(device, firmware, PSK, plaintext_fallback=True)
     device.join_and_check()
-    assert any("offers OTA encryption" in r.message for r in caplog.records)
+    assert device.received == firmware
+    assert any("fallback is removed in 2027.3.0" in r.message for r in caplog.records)
+
+
+# Remove before 2027.3.0
+def test_fallback_after_failed_handshake(
+    caplog: pytest.LogCaptureFixture, tmp_path: Path
+) -> None:
+    """A wrong key against an offering device reconnects in plaintext."""
+    pytest.importorskip("aioesphomeapi.noise")
+    firmware = b"firmware"
+    device = FakeEncryptedDevice(psk=OTHER_PSK, require_noise=False, connections=2)
+    with patch("time.sleep"), caplog.at_level(logging.WARNING):
+        rc = _run_ota(device, firmware, tmp_path, PSK)
+    device.join_and_check()
+    assert rc == 0
+    assert device.received == firmware
+    assert any("retried in plaintext" in r.message for r in caplog.records)
+
+
+# Remove before 2027.3.0
+def test_fallback_cannot_downgrade_a_requiring_device(
+    caplog: pytest.LogCaptureFixture, tmp_path: Path
+) -> None:
+    """The plaintext retry is refused by a device that requires encryption."""
+    pytest.importorskip("aioesphomeapi.noise")
+    device = FakeEncryptedDevice(psk=OTHER_PSK, require_noise=True, connections=2)
+    with patch("time.sleep"), caplog.at_level(logging.WARNING):
+        rc = _run_ota(device, b"firmware", tmp_path, PSK)
+    device.join_and_check()
+    assert rc == 1
+    assert any("requires an encrypted OTA" in r.message for r in caplog.records)
 
 
 @pytest.mark.parametrize("noise_psk", [None, PSK], ids=["plaintext", "encrypted"])

tests/unit_tests/test_main.py

@@ -2108,7 +2108,13 @@ def test_upload_program_ota_success(
         tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
     )
     mock_run_ota.assert_called_once_with(
-        ["192.168.1.100"], 3232, "secret", expected_firmware, OTA_TYPE_UPDATE_APP, None
+        ["192.168.1.100"],
+        3232,
+        "secret",
+        expected_firmware,
+        OTA_TYPE_UPDATE_APP,
+        None,
+        plaintext_fallback=False,
     )
 
 
@@ -2140,10 +2146,81 @@ def test_upload_program_ota_encryption_key(
         tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
     )
     mock_run_ota.assert_called_once_with(
-        ["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, key
+        ["192.168.1.100"],
+        3232,
+        None,
+        expected_firmware,
+        OTA_TYPE_UPDATE_APP,
+        key,
+        plaintext_fallback=False,
     )
 
 
+def test_upload_program_ota_api_key_opportunistic(
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """Without an ota encryption block the api key is tried with a plaintext
+    fallback (removed in 2027.3.0)."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+    mock_run_ota.return_value = (0, "192.168.1.100")
+
+    key = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+    config = {
+        CONF_API: {CONF_ENCRYPTION: {CONF_KEY: key}},
+        CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 3232}],
+    }
+    exit_code, _ = upload_program(config, MockArgs(), ["192.168.1.100"])
+
+    assert exit_code == 0
+    expected_firmware = (
+        tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
+    )
+    mock_run_ota.assert_called_once_with(
+        ["192.168.1.100"],
+        3232,
+        None,
+        expected_firmware,
+        OTA_TYPE_UPDATE_APP,
+        key,
+        plaintext_fallback=True,
+    )
+
+
+@pytest.mark.parametrize(
+    "api_conf",
+    [
+        {},
+        {CONF_ENCRYPTION: {}},
+        {CONF_ENCRYPTION: {CONF_KEY: "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="}},
+    ],
+    ids=["no_encryption", "runtime_key", "zeros_key"],
+)
+def test_upload_program_ota_no_usable_api_key_stays_plaintext(
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+    api_conf: dict[str, Any],
+) -> None:
+    """A missing, runtime provisioned, or all-zeros api key gives the
+    uploader nothing to try."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+    mock_run_ota.return_value = (0, "192.168.1.100")
+
+    config = {
+        CONF_API: api_conf,
+        CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 3232}],
+    }
+    exit_code, _ = upload_program(config, MockArgs(), ["192.168.1.100"])
+
+    assert exit_code == 0
+    assert mock_run_ota.call_args.args[5] is None
+    assert mock_run_ota.call_args.kwargs == {"plaintext_fallback": False}
+
+
 def test_upload_program_ota_encryption_without_key_fails_closed(
     mock_run_ota: Mock,
     mock_get_port_type: Mock,
@@ -2194,7 +2271,13 @@ def test_upload_program_ota_with_file_arg(
     assert exit_code == 0
     assert host == "192.168.1.100"
     mock_run_ota.assert_called_once_with(
-        ["192.168.1.100"], 3232, None, Path("custom.bin"), OTA_TYPE_UPDATE_APP, None
+        ["192.168.1.100"],
+        3232,
+        None,
+        Path("custom.bin"),
+        OTA_TYPE_UPDATE_APP,
+        None,
+        plaintext_fallback=False,
     )
 
 
@@ -2250,6 +2333,7 @@ def test_upload_program_ota_partition_table_with_file_arg(
         partition_file,
         OTA_TYPE_UPDATE_PARTITION_TABLE,
         None,
+        plaintext_fallback=False,
     )
 
 
@@ -2312,6 +2396,7 @@ def test_upload_program_ota_partition_table_mqttip(
         partition_file,
         OTA_TYPE_UPDATE_PARTITION_TABLE,
         None,
+        plaintext_fallback=False,
     )
 
 
@@ -2500,6 +2585,7 @@ def test_upload_program_ota_bootloader_with_file_arg(
         bootloader_file,
         OTA_TYPE_UPDATE_BOOTLOADER,
         None,
+        plaintext_fallback=False,
     )
 
 
@@ -2988,7 +3074,13 @@ def test_upload_program_ota_with_mqtt_resolution(
         tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
     )
     mock_run_ota.assert_called_once_with(
-        ["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, None
+        ["192.168.1.100"],
+        3232,
+        None,
+        expected_firmware,
+        OTA_TYPE_UPDATE_APP,
+        None,
+        plaintext_fallback=False,
     )
 
 
@@ -3038,7 +3130,13 @@ def test_upload_program_ota_with_mqtt_empty_broker(
         tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
     )
     mock_run_ota.assert_called_once_with(
-        ["192.168.1.50"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, None
+        ["192.168.1.50"],
+        3232,
+        None,
+        expected_firmware,
+        OTA_TYPE_UPDATE_APP,
+        None,
+        plaintext_fallback=False,
     )
     # Verify warning was logged
     assert "MQTT IP discovery failed" in caplog.text
@@ -5211,6 +5309,7 @@ def test_upload_program_ota_static_ip_with_mqttip(
         expected_firmware,
         OTA_TYPE_UPDATE_APP,
         None,
+        plaintext_fallback=False,
     )
 
 
@@ -5261,6 +5360,7 @@ def test_upload_program_ota_multiple_mqttip_resolves_once(
         expected_firmware,
         OTA_TYPE_UPDATE_APP,
         None,
+        plaintext_fallback=False,
     )
 
 
@@ -5438,7 +5538,13 @@ def test_upload_program_ota_mqtt_timeout_fallback(
         tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
     )
     mock_run_ota.assert_called_once_with(
-        ["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, None
+        ["192.168.1.100"],
+        3232,
+        None,
+        expected_firmware,
+        OTA_TYPE_UPDATE_APP,
+        None,
+        plaintext_fallback=False,
     )