Merge branch 'dev' into ota-encryption-offer-api-key

Co-authored-by: pre-commit-ci-lite[bot] <117423508+pre-commit-ci-lite[bot]@users.noreply.github.com>
Co-authored-by: Claude <noreply@anthropic.com>

AGENTS.md

@@ -553,6 +553,7 @@ file does, and it is the authority when they disagree. The most useful starting
     4.  **Lint:** Run `prek` to ensure code is compliant.
     5.  **Commit:** Commit your changes. There is no strict format for commit messages.
     6.  **Pull Request:** Submit a PR against the `dev` branch. The Pull Request title must start with a `[tag]` prefix. For component work, use the component name (e.g., `[display] Fix bug`, `[abc123] Add new component`); for changes to shared/core code that isn't tied to a single component, use `[core]` (e.g., `[core] Add validator`). Update documentation, examples, and add `CODEOWNERS` entries as needed. Pull requests should always be made using the `.github/PULL_REQUEST_TEMPLATE.md` template - fill out all sections completely without removing any parts of the template.
+    7.  **Comments:** When commenting on GitHub PRs or issues, don't tag contributors, especially bots. Avoid referring to list items (e.g. from reviews) with the form #nn - this will be interpreted by GitHub as a reference to issue or PR nn. Keep comments short and exclude irrelevant details, backstories, restatement of previous comments and anything that is already obvious to the reader.
 
 *   **Documentation Contributions:**
     *   Documentation is hosted in the separate `esphome/esphome.io` repository.

THREAT_MODEL.md

@@ -125,30 +125,47 @@ 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. With a
+  runtime provisioned key and no `provisioning:` window, whoever provisions
+  the key gains that upload path too; validation warns about the pair.
+- 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. With the `web_server`
+  or `prometheus` component the shared listener is always up, so the endpoint
+  stays reachable and validation warns about that combination;
+  `captive_portal:` alone brings the listener up only for the fallback AP
+  window, which is the intended recovery path, so that 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.

docker/Dockerfile

@@ -22,7 +22,7 @@ RUN \
     -r /requirements.txt
 
 # Install the ESPHome Device Builder dashboard.
-RUN uv pip install --no-cache-dir esphome-device-builder==1.14.0
+RUN uv pip install --no-cache-dir esphome-device-builder==1.14.2
 
 RUN \
     platformio settings set enable_telemetry No \

esphome/__main__.py

@@ -1335,12 +1335,14 @@ def _upload_via_native_api(
             break
 
     from esphome import espota2
+    from esphome.components.noise import static_encryption_key
 
     remote_port = int(ota_conf[CONF_PORT])
     password = ota_conf.get(CONF_PASSWORD)
     # 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 +1353,10 @@ 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_encryption_key(config.get(CONF_API) or {}):
+        # Remove before 2027.3.0: the api key is tried, falling back to plaintext
+        noise_psk = str(api_key)
+        plaintext_fallback = True
 
     def check_partition_access(option_string: str) -> None:
         if not ota_conf.get("allow_partition_access"):
@@ -1382,7 +1388,13 @@ def _upload_via_native_api(
         _validate_bootloader_binary(binary)
 
     return espota2.run_ota(
-        network_devices, remote_port, password, binary, ota_type, noise_psk
+        network_devices,
+        remote_port,
+        password,
+        binary,
+        ota_type,
+        noise_psk,
+        plaintext_fallback=plaintext_fallback,
     )
 
 

esphome/components/api/__init__.py

@@ -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

esphome/components/api/api_connection.cpp

@@ -2161,7 +2161,10 @@ void APIConnection::on_homeassistant_action_response(const HomeassistantActionRe
 bool APIConnection::send_noise_encryption_set_key_response_(const NoiseEncryptionSetKeyRequest &msg) {
   NoiseEncryptionSetKeyResponse resp;
   resp.success = false;
-
+#ifdef USE_API_NOISE_PSK_FROM_YAML
+  // A yaml key cannot be changed at runtime, so no decode or save path is built
+  ESP_LOGW(TAG, "Key set in YAML");
+#else
 #ifdef USE_PROVISIONING
   // Refuse to set a key once the provisioning window has closed (defense in depth;
   // such connections are already rejected at hello).
@@ -2196,6 +2199,7 @@ bool APIConnection::send_noise_encryption_set_key_response_(const NoiseEncryptio
     }
 #endif
   }
+#endif  // USE_API_NOISE_PSK_FROM_YAML
 
   return this->send_message(resp);
 }

esphome/components/api/api_frame_helper_noise.cpp

@@ -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;

esphome/components/api/api_server.cpp

@@ -41,13 +41,13 @@ void APIServer::setup() {
   ControllerRegistry::register_controller(this);
 
 #ifdef USE_API_NOISE
+  // Always reserve the slot: flash preferences are positional on esp8266, so
+  // a yaml key build must keep the layout of a runtime key build
   uint32_t hash = 88491486UL;
-
   this->noise_pref_ = global_preferences->make_preference<SavedNoisePsk>(hash, true);
-
 #ifndef USE_API_NOISE_PSK_FROM_YAML
-  // Only load saved PSK if not set from YAML
-  if (this->load_and_apply_noise_psk_()) {
+  // A cleared record loads fine but holds no key
+  if (this->load_and_apply_noise_psk_() && this->noise_ctx_.has_psk()) {
     ESP_LOGD(TAG, "Loaded saved Noise PSK");
   }
 #endif
@@ -550,6 +550,7 @@ const std::vector<APIServer::HomeAssistantStateSubscription> &APIServer::get_sta
 #endif
 
 #ifdef USE_API_NOISE
+#ifndef USE_API_NOISE_PSK_FROM_YAML
 bool APIServer::update_noise_psk_(const SavedNoisePsk &new_psk, const LogString *save_log_msg,
                                   const LogString *fail_log_msg, bool make_active) {
   if (!this->noise_pref_.save(&new_psk)) {
@@ -583,22 +584,19 @@ 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))
+  // 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;
+  // An unprovisioned device stores the reserved all-zeros key, which is no key
+  const bool has_key = !noise::NoiseContext::is_all_zeros(this->saved_psk_.psk);
+  this->noise_ctx_.set_psk(has_key ? this->saved_psk_.psk.data() : nullptr);
   return true;
 }
 
 bool APIServer::save_noise_psk(noise::psk_t psk, bool make_active) {
-#ifdef USE_API_NOISE_PSK_FROM_YAML
-  // When PSK is set from YAML, this function should never be called
-  // but if it is, reject the change
-  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;
   }
@@ -614,15 +612,8 @@ bool APIServer::save_noise_psk(noise::psk_t psk, bool make_active) {
   }
 #endif
   return result;
-#endif
 }
 bool APIServer::clear_noise_psk(bool make_active) {
-#ifdef USE_API_NOISE_PSK_FROM_YAML
-  // When PSK is set from YAML, this function should never be called
-  // but if it is, reject the change
-  ESP_LOGW(TAG, "Key set in YAML");
-  return false;
-#else
   SavedNoisePsk empty_psk{};
   bool result = this->update_noise_psk_(empty_psk, LOG_STR("Noise PSK cleared"), LOG_STR("Failed to clear Noise PSK"),
                                         make_active);
@@ -634,8 +625,8 @@ bool APIServer::clear_noise_psk(bool make_active) {
   }
 #endif
   return result;
-#endif
 }
+#endif  // USE_API_NOISE_PSK_FROM_YAML
 #endif
 
 #ifdef USE_HOMEASSISTANT_TIME

esphome/components/api/api_server.h

@@ -76,9 +76,14 @@ class APIServer final : public Component,
   APIBuffer &get_shared_buffer_ref() { return shared_write_buffer_; }
 
 #ifdef USE_API_NOISE
+#ifndef USE_API_NOISE_PSK_FROM_YAML
+  // Runtime key changes exist for the provisioning path only (not lambdas);
+  // with a yaml key they compile out
   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); }
+#endif
+  /// 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
 
@@ -275,10 +280,12 @@ class APIServer final : public Component,
 #endif
 
 #ifdef USE_API_NOISE
+#ifndef USE_API_NOISE_PSK_FROM_YAML
   bool update_noise_psk_(const SavedNoisePsk &new_psk, const LogString *save_log_msg, const LogString *fail_log_msg,
                          bool make_active);
   // Load saved PSK from preferences and apply it. Returns true on success.
   bool load_and_apply_noise_psk_();
+#endif  // USE_API_NOISE_PSK_FROM_YAML
 #endif  // USE_API_NOISE
 #ifdef USE_API_HOMEASSISTANT_STATES
   // Helper methods to reduce code duplication
@@ -358,6 +365,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
 };

esphome/components/esp32_ble/ble.cpp

@@ -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
   }
 }
 

esphome/components/esp32_ble/ble.h

@@ -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_;

esphome/components/esp32_ble_beacon/esp32_ble_beacon.cpp

@@ -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_() {

esphome/components/esp32_ble_server/__init__.py

@@ -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]:

esphome/components/esp32_ble_server/ble_server.cpp

@@ -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

esphome/components/esp32_ble_server/ble_server.h

@@ -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};

esphome/components/esp32_improv/esp32_improv_component.cpp

