Add tests for encrypted OTA

esphome/__main__.py

@@ -26,7 +26,9 @@ from esphome.const import (
     CONF_DEASSERT_RTS_DTR,
     CONF_DISABLED,
     CONF_DISCOVER_IP,
+    CONF_ENCRYPTION,
     CONF_ESPHOME,
+    CONF_KEY,
     CONF_LEVEL,
     CONF_LOG,
     CONF_LOG_TOPIC,
@@ -1323,6 +1325,17 @@ def _upload_via_native_api(
 
     remote_port = int(ota_conf[CONF_PORT])
     password = ota_conf.get(CONF_PASSWORD)
+    # Final validate resolved a bare `encryption:` block to the api key.
+    # Fail closed: if the block is present but no key was resolved, never
+    # fall back to a plaintext upload of an image that carries credentials.
+    noise_psk = None
+    if (encryption_conf := ota_conf.get(CONF_ENCRYPTION)) is not None:
+        noise_psk = encryption_conf.get(CONF_KEY)
+        if not noise_psk:
+            raise EsphomeError(
+                "OTA encryption is configured but no key was resolved; "
+                "set the key under 'ota: encryption:' or 'api: encryption:'"
+            )
 
     def check_partition_access(option_string: str) -> None:
         if not ota_conf.get("allow_partition_access"):
@@ -1353,7 +1366,9 @@ def _upload_via_native_api(
     if ota_type == espota2.OTA_TYPE_UPDATE_BOOTLOADER:
         _validate_bootloader_binary(binary)
 
-    return espota2.run_ota(network_devices, remote_port, password, binary, ota_type)
+    return espota2.run_ota(
+        network_devices, remote_port, password, binary, ota_type, noise_psk
+    )
 
 
 def _upload_via_web_server(
@@ -1362,6 +1377,16 @@ def _upload_via_web_server(
     from esphome import web_server_ota
     from esphome.web_server_helpers import get_web_server_connection
 
+    if any(
+        ota_item.get(CONF_PLATFORM) == CONF_ESPHOME
+        and ota_item.get(CONF_ENCRYPTION) is not None
+        for ota_item in config.get(CONF_OTA, [])
+    ):
+        _LOGGER.warning(
+            "This config has OTA encryption, but the web_server OTA path sends "
+            "the image over plaintext HTTP; use the esphome OTA platform to "
+            "keep it confidential"
+        )
     remote_port, username, password = get_web_server_connection(config)
     return web_server_ota.run_ota(
         network_devices, remote_port, username, password, binary

esphome/components/esphome/ota/__init__.py

@@ -1,12 +1,16 @@
 import logging
 
 import esphome.codegen as cg
+from esphome.components.noise import decode_encryption_key, encryption_schema
 from esphome.components.ota import BASE_OTA_SCHEMA, OTAComponent, ota_to_code
 from esphome.config_helpers import merge_config
 import esphome.config_validation as cv
 from esphome.const import (
+    CONF_API,
+    CONF_ENCRYPTION,
     CONF_ESPHOME,
     CONF_ID,
+    CONF_KEY,
     CONF_NUM_ATTEMPTS,
     CONF_OTA,
     CONF_PASSWORD,
@@ -15,6 +19,7 @@ from esphome.const import (
     CONF_REBOOT_TIMEOUT,
     CONF_SAFE_MODE,
     CONF_VERSION,
+    CONF_WEB_SERVER,
 )
 from esphome.core import CORE, coroutine_with_priority
 from esphome.coroutine import CoroPriority
@@ -30,7 +35,15 @@ CODEOWNERS = ["@esphome/core"]
 DEPENDENCIES = ["network"]
 
 
-AUTO_LOAD = ["sha256", "socket"]
+def AUTO_LOAD(config: ConfigType) -> list[str]:
+    """Auto-load noise only when encryption is configured."""
+    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
+    if not config or CONF_ENCRYPTION in config:
+        return base + ["noise"]
+    return base
 
 
 esphome = cg.esphome_ns.namespace("esphome")
@@ -67,11 +80,24 @@ def ota_esphome_final_validate(config):
                     CONF_PASSWORD in merged_ota_esphome_configs_by_port[conf_port]
                     and CONF_PASSWORD in ota_conf
                     and merged_ota_esphome_configs_by_port[conf_port][CONF_PASSWORD]
-                    != ota_conf.get(CONF_PASSWORD)
+                    != ota_conf[CONF_PASSWORD]
                 ):
                     raise cv.Invalid(
                         f"Found multiple configurations but {CONF_PASSWORD} is inconsistent"
                     )
+                # Encryption blocks conflict only when both pin a key; a bare
+                # `encryption:` (a package/device split) is compatible with a
+                # keyed one, and merge_config yields the keyed result
+                merged_key = (
+                    merged_ota_esphome_configs_by_port[conf_port]
+                    .get(CONF_ENCRYPTION, {})
+                    .get(CONF_KEY)
+                )
+                other_key = ota_conf.get(CONF_ENCRYPTION, {}).get(CONF_KEY)
+                if merged_key and other_key and merged_key != other_key:
+                    raise cv.Invalid(
+                        f"Found multiple configurations but {CONF_ENCRYPTION} is inconsistent"
+                    )
 
                 ports_with_merged_configs.append(conf_port)
                 merged_ota_esphome_configs_by_port[conf_port] = merge_config(
@@ -94,6 +120,73 @@ def ota_esphome_final_validate(config):
 
     new_ota_conf.extend(merged_ota_esphome_configs_by_port.values())
 
+    # There is one encryption key per device: when the api component has one,
+    # ota uses it, and an explicit ota key must match it. A bare `encryption:`
+    # block resolves to the api key here so both codegen and the upload CLI
+    # see the actual key.
+    api_conf = full_conf.get(CONF_API) or {}
+    api_key = api_conf.get(CONF_ENCRYPTION, {}).get(CONF_KEY)
+    has_web_server_ota = any(
+        conf.get(CONF_PLATFORM) == CONF_WEB_SERVER for conf in full_ota_conf
+    )
+    for ota_conf in merged_ota_esphome_configs_by_port.values():
+        # Merging same-port blocks can combine a password from one block with
+        # encryption from another; re-check the exclusion on the merged result.
+        _validate_no_password_with_encryption(ota_conf)
+        if (encryption_conf := ota_conf.get(CONF_ENCRYPTION)) is None:
+            continue
+        if has_web_server_ota:
+            # The web_server ota platform accepts the same image over plain
+            # HTTP with basic auth, a full bypass of the encryption.
+            if CONF_WEB_SERVER in full_conf:
+                # With the web_server component the endpoint is always on;
+                # fail closed like the password combination
+                raise cv.Invalid(
+                    f"'{CONF_OTA}' {CONF_ENCRYPTION} cannot be combined with the "
+                    f"'{CONF_WEB_SERVER}' component; its '{CONF_OTA}' platform "
+                    f"accepts the same image over plaintext HTTP, remove one of them"
+                )
+            # Without the component the platform is the captive_portal
+            # auto-load: the endpoint only exists while the fallback AP is
+            # active, so keep the recovery path and warn instead
+            _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,
+            )
+        if ota_key := encryption_conf.get(CONF_KEY):
+            if api_key and ota_key != api_key:
+                raise cv.Invalid(
+                    f"'{CONF_OTA}' {CONF_ENCRYPTION} {CONF_KEY} must match the "
+                    f"'{CONF_API}' {CONF_ENCRYPTION} {CONF_KEY}; omit the "
+                    f"'{CONF_OTA}' {CONF_KEY} to use the '{CONF_API}' one"
+                )
+        elif not api_key:
+            if CONF_ENCRYPTION in api_conf:
+                # A keyless `api: encryption:` block gets its key provisioned
+                # at runtime and stored in flash, so there is nothing to
+                # inherit at build time
+                raise cv.Invalid(
+                    f"the '{CONF_API}' {CONF_ENCRYPTION} {CONF_KEY} is provisioned at "
+                    f"runtime and cannot be inherited at build time; set an explicit "
+                    f"'{CONF_OTA}' {CONF_ENCRYPTION} {CONF_KEY}"
+                )
+            raise cv.Invalid(
+                f"'{CONF_OTA}' {CONF_ENCRYPTION} has no {CONF_KEY} and there is no "
+                f"'{CONF_API}' {CONF_ENCRYPTION} {CONF_KEY} to inherit; set one of them"
+            )
+        else:
+            encryption_conf[CONF_KEY] = api_key
+        # The device treats the all-zeros PSK as "no key configured" (it is the
+        # api provisioning sentinel), so letting it through would leave the OTA
+        # port accepting plaintext while the YAML says encryption. Fail closed.
+        if not any(decode_encryption_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"
+            )
+
     full_conf[CONF_OTA] = new_ota_conf
     fv.full_config.set(full_conf)
 
@@ -107,6 +200,17 @@ def ota_esphome_final_validate(config):
         )
 
 
+# Not cv.has_at_most_one_key: this message explains the why, and the check is
+# reused on merged same-port configs in final validate where schemas do not run
+def _validate_no_password_with_encryption(config: ConfigType) -> ConfigType:
+    if CONF_PASSWORD in config and CONF_ENCRYPTION in config:
+        raise cv.Invalid(
+            f"'{CONF_PASSWORD}' cannot be combined with '{CONF_ENCRYPTION}'; the "
+            f"encryption key already authenticates the uploader, remove '{CONF_PASSWORD}'"
+        )
+    return config
+
+
 def _consume_ota_sockets(config: ConfigType) -> ConfigType:
     """Register socket needs for OTA component."""
     from esphome.components import socket
@@ -134,6 +238,7 @@ CONFIG_SCHEMA = cv.All(
             ): cv.port,
             cv.Optional(CONF_ALLOW_PARTITION_ACCESS, default=False): cv.boolean,
             cv.Optional(CONF_PASSWORD): cv.sensitive(),
+            cv.Optional(CONF_ENCRYPTION): encryption_schema,
             cv.Optional(CONF_NUM_ATTEMPTS): cv.invalid(
                 f"'{CONF_SAFE_MODE}' (and its related configuration variables) has moved from 'ota' to its own component. See https://esphome.io/components/safe_mode"
             ),
@@ -147,12 +252,24 @@ CONFIG_SCHEMA = cv.All(
     )
     .extend(BASE_OTA_SCHEMA)
     .extend(cv.COMPONENT_SCHEMA),
+    _validate_no_password_with_encryption,
     _consume_ota_sockets,
 )
 
 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"]
+
+
 @coroutine_with_priority(CoroPriority.OTA_UPDATES)
 async def to_code(config: ConfigType) -> None:
     var = cg.new_Pvariable(config[CONF_ID])
@@ -171,6 +288,12 @@ 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]
+        cg.add_define("USE_OTA_ENCRYPTION")
+        cg.add(var.set_noise_psk(list(decode_encryption_key(key))))
+
     # Build flag so lwip_fast_select.c (a .c file that can't include defines.h) sees it.
     cg.add_build_flag("-DUSE_OTA_PLATFORM_ESPHOME")
 

esphome/components/esphome/ota/ota_esphome.cpp

@@ -27,7 +27,6 @@ namespace esphome {
 
 static const char *const TAG = "esphome.ota";
 static constexpr uint16_t OTA_BLOCK_SIZE = 8192;
-static constexpr size_t OTA_BUFFER_SIZE = 1024;                  // buffer size for OTA data transfer
 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
 
@@ -105,6 +104,11 @@ void ESPHomeOTAComponent::dump_config() {
     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"
@@ -148,8 +152,10 @@ void ESPHomeOTAComponent::loop() {
 static constexpr uint8_t CLIENT_FEATURE_SUPPORTS_COMPRESSION = 0x01;
 static constexpr uint8_t CLIENT_FEATURE_SUPPORTS_SHA256_AUTH = 0x02;
 static constexpr uint8_t CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL = 0x04;
+static constexpr uint8_t CLIENT_FEATURE_SUPPORTS_NOISE = 0x08;
 static constexpr uint8_t SERVER_FEATURE_SUPPORTS_COMPRESSION = 0x01;
 static constexpr uint8_t SERVER_FEATURE_SUPPORTS_PARTITION_ACCESS = 0x02;
+static constexpr uint8_t SERVER_FEATURE_SUPPORTS_NOISE = 0x04;
 
 void ESPHomeOTAComponent::handle_handshake_() {
   /// Handle the OTA handshake and authentication.
@@ -201,8 +207,7 @@ void ESPHomeOTAComponent::handle_handshake_() {
       }
 
       // Validate magic bytes
-      static const uint8_t MAGIC_BYTES[5] = {0x6C, 0x26, 0xF7, 0x5C, 0x45};
-      if (memcmp(this->handshake_buf_, MAGIC_BYTES, 5) != 0) {
+      if (memcmp(this->handshake_buf_, MAGIC_BYTES, sizeof(MAGIC_BYTES)) != 0) {
         ESP_LOGW(TAG, "Magic bytes mismatch! 0x%02X-0x%02X-0x%02X-0x%02X-0x%02X", this->handshake_buf_[0],
                  this->handshake_buf_[1], this->handshake_buf_[2], this->handshake_buf_[3], this->handshake_buf_[4]);
         this->send_error_and_cleanup_(ota::OTA_RESPONSE_ERROR_MAGIC);
@@ -234,6 +239,19 @@ 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) {
+        ESP_LOGW(TAG, "Client does not support encryption");
+        this->send_error_and_cleanup_(ota::OTA_RESPONSE_ERROR_ENCRYPTION_REQUIRED);
+        return;
+      }
+#endif
+
       this->transition_ota_state_(OTAState::FEATURE_ACK);
 
       const bool supports_compression =
@@ -249,6 +267,12 @@ void ESPHomeOTAComponent::handle_handshake_() {
         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()) {
+          this->handshake_buf_[1] |= SERVER_FEATURE_SUPPORTS_NOISE;
+        }
+        this->server_feature_flags_ = this->handshake_buf_[1];
 #endif
       } else {
         this->handshake_buf_[0] =
@@ -264,6 +288,18 @@ void ESPHomeOTAComponent::handle_handshake_() {
       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()) {
+        if (!this->noise_start_session_()) {
+          return;
+        }
+        this->transition_ota_state_(OTAState::NOISE_HANDSHAKE);
+        return;
+      }
+#endif
 #ifdef USE_OTA_PASSWORD
       // If password is set, move to auth phase
       if (!this->password_.empty()) {
@@ -301,6 +337,16 @@ void ESPHomeOTAComponent::handle_handshake_() {
       this->handle_data_();
       return;
 
+#ifdef USE_OTA_ENCRYPTION
+    case OTAState::NOISE_HANDSHAKE:
+      if (!this->handle_noise_handshake_()) {
+        return;
+      }
+      this->transition_ota_state_(OTAState::DATA);
+      this->handle_data_();
+      return;
+#endif
+
     default:
       break;
   }
@@ -360,12 +406,13 @@ void ESPHomeOTAComponent::handle_data_() {
   this->client_->setblocking(true);
 
   // Acknowledge auth OK - 1 byte
-  this->write_byte_(ota::OTA_RESPONSE_AUTH_OK);
+  this->data_write_byte_(ota::OTA_RESPONSE_AUTH_OK);
 
   if (this->extended_proto_) {
     // Read ota type, 1 byte
-    if (!this->readall_(buf, 1)) {
+    if (!this->data_readall_(buf, 1)) {
       this->log_read_error_(LOG_STR("OTA type"));
+      error_code = ota::OTA_RESPONSE_ERROR_UNKNOWN;
       goto error;  // NOLINT(cppcoreguidelines-avoid-goto)
     }
     ota_type = static_cast<ota::OTAType>(buf[0]);
@@ -373,8 +420,9 @@ void ESPHomeOTAComponent::handle_data_() {
   ESP_LOGV(TAG, "OTA type is 0x%02x", ota_type);
 
   // Read size, 4 bytes MSB first
-  if (!this->readall_(buf, 4)) {
+  if (!this->data_readall_(buf, 4)) {
     this->log_read_error_(LOG_STR("size"));
+    error_code = ota::OTA_RESPONSE_ERROR_UNKNOWN;
     goto error;  // NOLINT(cppcoreguidelines-avoid-goto)
   }
   ota_size = (static_cast<size_t>(buf[0]) << 24) | (static_cast<size_t>(buf[1]) << 16) |
@@ -404,11 +452,12 @@ void ESPHomeOTAComponent::handle_data_() {
     goto error;  // NOLINT(cppcoreguidelines-avoid-goto)
 
   // Acknowledge prepare OK - 1 byte
-  this->write_byte_(ota::OTA_RESPONSE_UPDATE_PREPARE_OK);
+  this->data_write_byte_(ota::OTA_RESPONSE_UPDATE_PREPARE_OK);
 
   // Read binary MD5, 32 bytes
-  if (!this->readall_(buf, 32)) {
+  if (!this->data_readall_(buf, 32)) {
     this->log_read_error_(LOG_STR("MD5 checksum"));
+    error_code = ota::OTA_RESPONSE_ERROR_UNKNOWN;
     goto error;  // NOLINT(cppcoreguidelines-avoid-goto)
   }
   sbuf[32] = '\0';
@@ -416,7 +465,7 @@ void ESPHomeOTAComponent::handle_data_() {
   this->backend_->set_update_md5(sbuf);
 
   // Acknowledge MD5 OK - 1 byte
-  this->write_byte_(ota::OTA_RESPONSE_BIN_MD5_OK);
+  this->data_write_byte_(ota::OTA_RESPONSE_BIN_MD5_OK);
 
   // Track when we last received data so a silently-vanished peer (no FIN/RST
   // delivered, e.g. uploader killed mid-transfer or NAT/router dropped state)
@@ -432,19 +481,37 @@ void ESPHomeOTAComponent::handle_data_() {
     }
     size_t remaining = ota_size - total;
     size_t requested = remaining < OTA_BUFFER_SIZE ? remaining : OTA_BUFFER_SIZE;
-    ssize_t read = this->client_->read(buf, requested);
-    if (read == -1) {
-      const int err = errno;
-      if (this->would_block_(err)) {
-        // read() already waited up to SO_RCVTIMEO for data, just feed WDT
-        App.feed_wdt();
-        continue;
+    ssize_t read;
+#ifdef USE_OTA_ENCRYPTION
+    if (this->noise_ != nullptr) {
+      // One frame per call; noise_read_data_ waits internally (readall_), so
+      // there is no would-block retry here and failures are already logged.
+      read = this->noise_read_data_(buf, requested);
+      if (read <= 0) {
+        // error_code still holds the last OK; report a real failure instead
+        error_code = ota::OTA_RESPONSE_ERROR_UNKNOWN;
+        goto error;  // NOLINT(cppcoreguidelines-avoid-goto)
+      }
+    } else
+#endif
+    {
+      read = this->client_->read(buf, requested);
+      if (read == -1) {
+        const int err = errno;
+        if (this->would_block_(err)) {
+          // read() already waited up to SO_RCVTIMEO for data, just feed WDT
+          App.feed_wdt();
+          continue;
+        }
+        ESP_LOGW(TAG, "Read err %d", err);
+        // error_code still holds the last OK; report a real failure instead
+        error_code = ota::OTA_RESPONSE_ERROR_UNKNOWN;
+        goto error;  // NOLINT(cppcoreguidelines-avoid-goto)
+      } else if (read == 0) {
+        ESP_LOGW(TAG, "Remote closed");
+        error_code = ota::OTA_RESPONSE_ERROR_UNKNOWN;
+        goto error;  // NOLINT(cppcoreguidelines-avoid-goto)
       }
-      ESP_LOGW(TAG, "Read err %d", err);
-      goto error;  // NOLINT(cppcoreguidelines-avoid-goto)
-    } else if (read == 0) {
-      ESP_LOGW(TAG, "Remote closed");
-      goto error;  // NOLINT(cppcoreguidelines-avoid-goto)
     }
 
     last_data_ms = millis();
@@ -456,7 +523,7 @@ void ESPHomeOTAComponent::handle_data_() {
     total += read;
 #if USE_OTA_VERSION == 2
     while (size_acknowledged + OTA_BLOCK_SIZE <= total || (total == ota_size && size_acknowledged < ota_size)) {
-      this->write_byte_(ota::OTA_RESPONSE_CHUNK_OK);
+      this->data_write_byte_(ota::OTA_RESPONSE_CHUNK_OK);
       size_acknowledged += OTA_BLOCK_SIZE;
     }
 #endif
@@ -475,7 +542,7 @@ void ESPHomeOTAComponent::handle_data_() {
   }
 
   // Acknowledge receive OK - 1 byte
-  this->write_byte_(ota::OTA_RESPONSE_RECEIVE_OK);
+  this->data_write_byte_(ota::OTA_RESPONSE_RECEIVE_OK);
 
   error_code = this->backend_->end();
   if (error_code != ota::OTA_RESPONSE_OK) {
@@ -484,10 +551,10 @@ void ESPHomeOTAComponent::handle_data_() {
   }
 
   // Acknowledge Update end OK - 1 byte
-  this->write_byte_(ota::OTA_RESPONSE_UPDATE_END_OK);
+  this->data_write_byte_(ota::OTA_RESPONSE_UPDATE_END_OK);
 
   // Read ACK
-  if (!this->readall_(buf, 1) || buf[0] != ota::OTA_RESPONSE_OK) {
+  if (!this->data_readall_(buf, 1) || buf[0] != ota::OTA_RESPONSE_OK) {
     this->log_read_error_(LOG_STR("ack"));
     // do not go to error, this is not fatal
   }
@@ -510,7 +577,7 @@ void ESPHomeOTAComponent::handle_data_() {
   App.safe_reboot();
 
 error:
-  this->write_byte_(static_cast<uint8_t>(error_code));
+  this->data_write_byte_(static_cast<uint8_t>(error_code));
 
   // Abort backend before cleanup - cleanup_connection_() destroys the backend.
   // Always call abort() unconditionally: backends register external partitions before
@@ -677,6 +744,9 @@ void ESPHomeOTAComponent::cleanup_connection_() {
   this->backend_ = nullptr;
 #ifdef USE_OTA_PASSWORD
   this->cleanup_auth_();
+#endif
+#ifdef USE_OTA_ENCRYPTION
+  this->noise_ = nullptr;
 #endif
   // Intentionally no disable_loop() — letting loop() run one more iteration catches
   // any connection that queued on the listener mid-session (otherwise the wake flag,

esphome/components/esphome/ota/ota_esphome.h

@@ -4,6 +4,9 @@
 #ifdef USE_OTA
 #include "esphome/components/ota/ota_backend_factory.h"
 #include "esphome/components/socket/socket.h"
+#ifdef USE_OTA_ENCRYPTION
+#include "esphome/components/noise/noise_handshake.h"
+#endif
 #include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
 #include "esphome/core/preferences.h"
@@ -24,7 +27,10 @@ class ESPHomeOTAComponent final : public ota::OTAComponent {
     AUTH_SEND,  // Sending authentication request
     AUTH_READ,  // Reading authentication data
 #endif          // USE_OTA_PASSWORD
-    DATA,       // BLOCKING! Processing OTA data (update, etc.)
+#ifdef USE_OTA_ENCRYPTION
+    NOISE_HANDSHAKE,  // Exchanging Noise handshake frames
+#endif
+    DATA,  // BLOCKING! Processing OTA data (update, etc.)
   };
 #ifdef USE_OTA_PASSWORD
   void set_auth_password(const std::string &password) { password_ = password; }
@@ -38,6 +44,10 @@ 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); }
+#endif
+
   /// Manually set the port OTA should listen on
   void set_port(uint16_t port) { this->port_ = port; }
 
@@ -63,6 +73,48 @@ class ESPHomeOTAComponent final : public ota::OTAComponent {
   bool writeall_(const uint8_t *buf, size_t len);
   inline bool write_byte_(uint8_t byte) { return this->writeall_(&byte, 1); }
 
+#ifdef USE_OTA_ENCRYPTION
+  // Heap-allocated only while an encrypted OTA session is active.
+  struct NoiseSession {
+    ~NoiseSession();
+    noise::NoiseResponderHandshake handshake;
+    NoiseCipherState *send_cipher{nullptr};
+    NoiseCipherState *recv_cipher{nullptr};
+    uint16_t frame_len{0};  // total frame size once the header is parsed, 0 until then
+    uint16_t frame_pos{0};  // bytes read or written so far
+    bool writing{false};    // a produced handshake frame is still being flushed
+    uint8_t frame_buf[noise::FRAME_HEADER_SIZE + 1 + noise::MAX_HANDSHAKE_SIZE];
+  };
+  bool noise_start_session_();
+  bool handle_noise_handshake_();
+  bool noise_try_read_frame_();
+  bool noise_try_write_frame_();
+  void noise_send_reject_(const LogString *reason);
+  ssize_t noise_decrypt_(uint8_t *buf, size_t len);
+  ssize_t noise_read_frame_blocking_(uint8_t *buf, size_t min_ciphertext, size_t max_ciphertext);
+  bool noise_readall_(uint8_t *buf, size_t len);
+  ssize_t noise_read_data_(uint8_t *buf, size_t capacity);
+  bool noise_write_byte_(uint8_t byte);
+#endif  // USE_OTA_ENCRYPTION
+
+  // Data-phase I/O dispatch: through the noise transport when a session is
+  // active, straight to the socket otherwise.
+  inline bool data_write_byte_(uint8_t byte) {
+#ifdef USE_OTA_ENCRYPTION
+    if (this->noise_ != nullptr)
+      return this->noise_write_byte_(byte);
+#endif
+    return this->write_byte_(byte);
+  }
+  // When encrypted, buf must have room for len + noise::MAC_SIZE bytes.
+  inline bool data_readall_(uint8_t *buf, size_t len) {
+#ifdef USE_OTA_ENCRYPTION
+    if (this->noise_ != nullptr)
+      return this->noise_readall_(buf, len);
+#endif
+    return this->readall_(buf, len);
+  }
+
   bool try_read_(size_t to_read, const LogString *desc);
   bool try_write_(size_t to_write, const LogString *desc);
 
@@ -91,6 +143,11 @@ class ESPHomeOTAComponent final : public ota::OTAComponent {
   std::string password_;
   std::unique_ptr<uint8_t[]> auth_buf_;
 #endif  // USE_OTA_PASSWORD
+#ifdef USE_OTA_ENCRYPTION
+  noise::NoiseContext noise_ctx_;
+  std::unique_ptr<NoiseSession> noise_;
+  uint8_t server_feature_flags_{0};  // as sent in the feature ack, bound into the prologue
+#endif                               // USE_OTA_ENCRYPTION
 
   socket::ListenSocket *server_{nullptr};
   std::unique_ptr<socket::Socket> client_;
@@ -98,6 +155,15 @@ class ESPHomeOTAComponent final : public ota::OTAComponent {
 
   uint32_t client_connect_time_{0};
   static constexpr size_t HANDSHAKE_BUF_SIZE = 5;
+  // Buffer size for OTA data transfer. The upload client derives its maximum
+  // encrypted frame plaintext from this (espota2.NOISE_MAX_PLAINTEXT is this
+  // minus the 16-byte MAC); both must change together.
+  static constexpr size_t OTA_BUFFER_SIZE = 1024;
+#ifdef USE_OTA_ENCRYPTION
+  // Shrinking the buffer would reject every frame a current CLI sends
+  static_assert(OTA_BUFFER_SIZE >= 1008 + noise::MAC_SIZE, "OTA_BUFFER_SIZE must fit a full encrypted data frame");
+#endif
+  static constexpr uint8_t MAGIC_BYTES[5] = {0x6C, 0x26, 0xF7, 0x5C, 0x45};
 #ifdef USE_OTA_PARTITIONS
   uint32_t running_app_offset_{0};
   size_t running_app_size_{0};

esphome/components/esphome/ota/ota_esphome_noise.cpp

@@ -0,0 +1,274 @@
+#include "ota_esphome.h"
+#ifdef USE_OTA
+#ifdef USE_OTA_ENCRYPTION
+#include "esphome/components/noise/noise.h"
+#include "esphome/components/ota/ota_backend.h"
+#include "esphome/core/log.h"
+
+#include <cstring>
+#include <new>
+
+#ifdef USE_ESP8266
+#include <pgmspace.h>
+#endif
+
+namespace esphome {
+
+static const char *const TAG = "esphome.ota";
+
+#ifdef USE_ESP8266
+static constexpr char OTA_NOISE_PROLOGUE_INIT[] PROGMEM = "NoiseOTAInit";
+#else
+static const char *const OTA_NOISE_PROLOGUE_INIT = "NoiseOTAInit";
+#endif
+static constexpr size_t OTA_NOISE_PROLOGUE_INIT_LEN = 12;  // strlen("NoiseOTAInit")
+
+ESPHomeOTAComponent::NoiseSession::~NoiseSession() {
+  if (this->send_cipher != nullptr) {
+    noise_cipherstate_free(this->send_cipher);
+  }
+  if (this->recv_cipher != nullptr) {
+    noise_cipherstate_free(this->recv_cipher);
+  }
+}
+
+/** Allocate the session and start the responder handshake.
+ *
+ * The prologue binds the whole plaintext preamble, so any tampering with the
+ * negotiation (a stripped feature flag, a changed version) breaks the first
+ * handshake MAC on either side:
+ *   "NoiseOTAInit" | magic(5) | OK,version | client_features | FEATURE_FLAGS,server_flags
+ */
+bool ESPHomeOTAComponent::noise_start_session_() {
+  // 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;
+  }
+
+  uint8_t prologue[OTA_NOISE_PROLOGUE_INIT_LEN + 5 + 2 + 1 + 2];
+#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
+  uint8_t *p = prologue + OTA_NOISE_PROLOGUE_INIT_LEN;
+  // Magic bytes, already validated in MAGIC_READ
+  std::memcpy(p, MAGIC_BYTES, sizeof(MAGIC_BYTES));
+  p += sizeof(MAGIC_BYTES);
+  // Our magic ack
+  *p++ = ota::OTA_RESPONSE_OK;
+  *p++ = USE_OTA_VERSION;
+  // The feature byte the client sent
+  *p++ = this->ota_features_;
+  // The feature ack we sent (noise requires the extended protocol)
+  *p++ = ota::OTA_RESPONSE_FEATURE_FLAGS;
+  *p++ = this->server_feature_flags_;
+
+  int err = this->noise_->handshake.init(this->noise_ctx_.get_psk(), prologue, sizeof(prologue));
+  if (err != 0) {
+    ESP_LOGW(TAG, "Handshake init: %s", LOG_STR_ARG(noise::noise_err_to_logstr(err)));
+    this->cleanup_connection_();
+    return false;
+  }
+  return true;
+}
+
+/** Drive the non-blocking handshake from loop(); returns true once the
+ * transport ciphers are ready and the session can enter the data phase.
+ * On failure the connection is cleaned up and false is returned.
+ */
+bool ESPHomeOTAComponent::handle_noise_handshake_() {
+  NoiseSession &s = *this->noise_;
+  while (true) {
+    if (s.writing) {
+      if (!this->noise_try_write_frame_()) {
+        return false;  // would block, or errored and cleaned up
+      }
+      s.writing = false;
+      s.frame_pos = 0;
+      s.frame_len = 0;
+    }
+    switch (s.handshake.action()) {
+      case noise::NoiseResponderHandshake::Action::ACTION_READ: {
+        if (!this->noise_try_read_frame_()) {
+          return false;
+        }
+        const uint16_t payload_len = s.frame_len - noise::FRAME_HEADER_SIZE;
+        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]);
+          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));
+          this->cleanup_connection_();
+          return false;
+        }
+        break;
+      }
+      case noise::NoiseResponderHandshake::Action::ACTION_WRITE: {
+        size_t msg_len = 0;
+        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)));
+          this->cleanup_connection_();
+          return false;
+        }
+        const uint16_t payload_len = msg_len + 1;
+        noise::write_frame_header(s.frame_buf, payload_len);
+        s.frame_buf[noise::FRAME_HEADER_SIZE] = noise::HANDSHAKE_STATUS_OK;
+        s.frame_len = noise::FRAME_HEADER_SIZE + payload_len;
+        s.frame_pos = 0;
+        s.writing = true;
+        break;
+      }
+      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)));
+          this->cleanup_connection_();
+          return false;
+        }
+        ESP_LOGD(TAG, "Noise handshake complete");
+        return true;
+      }
+      default: {
+        ESP_LOGW(TAG, "Bad handshake state");
+        this->cleanup_connection_();
+        return false;
+      }
+    }
+  }
+}
+
+/// 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;
+    }
+    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);
+      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.
+bool ESPHomeOTAComponent::noise_try_write_frame_() {
+  NoiseSession &s = *this->noise_;
+  while (s.frame_pos < s.frame_len) {
+    ssize_t written = this->client_->write(s.frame_buf + s.frame_pos, s.frame_len - s.frame_pos);
+    if (!this->handle_write_error_(written, LOG_STR("write noise frame"))) {
+      return false;
+    }
+    s.frame_pos += written;
+  }
+  return true;
+}
+
+/// Best-effort explicit reject frame so the client can log a readable reason.
+void ESPHomeOTAComponent::noise_send_reject_(const LogString *reason) {
+  uint8_t data[noise::FRAME_HEADER_SIZE + 1 + 32];
+  static_assert(sizeof(data) - noise::FRAME_HEADER_SIZE >= noise::MAC_FAILURE_PAYLOAD_SIZE,
+                "reject buffer must fit the MAC failure wire contract");
+  const size_t payload_len =
+      noise::format_reject_payload(data + noise::FRAME_HEADER_SIZE, sizeof(data) - noise::FRAME_HEADER_SIZE, reason);
+  noise::write_frame_header(data, payload_len);
+  this->client_->write(data, noise::FRAME_HEADER_SIZE + payload_len);  // Best effort, non-blocking
+}
+
+/// Decrypt a ciphertext in place; returns the plaintext size or -1.
+ssize_t ESPHomeOTAComponent::noise_decrypt_(uint8_t *buf, size_t len) {
+  NoiseBuffer mbuf;
+  noise_buffer_init(mbuf);
+  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)));
+    return -1;
+  }
+  return mbuf.size;
+}
+
+/** Blocking read of one frame whose ciphertext size must be within the given
+ * bounds, decrypted in place; returns the plaintext size, or -1 on error.
+ * buf needs max_ciphertext capacity.
+ */
+ssize_t ESPHomeOTAComponent::noise_read_frame_blocking_(uint8_t *buf, size_t min_ciphertext, size_t max_ciphertext) {
+  uint8_t header[noise::FRAME_HEADER_SIZE];
+  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);
+    return -1;
+  }
+  if (!this->readall_(buf, ciphertext_len)) {
+    return -1;
+  }
+  return this->noise_decrypt_(buf, ciphertext_len);
+}
+
+/** Blocking read of one frame whose plaintext must be exactly len bytes
+ * (control units are one unit per frame). buf needs len + noise::MAC_SIZE
+ * capacity; the plaintext lands at buf[0..len).
+ */
+bool ESPHomeOTAComponent::noise_readall_(uint8_t *buf, size_t len) {
+  return this->noise_read_frame_blocking_(buf, len + noise::MAC_SIZE, len + noise::MAC_SIZE) == (ssize_t) len;
+}
+
+/** Blocking read of one data-phase frame, decrypted in place; returns the
+ * plaintext size, or -1 on error. buf is the OTA_BUFFER_SIZE data buffer.
+ * The ciphertext must fit that buffer and its plaintext must fit what the
+ * caller accepts (the remaining image bytes).
+ */
+ssize_t ESPHomeOTAComponent::noise_read_data_(uint8_t *buf, size_t capacity) {
+  const size_t max_ciphertext = std::min(capacity + noise::MAC_SIZE, OTA_BUFFER_SIZE);
+  return this->noise_read_frame_blocking_(buf, noise::MAC_SIZE + 1, max_ciphertext);
+}
+
+/// Blocking write of one response byte as an encrypted frame.
+bool ESPHomeOTAComponent::noise_write_byte_(uint8_t byte) {
+  uint8_t frame[noise::FRAME_HEADER_SIZE + 1 + noise::MAC_SIZE];
+  frame[noise::FRAME_HEADER_SIZE] = byte;
+  NoiseBuffer mbuf;
+  noise_buffer_init(mbuf);
+  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)));
+    return false;
+  }
+  noise::write_frame_header(frame, mbuf.size);
+  return this->writeall_(frame, noise::FRAME_HEADER_SIZE + mbuf.size);
+}
+
+}  // namespace esphome
+#endif  // USE_OTA_ENCRYPTION
+#endif  // USE_OTA

