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

Key the web_server warning off the OTA platform and tell plaintext uploaders about the offer
Only warn about the web_server update endpoint when web_server is configured
2026-09-05 12:36:07 +00:00 · 2026-09-05 14:22:27 +02:00 · 2026-09-05 13:50:46 +02:00 · 2026-09-05 13:24:19 +02:00 · 2026-09-05 12:56:06 +02:00 · 2026-09-05 12:12:33 +02:00
54 changed files with 1382 additions and 240 deletions
@@ -125,30 +125,45 @@ design is optimal or that it will not change.
 ## OTA update encryption

 The `esphome` OTA platform optionally encrypts updates with the same Noise
-`NNpsk0` pattern the native API uses; one key protects the device. With an
-`encryption:` block configured 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.
-Both ends fail closed with no override: a device built with a key refuses
+`NNpsk0` pattern the native API uses; one key protects the device. A device
+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.
+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 keyed device accepting a
-plaintext or downgraded upload, getting past the MAC, or recovering image
-contents from captured traffic.
+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,
+or recovering image contents from captured traffic.

 The following are **not** vulnerabilities, by design:

- Plaintext OTA on a device with no `encryption:` block. That is the
-  documented default, authenticated (if at all) by the OTA password.
- The enablement window: turning encryption on takes one last upload of the
-  encryption-enabled firmware over the existing plaintext channel, with the
-  pre-existing plaintext exposure.
- The web OTA `/update` endpoint alongside encryption. The `web_server`
-  component keeps it always reachable, and `captive_portal:` auto-loads it
-  for the fallback AP window; validation warns about both combinations, and
-  the operator keeps the recovery path.
+- Plaintext OTA on a device with no `ota: encryption:` block, including one
+  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
+  the pre-existing plaintext exposure.
+- The web OTA `/update` endpoint alongside encryption. The `web_server` OTA
+  platform keeps it always reachable and validation warns about that
+  combination; `captive_portal:` auto-loads that platform only for the
+  fallback AP window, which is the intended recovery path, so that alone is
+  not warned about.
 - CLI retry behavior on transport or MAC failures; every attempt renegotiates
  a fresh handshake with fresh ephemerals, so retrying does not weaken
  authentication.
@@ -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,
    )


@@ -14,6 +14,7 @@ from esphome.components.noise import (  # noqa: F401
    ENCRYPTION_SCHEMA,
    decode_encryption_key,
    encryption_schema,
+    new_psk_progmem,
    validate_encryption_key,
 )
 from esphome.config_helpers import filter_source_files_from_defines, get_logger_level
@@ -589,8 +590,7 @@ async def to_code(config: ConfigType) -> None:

    if (encryption_config := config.get(CONF_ENCRYPTION, None)) is not None:
        if key := encryption_config.get(CONF_KEY):
-            decoded = decode_encryption_key(key)
-            cg.add(var.set_noise_psk(list(decoded)))
+            cg.add(var.set_noise_psk(new_psk_progmem(config[CONF_ID], key)))
            cg.add_define("USE_API_NOISE_PSK_FROM_YAML")
        else:
            # No key provided, but encryption desired