@@ -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

esphome/components/esp32_improv/esp32_improv_component.h

@@ -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);

esphome/components/esphome/ota/__init__.py

@@ -2,12 +2,12 @@ import logging
 
 import esphome.codegen as cg
 from esphome.components.noise import (
-    decode_encryption_key,
     encryption_schema,
-    is_reserved_key,
+    new_psk_progmem,
+    static_encryption_key,
 )
 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,
@@ -31,7 +31,6 @@ import esphome.final_validate as fv
 from esphome.types import ConfigType
 
 CONF_ALLOW_PARTITION_ACCESS = "allow_partition_access"
-CONF_CAPTIVE_PORTAL = "captive_portal"
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -41,11 +40,10 @@ 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
+    inherits it from the api component."""
     base = ["sha256", "socket"]
-    # A falsy config is a tooling probe for the maximal set (None from
-    # dependency resolution, {} from the components-graph platform probe);
-    # a validated config always carries defaults, never empty
+    # A falsy config is a tooling probe for the maximal set
     if not config or CONF_ENCRYPTION in config:
         return base + ["noise"]
     return base
@@ -132,12 +130,56 @@ 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 static_encryption_key(api_conf) is not None:
+            _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,
+            )
+        elif (
+            CONF_PASSWORD in ota_conf
+            and CONF_ENCRYPTION in api_conf
+            and not api_conf[CONF_ENCRYPTION].get(CONF_KEY)
+        ):
+            # The CLI still needs the password; whoever provisions the key skips it
+            _LOGGER.warning(
+                "The '%s' %s %s provisioned at runtime also authenticates OTA "
+                "uploads once provisioned; '%s' %s then only guards plaintext "
+                "uploads. Whoever provisions the key can upload firmware "
+                "without the password, so add a 'provisioning:' block to limit "
+                "when that is possible",
+                CONF_API,
+                CONF_ENCRYPTION,
+                CONF_KEY,
+                CONF_OTA,
+                CONF_PASSWORD,
+            )
+    # web_server and prometheus keep the shared listener up; the captive
+    # portal's copy only exists on the fallback AP and is the recovery path
+    if (
+        (CONF_WEB_SERVER in full_conf or "prometheus" in full_conf)
+        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)
+        _LOGGER.warning(
+            "OTA encryption does not cover the %s OTA platform; its "
+            "plaintext /update endpoint accepts the same image",
+            CONF_WEB_SERVER,
+        )
 
     full_conf[CONF_OTA] = new_ota_conf
     fv.full_config.set(full_conf)
@@ -152,33 +194,11 @@ def ota_esphome_final_validate(config: ConfigType) -> None:
         )
 
 
-def _warn_web_server_ota(full_conf: ConfigType) -> 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,
-        )
-
-
 def _resolve_encryption_key(encryption_conf: ConfigType, api_conf: ConfigType) -> None:
     """Resolve the one encryption key per device into the ota block.
 
     An explicit ota key must match the api key, a bare block inherits it,