esphome/components/ota/ota_backend.h

@@ -49,6 +49,7 @@ enum OTAResponseTypes {
   OTA_RESPONSE_ERROR_BOOTLOADER_VERIFY = 0x91,
   OTA_RESPONSE_ERROR_BOOTLOADER_UPDATE = 0x92,
   OTA_RESPONSE_ERROR_VERSION_DOWNGRADE = 0x93,
+  OTA_RESPONSE_ERROR_ENCRYPTION_REQUIRED = 0x94,
   OTA_RESPONSE_ERROR_UNKNOWN = 0xFF,
 };
 

esphome/core/defines.h

@@ -233,6 +233,7 @@
 #define USE_RUNTIME_IMAGE_JPEG
 #define USE_RUNTIME_STATS
 #define USE_OTA
+#define USE_OTA_ENCRYPTION
 #define USE_OTA_PASSWORD
 #define USE_OTA_VERSION 2
 #define USE_TIME_TIMEZONE

esphome/espota2.py

@@ -53,6 +53,7 @@ RESPONSE_ERROR_PARTITION_TABLE_UPDATE = 0x90
 RESPONSE_ERROR_BOOTLOADER_VERIFY = 0x91
 RESPONSE_ERROR_BOOTLOADER_UPDATE = 0x92
 RESPONSE_ERROR_VERSION_DOWNGRADE = 0x93
