[ota] Offer encryption with the api key so enabling it works over OTA (#18979)

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/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):