@@ -548,7 +548,7 @@ APIError APINoiseFrameHelper::write_frame_(const uint8_t *data, uint16_t len) {
 * @return 0 on success, -1 on error (check errno)
 */
 APIError APINoiseFrameHelper::init_handshake_() {
-  int err = this->handshake_.init(this->ctx_.get_psk(), prologue_.data(), prologue_.size());
+  int err = this->handshake_.init(this->ctx_, prologue_.data(), prologue_.size());
  APIError aerr = handle_noise_error_(err, LOG_STR("noise_handshake_init"), APIError::HANDSHAKESTATE_SETUP_FAILED);
  if (aerr != APIError::OK)
    return aerr;
@@ -583,11 +583,17 @@ bool APIServer::update_noise_psk_(const SavedNoisePsk &new_psk, const LogString
 }

 bool APIServer::load_and_apply_noise_psk_() {
-  SavedNoisePsk saved{};
-  if (!this->noise_pref_.load(&saved))
+#ifdef USE_API_NOISE_PSK_FROM_YAML
+  return false;
+#else
+  // Load into a temp so a failed read cannot disturb the key in use
+  SavedNoisePsk loaded{};
+  if (!this->noise_pref_.load(&loaded))
    return false;
-  this->set_noise_psk(saved.psk);
+  this->saved_psk_ = loaded;
+  this->noise_ctx_.set_psk(this->saved_psk_.psk.data());
  return true;
+#endif
 }

 bool APIServer::save_noise_psk(noise::psk_t psk, bool make_active) {
@@ -597,8 +603,7 @@ bool APIServer::save_noise_psk(noise::psk_t psk, bool make_active) {
  ESP_LOGW(TAG, "Key set in YAML");
  return false;
 #else
-  auto &old_psk = this->noise_ctx_.get_psk();
-  if (std::equal(old_psk.begin(), old_psk.end(), psk.begin())) {
+  if (this->saved_psk_.psk == psk) {
    ESP_LOGW(TAG, "New PSK matches old");
    return true;
  }
@@ -78,7 +78,8 @@ class APIServer final : public Component,
 #ifdef USE_API_NOISE
  bool save_noise_psk(noise::psk_t psk, bool make_active = true);
  bool clear_noise_psk(bool make_active = true);
-  void set_noise_psk(noise::psk_t psk) { this->noise_ctx_.set_psk(psk); }
+  /// 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); }
  noise::NoiseContext &get_noise_ctx() { return this->noise_ctx_; }
 #endif  // USE_API_NOISE

@@ -358,6 +359,9 @@ class APIServer final : public Component,

 #ifdef USE_API_NOISE
  noise::NoiseContext noise_ctx_;
+#ifndef USE_API_NOISE_PSK_FROM_YAML
+  SavedNoisePsk saved_psk_{};  // backs noise_ctx_ for a runtime provisioned key
+#endif
  ESPPreferenceObject noise_pref_;
 #endif  // USE_API_NOISE
 };
@@ -100,21 +100,38 @@ void ESP32BLE::disable() {
 #ifdef USE_ESP32_BLE_ADVERTISING
 void ESP32BLE::advertising_start() {
  this->advertising_init_();
-  if (!this->is_active())
+  this->advertising_ref_count_++;
+  this->advertising_refresh();
+}
+
+void ESP32BLE::advertising_stop() {
+  if (this->advertising_ref_count_ == 0)
    return;
-  this->advertising_->start();
+  this->advertising_ref_count_--;
+  this->advertising_refresh();
+}
+
+void ESP32BLE::advertising_refresh() {
+  if (this->advertising_ == nullptr || !this->is_active())
+    return;
+  // Advertise while any component still needs it, otherwise stop
+  if (this->advertising_ref_count_ == 0) {
+    this->advertising_->stop();
+  } else {
+    this->advertising_->start();
+  }
 }

 void ESP32BLE::advertising_set_service_data(const std::vector<uint8_t> &data) {
  this->advertising_init_();
  this->advertising_->set_service_data(data);
-  this->advertising_start();
+  this->advertising_refresh();
 }

 void ESP32BLE::advertising_set_manufacturer_data(const std::vector<uint8_t> &data) {
  this->advertising_init_();
  this->advertising_->set_manufacturer_data(data);
-  this->advertising_start();
+  this->advertising_refresh();
 }

 void ESP32BLE::advertising_set_service_data_and_name(std::span<const uint8_t> data, bool include_name) {
@@ -136,7 +153,7 @@ void ESP32BLE::advertising_set_service_data_and_name(std::span<const uint8_t> da
    this->advertising_->set_service_data(data);
  }

-  this->advertising_start();
+  this->advertising_refresh();
 }

 void ESP32BLE::advertising_register_raw_advertisement_callback(std::function<void(bool)> &&callback) {
@@ -147,13 +164,13 @@ void ESP32BLE::advertising_register_raw_advertisement_callback(std::function<voi
 void ESP32BLE::advertising_add_service_uuid(ESPBTUUID uuid) {
  this->advertising_init_();
  this->advertising_->add_service_uuid(uuid);
-  this->advertising_start();
+  this->advertising_refresh();
 }

 void ESP32BLE::advertising_remove_service_uuid(ESPBTUUID uuid) {
  this->advertising_init_();
  this->advertising_->remove_service_uuid(uuid);
-  this->advertising_start();
+  this->advertising_refresh();
 }
 #endif

@@ -575,6 +592,10 @@ void ESP32BLE::loop_handle_state_transition_not_active_() {
    }

    this->state_ = BLE_COMPONENT_STATE_ACTIVE;
+#ifdef USE_ESP32_BLE_ADVERTISING
+    // Requests made before the stack was up (or before it was re-enabled) take effect now
+    this->advertising_refresh();
+#endif
  }
 }

@@ -114,7 +114,17 @@ class ESP32BLE final : public Component {
  void set_name(const char *name) { this->name_ = name; }

 #ifdef USE_ESP32_BLE_ADVERTISING
+  /** Request advertising on behalf of a component.
+   *
+   * Requests are reference counted: advertising runs until every component that called
+   * advertising_start() has released it again with advertising_stop(). Each component must
+   * pair its calls, so nothing advertises until something actually asks for it.
+   */
  void advertising_start();
+  /// Release a request made with advertising_start(); advertising stops at the last release.
+  void advertising_stop();
+  /// Apply the current payload and request count: advertise while requested, otherwise stop.
+  void advertising_refresh();
  void advertising_set_service_data(const std::vector<uint8_t> &data);
  void advertising_set_manufacturer_data(const std::vector<uint8_t> &data);
  void advertising_set_appearance(uint16_t appearance) { this->appearance_ = appearance; }
@@ -226,6 +236,9 @@ class ESP32BLE final : public Component {
  // 1-byte aligned members (grouped together to minimize padding)
  BLEComponentState state_{BLE_COMPONENT_STATE_OFF};  // 1 byte (uint8_t enum)
  bool enable_on_boot_{};                             // 1 byte
+#ifdef USE_ESP32_BLE_ADVERTISING
+  uint8_t advertising_ref_count_{0};  // 1 byte, number of components requesting advertising
+#endif

 #ifdef ESPHOME_ESP32_BLE_EXTENDED_AUTH_PARAMS
  optional<esp_ble_auth_req_t> auth_req_mode_;
@@ -67,6 +67,8 @@ void ESP32BLEBeacon::setup() {
      this->on_advertise_();
    }
  });
+  // A beacon always needs the device to advertise, and never releases the request
+  global_ble->advertising_start();
 }

 void ESP32BLEBeacon::on_advertise_() {
@@ -596,6 +596,18 @@ async def to_code(config):
    cg.add(var.set_parent(parent))
    cg.add(parent.advertising_set_appearance(config[CONF_APPEARANCE]))
    cg.add(var.set_max_clients(config[CONF_MAX_CLIENTS]))
+    # Only advertise for the server itself when the configuration gives clients something to
+    # find. A server that is auto-loaded purely to host a runtime service (esp32_improv) stays
+    # silent until that service asks for advertising.
+    cg.add(
+        var.set_advertising_required(
+            CONF_MANUFACTURER_DATA in config
+            or any(
+                not uuid_is(service_config[CONF_UUID], DEVICE_INFORMATION_SERVICE_UUID)
+                for service_config in config[CONF_SERVICES]
+            )
+        )
+    )
    if CONF_MANUFACTURER_DATA in config:
        cg.add(var.set_manufacturer_data(config[CONF_MANUFACTURER_DATA]))
    for service_config in config[CONF_SERVICES]:
@@ -81,6 +81,7 @@ void BLEServer::loop() {
      if (this->device_information_service_->is_running()) {
        this->state_ = RUNNING;
        this->restart_advertising_();
+        this->request_advertising_();
        ESP_LOGD(TAG, "BLE server setup successfully");
      } else if (this->device_information_service_->is_created()) {
        this->device_information_service_->start();
@@ -98,6 +99,20 @@ void BLEServer::restart_advertising_() {
  }
 }

+void BLEServer::request_advertising_() {
+  if (!this->advertising_required_ || this->advertising_requested_)
+    return;
+  this->advertising_requested_ = true;
+  this->parent_->advertising_start();
+}
+
+void BLEServer::release_advertising_() {
+  if (!this->advertising_requested_)
+    return;
+  this->advertising_requested_ = false;
+  this->parent_->advertising_stop();
+}
+
 BLEService *BLEServer::create_service(ESPBTUUID uuid, bool advertise, uint16_t num_handles) {
 #if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERBOSE
  char uuid_buf[esp32_ble::UUID_STR_LEN];
@@ -170,7 +185,7 @@ void BLEServer::gatts_event_handler(esp_gatts_cb_event_t event, esp_gatt_if_t ga
      this->add_client_(param->connect.conn_id);
      // Resume advertising so additional clients can discover and connect
      if (this->client_count_ < this->max_clients_) {
-        this->parent_->advertising_start();
+        this->parent_->advertising_refresh();
      }
      this->dispatch_callbacks_(CallbackType::ON_CONNECT, param->connect.conn_id);
      break;
@@ -178,7 +193,7 @@ void BLEServer::gatts_event_handler(esp_gatts_cb_event_t event, esp_gatt_if_t ga
    case ESP_GATTS_DISCONNECT_EVT: {
      ESP_LOGD(TAG, "BLE Client disconnected");
      this->remove_client_(param->disconnect.conn_id);
-      this->parent_->advertising_start();
+      this->parent_->advertising_refresh();
      this->dispatch_callbacks_(CallbackType::ON_DISCONNECT, param->disconnect.conn_id);
      break;
    }
@@ -226,6 +241,8 @@ void BLEServer::remove_client_(uint16_t conn_id) {
 }

 void BLEServer::ble_before_disabled_event_handler() {
+  // Advertising is re-requested once the server is running again after BLE is re-enabled
+  this->release_advertising_();
  // Delete all clients
  this->client_count_ = 0;
  // Delete all services
@@ -38,6 +38,13 @@ class BLEServer final : public Component, public Parented<ESP32BLE> {
    this->restart_advertising_();
  }

+  /** Whether this server needs the device to advertise so clients can find and connect to it.
+   *
+   * False for a server that only hosts services created at runtime (e.g. esp32_improv), which
+   * request advertising themselves for as long as they need it.
+   */
+  void set_advertising_required(bool required) { this->advertising_required_ = required; }
+
  void set_max_clients(uint8_t max_clients) { this->max_clients_ = max_clients; }
  uint8_t get_max_clients() const { return this->max_clients_; }

@@ -82,6 +89,8 @@ class BLEServer final : public Component, public Parented<ESP32BLE> {
  };

  void restart_advertising_();
+  void request_advertising_();
+  void release_advertising_();

  int8_t find_client_index_(uint16_t conn_id) const;
  void add_client_(uint16_t conn_id);
@@ -93,6 +102,8 @@ class BLEServer final : public Component, public Parented<ESP32BLE> {
  std::vector<uint8_t> manufacturer_data_{};
  esp_gatt_if_t gatts_if_{0};
  bool registered_{false};
+  bool advertising_required_{true};
+  bool advertising_requested_{false};

  uint16_t clients_[USE_ESP32_BLE_MAX_CONNECTIONS]{};
  uint8_t client_count_{0};
@@ -112,6 +112,7 @@ void ESP32ImprovComponent::loop() {
      this->state_callback_.call(this->state_, this->error_state_);
 #endif
    }
+    this->release_advertising_();
    this->incoming_data_.clear();
    return;
  }
@@ -143,8 +144,9 @@ void ESP32ImprovComponent::loop() {
          ESP_LOGV(TAG, "Starting with device name advertising");
          this->advertising_device_name_ = true;
          this->last_name_adv_time_ = App.get_loop_component_start_time();
+          // Set the payload before requesting, so advertising starts exactly once
          esp32_ble::global_ble->advertising_set_service_data_and_name(std::span<const uint8_t>{}, true);
-          esp32_ble::global_ble->advertising_start();
+          this->request_advertising_();

          // Set initial state based on whether we have an authorizer
          this->set_state_(this->get_initial_state_(), false);
@@ -326,6 +328,8 @@ void ESP32ImprovComponent::stop() {
  this->set_timeout("end-service", STOP_ADVERTISING_DELAY, [this] {
    if (this->state_ == improv::STATE_STOPPED || this->service_ == nullptr)
      return;
+    // Release first so removing the service UUID does not restart advertising on the way out
+    this->release_advertising_();
    this->service_->stop();
    this->set_state_(improv::STATE_STOPPED);
  });
@@ -520,6 +524,20 @@ void ESP32ImprovComponent::update_advertising_type_() {
  }
 }

+void ESP32ImprovComponent::request_advertising_() {
+  if (this->advertising_requested_)
+    return;
+  this->advertising_requested_ = true;
+  esp32_ble::global_ble->advertising_start();
+}
+
+void ESP32ImprovComponent::release_advertising_() {
+  if (!this->advertising_requested_)
+    return;
+  this->advertising_requested_ = false;
+  esp32_ble::global_ble->advertising_stop();
+}
+
 improv::State ESP32ImprovComponent::get_initial_state_() const {
 #ifdef USE_BINARY_SENSOR
  // If we have an authorizer, start in awaiting authorization state
@@ -104,8 +104,11 @@ class ESP32ImprovComponent final : public Component, public improv_base::ImprovB
  bool status_indicator_state_{false};
  uint32_t last_name_adv_time_{0};
  bool advertising_device_name_{false};
+  bool advertising_requested_{false};
  void set_status_indicator_state_(bool state);
  void update_advertising_type_();
+  void request_advertising_();
+  void release_advertising_();

  void set_state_(improv::State state, bool update_advertising = true);
  void set_error_(improv::Error error);
@@ -1,13 +1,9 @@
 import logging

 import esphome.codegen as cg
-from esphome.components.noise import (
-    decode_encryption_key,
-    encryption_schema,
-    is_reserved_key,
-)
+from esphome.components.noise import encryption_schema, is_reserved_key, new_psk_progmem
 from esphome.components.ota import BASE_OTA_SCHEMA, OTAComponent, ota_to_code
-from esphome.config_helpers import merge_config
+from esphome.config_helpers import filter_source_files_from_defines, merge_config
 import esphome.config_validation as cv
 from esphome.const import (
    CONF_API,
@@ -41,7 +37,8 @@ DEPENDENCIES = ["network"]


 def AUTO_LOAD(config: ConfigType) -> list[str]:
-    """Auto-load noise only when encryption is configured."""
+    """Auto-load noise only when encryption is configured. The api key offer
+    path inherits noise from the api component's own AUTO_LOAD."""
    base = ["sha256", "socket"]
    # A falsy config is a tooling probe for the maximal set (None from
    # dependency resolution, {} from the components-graph platform probe);
@@ -132,12 +129,41 @@ 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)
-    if any(
-        conf.get(CONF_PLATFORM) == CONF_WEB_SERVER for conf in full_ota_conf
-    ) and any(
-        CONF_ENCRYPTION in conf for conf in merged_ota_esphome_configs_by_port.values()
+        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 "
+                "encryption with the '%s' %s %s, which authenticates any uploader "
+                "that takes it, and a password only matters for uploaders without "
+                "encryption support; remove '%s' and add '%s' under '%s' so "
+                "uploads use the key and encryption is required",
+                CONF_OTA,
+                CONF_PASSWORD,
+                CONF_API,
+                CONF_ENCRYPTION,
+                CONF_KEY,
+                CONF_PASSWORD,
+                CONF_ENCRYPTION,
+                CONF_OTA,
+            )
+    # The captive_portal auto-loads the web_server ota platform too, but that
+    # endpoint only exists while the fallback AP is active and is the intended
+    # recovery path, so it does not warn unless the web_server component is
+    # configured as well
+    captive_portal_only = (
+        CONF_CAPTIVE_PORTAL in full_conf and CONF_WEB_SERVER not in full_conf
+    )
+    if (
+        not captive_portal_only
+        and any(conf.get(CONF_PLATFORM) == CONF_WEB_SERVER for conf in full_ota_conf)
+        and any(
+            CONF_ENCRYPTION in conf
+            for conf in merged_ota_esphome_configs_by_port.values()
+        )
    ):
-        _warn_web_server_ota(full_conf)
+        _warn_web_server_ota()

    full_conf[CONF_OTA] = new_ota_conf
    fv.full_config.set(full_conf)
@@ -152,25 +178,30 @@ def ota_esphome_final_validate(config: ConfigType) -> None:
        )


-def _warn_web_server_ota(full_conf: ConfigType) -> None:
+def _warn_web_server_ota() -> None:
    """The web_server ota platform accepts the same image over plaintext HTTP
    with basic auth, bypassing the encryption; warn rather than fail so the
    operator keeps the recovery path."""
-    if CONF_CAPTIVE_PORTAL in full_conf and CONF_WEB_SERVER not in full_conf:
-        # The captive_portal auto-load: the endpoint only exists while the
-        # fallback AP is active
-        _LOGGER.warning(
-            "OTA encryption does not cover the %s OTA platform (auto-loaded "
-            "by captive_portal); the plaintext /update endpoint stays "
-            "reachable while the fallback AP is active",
-            CONF_WEB_SERVER,
-        )
-    else:
-        _LOGGER.warning(
-            "OTA encryption does not cover the %s OTA platform; its "
-            "plaintext /update endpoint accepts the same image",
-            CONF_WEB_SERVER,
-        )
+    _LOGGER.warning(
+        "OTA encryption does not cover the %s OTA platform; its "
+        "plaintext /update endpoint accepts the same image",
+        CONF_WEB_SERVER,
+    )
+
+
+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."""
+    key = api_conf.get(CONF_ENCRYPTION, {}).get(CONF_KEY)
+    if not key or is_reserved_key(key):
+        return None
+    return key


 def _resolve_encryption_key(encryption_conf: ConfigType, api_conf: ConfigType) -> None:
@@ -267,15 +298,9 @@ CONFIG_SCHEMA = cv.All(
 FINAL_VALIDATE_SCHEMA = ota_esphome_final_validate


-def FILTER_SOURCE_FILES() -> list[str]:
-    """Filter out the noise transport when no ota entry configures encryption."""
-    for ota_conf in CORE.config.get(CONF_OTA, []):
-        if (
-            ota_conf.get(CONF_PLATFORM) == CONF_ESPHOME
-            and ota_conf.get(CONF_ENCRYPTION) is not None
-        ):
-            return []
-    return ["ota_esphome_noise.cpp"]
+FILTER_SOURCE_FILES = filter_source_files_from_defines(
+    {"ota_esphome_noise.cpp": "USE_OTA_ENCRYPTION"}
+)


@coroutine_with_priority(CoroPriority.OTA_UPDATES)
@@ -296,11 +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.
+        # 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")
-        cg.add(var.set_noise_psk(list(decode_encryption_key(key))))
+        if encryption_conf is not None:
+            cg.add_define("USE_OTA_ENCRYPTION_REQUIRED")
+        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")
@@ -97,18 +97,24 @@ void ESPHomeOTAComponent::dump_config() {
  ESP_LOGCONFIG(TAG,
                "Over-The-Air updates:\n"
                "  Address: %s:%u\n"
-                "  Version: %d",
+                "  Version: %d"
+#ifdef USE_OTA_ENCRYPTION_REQUIRED
+                "\n  Encryption: required"
+#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");
  }
 #endif
-#ifdef USE_OTA_ENCRYPTION
-  if (this->noise_ctx_.has_psk()) {
-    ESP_LOGCONFIG(TAG, "  Encryption configured");
-  }
-#endif
 #ifdef USE_OTA_PARTITIONS
  ESP_LOGCONFIG(TAG,
                "  Partition access allowed\n"
@@ -150,10 +156,6 @@ void ESPHomeOTAComponent::loop() {
  this->handle_handshake_();
 }

-static constexpr uint8_t CLIENT_FEATURE_SUPPORTS_COMPRESSION = 0x01;
-static constexpr uint8_t CLIENT_FEATURE_SUPPORTS_SHA256_AUTH = 0x02;
-static constexpr uint8_t CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL = 0x04;
-static constexpr uint8_t CLIENT_FEATURE_SUPPORTS_NOISE = 0x08;
 static constexpr uint8_t SERVER_FEATURE_SUPPORTS_COMPRESSION = 0x01;
 static constexpr uint8_t SERVER_FEATURE_SUPPORTS_PARTITION_ACCESS = 0x02;
 static constexpr uint8_t SERVER_FEATURE_SUPPORTS_NOISE = 0x04;
@@ -241,12 +243,10 @@ void ESPHomeOTAComponent::handle_handshake_() {
      this->ota_features_ = this->handshake_buf_[0];
      ESP_LOGV(TAG, "Features: 0x%02X", this->ota_features_);

-#ifdef USE_OTA_ENCRYPTION
-      // Fail closed: with a PSK configured the client must negotiate encryption
-      // (which requires the extended protocol); refuse plaintext uploads.
-      static constexpr uint8_t NOISE_REQUIRED_FEATURES =
-          CLIENT_FEATURE_SUPPORTS_NOISE | CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL;
-      if (this->noise_ctx_.has_psk() && (this->ota_features_ & NOISE_REQUIRED_FEATURES) != NOISE_REQUIRED_FEATURES) {
+#ifdef USE_OTA_ENCRYPTION_REQUIRED
+      // Fail closed: an explicit `ota: encryption:` block means the client must
+      // negotiate encryption; refuse plaintext uploads
+      if ((this->ota_features_ & CLIENT_NOISE_FEATURES) != CLIENT_NOISE_FEATURES) {
        ESP_LOGW(TAG, "Client does not support encryption");
        this->send_error_and_cleanup_(ota::OTA_RESPONSE_ERROR_ENCRYPTION_REQUIRED);
        return;
@@ -261,8 +261,7 @@ void ESPHomeOTAComponent::handle_handshake_() {
      // Compose the feature-ack response. When the client negotiates the extended protocol we emit
      // a 2-byte response (marker + server feature flags); otherwise we emit the single-byte
      // legacy response.
-      this->extended_proto_ = (this->ota_features_ & CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL) != 0;
-      if (this->extended_proto_) {
+      if (this->extended_proto_()) {
        static_assert(HANDSHAKE_BUF_SIZE >= 2, "handshake_buf_ must hold the 2-byte extended-protocol feature ack");
        this->handshake_buf_[0] = ota::OTA_RESPONSE_FEATURE_FLAGS;
        this->handshake_buf_[1] = (supports_compression ? SERVER_FEATURE_SUPPORTS_COMPRESSION : 0);
@@ -270,7 +269,8 @@ void ESPHomeOTAComponent::handle_handshake_() {
        this->handshake_buf_[1] |= SERVER_FEATURE_SUPPORTS_PARTITION_ACCESS;
 #endif
 #ifdef USE_OTA_ENCRYPTION
-        if (this->noise_ctx_.has_psk()) {
+        // 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
@@ -284,15 +284,15 @@ void ESPHomeOTAComponent::handle_handshake_() {
    case OTAState::FEATURE_ACK: {
      static constexpr size_t STANDARD_PROTO_ACK_SIZE = 1;
      static constexpr size_t EXTENDED_PROTO_ACK_SIZE = 2;
-      const size_t ack_size = this->extended_proto_ ? EXTENDED_PROTO_ACK_SIZE : STANDARD_PROTO_ACK_SIZE;
+      const size_t ack_size = this->extended_proto_() ? EXTENDED_PROTO_ACK_SIZE : STANDARD_PROTO_ACK_SIZE;
      if (!this->try_write_(ack_size, LOG_STR("ack feature"))) {
        return;
      }
 #ifdef USE_OTA_ENCRYPTION
-      // With a PSK configured the rest of the session runs inside the noise
-      // transport; the client sends the first handshake frame next, so there
-      // is nothing to do until data arrives.
-      if (this->noise_ctx_.has_psk()) {
+      // 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->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])) {
@@ -412,7 +412,7 @@ void ESPHomeOTAComponent::handle_data_() {
  // Acknowledge auth OK - 1 byte
  this->data_write_byte_(ota::OTA_RESPONSE_AUTH_OK);

-  if (this->extended_proto_) {
+  if (this->extended_proto_()) {
    // Read ota type, 1 byte
    if (!this->data_readall_(buf, 1)) {
      this->log_read_error_(LOG_STR("OTA type"));
@@ -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,8 +47,9 @@ class ESPHomeOTAComponent final : public ota::OTAComponent {
  }
 #endif  // USE_OTA_PASSWORD

-#ifdef USE_OTA_ENCRYPTION
-  void set_noise_psk(noise::psk_t psk) { this->noise_ctx_.set_psk(psk); }
+#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

  /// Manually set the port OTA should listen on
@@ -85,9 +89,19 @@ 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_();
+  size_t noise_frame_payload_len_(const uint8_t *header, size_t min_len, size_t max_len);
  bool noise_try_write_frame_();
  void noise_send_reject_(const LogString *reason);
  ssize_t noise_decrypt_(uint8_t *buf, size_t len);
@@ -144,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

@@ -166,6 +182,16 @@ class ESPHomeOTAComponent final : public ota::OTAComponent {
                "OTA_BUFFER_SIZE must fit a full encrypted data frame");
 #endif
  static constexpr uint8_t MAGIC_BYTES[5] = {0x6C, 0x26, 0xF7, 0x5C, 0x45};
+  static constexpr uint8_t CLIENT_FEATURE_SUPPORTS_COMPRESSION = 0x01;
+  static constexpr uint8_t CLIENT_FEATURE_SUPPORTS_SHA256_AUTH = 0x02;
+  static constexpr uint8_t CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL = 0x04;
+  static constexpr uint8_t CLIENT_FEATURE_SUPPORTS_NOISE = 0x08;
+  // Noise needs the extended protocol: the prologue binds the 2-byte feature ack
+  static constexpr uint8_t CLIENT_NOISE_FEATURES =
+      CLIENT_FEATURE_SUPPORTS_NOISE | CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL;
+  // Derived from the feature byte rather than stored, which keeps the
+  // trailing byte members at a multiple of four
+  inline bool extended_proto_() const { return (this->ota_features_ & CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL) != 0; }
 #ifdef USE_OTA_PARTITIONS
  uint32_t running_app_offset_{0};
  size_t running_app_size_{0};
@@ -179,7 +205,6 @@ class ESPHomeOTAComponent final : public ota::OTAComponent {
  uint8_t auth_buf_pos_{0};
  uint8_t auth_type_{0};  // Store auth type to know which hasher to use
 #endif                    // USE_OTA_PASSWORD
-  bool extended_proto_{false};
 };

 }  // namespace esphome
@@ -3,6 +3,7 @@
 #ifdef USE_OTA_ENCRYPTION
 #include "esphome/components/noise/noise.h"
 #include "esphome/components/ota/ota_backend.h"
+#include "esphome/core/hal.h"
 #include "esphome/core/log.h"

 #include <cstring>
@@ -41,23 +42,15 @@ ESPHomeOTAComponent::NoiseSession::~NoiseSession() {
 */
 bool ESPHomeOTAComponent::noise_start_session_(uint8_t server_feature_flags) {
  // NOLINTNEXTLINE(clang-analyzer-cplusplus.NewDeleteLeaks)
-  this->noise_ = std::unique_ptr<NoiseSession>(new (std::nothrow) NoiseSession());
-  if (this->noise_ == nullptr) {
-    ESP_LOGW(TAG, "Session allocation failed");
-    this->cleanup_connection_();
-    return false;
-  }
-
+  // Default-init: the frame buffer is always written before it is read, so
+  // skip zeroing its 132 bytes
+  this->noise_ = std::unique_ptr<NoiseSession>(new (std::nothrow) NoiseSession);
  static constexpr size_t PROLOGUE_ACK_LEN = 2;  // OTA_RESPONSE_OK + version
  static constexpr size_t PROLOGUE_CLIENT_FEATURES_LEN = 1;
  static constexpr size_t PROLOGUE_FEATURE_ACK_LEN = 2;  // OTA_RESPONSE_FEATURE_FLAGS + server flags
  uint8_t prologue[OTA_NOISE_PROLOGUE_INIT_LEN + sizeof(MAGIC_BYTES) + PROLOGUE_ACK_LEN + PROLOGUE_CLIENT_FEATURES_LEN +
                   PROLOGUE_FEATURE_ACK_LEN];
-#ifdef USE_ESP8266
-  memcpy_P(prologue, OTA_NOISE_PROLOGUE_INIT, OTA_NOISE_PROLOGUE_INIT_LEN);
-#else
-  std::memcpy(prologue, OTA_NOISE_PROLOGUE_INIT, OTA_NOISE_PROLOGUE_INIT_LEN);
-#endif
+  progmem_memcpy(prologue, OTA_NOISE_PROLOGUE_INIT, OTA_NOISE_PROLOGUE_INIT_LEN);
  uint8_t *p = prologue + OTA_NOISE_PROLOGUE_INIT_LEN;
  // Magic bytes, already validated in MAGIC_READ
  std::memcpy(p, MAGIC_BYTES, sizeof(MAGIC_BYTES));
@@ -71,9 +64,14 @@ bool ESPHomeOTAComponent::noise_start_session_(uint8_t server_feature_flags) {
  *p++ = ota::OTA_RESPONSE_FEATURE_FLAGS;
  *p++ = server_feature_flags;

-  int err = this->noise_->handshake.init(this->noise_ctx_.get_psk(), 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, "Handshake init: %s", LOG_STR_ARG(noise::noise_err_to_logstr(err)));
+    ESP_LOGW(TAG, "Session init: %d", err);
    this->cleanup_connection_();
    return false;
  }
@@ -105,14 +103,16 @@ bool ESPHomeOTAComponent::handle_noise_handshake_() {
        s.frame_pos = 0;
        s.frame_len = 0;
        if (s.frame_buf[noise::FRAME_HEADER_SIZE] != noise::HANDSHAKE_STATUS_OK) {
-          ESP_LOGW(TAG, "Bad handshake error byte: %u", s.frame_buf[noise::FRAME_HEADER_SIZE]);
+          ESP_LOGV(TAG, "Bad handshake error byte: %u", s.frame_buf[noise::FRAME_HEADER_SIZE]);
          this->cleanup_connection_();
          return false;
        }
        int err = s.handshake.read_message(s.frame_buf + noise::FRAME_HEADER_SIZE + 1, payload_len - 1);
        if (err != 0) {
-          ESP_LOGW(TAG, "Handshake read: %s", LOG_STR_ARG(noise::noise_err_to_logstr(err)));
-          this->noise_send_reject_(noise::reject_reason_for(err));
+          // A MAC failure here almost always means the uploader has a different key
+          const LogString *reason = noise::reject_reason_for(err);
+          ESP_LOGW(TAG, "Handshake read: %s (%d)", LOG_STR_ARG(reason), err);
+          this->noise_send_reject_(reason);
          this->cleanup_connection_();
          return false;
        }
@@ -123,7 +123,7 @@ bool ESPHomeOTAComponent::handle_noise_handshake_() {
        int err =
            s.handshake.write_message(s.frame_buf + noise::FRAME_HEADER_SIZE + 1, noise::MAX_HANDSHAKE_SIZE, msg_len);
        if (err != 0) {
-          ESP_LOGW(TAG, "Handshake write: %s", LOG_STR_ARG(noise::noise_err_to_logstr(err)));
+          ESP_LOGW(TAG, "Handshake write: %d", err);
          this->cleanup_connection_();
          return false;
        }
@@ -138,7 +138,7 @@ bool ESPHomeOTAComponent::handle_noise_handshake_() {
      case noise::NoiseResponderHandshake::Action::ACTION_SPLIT: {
        int err = s.handshake.split(s.send_cipher, s.recv_cipher);
        if (err != 0) {
-          ESP_LOGW(TAG, "Handshake split: %s", LOG_STR_ARG(noise::noise_err_to_logstr(err)));
+          ESP_LOGW(TAG, "Handshake split: %d", err);
          this->cleanup_connection_();
          return false;
        }
@@ -154,33 +154,41 @@ bool ESPHomeOTAComponent::handle_noise_handshake_() {
  }
 }

+/// Payload length from a frame header, or 0 (logged) when the indicator or
+/// the length is out of range. Callers pass min_len >= 1 so 0 is never valid.
+size_t ESPHomeOTAComponent::noise_frame_payload_len_(const uint8_t *header, size_t min_len, size_t max_len) {
+  const size_t payload_len = encode_uint16(header[1], header[2]);
+  if (header[0] != noise::FRAME_INDICATOR || payload_len < min_len || payload_len > max_len) {
+    ESP_LOGW(TAG, "Bad frame: 0x%02X, %zu bytes", header[0], payload_len);
+    return 0;
+  }
+  return payload_len;
+}
+
 /// Non-blocking read of one handshake frame into the session buffer.
 bool ESPHomeOTAComponent::noise_try_read_frame_() {
  NoiseSession &s = *this->noise_;
-  while (s.frame_pos < noise::FRAME_HEADER_SIZE) {
-    ssize_t read = this->client_->read(s.frame_buf + s.frame_pos, noise::FRAME_HEADER_SIZE - s.frame_pos);
-    if (!this->handle_read_error_(read, LOG_STR("read noise header"))) {
-      return false;
+  while (true) {
+    // The header first, then the body once the header says how long it is
+    const uint16_t want = s.frame_len == 0 ? noise::FRAME_HEADER_SIZE : s.frame_len;
+    if (s.frame_pos < want) {
+      ssize_t read = this->client_->read(s.frame_buf + s.frame_pos, want - s.frame_pos);
+      if (!this->handle_read_error_(read, LOG_STR("read noise"))) {
+        return false;
+      }
+      s.frame_pos += read;
+      continue;
    }
-    s.frame_pos += read;
-  }
-  if (s.frame_len == 0) {
-    const uint16_t payload_len = encode_uint16(s.frame_buf[1], s.frame_buf[2]);
-    if (s.frame_buf[0] != noise::FRAME_INDICATOR || payload_len < 1 || payload_len > 1 + noise::MAX_HANDSHAKE_SIZE) {
-      ESP_LOGW(TAG, "Bad handshake frame: 0x%02X, %u bytes", s.frame_buf[0], payload_len);
+    if (s.frame_len != 0) {
+      return true;
+    }
+    const size_t payload_len = this->noise_frame_payload_len_(s.frame_buf, 1, 1 + noise::MAX_HANDSHAKE_SIZE);
+    if (payload_len == 0) {
      this->cleanup_connection_();
      return false;
    }
    s.frame_len = noise::FRAME_HEADER_SIZE + payload_len;
  }
-  while (s.frame_pos < s.frame_len) {
-    ssize_t read = this->client_->read(s.frame_buf + s.frame_pos, s.frame_len - s.frame_pos);
-    if (!this->handle_read_error_(read, LOG_STR("read noise frame"))) {
-      return false;
-    }
-    s.frame_pos += read;
-  }
-  return true;
 }

 /// Non-blocking write of the pending session-buffer frame.
@@ -214,7 +222,7 @@ ssize_t ESPHomeOTAComponent::noise_decrypt_(uint8_t *buf, size_t len) {
  noise_buffer_set_inout(mbuf, buf, len, len);
  int err = noise_cipherstate_decrypt(this->noise_->recv_cipher, &mbuf);
  if (err != 0) {
-    ESP_LOGW(TAG, "Decrypt: %s", LOG_STR_ARG(noise::noise_err_to_logstr(err)));
+    ESP_LOGW(TAG, "Decrypt: %d", err);
    return -1;
  }
  return mbuf.size;
@@ -229,9 +237,8 @@ ssize_t ESPHomeOTAComponent::noise_read_frame_blocking_(uint8_t *buf, size_t min
  if (!this->readall_(header, sizeof(header))) {
    return -1;
  }
-  const size_t ciphertext_len = encode_uint16(header[1], header[2]);
-  if (header[0] != noise::FRAME_INDICATOR || ciphertext_len < min_ciphertext || ciphertext_len > max_ciphertext) {
-    ESP_LOGW(TAG, "Bad frame: 0x%02X, %zu bytes", header[0], ciphertext_len);
+  const size_t ciphertext_len = this->noise_frame_payload_len_(header, min_ciphertext, max_ciphertext);
+  if (ciphertext_len == 0) {
    return -1;
  }
  if (!this->readall_(buf, ciphertext_len)) {
@@ -267,7 +274,7 @@ bool ESPHomeOTAComponent::noise_write_byte_(uint8_t byte) {
  noise_buffer_set_inout(mbuf, frame + noise::FRAME_HEADER_SIZE, 1, 1 + noise::MAC_SIZE);
  int err = noise_cipherstate_encrypt(this->noise_->send_cipher, &mbuf);
  if (err != 0) {
-    ESP_LOGW(TAG, "Encrypt: %s", LOG_STR_ARG(noise::noise_err_to_logstr(err)));
+    ESP_LOGW(TAG, "Encrypt: %d", err);
    return false;
  }
  noise::write_frame_header(frame, mbuf.size);
@@ -5,6 +5,8 @@ from typing import Any
 import esphome.codegen as cg
 import esphome.config_validation as cv
 from esphome.const import CONF_KEY
+from esphome.core import ID
+from esphome.cpp_generator import MockObj
 from esphome.types import ConfigType

 CODEOWNERS = ["@esphome/core"]
@@ -61,6 +63,15 @@ ENCRYPTION_SCHEMA = cv.Schema(
 )


+def new_psk_progmem(parent_id: ID, key: str) -> MockObj:
+    """Emit the decoded key as a PROGMEM array; the component keeps a pointer
+    so the key never occupies RAM."""
+    return cg.progmem_array(
+        ID(f"{parent_id.id}_psk", is_declaration=True, type=cg.uint8),
+        list(decode_encryption_key(key)),
+    )
+
+
 def encryption_schema(config: ConfigType | None) -> ConfigType:
    # A bare `encryption:` block is valid; a missing key means the consumer
    # falls back to its keyless behavior (api provisioning, ota inheriting
@@ -1,5 +1,6 @@
 #include "noise.h"
 #ifdef USE_NOISE
+#include "esphome/core/hal.h"
 #include "esphome/core/log.h"

 #include <algorithm>
@@ -15,6 +16,23 @@ namespace esphome::noise {

 static const char *const TAG = "noise";

+void NoiseContext::set_psk(const uint8_t *psk) {
+  this->psk_ = psk;
+  psk_t copy;
+  this->load_psk(copy);
+  if (is_all_zeros(copy)) {
+    this->psk_ = nullptr;
+  }
+}
+
+void NoiseContext::load_psk(psk_t &out) const {
+  if (this->psk_ == nullptr) {
+    out.fill(0);
+    return;
+  }
+  progmem_memcpy(out.data(), this->psk_, out.size());
+}
+
 const LogString *noise_err_to_logstr(int err) {
  if (err == NOISE_ERROR_NO_MEMORY)
    return LOG_STR("NO_MEMORY");
@@ -23,16 +23,16 @@ class NoiseContext {
    }
    return acc == 0;
  }
-  void set_psk(psk_t psk) {
-    this->psk_ = psk;
-    this->has_psk_ = !is_all_zeros(psk);
-  }
-  const psk_t &get_psk() const { return this->psk_; }
-  bool has_psk() const { return this->has_psk_; }
+  /// psk points at 32 bytes that outlive the context: a PROGMEM array from
+  /// codegen, or RAM owned by the caller for a runtime provisioned key.
+  /// The all-zeros key counts as no key.
+  void set_psk(const uint8_t *psk);
+  /// Copy the key out (flash-aware on ESP8266); all zeros when none is set.
+  void load_psk(psk_t &out) const;
+  bool has_psk() const { return this->psk_ != nullptr; }

 protected:
-  psk_t psk_{};
-  bool has_psk_{false};
+  const uint8_t *psk_{nullptr};
 };

 /// Convert a noise error code to a readable error
@@ -20,7 +20,7 @@ NoiseResponderHandshake::~NoiseResponderHandshake() {
  }
 }

-int NoiseResponderHandshake::init(const psk_t &psk, const uint8_t *prologue, size_t prologue_len) {
+int NoiseResponderHandshake::init(const NoiseContext &ctx, const uint8_t *prologue, size_t prologue_len) {
  if (this->handshake_ != nullptr) {
    noise_handshakestate_free(this->handshake_);
    this->handshake_ = nullptr;
@@ -44,6 +44,9 @@ int NoiseResponderHandshake::init(const psk_t &psk, const uint8_t *prologue, siz
    HANDSHAKE_STEP_LOG("noise_handshakestate_new_by_id", err);
    return err;
  }
+  // noise-c keeps its own copy, so the key only passes through the stack here
+  psk_t psk;
+  ctx.load_psk(psk);
  err = noise_handshakestate_set_pre_shared_key(this->handshake_, psk.data(), psk.size());
  if (err != 0) {
    HANDSHAKE_STEP_LOG("noise_handshakestate_set_pre_shared_key", err);
@@ -36,9 +36,9 @@ class NoiseResponderHandshake {
  NoiseResponderHandshake(const NoiseResponderHandshake &) = delete;
  NoiseResponderHandshake &operator=(const NoiseResponderHandshake &) = delete;

-  /// Create and start the handshake with the given PSK and prologue. A
-  /// repeated call frees the previous handshake state and starts over.
-  [[nodiscard]] int init(const psk_t &psk, const uint8_t *prologue, size_t prologue_len);
+  /// Create and start the handshake with the context's PSK and the prologue.
+  /// A repeated call frees the previous handshake state and starts over.
+  [[nodiscard]] int init(const NoiseContext &ctx, const uint8_t *prologue, size_t prologue_len);
  /// ACTION_FAILED is the catch-all: returned before init(), after split()
  /// has released the state, and when noise-c reports a failed handshake.
  [[nodiscard]] Action action() const;
@@ -244,6 +244,8 @@
 #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
 #define USE_TIME_TIMEZONE
@@ -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,19 +534,36 @@ def perform_ota(
    else:
        features = 0

-    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. "
-                "If the running firmware predates OTA encryption, first update "
-                "it without the 'ota: encryption:' block (over a trusted "
-                "network or via USB), then restore the block and upload again."
+                "The running firmware does not offer encryption (built before "
+                "ESPHome 2026.9.0 or without an 'api: encryption: key'). "
+                "If the config has an 'api: encryption: key', install once "
+                "with the 'ota: encryption:' block removed (that firmware "
+                "offers encryption), then restore the block and install "
+                "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 = (
@@ -549,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:
@@ -757,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

@@ -795,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)...",
@@ -815,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, ...);
@@ -847,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)
@@ -148,11 +148,14 @@ def wizard_file(**kwargs: Unpack[WizardFileKwargs]) -> str:
    if "api_encryption_key" in kwargs:
        config += f'  encryption:\n    key: "{kwargs["api_encryption_key"]}"\n'

-    # Configure OTA
+    # Configure OTA: the api key also secures OTA updates, a password is only
+    # for uploaders that do not support encryption
    config += "\nota:\n"
    config += "  - platform: esphome\n"
    if "ota_password" in kwargs:
        config += f'    password: "{kwargs["ota_password"]}"'
+    elif "api_encryption_key" in kwargs:
+        config += "    encryption:"

    # Configuring wifi
    config += "\n\nwifi:\n"
@@ -532,12 +535,13 @@ def wizard(path: Path) -> int:

        safe_print()
        safe_print(
-            f"Do you want to set a {color(AnsiFore.GREEN, 'password')} for OTA updates? "
-            "This can be insecure if you do not trust the WiFi network."
+            "The API encryption key also secures OTA updates. Do you want to set a "
+            f"{color(AnsiFore.GREEN, 'password')} for OTA updates instead? Only "
+            "older uploaders that do not support encryption need one."
        )
        safe_print()
        sleep(0.25)
-        safe_print("Press ENTER for no password")
+        safe_print("Press ENTER to use the encryption key")
        ota_password = safe_input(color(AnsiFore.BOLD_WHITE, "(password): "))
    else:
        ssid, psk = "", ""
@@ -27,7 +27,7 @@ bleak==3.0.2
 smpclient==7.2.0
 requests==2.34.2
 py7zr==1.1.3
-platformdirs==4.11.5  # native esp-idf toolchain global cache dir
+platformdirs==4.11.7  # native esp-idf toolchain global cache dir
 ninja==1.13.2  # native esp8266 arduino toolchain build driver
 filelock==3.32.5  # inter-process locks (PlatformIO cache heal, git clone cache); >=3.32 for FileLock(fallback_to_soft=...), older versions silently drop the kwarg

@@ -2,7 +2,7 @@ pylint==4.0.8
 flake8==7.3.0  # also change in .pre-commit-config.yaml when updating
 ruff==0.16.5  # also change in .pre-commit-config.yaml when updating
 pyupgrade==3.21.2  # also change in .pre-commit-config.yaml when updating
-prek==0.5.0  # also change in .github/workflows/ci.yml when updating
+prek==0.5.1  # also change in .github/workflows/ci.yml when updating

 # Unit tests
 pytest==9.1.1
@@ -0,0 +1,13 @@
+esphome:
+  name: test
+
+esp32:
+  variant: esp32
+
+wifi:
+  ssid: MySSID
+  password: password1
+
+# esp32_ble_server is only auto-loaded here, so it has no services of its own.
+esp32_improv:
+  authorizer: none
@@ -0,0 +1,9 @@
+esphome:
+  name: test
+
+esp32:
+  variant: esp32
+
+esp32_ble_server:
+  id: ble_server
+  manufacturer_data: [0x72, 0x04, 0x00, 0x23]
@@ -0,0 +1,14 @@
+esphome:
+  name: test
+
+esp32:
+  variant: esp32
+
+esp32_ble_server:
+  id: ble_server
+  services:
+    - uuid: 2a24b789-7aab-4535-af3e-ee76a35cc12d
+      characteristics:
+        - uuid: cad48e28-7fbe-41cf-bae9-d77a6c233423
+          read: true
+          value: [1, 2, 3, 4]
@@ -1,5 +1,10 @@
 """Tests for esp32_ble_server configuration helpers."""

+from __future__ import annotations
+
+from collections.abc import Callable
+from pathlib import Path
+
 import pytest

 from esphome.components.esp32_ble_server import (
@@ -45,3 +50,26 @@ def test_uuid_is_matches_descriptor_short_strings(uuid16) -> None:
    assert uuid_is(uuid16, uuid16)
    assert uuid_is(f"{uuid16:04X}", uuid16)
    assert uuid_is(f"{uuid16:08X}", uuid16)
+
+
+@pytest.mark.parametrize(
+    ("config_file", "required"),
+    [
+        # Auto-loaded by esp32_improv only: nothing to find until Improv asks for it
+        ("improv_only.yaml", False),
+        # The configuration defines a service clients are meant to connect to
+        ("own_service.yaml", True),
+        # Manufacturer data is only useful if it is actually broadcast
+        ("manufacturer_data_only.yaml", True),
+    ],
+)
+def test_advertising_required(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+    config_file: str,
+    required: bool,
+) -> None:
+    """The server only requests advertising when the configuration needs it."""
+    main_cpp = generate_main(component_config_path(config_file))
+
+    assert f"set_advertising_required({str(required).lower()})" in main_cpp
@@ -2,6 +2,7 @@

 from __future__ import annotations

+from collections.abc import Callable
 import logging
 from typing import Any

@@ -11,6 +12,8 @@ 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,
 )
@@ -316,12 +319,12 @@ def test_encryption_with_web_server_ota_warns(
        fv.full_config.reset(token)


-def test_encryption_with_captive_portal_web_server_ota_warns(
+def test_encryption_with_captive_portal_does_not_warn(
    caplog: pytest.LogCaptureFixture,
 ) -> None:
    """captive_portal auto-loads the web_server ota platform without the
-    web_server component; encryption stays usable and only warns, so the
-    fallback AP recovery path is not lost."""
+    web_server component; its endpoint only exists while the fallback AP is
+    active and is the intended recovery path, so there is no warning."""
    full_conf = {
        "captive_portal": {},
        CONF_OTA: [
@@ -333,7 +336,10 @@ def test_encryption_with_captive_portal_web_server_ota_warns(
    try:
        with caplog.at_level(logging.WARNING):
            ota_esphome_final_validate({})
-        assert any("captive_portal" in record.message for record in caplog.records)
+        assert not any(
+            "OTA encryption does not cover" in record.message
+            for record in caplog.records
+        )
        esphome_conf = next(
            conf
            for conf in fv.full_config.get()[CONF_OTA]
@@ -344,6 +350,93 @@ def test_encryption_with_captive_portal_web_server_ota_warns(
        fv.full_config.reset(token)


+@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: api_conf,
+        CONF_OTA: [_make_ota_config(port=3232, **{CONF_PASSWORD: "pw"})],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with caplog.at_level(logging.WARNING):
+            ota_esphome_final_validate({})
+        assert any("wastes significant flash" in r.message for r in caplog.records)
+    finally:
+        fv.full_config.reset(token)
+
+
+@pytest.mark.parametrize(
+    "api_conf",
+    [{}, {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]
+) -> None:
+    """Without a build-time api key there is no offer, so nothing to warn about."""
+    full_conf = {
+        CONF_API: api_conf,
+        CONF_OTA: [_make_ota_config(port=3232, **{CONF_PASSWORD: "pw"})],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with caplog.at_level(logging.WARNING):
+            ota_esphome_final_validate({})
+        assert not any("wastes significant flash" in r.message for r in caplog.records)
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_web_server_component_without_ota_platform_does_not_warn(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """The web_server component alone has no /update endpoint."""
+    full_conf = {
+        "web_server": {},
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}})
+        ],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with caplog.at_level(logging.WARNING):
+            ota_esphome_final_validate({})
+        assert not any(
+            "OTA encryption does not cover" in r.message for r in caplog.records
+        )
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_web_server_ota_platform_alone_warns(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """An explicit web_server ota platform exposes /update permanently, with
+    or without the web_server component."""
+    full_conf = {
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}}),
+            {CONF_PLATFORM: "web_server", CONF_ID: ID("ota_ws", is_manual=False)},
+        ],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with caplog.at_level(logging.WARNING):
+            ota_esphome_final_validate({})
+        assert any("plaintext /update" in r.message for r in caplog.records)
+    finally:
+        fv.full_config.reset(token)
+
+
 def test_web_server_ota_without_encryption_unaffected() -> None:
    """web_server ota stays valid alongside an unencrypted esphome entry."""
    full_conf = {
@@ -370,20 +463,70 @@ def test_auto_load_pulls_noise_only_for_encryption() -> None:
    assert "noise" in AUTO_LOAD({})


-def test_filter_source_files_excludes_noise_without_encryption() -> None:
-    """The noise transport source compiles only for encrypted builds."""
-    old_config = CORE.config
-    try:
-        CORE.config = {CONF_OTA: [_make_ota_config(port=3232)]}
-        assert FILTER_SOURCE_FILES() == ["ota_esphome_noise.cpp"]
-        CORE.config = {
-            CONF_OTA: [
-                _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: API_KEY}})
-            ]
-        }
-        assert FILTER_SOURCE_FILES() == []
-    finally:
-        CORE.config = old_config
+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
+    assert _api_static_key({CONF_ENCRYPTION: {}}) is None
+    assert _api_static_key({CONF_ENCRYPTION: {CONF_KEY: ZEROS_KEY}}) is None
+    assert _api_static_key({CONF_ENCRYPTION: {CONF_KEY: API_KEY}}) == API_KEY
+
+
+@pytest.mark.parametrize(
+    ("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", "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_FROM_API"},
+        ),
+        # The ota encryption block is what makes the device refuse plaintext
+        (
+            "encryption_required",
+            {"USE_OTA_ENCRYPTION", "USE_OTA_ENCRYPTION_REQUIRED"},
+            {"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"},
+        ),
+    ],
+)
+def test_encryption_offer_codegen(
+    generate_main: Callable[[str], str],
+    yaml_name: str,
+    defines_present: set[str],
+    defines_absent: set[str],
+) -> None:
+    main_cpp = generate_main(
+        f"tests/component_tests/ota/test_esphome_ota_{yaml_name}.yaml"
+    )
+    defines = {define.name for define in CORE.defines}
+    assert defines_present <= defines
+    assert not (defines_absent & defines)
+    encrypted = "USE_OTA_ENCRYPTION" in defines_present
+    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"])


 def test_password_with_encryption_rejected() -> None:
@@ -0,0 +1,11 @@
+esphome:
+  name: ota-offer
+
+host:
+
+api:
+  encryption:
+    key: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+
+ota:
+  - platform: esphome
@@ -0,0 +1,12 @@
+esphome:
+  name: ota-offer-password
+
+host:
+
+api:
+  encryption:
+    key: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+
+ota:
+  - platform: esphome
+    password: "superlongpasswordthatnoonewillknow"
@@ -0,0 +1,12 @@
+esphome:
+  name: ota-encryption-required
+
+host:
+
+api:
+  encryption:
+    key: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+
+ota:
+  - platform: esphome
+    encryption:
@@ -0,0 +1,10 @@
+esphome:
+  name: ota-runtime-key
+
+host:
+
+api:
+  encryption:
+
+ota:
+  - platform: esphome
@@ -68,6 +68,14 @@ class Initiator {

 static const uint8_t PROLOGUE[] = {'t', 'e', 's', 't', 'p', 'r', 'o', 'l', 'o', 'g', 'u', 'e'};

+// The context only points at the key and init() copies it before returning,
+// so a temporary context over a temporary key is safe within one call
+static NoiseContext ctx_for(const psk_t &psk) {
+  NoiseContext ctx;
+  ctx.set_psk(psk.data());
+  return ctx;
+}
+
 static psk_t make_psk(uint8_t seed) {
  psk_t psk;
  for (size_t i = 0; i < psk.size(); i++) {
@@ -102,7 +110,7 @@ TEST(NoiseResponderHandshakeTest, MessageMethodsErrorBeforeInit) {
 TEST(NoiseResponderHandshakeTest, FullHandshakeAndTransportRoundTrip) {
  const psk_t psk = make_psk(7);
  NoiseResponderHandshake responder;
-  ASSERT_EQ(responder.init(psk, PROLOGUE, sizeof(PROLOGUE)), 0);
+  ASSERT_EQ(responder.init(ctx_for(psk), PROLOGUE, sizeof(PROLOGUE)), 0);
  EXPECT_EQ(responder.action(), Action::ACTION_READ);

  Initiator initiator(psk, PROLOGUE, sizeof(PROLOGUE));
@@ -155,8 +163,8 @@ TEST(NoiseResponderHandshakeTest, ReInitRestartsHandshake) {
  // proves the restart took effect; the old state surviving would fail the
  // MAC here.
  NoiseResponderHandshake responder;
-  ASSERT_EQ(responder.init(make_psk(7), PROLOGUE, sizeof(PROLOGUE)), 0);
-  ASSERT_EQ(responder.init(make_psk(9), PROLOGUE, sizeof(PROLOGUE)), 0);
+  ASSERT_EQ(responder.init(ctx_for(make_psk(7)), PROLOGUE, sizeof(PROLOGUE)), 0);
+  ASSERT_EQ(responder.init(ctx_for(make_psk(9)), PROLOGUE, sizeof(PROLOGUE)), 0);
  EXPECT_EQ(responder.action(), Action::ACTION_READ);

  Initiator initiator(make_psk(9), PROLOGUE, sizeof(PROLOGUE));
@@ -168,7 +176,7 @@ TEST(NoiseResponderHandshakeTest, ReInitRestartsHandshake) {

 TEST(NoiseResponderHandshakeTest, WrongPskFailsWithMacFailure) {
  NoiseResponderHandshake responder;
-  ASSERT_EQ(responder.init(make_psk(7), PROLOGUE, sizeof(PROLOGUE)), 0);
+  ASSERT_EQ(responder.init(ctx_for(make_psk(7)), PROLOGUE, sizeof(PROLOGUE)), 0);

  Initiator initiator(make_psk(200), PROLOGUE, sizeof(PROLOGUE));
  uint8_t msg[MAX_HANDSHAKE_SIZE];
@@ -185,7 +193,7 @@ TEST(NoiseResponderHandshakeTest, MismatchedPrologueFailsWithMacFailure) {
  // tampered preamble must fail even with the right key.
  const psk_t psk = make_psk(7);
  NoiseResponderHandshake responder;
-  ASSERT_EQ(responder.init(psk, PROLOGUE, sizeof(PROLOGUE)), 0);
+  ASSERT_EQ(responder.init(ctx_for(psk), PROLOGUE, sizeof(PROLOGUE)), 0);

  static const uint8_t TAMPERED[] = {'x'};
  Initiator initiator(psk, TAMPERED, sizeof(TAMPERED));
@@ -17,12 +17,16 @@ TEST(NoiseContextTest, AllZerosPskIsReserved) {
  EXPECT_FALSE(NoiseContext::is_all_zeros(psk));

  NoiseContext ctx;
+  psk_t loaded;
  EXPECT_FALSE(ctx.has_psk());
-  ctx.set_psk(zeros);
+  ctx.load_psk(loaded);
+  EXPECT_EQ(loaded, zeros);
+  ctx.set_psk(zeros.data());
  EXPECT_FALSE(ctx.has_psk());
-  ctx.set_psk(psk);
+  ctx.set_psk(psk.data());
  EXPECT_TRUE(ctx.has_psk());
-  EXPECT_EQ(ctx.get_psk(), psk);
+  ctx.load_psk(loaded);
+  EXPECT_EQ(loaded, psk);
 }

 TEST(WireFormatTest, FrameHeaderIsIndicatorPlusBigEndianLength) {
@@ -0,0 +1,12 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+api:
+  encryption:
+    key: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+
+ota:
+  - platform: esphome
+    port: 3290
+    password: "superlongpasswordthatnoonewillknow"
@@ -0,0 +1,10 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+api:
+  encryption:
+
+ota:
+  - platform: esphome
+    port: 3291
@@ -0,0 +1,2 @@
+packages:
+  ota: !include api_key_offer.yaml
@@ -0,0 +1,2 @@
+packages:
+  ota: !include api_key_offer.yaml
@@ -0,0 +1,2 @@
+packages:
+  ota: !include api_runtime_key.yaml
@@ -0,0 +1,2 @@
+packages:
+  ota: !include api_runtime_key.yaml
@@ -0,0 +1,6 @@
+esphome:
+  name: zero-psk-provision-test
+host:
+api:
+  encryption:
+logger:
@@ -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
@@ -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
@@ -0,0 +1,10 @@
+esphome:
+  name: host-ota-test
+host:
+api:
+  encryption:
+ota:
+  - platform: esphome
+    port: __OTA_PORT__
+logger:
+  level: DEBUG
@@ -10,11 +10,19 @@ from __future__ import annotations

 import asyncio
 import base64
+import socket

 from aioesphomeapi import InvalidEncryptionKeyAPIError, RequiresEncryptionAPIError
 import pytest

-from .types import APIClientConnectedFactory, RunCompiledFunction
+from .conftest import run_binary_and_wait_for_port
+from .const import LOCALHOST
+from .types import (
+    APIClientConnectedFactory,
+    CompileFunction,
+    ConfigWriter,
+    RunCompiledFunction,
+)

 # The well-known provisioning PSK: base64 of 32 zero bytes
 ZERO_PSK = base64.b64encode(bytes(32)).decode()
@@ -125,3 +133,40 @@ async def test_api_zero_psk_provisioning_plaintext(
        with pytest.raises(InvalidEncryptionKeyAPIError):
            async with api_client_connected(noise_psk=ZERO_PSK, timeout=5) as client:
                await client.device_info()
+
+
+@pytest.mark.asyncio
+async def test_api_zero_psk_provisioning_persists_across_restart(
+    yaml_config: str,
+    write_yaml_config: ConfigWriter,
+    compile_esphome: CompileFunction,
+    reserved_tcp_port: tuple[int, socket.socket],
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """A key provisioned over the api is loaded from preferences on the next
+    boot, so the device comes back requiring that key."""
+    port, port_socket = reserved_tcp_port
+    config_path = await write_yaml_config(yaml_config)
+    binary_path = await compile_esphome(config_path)
+    port_socket.close()
+
+    async with (
+        run_binary_and_wait_for_port(binary_path, LOCALHOST, port),
+        api_client_connected(noise_psk=ZERO_PSK) as client,
+    ):
+        # The key is saved and synced before the response is sent
+        assert await client.noise_encryption_set_key(NEW_KEY) is True
+
+    lines: list[str] = []
+    async with run_binary_and_wait_for_port(
+        binary_path, LOCALHOST, port, line_callback=lines.append
+    ):
+        async with api_client_connected(noise_psk=NEW_KEY.decode()) as client:
+            device_info = await client.device_info()
+            assert device_info.api_encryption_provisionable is False
+
+        with pytest.raises(InvalidEncryptionKeyAPIError):
+            async with api_client_connected(noise_psk=ZERO_PSK, timeout=5) as client:
+                await client.device_info()
+
+    assert any("Loaded saved Noise PSK" in line for line in lines)
@@ -8,9 +8,12 @@ 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

 import pytest
@@ -19,9 +22,10 @@ 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="


@contextmanager
@@ -121,7 +125,6 @@ async def test_host_ota_encrypted(
 ) -> None:
    """Encrypted self-OTA succeeds; a plaintext upload to the same device fails."""
    pytest.importorskip("aioesphomeapi.noise")
-    noise_psk = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
    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))
@@ -142,25 +145,13 @@ async def test_host_ota_encrypted(
            pid_before = proc.pid

            # A plaintext upload must be refused with the device unharmed
-            rc, _ = await loop.run_in_executor(
-                None, espota2.run_ota, LOCALHOST, ota_port, None, binary_path
-            )
+            rc = await _run_ota(ota_port, None, binary_path, None)
            assert rc == 1, "plaintext upload to an encrypted device must fail"
            await asyncio.sleep(0.5)
            assert proc.returncode is None, "process died on rejected plaintext OTA"

            # The encrypted upload goes through and the device re-execs
-            rc, _ = await loop.run_in_executor(
-                None,
-                functools.partial(
-                    espota2.run_ota,
-                    LOCALHOST,
-                    ota_port,
-                    None,
-                    binary_path,
-                    noise_psk=noise_psk,
-                ),
-            )
+            rc = await _run_ota(ota_port, None, binary_path, API_KEY)
            assert rc == 0, "encrypted OTA reported failure"
            await asyncio.wait_for(rebooted, timeout=10.0)
            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
@@ -168,6 +159,217 @@ async def test_host_ota_encrypted(
            assert proc.pid == pid_before


+class _RebootCounter:
+    """Counts safe reboots so a test can wait for the nth one."""
+
+    def __init__(self) -> None:
+        self._seen = asyncio.Event()
+        self.count = 0
+
+    def on_log(self, line: str) -> None:
+        if "Rebooting safely" in line:
+            self.count += 1
+            self._seen.set()
+
+    async def wait(self, count: int, timeout: float = 10.0) -> None:
+        async with asyncio.timeout(timeout):
+            while self.count < count:
+                self._seen.clear()
+                await self._seen.wait()
+
+
+async def _run_ota(
+    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(
+        None,
+        functools.partial(
+            espota2.run_ota,
+            LOCALHOST,
+            ota_port,
+            password,
+            binary_path,
+            noise_psk=noise_psk,
+            plaintext_fallback=plaintext_fallback,
+        ),
+    )
+    return rc
+
+
+@pytest.mark.asyncio
+async def test_host_ota_api_key_offer_with_password(
+    yaml_config: str,
+    write_yaml_config: ConfigWriter,
+    compile_esphome: CompileFunction,
+    reserved_tcp_port: tuple[int, socket.socket],
+) -> None:
+    """With only an api key the device offers encryption without requiring
+    it: the password still guards plaintext uploads, and the key alone
+    authenticates an encrypted one, which is the enablement path for
+    `ota: encryption:`."""
+    pytest.importorskip("aioesphomeapi.noise")
+    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
+
+            rc = await _run_ota(ota_port, None, binary_path, None)
+            assert rc == 1, "plaintext upload without the password must fail"
+            await asyncio.sleep(0.5)
+            assert proc.returncode is None, "process died on rejected upload"
+
+            rc = await _run_ota(ota_port, "hunter2", binary_path, None)
+            assert rc == 0, "plaintext upload with the password must succeed"
+            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)
+            assert rc == 0, "encrypted upload with the api key must succeed"
+            await reboots.wait(2)
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            assert proc.returncode is None, "process exited instead of execing"
+            assert proc.pid == pid_before
+            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,
@@ -10,8 +10,10 @@ when the installed aioesphomeapi predates the noise module.
 from __future__ import annotations

 import base64
+from collections.abc import Callable
 import hashlib
 import io
+import logging
 from pathlib import Path
 import socket
 import sys
@@ -65,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
@@ -81,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:
@@ -109,8 +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 (self.offer_noise and noise_negotiated):
-            return  # the client fails closed; nothing further arrives
+        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

        from cryptography.exceptions import InvalidTag
        from noise.connection import NoiseConnection
@@ -149,6 +169,20 @@ class FakeEncryptedDevice(threading.Thread):
            assert len(plaintext) == length, "control units must be one per frame"
            return plaintext

+        def recv_data(_remaining: int) -> bytes:
+            plaintext = proto.decrypt(_recv_frame(sock))
+            assert 0 < len(plaintext) <= espota2.NOISE_MAX_PLAINTEXT
+            return plaintext
+
+        self._transfer(send_byte, recv_unit, recv_data)
+
+    def _transfer(
+        self,
+        send_byte: Callable[[int], None],
+        recv_unit: Callable[[int], bytes],
+        recv_data: Callable[[int], bytes],
+    ) -> None:
+        """The post-handshake exchange, identical over both transports."""
        send_byte(espota2.RESPONSE_AUTH_OK)
        recv_unit(1)  # ota type
        size = int.from_bytes(recv_unit(4), "big")
@@ -159,9 +193,7 @@ class FakeEncryptedDevice(threading.Thread):
        received = b""
        acked = 0
        while len(received) < size:
-            plaintext = proto.decrypt(_recv_frame(sock))
-            assert 0 < len(plaintext) <= espota2.NOISE_MAX_PLAINTEXT
-            received += plaintext
+            received += recv_data(size - len(received))
            if self.version >= espota2.OTA_VERSION_2_0:
                while acked + espota2.UPLOAD_BLOCK_SIZE <= len(received) or (
                    len(received) == size and acked < size
@@ -176,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)
@@ -184,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")
@@ -240,6 +298,63 @@ def test_client_fails_closed_when_device_lacks_encryption() -> None:
    device.join_and_check()


+# 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, firmware, PSK, plaintext_fallback=True)
+    device.join_and_check()
+    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"])
+def test_offering_device_accepts_either_transport(noise_psk: str | None) -> None:
+    """A device that offers but does not require encryption takes a plaintext
+    upload from a keyless client and an encrypted one from a keyed client."""
+    if noise_psk:
+        pytest.importorskip("aioesphomeapi.noise")
+    firmware = bytes(range(256)) * 40
+    device = FakeEncryptedDevice(offer_noise=True, require_noise=False)
+    with patch("time.sleep"):
+        _upload(device, firmware, noise_psk)
+    device.join_and_check()
+    assert device.received == firmware
+
+
 def test_plaintext_client_gets_encryption_required_error() -> None:
    """A client without a key gets the device's 0x94 error message."""
    device = FakeEncryptedDevice()
@@ -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,
    )


@@ -101,6 +101,25 @@ def test_config_file_should_include_ota(default_config: dict[str, Any]):
    assert "ota:" in config


+def test_config_file_should_use_encryption_when_api_key_set(
+    default_config: dict[str, Any],
+):
+    """
+    With an API encryption key and no OTA password the OTA block reuses the key
+    """
+    # Given
+    default_config["api_encryption_key"] = (
+        "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+    )
+
+    # When
+    config = wz.wizard_file(**default_config)
+
+    # Then
+    assert "ota:\n  - platform: esphome\n    encryption:" in config
+    assert "password" not in config.split("ota:")[1].split("wifi:")[0]
+
+
 def test_config_file_should_include_ota_when_password_set(
    default_config: dict[str, Any],
 ):
Author	SHA1	Message	Date
J. Nick Koston	034813ac25	Offer with a provisioned api key and let the CLI use the api key with a plaintext fallback until 2027.3.0	2026-09-05 14:22:27 +02:00
J. Nick Koston	5e6d74f170	Key the web_server warning off the OTA platform and tell plaintext uploaders about the offer	2026-09-05 13:50:46 +02:00
J. Nick Koston	7d5690b881	Only warn about the web_server update endpoint when web_server is configured	2026-09-05 13:24:19 +02:00
J. Nick Koston	1b99443a97	Guard the session start against a missing key, reword the warning, and let the wizard use the key for OTA	2026-09-05 12:56:06 +02:00
J. Nick Koston	e3e755850f	Put the measured password cost in the warning	2026-09-05 12:12:33 +02:00
J. Nick Koston	69870a8801	Say why the OTA password should go	2026-09-05 12:10:46 +02:00
J. Nick Koston	b4c6a40593	Warn that an api key supersedes the OTA password	2026-09-05 12:09:53 +02:00
J. Nick Koston	3601a1aad8	Use progmem_memcpy for the key and prologue copies and load the saved psk through a temp	2026-09-05 12:05:18 +02:00
J. Nick Koston	4f7ba7fa29	Read the noise handshake frame in one loop	2026-09-05 11:52:45 +02:00
J. Nick Koston	d6176bcb7d	Test that a key provisioned over the api survives a restart	2026-09-05 11:47:32 +02:00
J. Nick Koston	fad229a7cb	Keep the api and ota noise keys in flash behind one context	2026-09-05 11:45:59 +02:00
J. Nick Koston	5c047ba485	Keep the OTA key in flash and derive the extended protocol flag	2026-09-05 11:38:55 +02:00
J. Nick Koston	ef3b847a63	Pass the noise psk by reference and share the frame header check	2026-09-05 11:32:32 +02:00
J. Nick Koston	8d60f03ff3	Trim the noise glue and drive the source filter from the define	2026-09-05 11:28:00 +02:00
J. Nick Koston	029f6d4bc3	[ota] Offer encryption with the api key so enabling it works over OTA	2026-09-05 11:13:43 +02:00
dependabot[bot]anddependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	d1829c495d	Bump prek from 0.5.0 to 0.5.1 (#18977 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-09-04 19:05:36 -04:00
dependabot[bot]anddependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	ce87bf9b17	Bump platformdirs from 4.11.5 to 4.11.7 (#18976 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-09-04 19:05:26 -04:00
Jesse Hills	51ea97deff	[esp32_ble] Reference count BLE advertising (#18943 )	2026-09-05 08:38:55 +12:00