+RESPONSE_ERROR_ENCRYPTION_REQUIRED = 0x94
 RESPONSE_ERROR_UNKNOWN = 0xFF
 
 OTA_VERSION_1_0 = 1
@@ -63,8 +64,17 @@ MAGIC_BYTES = [0x6C, 0x26, 0xF7, 0x5C, 0x45]
 CLIENT_FEATURE_SUPPORTS_COMPRESSION = 0x01
 CLIENT_FEATURE_SUPPORTS_SHA256_AUTH = 0x02
 CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL = 0x04
+CLIENT_FEATURE_SUPPORTS_NOISE = 0x08
 SERVER_FEATURE_SUPPORTS_COMPRESSION = 0x01
 SERVER_FEATURE_SUPPORTS_PARTITION_ACCESS = 0x02
+SERVER_FEATURE_SUPPORTS_NOISE = 0x04
+
+NOISE_FRAME_INDICATOR = 0x01
+NOISE_HANDSHAKE_OK = 0x00
+# The device decrypts frames in its 1024-byte transfer buffer; the 16-byte
+# ChaCha20-Poly1305 MAC leaves this much plaintext per frame.
+NOISE_MAX_PLAINTEXT = 1008
+NOISE_PROLOGUE_INIT = b"NoiseOTAInit"
 
 # OTA types this client knows how to send. Future PRs that add bootloader/partition
 # updates extend this set. Anything outside the set is rejected up front so callers