-    a runtime provisioned api key cannot be inherited, and the all-zeros
-    provisioning sentinel is rejected (the device treats it as no key).
+    a runtime provisioned api key cannot be inherited.
     """
     api_key = api_conf.get(CONF_ENCRYPTION, {}).get(CONF_KEY)
     if ota_key := encryption_conf.get(CONF_KEY):
@@ -201,11 +221,6 @@ def _resolve_encryption_key(encryption_conf: ConfigType, api_conf: ConfigType) -
         )
     else:
         encryption_conf[CONF_KEY] = api_key
-    if is_reserved_key(encryption_conf[CONF_KEY]):
-        raise cv.Invalid(
-            f"The all-zeros {CONF_KEY} is reserved and provides no protection; "
-            f"generate a real key with: openssl rand -base64 32"
-        )
 
 
 # Also called on merged same-port configs in final validate, where schemas
@@ -267,15 +282,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 +305,24 @@ async def to_code(config: ConfigType) -> None:
     if config.get(CONF_ALLOW_PARTITION_ACCESS):
         cg.add_define("USE_OTA_PARTITIONS")
 
-    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]
+    # One key per device: an api encryption block supplies it (static or
+    # runtime) and offers; the ota block only adds the requirement
+    api_conf = CORE.config.get(CONF_API) or {}
+    encryption_conf = config.get(CONF_ENCRYPTION)
+    own_key = None
+    if encryption_conf is not None and static_encryption_key(api_conf) is None:
+        own_key = encryption_conf[CONF_KEY]
+    if own_key is not None:
         cg.add_define("USE_OTA_ENCRYPTION")
-        cg.add(var.set_noise_psk(list(decode_encryption_key(key))))
+        cg.add(var.set_noise_psk(new_psk_progmem(config[CONF_ID], own_key)))
+    elif CONF_ENCRYPTION in api_conf:
+        cg.add_define("USE_OTA_ENCRYPTION")
+        cg.add_define("USE_OTA_ENCRYPTION_FROM_API")
+        if static_encryption_key(api_conf) is None:
+            # The key arrives at runtime, so the offer has to look for it
+            cg.add_define("USE_OTA_ENCRYPTION_PROVISIONED")
+    if encryption_conf is not None:
+        cg.add_define("USE_OTA_ENCRYPTION_REQUIRED")
 
     # Build flag so lwip_fast_select.c (a .c file that can't include defines.h) sees it.
     cg.add_build_flag("-DUSE_OTA_PLATFORM_ESPHOME")

esphome/components/esphome/ota/ota_esphome.cpp

@@ -1,4 +1,7 @@
 #include "ota_esphome.h"
+#ifdef USE_OTA_ENCRYPTION_FROM_API
+#include "esphome/components/api/api_server.h"
+#endif
 #ifdef USE_OTA
 #ifdef USE_OTA_PASSWORD
 #include "esphome/components/sha256/sha256.h"
@@ -26,6 +29,16 @@
 namespace esphome {
 
 static const char *const TAG = "esphome.ota";
+
+#ifdef USE_OTA_ENCRYPTION
+const noise::NoiseContext &ESPHomeOTAComponent::noise_context_() const {
+#ifdef USE_OTA_ENCRYPTION_FROM_API
+  return api::global_api_server->get_noise_ctx();
+#else
+  return this->noise_ctx_;
+#endif
+}
+#endif
 static constexpr uint16_t OTA_BLOCK_SIZE = 8192;
 static constexpr uint32_t OTA_SOCKET_TIMEOUT_HANDSHAKE = 20000;  // milliseconds for initial handshake
 static constexpr uint32_t OTA_SOCKET_TIMEOUT_DATA = 90000;       // milliseconds for data transfer
@@ -97,18 +110,30 @@ void ESPHomeOTAComponent::dump_config() {
   ESP_LOGCONFIG(TAG,
                 "Over-The-Air updates:\n"
                 "  Address: %s:%u\n"
-                "  Version: %d",
-                network::get_use_address_to(addr_buf), this->port_, USE_OTA_VERSION);
+                "  Version: %d"
+#ifdef USE_OTA_ENCRYPTION
+                "\n  Encryption: %s"
+#endif
+                ,
+                network::get_use_address_to(addr_buf), this->port_, USE_OTA_VERSION
+#ifdef USE_OTA_ENCRYPTION_REQUIRED
+                ,
+                LOG_STR_LITERAL("required")
+#elif defined(USE_OTA_ENCRYPTION_PROVISIONED)
+                // A runtime provisioned key may not exist yet
+                ,
+                this->noise_context_().has_psk() ? LOG_STR_LITERAL("offered, plaintext accepted")
+                                                 : LOG_STR_LITERAL("offered once the api key is provisioned")
+#elif defined(USE_OTA_ENCRYPTION)
+                ,
+                LOG_STR_LITERAL("offered, plaintext accepted")
+#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"
@@ -154,10 +179,22 @@ 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;
 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;
 
+inline bool ESPHomeOTAComponent::extended_proto_() const {
+#ifdef USE_OTA_ENCRYPTION_REQUIRED
+  // FEATURE_READ already refused every client without the extended protocol
+  return true;
+#else
+  return (this->ota_features_ & CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL) != 0;
+#endif
+}
+
 void ESPHomeOTAComponent::handle_handshake_() {
   /// Handle the OTA handshake and authentication.
   ///
@@ -241,12 +278,9 @@ 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
+      // `ota: encryption:` requires the client to negotiate encryption
+      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,18 +295,21 @@ 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);
 #ifdef USE_OTA_PARTITIONS
         this->handshake_buf_[1] |= SERVER_FEATURE_SUPPORTS_PARTITION_ACCESS;
 #endif
-#ifdef USE_OTA_ENCRYPTION
-        if (this->noise_ctx_.has_psk()) {
+#ifdef USE_OTA_ENCRYPTION_PROVISIONED
+        // A runtime provisioned key may not exist yet
+        if (this->noise_context_().has_psk()) {
           this->handshake_buf_[1] |= SERVER_FEATURE_SUPPORTS_NOISE;
         }
+#elif defined(USE_OTA_ENCRYPTION)
+        // A yaml key always exists: validation rejects the all-zeros key
+        this->handshake_buf_[1] |= SERVER_FEATURE_SUPPORTS_NOISE;
 #endif
       } else {
         this->handshake_buf_[0] =
@@ -284,15 +321,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()) {
+      // Latch the offer actually sent: a key activating between the two
+      // states must not start a session the client never expects
+      if ((this->handshake_buf_[1] & SERVER_FEATURE_SUPPORTS_NOISE) != 0 &&
+          (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 +449,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"));

esphome/components/esphome/ota/ota_esphome.h

@@ -44,8 +44,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 +86,12 @@ 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 api server's live context when the api has encryption, else our own
+  const noise::NoiseContext &noise_context_() const;
   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 +148,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 +172,8 @@ 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};
+  // Derived from the feature byte; storing it would pad the trailing bytes
+  bool extended_proto_() const;
 #ifdef USE_OTA_PARTITIONS
   uint32_t running_app_offset_{0};
   size_t running_app_size_{0};
@@ -179,7 +187,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

esphome/components/esphome/ota/ota_esphome_noise.cpp

@@ -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>
@@ -40,24 +41,17 @@ ESPHomeOTAComponent::NoiseSession::~NoiseSession() {
  *   "NoiseOTAInit" | magic(5) | OK,version | client_features | FEATURE_FLAGS,server_flags
  */
 bool ESPHomeOTAComponent::noise_start_session_(uint8_t server_feature_flags) {
+  // A provisioned key cleared between the offer and here is not guarded: the
+  // session runs on the zero key load_psk fills in and fails the client's MAC.
+  // Default-init: the frame buffer is written before it is read
   // 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;
-  }
-
+  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 +65,13 @@ 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));
+  // The caller only starts a session when the context holds a key
+  int err = this->noise_ == nullptr ? NOISE_ERROR_NO_MEMORY
+                                    : this->noise_->handshake.init(this->noise_context_(), prologue, sizeof(prologue));
   if (err != 0) {
-    ESP_LOGW(TAG, "Handshake init: %s", LOG_STR_ARG(noise::noise_err_to_logstr(err)));
+    // Raw noise codes throughout: the name table would cost flash in builds
+    // where only the OTA uses noise
+    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_LOGW(TAG, "Client rejected the handshake: %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);

esphome/components/noise/__init__.py

@@ -4,7 +4,9 @@ from typing import Any
 
 import esphome.codegen as cg
 import esphome.config_validation as cv
-from esphome.const import CONF_KEY
+from esphome.const import CONF_ENCRYPTION, CONF_KEY
+from esphome.core import ID
+from esphome.cpp_generator import MockObj
 from esphome.types import ConfigType
 
 CODEOWNERS = ["@esphome/core"]
@@ -23,6 +25,14 @@ def validate_encryption_key(value: Any) -> str:
 
     if len(decoded) != 32:
         raise cv.Invalid("Encryption key must be base64 and 32 bytes long")
+    if not any(decoded):
+        # The device treats the all-zeros key as no key at all (it is the
+        # provisioning sentinel), so it must never reach a build
+        raise cv.Invalid(
+            f"The all-zeros {CONF_KEY} is reserved and provides no protection; "
+            f"omit the {CONF_KEY} to provision it at runtime, or generate a real "
+            "key with: openssl rand -base64 32"
+        )
 
     # Return original data for roundtrip conversion
     return value
@@ -45,15 +55,6 @@ def decode_encryption_key(value: str) -> bytes:
     return decoded
 
 
-def is_reserved_key(value: str) -> bool:
-    """Whether the key is the reserved all-zeros provisioning sentinel.
-
-    The device treats it as no key configured, so consumers that require a
-    real key must reject it.
-    """
-    return not any(decode_encryption_key(value))
-
-
 ENCRYPTION_SCHEMA = cv.Schema(
     {
         cv.Optional(CONF_KEY): cv.sensitive(validate_encryption_key),
@@ -61,6 +62,21 @@ ENCRYPTION_SCHEMA = cv.Schema(
 )
 
 
+def static_encryption_key(conf: ConfigType) -> str | None:
+    """The build time key of a component config; None without one or when
+    the key is provisioned at runtime."""
+    return (conf.get(CONF_ENCRYPTION) or {}).get(CONF_KEY) or None
+
+
+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

esphome/components/noise/noise.cpp

@@ -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,14 @@ namespace esphome::noise {
 
 static const char *const TAG = "noise";
 
+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");

esphome/components/noise/noise.h

@@ -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 (PROGMEM or caller owned
+  /// RAM); nullptr means no key. Runtime callers map the all-zeros key to
+  /// nullptr themselves; validation keeps it out of yaml.
+  void set_psk(const uint8_t *psk) { this->psk_ = 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

esphome/components/noise/noise_handshake.cpp

@@ -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);

esphome/components/noise/noise_handshake.h

@@ -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;

esphome/components/usb_uart/usb_uart.cpp

@@ -434,11 +434,12 @@ void USBUartTypeCdcAcm::on_connected() {
       auto err_comm = usb_host_interface_claim(this->handle_, this->device_handle_,
                                                channel->cdc_dev_.interrupt_interface_number, 0);
       if (err_comm != ESP_OK) {
+        // Continue anyway: the interface number stays valid for CDC request addressing
         ESP_LOGW(TAG, "Could not claim comm interface %d: %s", channel->cdc_dev_.interrupt_interface_number,
                  esp_err_to_name(err_comm));
-        channel->cdc_dev_.interrupt_interface_number = 0xFF;  // Mark as unavailable, but continue anyway
       } else {
         ESP_LOGD(TAG, "Claimed comm interface %d", channel->cdc_dev_.interrupt_interface_number);
+        channel->cdc_dev_.interrupt_interface_claimed = true;
       }
     }
     auto err =
@@ -465,14 +466,15 @@ void USBUartTypeCdcAcm::on_disconnected() {
       usb_host_endpoint_halt(this->device_handle_, channel->cdc_dev_.out_ep->bEndpointAddress);
       usb_host_endpoint_flush(this->device_handle_, channel->cdc_dev_.out_ep->bEndpointAddress);
     }
-    if (channel->cdc_dev_.notify_ep != nullptr) {
+    // Only tear down the notify pipe when we claimed its interface ourselves;
+    // no transfer is ever submitted on it, so there is nothing else to cancel.
+    if (channel->cdc_dev_.notify_ep != nullptr && channel->cdc_dev_.interrupt_interface_claimed) {
       usb_host_endpoint_halt(this->device_handle_, channel->cdc_dev_.notify_ep->bEndpointAddress);
       usb_host_endpoint_flush(this->device_handle_, channel->cdc_dev_.notify_ep->bEndpointAddress);
     }
-    if (channel->cdc_dev_.interrupt_interface_number != 0xFF &&
-        channel->cdc_dev_.interrupt_interface_number != channel->cdc_dev_.bulk_interface_number) {
+    if (channel->cdc_dev_.interrupt_interface_claimed) {
       usb_host_interface_release(this->handle_, this->device_handle_, channel->cdc_dev_.interrupt_interface_number);
-      channel->cdc_dev_.interrupt_interface_number = 0xFF;
+      channel->cdc_dev_.interrupt_interface_claimed = false;
     }
     usb_host_interface_release(this->handle_, this->device_handle_, channel->cdc_dev_.bulk_interface_number);
     // Reset the input and output started flags to their initial state to avoid the possibility of spurious restarts

esphome/components/usb_uart/usb_uart.h

@@ -34,7 +34,10 @@ struct CdcEps {
   const usb_ep_desc_t *in_ep;
   const usb_ep_desc_t *out_ep;
   uint8_t bulk_interface_number;
+  // Also the wIndex target for CDC class requests (SET_LINE_CODING etc.), so it
+  // must remain valid even when the interface itself is not claimed.
   uint8_t interrupt_interface_number;
+  bool interrupt_interface_claimed{false};
 };
 
 enum CH34xChipType : uint8_t {

esphome/components/wifi/__init__.py

@@ -66,13 +66,14 @@ from esphome.const import (
 )
 from esphome.core import (
     CORE,
+    ID,
     CoroPriority,
     EsphomeError,
     HexInt,
     coroutine_with_priority,
 )
 import esphome.final_validate as fv
-from esphome.types import ConfigType
+from esphome.types import ConfigType, TemplateArgsType
 
 from . import wpa2_eap
 
@@ -208,6 +209,7 @@ WiFiEnabledCondition = wifi_ns.class_("WiFiEnabledCondition", Condition)
 WiFiAPActiveCondition = wifi_ns.class_("WiFiAPActiveCondition", Condition)
 WiFiEnableAction = wifi_ns.class_("WiFiEnableAction", automation.Action)
 WiFiDisableAction = wifi_ns.class_("WiFiDisableAction", automation.Action)
+WiFiRoamAction = wifi_ns.class_("WiFiRoamAction", automation.Action)
 WiFiConfigureAction = wifi_ns.class_(
     "WiFiConfigureAction", automation.Action, cg.Component
 )
@@ -820,6 +822,18 @@ async def wifi_disable_to_code(config, action_id, template_arg, args):
     return cg.new_Pvariable(action_id, template_arg)
 
 
+@automation.register_action(
+    "wifi.roam", WiFiRoamAction, cv.Schema({}), synchronous=True
+)
+async def wifi_roam_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> cg.MockObj:
+    return cg.new_Pvariable(action_id, template_arg)
+
+
 KEEP_SCAN_RESULTS_KEY = "wifi_keep_scan_results"
 RUNTIME_POWER_SAVE_KEY = "wifi_runtime_power_save"
 RUNTIME_ROAMING_SUPPRESSION_KEY = "wifi_runtime_roaming_suppression"

esphome/components/wifi/automation.h

@@ -31,6 +31,11 @@ template<typename... Ts> class WiFiDisableAction final : public Action<Ts...> {
   void play(const Ts &...x) override { global_wifi_component->disable(); }
 };
 
+template<typename... Ts> class WiFiRoamAction final : public Action<Ts...> {
+ public:
+  void play(const Ts &...x) override { global_wifi_component->force_roam_check(); }
+};
+
 template<typename... Ts> class WiFiConfigureAction final : public Action<Ts...>, public Component {
  public:
   TEMPLATABLE_VALUE(std::string, ssid)

esphome/components/wifi/wifi_component.cpp

@@ -846,17 +846,18 @@ void WiFiComponent::loop() {
           this->notify_connect_state_listeners_();
 #endif
 
-          // Post-connect roaming: check for better AP
-          if (this->post_connect_roaming_) {
-            if (this->is_roaming_scan_active()) {
-              if (this->scan_done_) {
-                this->process_roaming_scan_();
-              }
-              // else: scan in progress, wait
-            } else if (this->roaming_state_ == RoamingState::IDLE && this->roaming_attempts_ < ROAMING_MAX_ATTEMPTS &&
-                       now - this->roaming_last_check_ >= ROAMING_CHECK_INTERVAL && !this->roaming_suppressed_()) {
-              this->check_roaming_(now);
+          // Post-connect roaming: check for better AP. A scan may have been started by an
+          // explicit force_roam_check() even when post_connect_roaming_ is disabled, so the
+          // scan must always be consumed here to avoid leaving roaming_state_ stuck.
+          if (this->is_roaming_scan_active()) {
+            if (this->scan_done_) {
+              this->process_roaming_scan_();
             }
+            // else: scan in progress, wait
+          } else if (this->post_connect_roaming_ && this->roaming_state_ == RoamingState::IDLE &&
+                     this->roaming_attempts_ < ROAMING_MAX_ATTEMPTS &&
+                     now - this->roaming_last_check_ >= ROAMING_CHECK_INTERVAL && !this->roaming_suppressed_()) {
+            this->check_roaming_(now);
           }
         }
         break;
@@ -2463,6 +2464,17 @@ void WiFiComponent::notify_scan_results_listeners_() {
 }
 #endif  // USE_WIFI_SCAN_RESULTS_LISTENERS
 
+void WiFiComponent::force_roam_check() {
+  if (!this->is_connected() || this->roaming_state_ != RoamingState::IDLE || this->roaming_suppressed_()) {
+    ESP_LOGD(TAG, "Roam check requested, but not able to check now");
+    return;
+  }
+  // Reset the attempt counter so a prior run of failed roams doesn't block this explicit request
+  // Note that this re-arms automatic roaming if enabled.
+  this->roaming_attempts_ = 0;
+  this->check_roaming_(millis());
+}
+
 void WiFiComponent::check_roaming_(uint32_t now) {
   // Guard: not for hidden networks (may not appear in scan)
   const WiFiAP *selected = this->get_selected_sta_();
@@ -2484,7 +2496,11 @@ void WiFiComponent::check_roaming_(uint32_t now) {
 
   ESP_LOGD(TAG, "Roam scan (%d dBm, attempt %u/%u)", rssi, this->roaming_attempts_, ROAMING_MAX_ATTEMPTS);
   this->roaming_state_ = RoamingState::SCANNING;
-  this->wifi_scan_start_(this->passive_scan_);
+  if (!this->wifi_scan_start_(this->passive_scan_)) {
+    // Scan failed to start (e.g. busy) - don't get stuck in SCANNING forever
+    ESP_LOGD(TAG, "Roam scan failed to start");
+    this->roaming_state_ = RoamingState::IDLE;
+  }
 }
 
 void WiFiComponent::process_roaming_scan_() {

esphome/components/wifi/wifi_component.h

@@ -565,6 +565,12 @@ class WiFiComponent final : public Component {
   void set_keep_scan_results(bool keep_scan_results) { this->keep_scan_results_ = keep_scan_results; }
   void set_post_connect_roaming(bool enabled) { this->post_connect_roaming_ = enabled; }
 
+  /** Force an immediate post-connect roaming check, bypassing the periodic interval and the
+   * per-connection attempt limit. Does nothing (besides a debug log) if not connected, if a
+   * roam scan or connect is already in progress, or if roaming is currently suppressed.
+   */
+  void force_roam_check();
+
 #ifdef USE_WIFI_CONNECT_TRIGGER
   Trigger<> *get_connect_trigger() { return &this->connect_trigger_; }
 #endif

esphome/core/defines.h

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

esphome/espota2.py

@@ -202,6 +202,49 @@ 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."""
+
+
+# Remove before 2027.3.0
+PLAINTEXT_FALLBACK_NOTICE = (
+    "A device with an api encryption key offers encryption after this "
+    "install; add 'encryption:' under 'ota: platform: esphome' to require it. "
+    "This plaintext fallback is removed in 2027.3.0."
+)
+
+
+# Remove before 2027.3.0
+class _EncryptionAttempt:
+    """The key an upload tries and whether it may fall back to plaintext;
+    a rejected handshake falls back at once, a transport fault only on repeat."""
+
+    def __init__(self, noise_psk: str | None, plaintext_fallback: bool) -> None:
+        self.noise_psk = noise_psk
+        self.plaintext_fallback = plaintext_fallback
+        self.handshake_faults = 0
+
+    def handshake_fault_falls_back(self) -> bool:
+        self.handshake_faults += 1
+        return self.plaintext_fallback and self.handshake_faults >= 2
+
+    def downgrade(self, reason: str) -> None:
+        _LOGGER.warning(
+            "%s. Retrying in plaintext; a device that requires encryption "
+            "refuses it. %s",
+            reason,
+            PLAINTEXT_FALLBACK_NOTICE,
+        )
+        self.noise_psk = None
+        self.plaintext_fallback = False
+
+
+# Remove before 2027.3.0: only the fallback decision needs this distinction
+class OTAHandshakeNetworkError(OTANetworkError):
+    """A transport failure inside the noise handshake; retrying encrypted may succeed."""
+
+
 def _committed_error(err: OTANetworkError) -> OTAError:
     """Wrap a network failure that happened once the device had the full image.
 
@@ -464,6 +507,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 +572,28 @@ 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: older firmware that cannot encrypt still
+            # gets its update on this connection
+            _LOGGER.warning(
+                "The device did not offer OTA encryption; continuing in plaintext. %s",
+                PLAINTEXT_FALLBACK_NOTICE,
+            )
+            noise_psk = None
+        else:
+            # Fail closed: an attacker could otherwise strip the offer and
+            # capture the image (wifi credentials, api 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 predates ESPHome 2026.9.0 or has no "
+                "'api: encryption: key'. With an api key, install once "
+                "without the 'ota: encryption:' block (that build offers "
+                "encryption), then restore it; 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 +602,18 @@ def perform_ota(
             + bytes([RESPONSE_OK, version, features_to_send])
             + bytes([RESPONSE_FEATURE_FLAGS, features])
         )
+        # Built outside the try: a local failure must never downgrade the upload
         sock = NoiseSocketWrapper(sock, noise_psk, prologue)
-        sock.do_handshake()
+        try:
+            sock.do_handshake()
+        except OTANetworkError as err:
+            # A transport fault: retry encrypted before considering plaintext
+            raise OTAHandshakeNetworkError(str(err)) from err
+        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 +820,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,7 +859,9 @@ def run_ota_impl_(
     total_attempts = len(res) + EXTRA_UPLOAD_ATTEMPTS
     last_error = ""
     reached_device = False
-    for attempt in range(total_attempts):
+    attempt = 0
+    encryption = _EncryptionAttempt(noise_psk, plaintext_fallback)
+    while attempt < total_attempts:
         af, socktype, _, _, sa = res[attempt % len(res)]
         if reached_device or attempt >= len(res):
             _LOGGER.info(
@@ -815,17 +881,40 @@ 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
             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,
+                    encryption.noise_psk,
+                    encryption.plaintext_fallback,
+                )
+            except OTAEncryptionFallback as err:
+                # Same address and attempt budget: not a network retry
+                last_error = str(err)
+                encryption.downgrade(last_error)
+                continue
+            except OTAHandshakeNetworkError as err:
+                last_error = str(err)
+                if encryption.handshake_fault_falls_back():
+                    encryption.downgrade(last_error)
+                    continue
+                _LOGGER.warning("%s", last_error)
+                attempt += 1
+                continue
             except OTANetworkError as err:
                 # Transient network failure; retry
                 last_error = str(err)
                 _LOGGER.warning("%s", last_error)
+                attempt += 1
                 continue
             except OTAError as err:
                 # Device-reported error (wrong password, wrong flash size, ...);
@@ -847,10 +936,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)

esphome/wizard.py

@@ -148,11 +148,13 @@ 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
+    # The api key also secures OTA; a password only serves older uploaders
     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"
@@ -529,20 +531,9 @@ def wizard(path: Path) -> int:
         safe_print()
         safe_print("You'll need this key when adding the device to Home Assistant.")
         sleep(1)
-
-        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."
-        )
-        safe_print()
-        sleep(0.25)
-        safe_print("Press ENTER for no password")
-        ota_password = safe_input(color(AnsiFore.BOLD_WHITE, "(password): "))
     else:
         ssid, psk = "", ""
         api_encryption_key = None
-        ota_password = ""
 
     kwargs = {
         "path": path,
@@ -553,10 +544,9 @@ def wizard(path: Path) -> int:
         "psk": psk,
         "type": "basic",
     }
+    # The api key also secures OTA updates, so the wizard sets no OTA password
     if api_encryption_key:
         kwargs["api_encryption_key"] = api_encryption_key
-    if ota_password:
-        kwargs["ota_password"] = ota_password
 
     if not wizard_write(**kwargs):
         return 1

requirements.txt

@@ -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
 

requirements_test.txt

@@ -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

tests/component_tests/esp32_ble_server/config/improv_only.yaml

@@ -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

tests/component_tests/esp32_ble_server/config/manufacturer_data_only.yaml

@@ -0,0 +1,9 @@
+esphome:
+  name: test
+
+esp32:
+  variant: esp32
+
+esp32_ble_server:
+  id: ble_server
+  manufacturer_data: [0x72, 0x04, 0x00, 0x23]

tests/component_tests/esp32_ble_server/config/own_service.yaml

@@ -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]

tests/component_tests/esp32_ble_server/test_esp32_ble_server.py

@@ -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

tests/component_tests/noise/test_encryption_key.py

@@ -5,11 +5,7 @@ from __future__ import annotations
 import pytest
 
 from esphome import config_validation as cv
-from esphome.components.noise import (
-    decode_encryption_key,
-    is_reserved_key,
-    validate_encryption_key,
-)
+from esphome.components.noise import decode_encryption_key, validate_encryption_key
 
 KEY = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
 
@@ -41,6 +37,8 @@ def test_decode_encryption_key_rejects_short_decode() -> None:
         decode_encryption_key("AAECAw==")
 
 
-def test_is_reserved_key() -> None:
-    assert is_reserved_key("AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=")
-    assert not is_reserved_key(KEY)
+def test_validate_encryption_key_rejects_all_zeros() -> None:
+    """The all-zeros key is the provisioning sentinel the device treats as no
+    key, so it never reaches a build."""
+    with pytest.raises(cv.Invalid, match="all-zeros key is reserved"):
+        validate_encryption_key("AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=")

tests/component_tests/ota/test_esphome_ota.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from collections.abc import Callable
 import logging
 from typing import Any
 
@@ -14,6 +15,7 @@ from esphome.components.esphome.ota import (
     _validate_no_password_with_encryption,
     ota_esphome_final_validate,
 )
+from esphome.components.noise import static_encryption_key
 from esphome.const import (
     CONF_API,
     CONF_ENCRYPTION,
@@ -115,7 +117,6 @@ def test_non_esphome_ota_unaffected() -> None:
 
 API_KEY = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
 OTHER_KEY = "AQIDBAUGBwgJCgsMDQ4PEBESExQVFhcYGRobHB0eHyA="
-ZEROS_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
 
 
 def test_encryption_key_inherited_from_api() -> None:
@@ -197,36 +198,6 @@ def test_encryption_without_any_key_rejected() -> None:
         fv.full_config.reset(token)
 
 
-def test_encryption_explicit_all_zeros_key_rejected() -> None:
-    """The all-zeros key is the provisioning sentinel; the device would treat
-    it as no PSK and accept plaintext, so it must fail validation."""
-    full_conf = {
-        CONF_OTA: [
-            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: ZEROS_KEY}})
-        ],
-    }
-    token = fv.full_config.set(full_conf)
-    try:
-        with pytest.raises(cv.Invalid, match="all-zeros key is reserved"):
-            ota_esphome_final_validate({})
-    finally:
-        fv.full_config.reset(token)
-
-
-def test_encryption_inherited_all_zeros_key_rejected() -> None:
-    """An all-zeros api key must not silently disable ota encryption either."""
-    full_conf = {
-        CONF_API: {CONF_ENCRYPTION: {CONF_KEY: ZEROS_KEY}},
-        CONF_OTA: [_make_ota_config(port=3232, **{CONF_ENCRYPTION: {}})],
-    }
-    token = fv.full_config.set(full_conf)
-    try:
-        with pytest.raises(cv.Invalid, match="all-zeros key is reserved"):
-            ota_esphome_final_validate({})
-    finally:
-        fv.full_config.reset(token)
-
-
 def test_encryption_key_mismatch_between_merged_configs_rejected() -> None:
     """Same-port configs with different encryption keys raise."""
     full_conf = {
@@ -295,13 +266,14 @@ def test_encryption_explicit_key_with_runtime_provisioned_api_accepted() -> None
         fv.full_config.reset(token)
 
 
+@pytest.mark.parametrize("component", ["web_server", "prometheus"])
 def test_encryption_with_web_server_ota_warns(
-    caplog: pytest.LogCaptureFixture,
+    caplog: pytest.LogCaptureFixture, component: str
 ) -> None:
-    """With the web_server component the plaintext /update endpoint is always
-    on; the combination validates with a warning."""
+    """web_server and prometheus keep the shared listener up, so the
+    plaintext /update endpoint is always on and the combination warns."""
     full_conf = {
-        "web_server": {},
+        component: {},
         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)},
@@ -316,12 +288,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 +305,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 +319,100 @@ def test_encryption_with_captive_portal_web_server_ota_warns(
         fv.full_config.reset(token)
 
 
+def test_password_with_api_key_warns(caplog: pytest.LogCaptureFixture) -> None:
+    """A static api key makes the device offer encryption and the CLI take
+    it, so the password is dead weight; the config validates with a warning."""
+    full_conf = {
+        CONF_API: {CONF_ENCRYPTION: {CONF_KEY: API_KEY}},
+        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)
+
+
+def test_password_with_runtime_api_key_warns_differently(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """The CLI still needs the password, but the provisioned key also
+    authenticates uploads; the warning says so without the flash advice."""
+    full_conf = {
+        CONF_API: {CONF_ENCRYPTION: {}},
+        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({})
+        messages = [r.message for r in caplog.records]
+        assert any("provisioned at runtime also authenticates" in m for m in messages)
+        assert not any("wastes significant flash" in m for m in messages)
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_password_without_api_key_no_warning(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Without an api key there is no offer, so nothing to warn about."""
+    full_conf = {
+        CONF_API: {},
+        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("authenticates" 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_does_not_warn(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Only the web_server component starts the shared listener, so the ota
+    platform on its own never exposes /update."""
+    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 not 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 +439,87 @@ 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_static_encryption_key() -> None:
+    """Only a build-time key counts; a runtime provisioned one does not."""
+    assert static_encryption_key({}) is None
+    assert static_encryption_key({CONF_ENCRYPTION: {}}) is None
+    assert static_encryption_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;
+        # the device uses the api server's key, not a copy
+        (
+            "api_key_offer",
+            {"USE_OTA_ENCRYPTION", "USE_OTA_ENCRYPTION_FROM_API"},
+            {"USE_OTA_ENCRYPTION_REQUIRED", "USE_OTA_ENCRYPTION_PROVISIONED"},
+        ),
+        # A password still guards plaintext uploads on an offering device
+        (
+            "api_key_offer_password",
+            {"USE_OTA_ENCRYPTION", "USE_OTA_ENCRYPTION_FROM_API", "USE_OTA_PASSWORD"},
+            {"USE_OTA_ENCRYPTION_REQUIRED", "USE_OTA_ENCRYPTION_PROVISIONED"},
+        ),
+        # 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",
+            },
+            {"USE_OTA_ENCRYPTION_PROVISIONED"},
+        ),
+        # Without api encryption the ota key is the device's own
+        (
+            "own_key",
+            {"USE_OTA_ENCRYPTION", "USE_OTA_ENCRYPTION_REQUIRED"},
+            {"USE_OTA_ENCRYPTION_FROM_API", "USE_OTA_ENCRYPTION_PROVISIONED"},
+        ),
+        # 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_PROVISIONED",
+            },
+            {"USE_OTA_ENCRYPTION_REQUIRED"},
+        ),
+        # No api encryption at all keeps the noise glue out of the build
+        (
+            "plain",
+            set(),
+            {
+                "USE_OTA_ENCRYPTION",
+                "USE_OTA_ENCRYPTION_REQUIRED",
+                "USE_OTA_ENCRYPTION_FROM_API",
+                "USE_OTA_ENCRYPTION_PROVISIONED",
+            },
+        ),
+    ],
+)
+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:

tests/component_tests/ota/test_esphome_ota_api_key_offer.yaml

@@ -0,0 +1,11 @@
+esphome:
+  name: ota-offer
+
+host:
+
+api:
+  encryption:
+    key: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+
+ota:
+  - platform: esphome

tests/component_tests/ota/test_esphome_ota_api_key_offer_password.yaml

@@ -0,0 +1,12 @@
+esphome:
+  name: ota-offer-password
+
+host:
+
+api:
+  encryption:
+    key: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+
+ota:
+  - platform: esphome
+    password: "superlongpasswordthatnoonewillknow"

tests/component_tests/ota/test_esphome_ota_encryption_required.yaml

@@ -0,0 +1,12 @@
+esphome:
+  name: ota-encryption-required
+
+host:
+
+api:
+  encryption:
+    key: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+
+ota:
+  - platform: esphome
+    encryption:

tests/component_tests/ota/test_esphome_ota_own_key.yaml

@@ -0,0 +1,11 @@
+esphome:
+  name: ota-own-key
+
+host:
+
+api:
+
+ota:
+  - platform: esphome
+    encryption:
+      key: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="

tests/component_tests/ota/test_esphome_ota_plain.yaml

@@ -0,0 +1,9 @@
+esphome:
+  name: ota-plain
+
+host:
+
+api:
+
+ota:
+  - platform: esphome

tests/component_tests/ota/test_esphome_ota_runtime_api_key.yaml

@@ -0,0 +1,10 @@
+esphome:
+  name: ota-runtime-key
+
+host:
+
+api:
+  encryption:
+
+ota:
+  - platform: esphome

tests/components/noise/test_noise_handshake.cpp

@@ -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));

tests/components/noise/test_noise_primitives.cpp

@@ -17,12 +17,17 @@ TEST(NoiseContextTest, AllZerosPskIsReserved) {
   EXPECT_FALSE(NoiseContext::is_all_zeros(psk));
 
   NoiseContext ctx;
+  psk_t loaded;
   EXPECT_FALSE(ctx.has_psk());
-  ctx.set_psk(zeros);
-  EXPECT_FALSE(ctx.has_psk());
-  ctx.set_psk(psk);
+  ctx.load_psk(loaded);
+  EXPECT_EQ(loaded, zeros);
+  ctx.set_psk(psk.data());
   EXPECT_TRUE(ctx.has_psk());
-  EXPECT_EQ(ctx.get_psk(), psk);
+  ctx.load_psk(loaded);
+  EXPECT_EQ(loaded, psk);
+  // Callers map the reserved key to nullptr; the context just stores what it is given
+  ctx.set_psk(nullptr);
+  EXPECT_FALSE(ctx.has_psk());
 }
 
 TEST(WireFormatTest, FrameHeaderIsIndicatorPlusBigEndianLength) {

tests/components/ota/api_key_offer.yaml

@@ -0,0 +1,12 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+api:
+  encryption:
+    key: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+
+ota:
+  - platform: esphome
+    port: 3290
+    password: "superlongpasswordthatnoonewillknow"

tests/components/ota/api_runtime_key.yaml

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

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

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

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

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

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

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

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

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

tests/components/wifi/common.yaml

@@ -14,6 +14,7 @@ esphome:
           condition: wifi.ap_active
           then:
             - logger.log: "WiFi AP is active!"
+      - wifi.roam
 
 wifi:
   networks:

tests/integration/conftest.py

@@ -162,6 +162,13 @@ def integration_test_dir() -> Generator[Path]:
         yield Path(tmpdir)
 
 
+@pytest.fixture
+def isolated_preferences(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    """Host preferences persist per device name; give the test its own so a
+    provisioned key never leaks into another run."""
+    monkeypatch.setenv("ESPHOME_PREFDIR", str(tmp_path / "prefs"))
+
+
 @pytest.fixture
 def reserved_tcp_port() -> Generator[tuple[int, socket.socket]]:
     """Reserve an unused TCP port by holding the socket open."""

tests/integration/const.py

@@ -9,6 +9,13 @@ API_CONNECTION_TIMEOUT = 30.0  # seconds
 PORT_WAIT_TIMEOUT = 30.0  # seconds
 PORT_POLL_INTERVAL = 0.1  # seconds
 
+# The well-known all-zeros provisioning PSK, a key to provision over it, and
+# the time the device takes to activate a newly saved key (100 ms timer plus
+# margin)
+ZERO_PSK = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
+PROVISIONING_PSK = b"bm5ubm5ubm5ubm5ubm5ubm5ubm5ubm5ubm5ubm5ubm4="
+KEY_ACTIVATION_DELAY = 0.5  # seconds
+
 # Process shutdown timeouts
 SIGINT_TIMEOUT = 5.0  # seconds
 SIGTERM_TIMEOUT = 2.0  # seconds

tests/integration/fixtures/host_ota_api_key_offer_with_password.yaml

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

tests/integration/fixtures/host_ota_provisioned_api_key.yaml

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

tests/integration/test_api_zero_psk_provisioning.py

@@ -10,34 +10,40 @@ 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 KEY_ACTIVATION_DELAY, LOCALHOST, PROVISIONING_PSK, ZERO_PSK
+from .types import (
+    APIClientConnectedFactory,
+    CompileFunction,
+    ConfigWriter,
+    RunCompiledFunction,
+)
 
-# The well-known provisioning PSK: base64 of 32 zero bytes
-ZERO_PSK = base64.b64encode(bytes(32)).decode()
-# A real key to provision
-NEW_KEY = base64.b64encode(b"n" * 32)
-# Time for the device to activate a newly saved key (100ms timer plus margin)
-KEY_ACTIVATION_DELAY = 0.5
-
-
-@pytest.fixture(autouse=True)
-def isolated_preferences(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None:
-    """Keep host preferences per-test so every run starts unprovisioned."""
-    monkeypatch.setenv("ESPHOME_PREFDIR", str(tmp_path / "prefs"))
+pytestmark = pytest.mark.usefixtures("isolated_preferences")
+NEW_KEY = PROVISIONING_PSK
 
 
 @pytest.mark.asyncio
 async def test_api_zero_psk_provisioning(
     yaml_config: str,
-    run_compiled: RunCompiledFunction,
+    write_yaml_config: ConfigWriter,
+    compile_esphome: CompileFunction,
+    reserved_tcp_port: tuple[int, socket.socket],
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
-    """Exercise the reject paths, then provision a key over the zero-PSK channel."""
-    async with run_compiled(yaml_config):
+    """Exercise the reject paths, provision a key over the zero-PSK channel,
+    and check the key comes back from preferences on the next boot."""
+    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):
         # --- Pre-provisioning reject paths (device state is unchanged) ---
 
         # A wrong (non-zero) PSK fails against the zero provisioning PSK
@@ -97,6 +103,19 @@ async def test_api_zero_psk_provisioning(
             async with api_client_connected(timeout=5) as client:
                 await client.device_info()
 
+    # The key is loaded from preferences on the next boot
+    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)
+
 
 @pytest.mark.asyncio
 async def test_api_zero_psk_provisioning_plaintext(

tests/integration/test_host_ota.py

@@ -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
+from dataclasses import dataclass
 import functools
+from pathlib import Path
 import socket
 
 import pytest
@@ -18,10 +21,18 @@ import pytest
 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 .const import (
+    KEY_ACTIVATION_DELAY,
+    LOCALHOST,
+    PORT_POLL_INTERVAL,
+    PORT_WAIT_TIMEOUT,
+    PROVISIONING_PSK,
+    ZERO_PSK,
+)
+from .types import APIClientConnectedFactory, CompileFunction, ConfigWriter
 
 DEVICE_NAME = "host-ota-test"
+API_KEY = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
 
 
 @contextmanager
@@ -35,6 +46,14 @@ def _reserve_port() -> Generator[tuple[int, socket.socket]]:
         s.close()
 
 
+async def _wait_for_line(lines: list[str], needle: str, timeout: float = 5.0) -> None:
+    """The config dump prints after every setup, a little after the api port
+    opens, so wait for it rather than assert on the lines seen so far."""
+    async with asyncio.timeout(timeout):
+        while not any(needle in line for line in lines):
+            await asyncio.sleep(PORT_POLL_INTERVAL)
+
+
 async def _wait_for_port(host: str, port: int, timeout: float) -> None:
     """Poll until a TCP port accepts connections, or raise TimeoutError."""
     loop = asyncio.get_running_loop()
@@ -51,6 +70,102 @@ async def _wait_for_port(host: str, port: int, timeout: float) -> None:
     raise TimeoutError(f"Port {port} on {host} did not open within {timeout}s")
 
 
+async def _build(
+    yaml_config: str,
+    write_yaml_config: ConfigWriter,
+    compile_esphome: CompileFunction,
+    reserved_tcp_port: tuple[int, socket.socket],
+) -> tuple[int, int, Path]:
+    """Reserve an OTA port, compile the fixture with it, and release both
+    ports right before the binary is started."""
+    api_port, api_socket = reserved_tcp_port
+    with _reserve_port() as (ota_port, ota_socket):
+        config_path = await write_yaml_config(
+            yaml_config.replace("__OTA_PORT__", str(ota_port))
+        )
+        binary_path = await compile_esphome(config_path)
+        api_socket.close()
+        ota_socket.close()
+    return api_port, ota_port, binary_path
+
+
+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
+
+
+@dataclass
+class _Device:
+    """A running host binary and the checks every successful OTA repeats:
+    a safe reboot, the api port back up, and the pid preserved by execv."""
+
+    api_port: int
+    ota_port: int
+    binary_path: Path
+    proc: asyncio.subprocess.Process | None = None
+    reboots: int = 0
+
+    def __post_init__(self) -> None:
+        self._rebooted = asyncio.Event()
+
+    def on_log(self, line: str) -> None:
+        if "Rebooting safely" in line:
+            self.reboots += 1
+            self._rebooted.set()
+
+    async def wait_reboot(self, count: int, timeout: float = 10.0) -> None:
+        async with asyncio.timeout(timeout):
+            while self.reboots < count:
+                self._rebooted.clear()
+                await self._rebooted.wait()
+
+    async def ota(
+        self,
+        password: str | None,
+        noise_psk: str | None,
+        msg: str,
+        plaintext_fallback: bool = False,
+    ) -> None:
+        """Upload, then expect the re-exec with the pid preserved."""
+        pid_before = self.proc.pid
+        expected_reboots = self.reboots + 1
+        rc = await _run_ota(
+            self.ota_port, password, self.binary_path, noise_psk, plaintext_fallback
+        )
+        assert rc == 0, msg
+        await self.wait_reboot(expected_reboots)
+        await _wait_for_port(LOCALHOST, self.api_port, PORT_WAIT_TIMEOUT)
+        assert self.proc.returncode is None, "process exited instead of execing"
+        assert self.proc.pid == pid_before
+
+    async def refused_ota(
+        self, password: str | None, noise_psk: str | None, msg: str
+    ) -> None:
+        """Upload must fail and the device must keep running."""
+        rc = await _run_ota(self.ota_port, password, self.binary_path, noise_psk)
+        assert rc == 1, msg
+        await asyncio.sleep(0.5)
+        assert self.proc.returncode is None, "process died on rejected OTA"
+
+
 @pytest.mark.asyncio
 async def test_host_ota_self_update(
     yaml_config: str,
@@ -59,57 +174,34 @@ async def test_host_ota_self_update(
     reserved_tcp_port: tuple[int, socket.socket],
 ) -> None:
     """Self-OTA: upload the running binary back to itself, expect re-exec."""
-    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()
+    dev = _Device(
+        *await _build(
+            yaml_config, write_yaml_config, compile_esphome, reserved_tcp_port
+        )
+    )
+    staged = asyncio.Event()
 
-        loop = asyncio.get_running_loop()
-        ota_staged = loop.create_future()
-        rebooted = loop.create_future()
+    def on_log(line: str) -> None:
+        if "OTA staged at" in line:
+            staged.set()
+        dev.on_log(line)
 
-        def on_log(line: str) -> None:
-            if not ota_staged.done() and "OTA staged at" in line:
-                ota_staged.set_result(True)
-            if not rebooted.done() and "Rebooting safely" in line:
-                rebooted.set_result(True)
+    async with run_binary(dev.binary_path, line_callback=on_log) as (proc, _lines):
+        dev.proc = proc
+        await _wait_for_port(LOCALHOST, dev.api_port, PORT_WAIT_TIMEOUT)
+        async with wait_and_connect_api_client(port=dev.api_port) as client:
+            info_before = await client.device_info()
+            assert info_before.name == DEVICE_NAME
 
-        async with run_binary(binary_path, line_callback=on_log) as (proc, _lines):
-            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
-            pid_before = proc.pid
-            async with wait_and_connect_api_client(port=api_port) as client:
-                info_before = await client.device_info()
-                assert info_before.name == DEVICE_NAME
+        await dev.ota(None, None, "espota2 reported failure")
+        assert staged.is_set()
 
-            # espota2 is blocking; run in executor.
-            rc, _ = await loop.run_in_executor(
-                None, espota2.run_ota, LOCALHOST, ota_port, None, binary_path
-            )
-            assert rc == 0, "espota2 reported failure"
+        async with wait_and_connect_api_client(port=dev.api_port) as client:
+            info_after = await client.device_info()
+            assert info_after.name == info_before.name
 
-            await asyncio.wait_for(ota_staged, timeout=10.0)
-            await asyncio.wait_for(rebooted, timeout=10.0)
-            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
-
-            # execv preserves pid; mismatch means external respawn.
-            assert proc.returncode is None, "process exited instead of execing"
-            assert proc.pid == pid_before
-
-            async with wait_and_connect_api_client(port=api_port) as client:
-                info_after = await client.device_info()
-                assert info_after.name == DEVICE_NAME
-                assert info_after.name == info_before.name
-
-            # Second OTA: catches FD_CLOEXEC regressions (EADDRINUSE on rebind).
-            rc, _ = await loop.run_in_executor(
-                None, espota2.run_ota, LOCALHOST, ota_port, None, binary_path
-            )
-            assert rc == 0, "second OTA failed -- listener leaked across execv"
-            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
-            assert proc.pid == pid_before
+        # Second OTA: catches FD_CLOEXEC regressions (EADDRINUSE on rebind).
+        await dev.ota(None, None, "second OTA failed -- listener leaked across execv")
 
 
 @pytest.mark.asyncio
@@ -121,51 +213,110 @@ 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))
-        config_path = await write_yaml_config(yaml_config)
-        binary_path = await compile_esphome(config_path)
-        api_socket.close()
-        ota_socket.close()
+    dev = _Device(
+        *await _build(
+            yaml_config, write_yaml_config, compile_esphome, reserved_tcp_port
+        )
+    )
+    async with run_binary(dev.binary_path, line_callback=dev.on_log) as (proc, _lines):
+        dev.proc = proc
+        await _wait_for_port(LOCALHOST, dev.api_port, PORT_WAIT_TIMEOUT)
+        await dev.refused_ota(
+            None, None, "plaintext upload to an encrypted device must fail"
+        )
+        await dev.ota(None, API_KEY, "encrypted OTA reported failure")
 
-        loop = asyncio.get_running_loop()
-        rebooted = loop.create_future()
 
-        def on_log(line: str) -> None:
-            if not rebooted.done() and "Rebooting safely" in line:
-                rebooted.set_result(True)
+@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],
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """With only an api key the device offers encryption without requiring
+    it: the password still guards plaintext uploads, the key alone
+    authenticates an encrypted one, and until 2027.3.0 a failed encrypted
+    attempt falls back to plaintext."""
+    pytest.importorskip("aioesphomeapi.noise")
+    wrong_key = base64.b64encode(b"w" * 32).decode()
+    dev = _Device(
+        *await _build(
+            yaml_config, write_yaml_config, compile_esphome, reserved_tcp_port
+        )
+    )
+    async with run_binary(dev.binary_path, line_callback=dev.on_log) as (proc, lines):
+        dev.proc = proc
+        await _wait_for_port(LOCALHOST, dev.api_port, PORT_WAIT_TIMEOUT)
+        await _wait_for_line(lines, "Encryption: offered")
 
-        async with run_binary(binary_path, line_callback=on_log) as (proc, _lines):
-            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
-            pid_before = proc.pid
+        await dev.refused_ota(
+            None, None, "plaintext upload without the password must fail"
+        )
+        await dev.ota(
+            "hunter2", None, "plaintext upload with the password must succeed"
+        )
+        await dev.ota(None, API_KEY, "encrypted upload with the api key must succeed")
 
-            # 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
+        # Remove before 2027.3.0: a wrong key falls back to plaintext, which
+        # the password still guards
+        with caplog.at_level("WARNING", logger="esphome.espota2"):
+            await dev.ota(
+                "hunter2",
+                wrong_key,
+                "the plaintext retry with the password must succeed",
+                plaintext_fallback=True,
             )
-            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"
+        assert any("Retrying in plaintext" in r.message for r in caplog.records)
+        await dev.ota(
+            None,
+            API_KEY,
+            "the right api key encrypts without touching the fallback",
+            plaintext_fallback=True,
+        )
 
-            # 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,
-                ),
-            )
-            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)
-            assert proc.returncode is None, "process exited instead of execing"
-            assert proc.pid == pid_before
+
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("isolated_preferences")
+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,
+) -> None:
+    """A key provisioned over the api feeds the OTA offer: plaintext works
+    while unprovisioned, the provisioned key encrypts, the key loaded from
+    preferences on the next boot keeps encrypting, and plaintext stays
+    accepted because only the ota block requires encryption."""
+    pytest.importorskip("aioesphomeapi.noise")
+    dev = _Device(
+        *await _build(
+            yaml_config, write_yaml_config, compile_esphome, reserved_tcp_port
+        )
+    )
+    async with run_binary(dev.binary_path, line_callback=dev.on_log) as (proc, lines):
+        dev.proc = proc
+        await _wait_for_port(LOCALHOST, dev.api_port, PORT_WAIT_TIMEOUT)
+        await _wait_for_line(lines, "once the api key is provisioned")
+
+        await dev.ota(
+            None, None, "plaintext upload to an unprovisioned device must succeed"
+        )
+
+        async with api_client_connected(
+            port=dev.api_port, noise_psk=ZERO_PSK
+        ) as client:
+            assert await client.noise_encryption_set_key(PROVISIONING_PSK) is True
+        await asyncio.sleep(KEY_ACTIVATION_DELAY)
+
+        key = PROVISIONING_PSK.decode()
+        await dev.ota(
+            None, key, "encrypted upload with the provisioned key must succeed"
+        )
+        await dev.ota(None, key, "the key loaded at boot must feed the OTA offer")
+        await dev.ota(None, None, "plaintext must stay accepted on an offering device")
 
 
 @pytest.mark.asyncio
@@ -177,33 +328,25 @@ async def test_host_ota_rejects_garbage(
     integration_test_dir,
 ) -> None:
     """Bogus payload is rejected and the device keeps running."""
-    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)
+    dev = _Device(
+        *await _build(
+            yaml_config, write_yaml_config, compile_esphome, reserved_tcp_port
+        )
+    )
+    # 192 bytes that are neither ELF nor Mach-O.
+    bogus_path = integration_test_dir / "bogus.bin"
+    bogus_path.write_bytes(b"NOT-AN-EXECUTABLE-AT-ALL" * 8)
 
-        # 192 bytes that are neither ELF nor Mach-O.
-        bogus_path = integration_test_dir / "bogus.bin"
-        bogus_path.write_bytes(b"NOT-AN-EXECUTABLE-AT-ALL" * 8)
+    async with run_binary(dev.binary_path) as (proc, _lines):
+        dev.proc = proc
+        await _wait_for_port(LOCALHOST, dev.api_port, PORT_WAIT_TIMEOUT)
+        pid_before = proc.pid
+        rc = await _run_ota(dev.ota_port, None, bogus_path, None)
+        assert rc == 1
+        await asyncio.sleep(0.5)
+        assert proc.returncode is None, "process died on rejected OTA"
+        assert proc.pid == pid_before
 
-        api_socket.close()
-        ota_socket.close()
-
-        async with run_binary(binary_path) as (proc, _lines):
-            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
-            pid_before = proc.pid
-
-            loop = asyncio.get_running_loop()
-            rc, _ = await loop.run_in_executor(
-                None, espota2.run_ota, LOCALHOST, ota_port, None, bogus_path
-            )
-            assert rc == 1
-
-            await asyncio.sleep(0.5)
-            assert proc.returncode is None, "process died on rejected OTA"
-            assert proc.pid == pid_before
-
-            async with wait_and_connect_api_client(port=api_port) as client:
-                info = await client.device_info()
-                assert info.name == DEVICE_NAME
+        async with wait_and_connect_api_client(port=dev.api_port) as client:
+            info = await client.device_info()
+            assert info.name == DEVICE_NAME

tests/unit_tests/test_espota2_noise.py

@@ -10,12 +10,15 @@ 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
 import threading
+from typing import Any
 from unittest.mock import Mock, patch
 
 import pytest
@@ -65,8 +68,12 @@ class FakeEncryptedDevice(threading.Thread):
         offer_noise: bool = True,
         require_noise: bool = True,
         prologue_features_override: int | None = None,
+        connections: int = 1,
+        drop_handshakes: int = 0,
     ) -> None:
         super().__init__(daemon=True)
+        self.connections = connections
+        self.drop_handshakes = drop_handshakes  # hang up mid-handshake this many times
         self.psk = psk
         self.version = version
         self.offer_noise = offer_noise
@@ -81,10 +88,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 +117,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
@@ -134,6 +157,9 @@ class FakeEncryptedDevice(threading.Thread):
 
         msg1 = _recv_frame(sock)
         assert msg1[0] == 0x00
+        if self.drop_handshakes > 0:
+            self.drop_handshakes -= 1
+            return  # a transport fault: the socket closes with no reply
         try:
             proto.read_message(msg1[1:])
         except InvalidTag:
@@ -149,6 +175,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 +199,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 +214,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 +225,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 +304,56 @@ 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
+@pytest.mark.parametrize(
+    ("device_kwargs", "expected_rc", "fell_back"),
+    [
+        # A wrong key against an offering device reconnects in plaintext
+        ({"psk": OTHER_PSK, "require_noise": False, "connections": 2}, 0, True),
+        # The plaintext retry is refused by a device that requires encryption
+        ({"psk": OTHER_PSK, "require_noise": True, "connections": 2}, 1, True),
+        # A dropped connection inside the handshake is retried encrypted
+        ({"require_noise": False, "connections": 2, "drop_handshakes": 1}, 0, False),
+        # A second transport fault inside the handshake falls back
+        ({"require_noise": False, "connections": 3, "drop_handshakes": 2}, 0, True),
+    ],
+    ids=["wrong_key", "wrong_key_required", "one_fault", "two_faults"],
+)
+def test_fallback_through_the_retry_loop(
+    caplog: pytest.LogCaptureFixture,
+    tmp_path: Path,
+    device_kwargs: dict[str, Any],
+    expected_rc: int,
+    fell_back: bool,
+) -> None:
+    pytest.importorskip("aioesphomeapi.noise")
+    firmware = b"firmware"
+    device = FakeEncryptedDevice(**device_kwargs)
+    with patch("time.sleep"), caplog.at_level(logging.WARNING):
+        rc = _run_ota(device, firmware, tmp_path, PSK)
+    device.join_and_check()
+    assert rc == expected_rc
+    assert (device.received == firmware) is (expected_rc == 0)
+    assert (
+        any("Retrying in plaintext" in r.message for r in caplog.records) is fell_back
+    )
+    if expected_rc == 1:
+        assert any("requires an encrypted OTA" in r.message for r in caplog.records)
+
+
 def test_plaintext_client_gets_encryption_required_error() -> None:
     """A client without a key gets the device's 0x94 error message."""
     device = FakeEncryptedDevice()

tests/unit_tests/test_main.py

@@ -2108,7 +2108,13 @@ def test_upload_program_ota_success(
         tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
     )
     mock_run_ota.assert_called_once_with(
-        ["192.168.1.100"], 3232, "secret", expected_firmware, OTA_TYPE_UPDATE_APP, None
+        ["192.168.1.100"],
+        3232,
+        "secret",
+        expected_firmware,
+        OTA_TYPE_UPDATE_APP,
+        None,
+        plaintext_fallback=False,
     )
 
 
@@ -2140,10 +2146,77 @@ 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: {}}],
+    ids=["no_encryption", "runtime_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 or runtime provisioned 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 +2267,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 +2329,7 @@ def test_upload_program_ota_partition_table_with_file_arg(
         partition_file,
         OTA_TYPE_UPDATE_PARTITION_TABLE,
         None,
+        plaintext_fallback=False,
     )
 
 
@@ -2312,6 +2392,7 @@ def test_upload_program_ota_partition_table_mqttip(
         partition_file,
         OTA_TYPE_UPDATE_PARTITION_TABLE,
         None,
+        plaintext_fallback=False,
     )
 
 
@@ -2500,6 +2581,7 @@ def test_upload_program_ota_bootloader_with_file_arg(
         bootloader_file,
         OTA_TYPE_UPDATE_BOOTLOADER,
         None,
+        plaintext_fallback=False,
     )
 
 
@@ -2988,7 +3070,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 +3126,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 +5305,7 @@ def test_upload_program_ota_static_ip_with_mqttip(
         expected_firmware,
         OTA_TYPE_UPDATE_APP,
         None,
+        plaintext_fallback=False,
     )
 
 
@@ -5261,6 +5356,7 @@ def test_upload_program_ota_multiple_mqttip_resolves_once(
         expected_firmware,
         OTA_TYPE_UPDATE_APP,
         None,
+        plaintext_fallback=False,
     )
 
 
@@ -5438,7 +5534,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,
     )
 
 

