[ota] Offer encryption with the api key so enabling it works over OTA

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_api_key_offers_encryption.yaml

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

tests/integration/test_host_ota.py

@@ -168,6 +168,143 @@ async def test_host_ota_encrypted(
             assert proc.pid == pid_before
 
 
+class _RebootCounter:
+    """Counts safe reboots so a test can wait for the nth one."""
+
+    def __init__(self) -> None:
+        self._seen = asyncio.Event()
+        self.count = 0
+
+    def on_log(self, line: str) -> None:
+        if "Rebooting safely" in line:
+            self.count += 1
+            self._seen.set()
+
+    async def wait(self, count: int, timeout: float = 10.0) -> None:
+        async with asyncio.timeout(timeout):
+            while self.count < count:
+                self._seen.clear()
+                await self._seen.wait()
+
+
+@pytest.mark.asyncio
+async def test_host_ota_api_key_offers_encryption(
+    yaml_config: str,
+    write_yaml_config: ConfigWriter,
+    compile_esphome: CompileFunction,
+    reserved_tcp_port: tuple[int, socket.socket],
+) -> None:
+    """With only an api key the device takes both a plaintext upload and an
+    encrypted one using that key, which is the enablement path for
+    `ota: encryption:`."""
+    pytest.importorskip("aioesphomeapi.noise")
+    api_key = "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()
+        reboots = _RebootCounter()
+
+        async with run_binary(binary_path, line_callback=reboots.on_log) as (
+            proc,
+            lines,
+        ):
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            pid_before = proc.pid
+
+            rc, _ = await loop.run_in_executor(
+                None, espota2.run_ota, LOCALHOST, ota_port, None, binary_path
+            )
+            assert rc == 0, "plaintext upload to an offering device must succeed"
+            await reboots.wait(1)
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            assert proc.pid == pid_before
+
+            rc, _ = await loop.run_in_executor(
+                None,
+                functools.partial(
+                    espota2.run_ota,
+                    LOCALHOST,
+                    ota_port,
+                    None,
+                    binary_path,
+                    noise_psk=api_key,
+                ),
+            )
+            assert rc == 0, "encrypted upload with the api key must succeed"
+            await reboots.wait(2)
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            assert proc.returncode is None, "process exited instead of execing"
+            assert proc.pid == pid_before
+            assert any("Encryption: offered" in line for line in lines)
+
+
+@pytest.mark.asyncio
+async def test_host_ota_api_key_offer_with_password(
+    yaml_config: str,
+    write_yaml_config: ConfigWriter,
+    compile_esphome: CompileFunction,
+    reserved_tcp_port: tuple[int, socket.socket],
+) -> None:
+    """The OTA password still guards plaintext uploads on an offering device
+    while the api key alone authenticates an encrypted one."""
+    pytest.importorskip("aioesphomeapi.noise")
+    api_key = "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()
+        reboots = _RebootCounter()
+
+        async with run_binary(binary_path, line_callback=reboots.on_log) as (
+            proc,
+            _lines,
+        ):
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            pid_before = proc.pid
+
+            rc, _ = await loop.run_in_executor(
+                None, espota2.run_ota, LOCALHOST, ota_port, None, binary_path
+            )
+            assert rc == 1, "plaintext upload without the password must fail"
+            await asyncio.sleep(0.5)
+            assert proc.returncode is None, "process died on rejected upload"
+
+            rc, _ = await loop.run_in_executor(
+                None, espota2.run_ota, LOCALHOST, ota_port, "hunter2", binary_path
+            )
+            assert rc == 0, "plaintext upload with the password must succeed"
+            await reboots.wait(1)
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            assert proc.pid == pid_before
+
+            rc, _ = await loop.run_in_executor(
+                None,
+                functools.partial(
+                    espota2.run_ota,
+                    LOCALHOST,
+                    ota_port,
+                    None,
+                    binary_path,
+                    noise_psk=api_key,
+                ),
+            )
+            assert rc == 0, "encrypted upload with the api key must succeed"
+            await reboots.wait(2)
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            assert proc.pid == pid_before
+
+
 @pytest.mark.asyncio
 async def test_host_ota_rejects_garbage(
     yaml_config: str,