@@ -171,6 +181,12 @@ _ERROR_MESSAGES: dict[int, str] = {
         "enabled: the new firmware's version must be newer than the version the "
         "device is currently running."
     ),
+    RESPONSE_ERROR_ENCRYPTION_REQUIRED: (
+        "The device requires an encrypted OTA connection but this upload has no "
+        "encryption key. Add 'encryption:' to the 'ota: platform: esphome' section "
+        "of the YAML this upload uses, or update your esphome installation if it "
+        "predates OTA encryption."
+    ),
     RESPONSE_ERROR_UNKNOWN: "Unknown error from ESP",
 }
 
@@ -305,12 +321,155 @@ def send_check(
         raise OTANetworkError(f"sending {msg}: {err}") from err
 
 
+class NoiseSocketWrapper:
+    """Runs the OTA session inside a Noise (ChaCha20-Poly1305) transport.
+
+    Presents the small subset of the socket API perform_ota uses (recv,
+    sendall, and the settimeout/setsockopt/close pass-throughs), so the rest
+    of the upload flow works unchanged. Frames on the wire are
+    indicator 0x01, 16-bit big-endian length, ciphertext; recv() serves the
+    decrypted stream from an internal buffer one frame at a time. Writes keep
+    each sendall() unit within one frame when it fits (the device expects
+    control units one per frame) and split larger data blocks at
+    NOISE_MAX_PLAINTEXT.
+    """
+
+    def __init__(self, sock: socket.socket, psk: str, prologue: bytes) -> None:
+        # Deliberately lazy: the noise stack (noiseprotocol, cryptography) is
+        # only imported when an encrypted upload actually runs.
+        try:
+            from aioesphomeapi.noise import NoiseHandshake
+        except ImportError as err:
+            raise OTAError(
+                "OTA encryption requires a newer aioesphomeapi; update your "
+                "esphome installation (pip install -U esphome) and retry"
+            ) from err
+        from cryptography.exceptions import InvalidTag
+
+        self._invalid_tag = InvalidTag
+        self._sock = sock
+        try:
+            self._handshake = NoiseHandshake(psk, prologue)
+        except ValueError as err:
+            raise OTAError(f"Invalid OTA encryption key: {err}") from err
+        self._encrypt = None
+        self._decrypt = None
+        self._buffer = b""
+
+    # Only harmless socket controls pass through; anything that moves bytes
+    # must go through the encrypted recv/sendall. Byte-moving socket methods
+    # (send, recv_into, ...) are deliberately not defined, so reaching for
+    # one raises AttributeError instead of leaking plaintext.
+    def settimeout(self, timeout: float | None) -> None:
+        self._sock.settimeout(timeout)
+
+    def setsockopt(self, level: int, optname: int, value: int) -> None:
+        self._sock.setsockopt(level, optname, value)
+
+    def close(self) -> None:
+        self._sock.close()
+
+    def do_handshake(self) -> None:
+        """Run the two-message NNpsk0 handshake and set up the transport ciphers."""
+        try:
+            self._send_frame(
+                bytes([NOISE_HANDSHAKE_OK]) + self._handshake.write_message()
+            )
+            payload = self._recv_frame()
+        except OSError as err:
+            raise OTANetworkError(f"noise handshake: {err}") from err
+        if not payload:
+            raise OTANetworkError("Device closed connection during the noise handshake")
+        if payload[0] != NOISE_HANDSHAKE_OK:
+            reason = payload[1:].decode("utf-8", "replace")
+            if reason == "Handshake MAC failure":
+                raise OTAError(
+                    "Device rejected the handshake; is the OTA encryption key correct?"
+                )
+            raise OTAError(f"Device rejected the noise handshake: {reason}")
+        try:
+            self._handshake.read_message(payload[1:])
+        except (ValueError, self._invalid_tag) as err:
+            # InvalidTag is a wrong key; ValueError covers a device sending an
+            # invalid curve point, which cryptography rejects during the DH
+            raise OTAError(
+                "Noise handshake failed; is the OTA encryption key correct?"
+            ) from err
+        self._encrypt, self._decrypt = self._handshake.get_ciphers()
+
+    def sendall(self, data: bytes) -> None:
+        frames: list[bytes] = []
+        for offset in range(0, len(data), NOISE_MAX_PLAINTEXT):
+            ciphertext = self._encrypt.encrypt(
+                data[offset : offset + NOISE_MAX_PLAINTEXT]
+            )
+            frames.append(self._frame_header(len(ciphertext)))
+            frames.append(ciphertext)
+        self._sock.sendall(b"".join(frames))
+
+    def recv(self, amount: int) -> bytes:
+        if not self._buffer:
+            ciphertext = self._recv_frame()
+            if not ciphertext:
+                return b""  # connection closed at a frame boundary
+            try:
+                self._buffer = self._decrypt.decrypt(ciphertext)
+            except self._invalid_tag as err:
+                # A fresh connection renegotiates the session, so this is
+                # retryable like other transport failures. The message names
+                # the MAC so repeated failures read as tampering or a cipher
+                # desync, not a flaky link.
+                raise OTANetworkError(
+                    "Noise decryption failed (MAC mismatch); frame corrupted or tampered"
+                ) from err
+            if not self._buffer:
+                # A MAC-only frame decrypts to nothing; reject it so recv's
+                # b"" always means the peer closed
+                raise OTANetworkError("Device sent an empty noise frame")
+        data = self._buffer[:amount]
+        self._buffer = self._buffer[amount:]
+        return data
+
+    @staticmethod
+    def _frame_header(length: int) -> bytes:
+        return bytes([NOISE_FRAME_INDICATOR, (length >> 8) & 0xFF, length & 0xFF])
+
+    def _send_frame(self, payload: bytes) -> None:
+        self._sock.sendall(self._frame_header(len(payload)) + payload)
+
+    def _recv_frame(self) -> bytes:
+        header = self._recv_exact(3, closed_ok=True)
+        if not header:
+            return b""  # connection closed at a frame boundary
+        # A malformed frame is a broken transport, not a device-reported
+        # error; raise the retryable class so the send-failure probe keeps its
+        # semantics and a fresh session is tried
+        if header[0] != NOISE_FRAME_INDICATOR:
+            raise OTANetworkError(f"Bad noise frame indicator 0x{header[0]:02X}")
+        length = (header[1] << 8) | header[2]
+        if length == 0:
+            raise OTANetworkError("Device sent an empty noise frame")
+        return self._recv_exact(length)
+
+    def _recv_exact(self, amount: int, closed_ok: bool = False) -> bytes:
+        data = b""
+        while len(data) < amount:
+            chunk = self._sock.recv(amount - len(data))
+            if not chunk:
+                if closed_ok and not data:
+                    return b""
+                raise OSError("connection closed inside a noise frame")
+            data += chunk
+        return data
+
+
 def perform_ota(
     sock: socket.socket,
     password: str | None,
     file_handle: io.IOBase,
     filename: Path,
     ota_type: int = OTA_TYPE_UPDATE_APP,
+    noise_psk: str | None = None,
 ) -> None:
     # Validate ota_type up front. It travels as a single byte on the wire, and
     # passing an out-of-range value would only surface as a ValueError from
@@ -325,6 +484,11 @@ def perform_ota(
             f"Unsupported OTA type 0x{ota_type:02X}; this ESPHome supports: {supported}"
         )
 
+    if noise_psk is not None and not noise_psk:
+        raise OTAError(
+            "An empty OTA encryption key was provided; refusing to upload in plaintext"
+        )
+
     file_contents = file_handle.read()
     file_size = len(file_contents)
     _LOGGER.info("Uploading %s (%s bytes)", filename, file_size)
@@ -347,6 +511,8 @@ def perform_ota(
         | CLIENT_FEATURE_SUPPORTS_SHA256_AUTH
         | CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL
     )
+    if noise_psk:
+        features_to_send |= CLIENT_FEATURE_SUPPORTS_NOISE
     send_check(sock, features_to_send, "features")
     features = receive_exactly(
         sock,
@@ -369,6 +535,31 @@ 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):
+            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 prologue binds every negotiation byte both sides saw, so any
+        # tampering with the plaintext preamble breaks the handshake.
+        prologue = (
+            NOISE_PROLOGUE_INIT
+            + bytes(MAGIC_BYTES)
+            + bytes([RESPONSE_OK, version, features_to_send])
+            + bytes([RESPONSE_FEATURE_FLAGS, features])
+        )
+        sock = NoiseSocketWrapper(sock, noise_psk, prologue)
+        sock.do_handshake()
+        _LOGGER.info("Encrypted connection established")
+
     if ota_type != OTA_TYPE_UPDATE_APP:
         # Any non-app OTA type requires the extended protocol and the
         # partition-access server feature. Reject up front so the user gets
@@ -572,6 +763,7 @@ def run_ota_impl_(
     password: str | None,
     filename: Path,
     ota_type: int = OTA_TYPE_UPDATE_APP,
+    noise_psk: str | None = None,
 ) -> tuple[int, str | None]:
     from esphome.core import CORE
 
@@ -636,7 +828,7 @@ def run_ota_impl_(
         reached_device = True
         with contextlib.closing(sock), Path(filename).open("rb") as file_handle:
             try:
-                perform_ota(sock, password, file_handle, filename, ota_type)
+                perform_ota(sock, password, file_handle, filename, ota_type, noise_psk)
             except OTANetworkError as err:
                 # Transient network failure; retry
                 last_error = str(err)
@@ -661,9 +853,12 @@ def run_ota(
     password: str | None,
     filename: Path,
     ota_type: int = OTA_TYPE_UPDATE_APP,
+    noise_psk: str | None = None,
 ) -> tuple[int, str | None]:
     try:
-        return run_ota_impl_(remote_host, remote_port, password, filename, ota_type)
+        return run_ota_impl_(
+            remote_host, remote_port, password, filename, ota_type, noise_psk
+        )
     except OTAError as err:
         _LOGGER.error(err)
         return 1, None

tests/component_tests/ota/test_esphome_ota.py

@@ -8,17 +8,25 @@ from typing import Any
 import pytest
 
 from esphome import config_validation as cv
-from esphome.components.esphome.ota import ota_esphome_final_validate
+from esphome.components.esphome.ota import (
+    AUTO_LOAD,
+    FILTER_SOURCE_FILES,
+    _validate_no_password_with_encryption,
+    ota_esphome_final_validate,
+)
 from esphome.const import (
+    CONF_API,
+    CONF_ENCRYPTION,
     CONF_ESPHOME,
     CONF_ID,
+    CONF_KEY,
     CONF_OTA,
     CONF_PASSWORD,
     CONF_PLATFORM,
     CONF_PORT,
     CONF_VERSION,
 )
-from esphome.core import ID
+from esphome.core import CORE, ID
 import esphome.final_validate as fv
 
 
@@ -103,3 +111,301 @@ def test_non_esphome_ota_unaffected() -> None:
         assert len(updated[CONF_OTA]) == 3
     finally:
         fv.full_config.reset(token)
+
+
+API_KEY = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+OTHER_KEY = "AQIDBAUGBwgJCgsMDQ4PEBESExQVFhcYGRobHB0eHyA="
+ZEROS_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
+
+
+def test_encryption_key_inherited_from_api() -> None:
+    """A bare encryption block resolves to the api encryption key."""
+    full_conf = {
+        CONF_API: {CONF_ENCRYPTION: {CONF_KEY: API_KEY}},
+        CONF_OTA: [_make_ota_config(port=3232, **{CONF_ENCRYPTION: {}})],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        ota_esphome_final_validate({})
+        updated = fv.full_config.get()
+        assert updated[CONF_OTA][0][CONF_ENCRYPTION][CONF_KEY] == API_KEY
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_explicit_key_matching_api_accepted() -> None:
+    """An explicit ota key equal to the api key validates."""
+    full_conf = {
+        CONF_API: {CONF_ENCRYPTION: {CONF_KEY: API_KEY}},
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: API_KEY}})
+        ],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        ota_esphome_final_validate({})
+        updated = fv.full_config.get()
+        assert updated[CONF_OTA][0][CONF_ENCRYPTION][CONF_KEY] == API_KEY
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_key_differing_from_api_rejected() -> None:
+    """There is one key per device; an ota key differing from the api key raises."""
+    full_conf = {
+        CONF_API: {CONF_ENCRYPTION: {CONF_KEY: API_KEY}},
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}})
+        ],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with pytest.raises(cv.Invalid, match="must match the 'api' encryption key"):
+            ota_esphome_final_validate({})
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_explicit_key_without_api_encryption_accepted() -> None:
+    """An explicit ota key with a plaintext api has nothing to match; it stands."""
+    full_conf = {
+        CONF_API: {},
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}})
+        ],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        ota_esphome_final_validate({})
+        updated = fv.full_config.get()
+        assert updated[CONF_OTA][0][CONF_ENCRYPTION][CONF_KEY] == OTHER_KEY
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_without_any_key_rejected() -> None:
+    """A bare encryption block with no api key to inherit raises."""
+    full_conf = {
+        CONF_API: {},
+        CONF_OTA: [_make_ota_config(port=3232, **{CONF_ENCRYPTION: {}})],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with pytest.raises(cv.Invalid, match="no 'api' encryption key to inherit"):
+            ota_esphome_final_validate({})
+    finally:
+        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 = {
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: API_KEY}}),
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}}),
+        ]
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with pytest.raises(cv.Invalid, match="encryption is inconsistent"):
+            ota_esphome_final_validate({})
+    finally:
+        fv.full_config.reset(token)
+
+
+@pytest.mark.parametrize("keyed_first", [True, False])
+def test_encryption_bare_and_keyed_blocks_merge(keyed_first: bool) -> None:
+    """A bare encryption block (package/device split) is compatible with a
+    keyed one on the same port; the merge resolves to the keyed result."""
+    keyed = _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}})
+    bare = _make_ota_config(port=3232, **{CONF_ENCRYPTION: {}})
+    full_conf = {
+        CONF_OTA: [keyed, bare] if keyed_first else [bare, keyed],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        ota_esphome_final_validate({})
+        updated = fv.full_config.get()
+        assert len(updated[CONF_OTA]) == 1
+        assert updated[CONF_OTA][0][CONF_ENCRYPTION][CONF_KEY] == OTHER_KEY
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_runtime_provisioned_api_key_not_inheritable() -> None:
+    """A keyless api encryption block provisions its key at runtime; a bare
+    ota encryption block cannot inherit it and the message says so."""
+    full_conf = {
+        CONF_API: {CONF_ENCRYPTION: {}},
+        CONF_OTA: [_make_ota_config(port=3232, **{CONF_ENCRYPTION: {}})],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with pytest.raises(cv.Invalid, match="provisioned at runtime"):
+            ota_esphome_final_validate({})
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_explicit_key_with_runtime_provisioned_api_accepted() -> None:
+    """The documented remedy for a runtime-provisioned api key: set an
+    explicit ota key."""
+    full_conf = {
+        CONF_API: {CONF_ENCRYPTION: {}},
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}})
+        ],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        ota_esphome_final_validate({})
+        updated = fv.full_config.get()
+        assert updated[CONF_OTA][0][CONF_ENCRYPTION][CONF_KEY] == OTHER_KEY
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_with_web_server_ota_rejected() -> None:
+    """With the web_server component the plaintext /update endpoint is always
+    on, a full bypass of the encryption; the combination fails closed."""
+    full_conf = {
+        "web_server": {},
+        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 pytest.raises(cv.Invalid, match="plaintext HTTP"):
+            ota_esphome_final_validate({})
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_with_captive_portal_web_server_ota_warns(
+    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."""
+    full_conf = {
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}}),
+            {CONF_PLATFORM: "web_server", CONF_ID: ID("ota_ws", is_manual=False)},
+        ],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with caplog.at_level(logging.WARNING):
+            ota_esphome_final_validate({})
+        assert any("captive_portal" in record.message for record in caplog.records)
+        esphome_conf = next(
+            conf
+            for conf in fv.full_config.get()[CONF_OTA]
+            if conf.get(CONF_PLATFORM) == CONF_ESPHOME
+        )
+        assert esphome_conf[CONF_ENCRYPTION][CONF_KEY] == OTHER_KEY
+    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 = {
+        CONF_OTA: [
+            _make_ota_config(port=3232),
+            {CONF_PLATFORM: "web_server", CONF_ID: ID("ota_ws", is_manual=False)},
+        ],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        ota_esphome_final_validate({})
+        assert len(fv.full_config.get()[CONF_OTA]) == 2
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_auto_load_pulls_noise_only_for_encryption() -> None:
+    """A plain ota entry must never pull noise-c into the build."""
+    assert AUTO_LOAD({CONF_PORT: 3232}) == ["sha256", "socket"]
+    assert "noise" in AUTO_LOAD({CONF_ENCRYPTION: {}})
+    # Tooling probes must get the maximal set: None from dependency
+    # resolution, {} from the components-graph platform probe
+    assert "noise" in AUTO_LOAD(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_password_with_encryption_rejected() -> None:
+    """The password and encryption options are mutually exclusive."""
+    config = {CONF_PASSWORD: "pw", CONF_ENCRYPTION: {CONF_KEY: API_KEY}}
+    with pytest.raises(cv.Invalid, match="cannot be combined"):
+        _validate_no_password_with_encryption(config)
+
+
+def test_password_alone_accepted() -> None:
+    """A password without encryption still validates."""
+    config = {CONF_PASSWORD: "pw"}
+    assert _validate_no_password_with_encryption(config) is config
+
+
+def test_merged_password_and_encryption_rejected() -> None:
+    """A password block and an encryption block merged on one port raise."""
+    full_conf = {
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_PASSWORD: "pw"}),
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: API_KEY}}),
+        ]
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with pytest.raises(cv.Invalid, match="cannot be combined"):
+            ota_esphome_final_validate({})
+    finally:
+        fv.full_config.reset(token)

tests/components/ota/encryption.yaml

@@ -0,0 +1,9 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+ota:
+  - platform: esphome
+    port: 3288
+    encryption:
+      key: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="

tests/components/ota/encryption_inherit.yaml

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

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

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

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

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

tests/components/ota/test-encryption.rp2040-ard.yaml

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

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

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

tests/integration/fixtures/host_ota_encrypted.yaml

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

tests/integration/test_host_ota.py

@@ -10,6 +10,7 @@ from __future__ import annotations
 import asyncio
 from collections.abc import Generator
 from contextlib import contextmanager
+import functools
 import socket
 
 import pytest
@@ -111,6 +112,62 @@ async def test_host_ota_self_update(
             assert proc.pid == pid_before
 
 
+@pytest.mark.asyncio
+async def test_host_ota_encrypted(
+    yaml_config: str,
+    write_yaml_config: ConfigWriter,
+    compile_esphome: CompileFunction,
+    reserved_tcp_port: tuple[int, socket.socket],
+) -> 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()
+
+        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)
+
+        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
+
+            # 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
+            )
+            assert rc == 1, "plaintext upload to an encrypted device must fail"
+            await asyncio.sleep(0.5)
+            assert proc.returncode is None, "process died on rejected plaintext OTA"
+
+            # The encrypted upload goes through and the device re-execs
+            rc, _ = await loop.run_in_executor(
+                None,
+                functools.partial(
+                    espota2.run_ota,
+                    LOCALHOST,
+                    ota_port,
+                    None,
+                    binary_path,
+                    noise_psk=noise_psk,
+                ),
+            )
+            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
 async def test_host_ota_rejects_garbage(
     yaml_config: str,

tests/unit_tests/test_espota2_noise.py

@@ -0,0 +1,407 @@
+"""Unit tests for encrypted OTA uploads in esphome.espota2.
+
+A fake device implementing the responder side of the wire protocol (via
+noiseprotocol, which esphome already has through aioesphomeapi) serves a real
+TCP loopback connection, so these exercise the actual handshake, framing, and
+cipher interop of the client code. Tests that need the client-side crypto skip
+when the installed aioesphomeapi predates the noise module.
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import io
+from pathlib import Path
+import socket
+import sys
+import threading
+from unittest.mock import Mock, patch
+
+import pytest
+
+from esphome import espota2
+
+PSK = base64.b64encode(bytes(range(32))).decode()
+OTHER_PSK = base64.b64encode(bytes(range(1, 33))).decode()
+
+MAGIC = bytes(espota2.MAGIC_BYTES)
+
+
+def _recv_exact(sock: socket.socket, amount: int) -> bytes:
+    data = b""
+    while len(data) < amount:
+        chunk = sock.recv(amount - len(data))
+        if not chunk:
+            raise ConnectionError("client closed")
+        data += chunk
+    return data
+
+
+def _frame(payload: bytes) -> bytes:
+    return (
+        bytes([espota2.NOISE_FRAME_INDICATOR, len(payload) >> 8, len(payload) & 0xFF])
+        + payload
+    )
+
+
+def _send_frame(sock: socket.socket, payload: bytes) -> None:
+    sock.sendall(_frame(payload))
+
+
+def _recv_frame(sock: socket.socket) -> bytes:
+    header = _recv_exact(sock, 3)
+    assert header[0] == 0x01
+    return _recv_exact(sock, (header[1] << 8) | header[2])
+
+
+class FakeEncryptedDevice(threading.Thread):
+    """Responder side of the encrypted OTA wire protocol."""
+
+    def __init__(
+        self,
+        psk: str = PSK,
+        version: int = 2,
+        offer_noise: bool = True,
+        require_noise: bool = True,
+        prologue_features_override: int | None = None,
+    ) -> None:
+        super().__init__(daemon=True)
+        self.psk = psk
+        self.version = version
+        self.offer_noise = offer_noise
+        self.require_noise = require_noise
+        self.prologue_features_override = prologue_features_override
+        self.received: bytes | None = None
+        self.error: Exception | None = None
+        self.listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.listener.bind(("127.0.0.1", 0))
+        self.listener.listen(1)
+        self.port = self.listener.getsockname()[1]
+
+    def run(self) -> None:
+        try:
+            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:
+            self.listener.close()
+
+    def join_and_check(self) -> None:
+        self.join(timeout=10)
+        assert not self.is_alive(), "fake device did not finish"
+        if self.error is not None:
+            raise self.error
+
+    def _serve(self, sock: socket.socket) -> None:
+        assert _recv_exact(sock, 5) == MAGIC
+        sock.sendall(bytes([espota2.RESPONSE_OK, self.version]))
+        features = _recv_exact(sock, 1)[0]
+        noise_negotiated = bool(
+            features & espota2.CLIENT_FEATURE_SUPPORTS_NOISE
+            and features & espota2.CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL
+        )
+        if self.require_noise and not noise_negotiated:
+            sock.sendall(bytes([espota2.RESPONSE_ERROR_ENCRYPTION_REQUIRED]))
+            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
+
+        from cryptography.exceptions import InvalidTag
+        from noise.connection import NoiseConnection
+
+        prologue_features = (
+            features
+            if self.prologue_features_override is None
+            else self.prologue_features_override
+        )
+        prologue = (
+            espota2.NOISE_PROLOGUE_INIT
+            + MAGIC
+            + bytes([espota2.RESPONSE_OK, self.version, prologue_features])
+            + bytes([espota2.RESPONSE_FEATURE_FLAGS, server_flags])
+        )
+        proto = NoiseConnection.from_name(b"Noise_NNpsk0_25519_ChaChaPoly_SHA256")
+        proto.set_as_responder()
+        proto.set_psks(base64.b64decode(self.psk))
+        proto.set_prologue(prologue)
+        proto.start_handshake()
+
+        msg1 = _recv_frame(sock)
+        assert msg1[0] == 0x00
+        try:
+            proto.read_message(msg1[1:])
+        except InvalidTag:
+            _send_frame(sock, b"\x01Handshake MAC failure")
+            return
+        _send_frame(sock, b"\x00" + bytes(proto.write_message()))
+
+        def send_byte(byte: int) -> None:
+            _send_frame(sock, proto.encrypt(bytes([byte])))
+
+        def recv_unit(length: int) -> bytes:
+            plaintext = proto.decrypt(_recv_frame(sock))
+            assert len(plaintext) == length, "control units must be one per frame"
+            return plaintext
+
+        send_byte(espota2.RESPONSE_AUTH_OK)
+        recv_unit(1)  # ota type
+        size = int.from_bytes(recv_unit(4), "big")
+        send_byte(espota2.RESPONSE_UPDATE_PREPARE_OK)
+        md5_hex = recv_unit(32)
+        send_byte(espota2.RESPONSE_BIN_MD5_OK)
+
+        received = b""
+        acked = 0
+        while len(received) < size:
+            plaintext = proto.decrypt(_recv_frame(sock))
+            assert 0 < len(plaintext) <= espota2.NOISE_MAX_PLAINTEXT
+            received += plaintext
+            if self.version >= espota2.OTA_VERSION_2_0:
+                while acked + espota2.UPLOAD_BLOCK_SIZE <= len(received) or (
+                    len(received) == size and acked < size
+                ):
+                    send_byte(espota2.RESPONSE_CHUNK_OK)
+                    acked += espota2.UPLOAD_BLOCK_SIZE
+        assert hashlib.md5(received).hexdigest().encode() == md5_hex
+        send_byte(espota2.RESPONSE_RECEIVE_OK)
+        send_byte(espota2.RESPONSE_UPDATE_END_OK)
+        assert recv_unit(1) == bytes([espota2.RESPONSE_OK])
+        self.received = received
+
+
+def _upload(
+    device: FakeEncryptedDevice, firmware: bytes, noise_psk: str | None
+) -> None:
+    device.start()
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sock.settimeout(10)
+    sock.connect(("127.0.0.1", device.port))
+    try:
+        espota2.perform_ota(
+            sock, None, io.BytesIO(firmware), Path("firmware.bin"), noise_psk=noise_psk
+        )
+    finally:
+        sock.close()
+
+
+def test_encrypted_upload_success() -> None:
+    """A full encrypted v2 upload spanning several 8192-byte blocks."""
+    pytest.importorskip("aioesphomeapi.noise")
+    firmware = bytes(range(256)) * 80  # 20480 bytes, crosses chunk-ack boundaries
+    device = FakeEncryptedDevice()
+    with patch("time.sleep"):
+        _upload(device, firmware, PSK)
+    device.join_and_check()
+    assert device.received == firmware
+
+
+def test_encrypted_upload_version_1() -> None:
+    """Version 1 protocol (no chunk acks) works through the noise transport."""
+    pytest.importorskip("aioesphomeapi.noise")
+    firmware = b"v1 firmware image" * 100
+    device = FakeEncryptedDevice(version=1)
+    with patch("time.sleep"):
+        _upload(device, firmware, PSK)
+    device.join_and_check()
+    assert device.received == firmware
+
+
+def test_wrong_key_fails_with_clear_error() -> None:
+    """A key mismatch surfaces the device's handshake reject readably."""
+    pytest.importorskip("aioesphomeapi.noise")
+    device = FakeEncryptedDevice(psk=OTHER_PSK)
+    with pytest.raises(espota2.OTAError, match="encryption key correct"):
+        _upload(device, b"firmware", PSK)
+    device.join_and_check()
+
+
+def test_tampered_negotiation_breaks_handshake() -> None:
+    """A negotiation byte differing between the sides breaks the prologue MAC."""
+    pytest.importorskip("aioesphomeapi.noise")
+    device = FakeEncryptedDevice(
+        prologue_features_override=espota2.CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL
+    )
+    with pytest.raises(espota2.OTAError, match="encryption key correct"):
+        _upload(device, b"firmware", PSK)
+    device.join_and_check()
+
+
+def test_client_fails_closed_when_device_lacks_encryption() -> None:
+    """With a key configured, a device not offering noise aborts the upload."""
+    device = FakeEncryptedDevice(offer_noise=False, require_noise=False)
+    with pytest.raises(espota2.OTAError, match="refusing to send the image"):
+        _upload(device, b"firmware", PSK)
+    device.join_and_check()
+
+
+def test_plaintext_client_gets_encryption_required_error() -> None:
+    """A client without a key gets the device's 0x94 error message."""
+    device = FakeEncryptedDevice()
+    with pytest.raises(espota2.OTAError, match="requires an encrypted OTA"):
+        _upload(device, b"firmware", None)
+    device.join_and_check()
+
+
+def test_missing_aioesphomeapi_noise_module_message() -> None:
+    """An aioesphomeapi without the noise module produces a clear error."""
+    with (
+        patch.dict(sys.modules, {"aioesphomeapi.noise": None}),
+        pytest.raises(espota2.OTAError, match="requires a newer aioesphomeapi"),
+    ):
+        espota2.NoiseSocketWrapper(Mock(), PSK, b"prologue")
+
+
+class ScriptedSocket:
+    """Serves scripted recv chunks; b"" means the peer closed."""
+
+    def __init__(self, *chunks: bytes | Exception) -> None:
+        self.chunks = list(chunks)
+        self.sent: list[bytes] = []
+
+    def sendall(self, data: bytes) -> None:
+        self.sent.append(data)
+
+    def settimeout(self, timeout: float) -> None:
+        pass
+
+    def recv(self, amount: int) -> bytes:
+        if not self.chunks:
+            return b""
+        chunk = self.chunks[0]
+        if isinstance(chunk, Exception):
+            self.chunks.pop(0)
+            raise chunk
+        take, rest = chunk[:amount], chunk[amount:]
+        if rest:
+            self.chunks[0] = rest
+        else:
+            self.chunks.pop(0)
+        return take
+
+
+def _wrapper(*chunks: bytes | Exception) -> espota2.NoiseSocketWrapper:
+    pytest.importorskip("aioesphomeapi.noise")
+    return espota2.NoiseSocketWrapper(ScriptedSocket(*chunks), PSK, b"prologue")
+
+
+def test_wrapper_rejects_malformed_psk() -> None:
+    pytest.importorskip("aioesphomeapi.noise")
+    with pytest.raises(espota2.OTAError, match="Invalid OTA encryption key"):
+        espota2.NoiseSocketWrapper(ScriptedSocket(), "not-base64!!!", b"prologue")
+
+
+def test_handshake_socket_error_is_network_error() -> None:
+    wrapper = _wrapper(OSError("boom"))
+    with pytest.raises(espota2.OTANetworkError, match="noise handshake"):
+        wrapper.do_handshake()
+
+
+def test_handshake_closed_at_frame_boundary() -> None:
+    wrapper = _wrapper()
+    with pytest.raises(espota2.OTANetworkError, match="closed connection during"):
+        wrapper.do_handshake()
+
+
+def test_handshake_reject_with_other_reason() -> None:
+    wrapper = _wrapper(_frame(b"\x01Handshake error"))
+    with pytest.raises(
+        espota2.OTAError, match="rejected the noise handshake: Handshake error"
+    ):
+        wrapper.do_handshake()
+
+
+def test_handshake_garbage_second_message() -> None:
+    """A valid-looking point with a garbage MAC fails cleanly."""
+    wrapper = _wrapper(_frame(b"\x00" + bytes(range(48))))
+    with pytest.raises(
+        espota2.OTAError, match="handshake failed; is the OTA encryption key"
+    ):
+        wrapper.do_handshake()
+
+
+def test_handshake_invalid_curve_point() -> None:
+    """An all-zero x25519 point is rejected as a clean error, not a crash."""
+    wrapper = _wrapper(_frame(b"\x00" + bytes(48)))
+    with pytest.raises(
+        espota2.OTAError, match="handshake failed; is the OTA encryption key"
+    ):
+        wrapper.do_handshake()
+
+
+def test_recv_closed_at_frame_boundary_returns_empty() -> None:
+    wrapper = _wrapper()
+    assert wrapper.recv(1) == b""
+
+
+def test_recv_corrupt_frame_is_retryable_network_error() -> None:
+    from cryptography.exceptions import InvalidTag
+
+    wrapper = _wrapper(_frame(b"ciphertext"))
+    wrapper._decrypt = Mock(decrypt=Mock(side_effect=InvalidTag()))
+    with pytest.raises(espota2.OTANetworkError, match="decryption failed"):
+        wrapper.recv(1)
+
+
+def test_wrapper_blocks_unencrypted_socket_methods() -> None:
+    """Byte-moving socket methods must not bypass the encrypted transport."""
+    wrapper = _wrapper()
+    # The harmless socket controls pass through to the wrapped socket
+    wrapper._sock = Mock()
+    wrapper.settimeout(1)
+    wrapper._sock.settimeout.assert_called_once_with(1)
+    wrapper.setsockopt(6, 1, 1)
+    wrapper._sock.setsockopt.assert_called_once_with(6, 1, 1)
+    wrapper.close()
+    wrapper._sock.close.assert_called_once_with()
+    with pytest.raises(AttributeError):
+        _ = wrapper.send
+    with pytest.raises(AttributeError):
+        _ = wrapper.recv_into
+
+
+def test_recv_empty_plaintext_frame_is_protocol_error() -> None:
+    """A MAC-only frame decrypts to nothing; b'' from recv must mean close."""
+    wrapper = _wrapper(_frame(bytes(16)))
+    wrapper._decrypt = Mock(decrypt=Mock(return_value=b""))
+    with pytest.raises(espota2.OTANetworkError, match="empty noise frame"):
+        wrapper.recv(1)
+
+
+def test_recv_frame_bad_indicator_is_retryable() -> None:
+    wrapper = _wrapper(b"\x02\x00\x01x")
+    with pytest.raises(espota2.OTANetworkError, match="Bad noise frame indicator"):
+        wrapper._recv_frame()
+
+
+def test_recv_frame_zero_length_is_retryable() -> None:
+    wrapper = _wrapper(bytes([espota2.NOISE_FRAME_INDICATOR, 0, 0]))
+    with pytest.raises(espota2.OTANetworkError, match="empty noise frame"):
+        wrapper._recv_frame()
+
+
+def test_perform_ota_blank_key_refuses_plaintext() -> None:
+    with pytest.raises(espota2.OTAError, match="empty OTA encryption key"):
+        espota2.perform_ota(
+            ScriptedSocket(), None, io.BytesIO(b"x"), Path("f.bin"), noise_psk=""
+        )
+
+
+def test_recv_exact_closed_mid_frame() -> None:
+    wrapper = _wrapper(_frame(b"partial")[:5])
+    with pytest.raises(OSError, match="closed inside a noise frame"):
+        wrapper._recv_frame()
+
+
+def test_recv_serves_buffered_plaintext_without_new_frame() -> None:
+    """A second recv drains the decrypted buffer without reading another frame."""
+    wrapper = _wrapper(_frame(b"ciphertext"))
+    wrapper._decrypt = Mock(decrypt=Mock(return_value=b"AB"))
+    assert wrapper.recv(1) == b"A"  # reads and decrypts one frame
+    assert wrapper.recv(1) == b"B"  # served from the buffer, no new frame
+    wrapper._decrypt.decrypt.assert_called_once()

tests/unit_tests/test_main.py

@@ -82,7 +82,9 @@ from esphome.const import (
     CONF_BAUD_RATE,
     CONF_BROKER,
     CONF_DISABLED,
+    CONF_ENCRYPTION,
     CONF_ESPHOME,
+    CONF_KEY,
     CONF_LEVEL,
     CONF_LOG,
     CONF_LOG_TOPIC,
@@ -2104,10 +2106,65 @@ 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
+        ["192.168.1.100"], 3232, "secret", expected_firmware, OTA_TYPE_UPDATE_APP, None
     )
 
 
+def test_upload_program_ota_encryption_key(
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """The resolved encryption key is passed through to run_ota."""
+    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_OTA: [
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 3232,
+                CONF_ENCRYPTION: {CONF_KEY: key},
+            }
+        ]
+    }
+    exit_code, host = upload_program(config, MockArgs(), ["192.168.1.100"])
+
+    assert exit_code == 0
+    assert host == "192.168.1.100"
+    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
+    )
+
+
+def test_upload_program_ota_encryption_without_key_fails_closed(
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """An encryption block with no resolved key must never upload plaintext."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+
+    config = {
+        CONF_OTA: [
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 3232,
+                CONF_ENCRYPTION: {},
+            }
+        ]
+    }
+    with pytest.raises(EsphomeError, match="no key was resolved"):
+        upload_program(config, MockArgs(), ["192.168.1.100"])
+    mock_run_ota.assert_not_called()
+
+
 def test_upload_program_ota_with_file_arg(
     mock_run_ota: Mock,
     mock_get_port_type: Mock,
@@ -2135,7 +2192,7 @@ 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
+        ["192.168.1.100"], 3232, None, Path("custom.bin"), OTA_TYPE_UPDATE_APP, None
     )
 
 
@@ -2190,6 +2247,7 @@ def test_upload_program_ota_partition_table_with_file_arg(
         None,
         partition_file,
         OTA_TYPE_UPDATE_PARTITION_TABLE,
+        None,
     )
 
 
@@ -2251,6 +2309,7 @@ def test_upload_program_ota_partition_table_mqttip(
         None,
         partition_file,
         OTA_TYPE_UPDATE_PARTITION_TABLE,
+        None,
     )
 
 
@@ -2438,6 +2497,7 @@ def test_upload_program_ota_bootloader_with_file_arg(
         None,
         bootloader_file,
         OTA_TYPE_UPDATE_BOOTLOADER,
+        None,
     )
 
 
@@ -2600,6 +2660,42 @@ def test_has_web_server_logging_respects_log_disabled() -> None:
     assert has_web_server_logging() is False
 
 
+def test_upload_program_web_server_warns_when_encryption_configured(
+    mock_run_web_server_ota: Mock,
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Explicitly picking web_server OTA on an encrypted config warns about
+    the plaintext upload path."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+    mock_run_web_server_ota.return_value = (0, "192.168.1.100")
+
+    config = {
+        CONF_OTA: [
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 3232,
+                CONF_ENCRYPTION: {CONF_KEY: "test_key"},
+            },
+            {CONF_PLATFORM: CONF_WEB_SERVER},
+        ],
+        CONF_WEB_SERVER: {
+            CONF_PORT: 80,
+            CONF_AUTH: {CONF_USERNAME: "admin", CONF_PASSWORD: "pw"},
+        },
+    }
+    args = MockArgs(ota_platform=CONF_WEB_SERVER)
+    with caplog.at_level(logging.WARNING):
+        exit_code, _ = upload_program(config, args, ["192.168.1.100"])
+
+    assert exit_code == 0
+    assert any("plaintext HTTP" in record.message for record in caplog.records)
+    mock_run_ota.assert_not_called()
+
+
 def test_upload_program_web_server_only_auto_dispatches(
     mock_run_web_server_ota: Mock,
     mock_run_ota: Mock,
@@ -2890,7 +2986,7 @@ 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
+        ["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, None
     )
 
 
@@ -2940,7 +3036,7 @@ 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
+        ["192.168.1.50"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, None
     )
     # Verify warning was logged
     assert "MQTT IP discovery failed" in caplog.text
@@ -5115,6 +5211,7 @@ def test_upload_program_ota_static_ip_with_mqttip(
         None,
         expected_firmware,
         OTA_TYPE_UPDATE_APP,
+        None,
     )
 
 
@@ -5164,6 +5261,7 @@ def test_upload_program_ota_multiple_mqttip_resolves_once(
         None,
         expected_firmware,
         OTA_TYPE_UPDATE_APP,
+        None,
     )
 
 
@@ -5341,7 +5439,7 @@ 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
+        ["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, None
     )