tests/unit_tests/test_wizard.py

@@ -37,7 +37,6 @@ def wizard_answers() -> list[str]:
         "nodemcuv2",  # board
         "SSID",  # ssid
         "psk",  # wifi password
-        "",  # ota password (empty for no password)
     ]
 
 
@@ -101,6 +100,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],
 ):
@@ -630,15 +648,15 @@ def test_wizard_write_protects_existing_config(
     assert config_file.read_text() == original_content
 
 
-def test_wizard_accepts_ota_password(
+def test_wizard_uses_the_api_key_for_ota(
     tmp_path: Path, monkeypatch: MonkeyPatch, wizard_answers: list[str]
 ):
     """
-    The wizard should pass ota_password to wizard_write when the user provides one
+    The wizard generates an api key and does not ask for an OTA password;
+    the key secures OTA updates
     """
 
     # Given
-    wizard_answers[5] = "my_ota_password"  # Set OTA password
     config_file = tmp_path / "test.yaml"
     input_mock = MagicMock(side_effect=wizard_answers)
     monkeypatch.setattr("builtins.input", input_mock)
@@ -653,8 +671,9 @@ def test_wizard_accepts_ota_password(
     # Then
     assert retval == 0
     call_kwargs = wizard_write_mock.call_args.kwargs
-    assert "ota_password" in call_kwargs
-    assert call_kwargs["ota_password"] == "my_ota_password"
+    assert "api_encryption_key" in call_kwargs
+    assert "ota_password" not in call_kwargs
+    assert input_mock.call_count == len(wizard_answers)
 
 
 def test_wizard_accepts_rpipico_board(tmp_path: Path, monkeypatch: MonkeyPatch):