Provision ESPHome encryption keys over an encrypted zero-PSK connection (#176144)

homeassistant/components/esphome/__init__.py

@@ -2,7 +2,7 @@
 
 import logging
 
-from aioesphomeapi import APIClient, APIConnectionError
+from aioesphomeapi import APIConnectionError
 
 from homeassistant.components import zeroconf
 from homeassistant.components.bluetooth import async_remove_scanner
@@ -11,13 +11,7 @@ from homeassistant.components.usb import (
     USBDevice,
     async_register_serial_port_scanner,
 )
-from homeassistant.const import (
-    CONF_HOST,
-    CONF_PASSWORD,
-    CONF_PORT,
-    EVENT_HOMEASSISTANT_STOP,
-    __version__ as ha_version,
-)
+from homeassistant.const import CONF_HOST, CONF_PASSWORD, EVENT_HOMEASSISTANT_STOP
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.issue_registry import async_delete_issue
@@ -29,15 +23,18 @@ from .const import CONF_BLUETOOTH_MAC_ADDRESS, CONF_NOISE_PSK, DOMAIN
 from .domain_data import DomainData
 from .encryption_key_storage import async_get_encryption_key_storage
 from .entry_data import ESPHomeConfigEntry, RuntimeEntryData
-from .manager import DEVICE_CONFLICT_ISSUE_FORMAT, ESPHomeManager, cleanup_instance
+from .manager import (
+    DEVICE_CONFLICT_ISSUE_FORMAT,
+    ESPHomeManager,
+    async_create_api_client,
+    cleanup_instance,
+)
 from .websocket_api import async_setup as async_setup_websocket_api
 
 _LOGGER = logging.getLogger(__name__)
 
 CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
 
-CLIENT_INFO = f"Home Assistant {ha_version}"
-
 
 @callback
 def _async_scan_serial_ports(
@@ -90,20 +87,12 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
 async def async_setup_entry(hass: HomeAssistant, entry: ESPHomeConfigEntry) -> bool:
     """Set up the esphome component."""
     host: str = entry.data[CONF_HOST]
-    port: int = entry.data[CONF_PORT]
     password: str | None = entry.data[CONF_PASSWORD]
-    noise_psk: str | None = entry.data.get(CONF_NOISE_PSK)
 
     zeroconf_instance = await zeroconf.async_get_instance(hass)
 
-    cli = APIClient(
-        host,
-        port,
-        password,
-        client_info=CLIENT_INFO,
-        zeroconf_instance=zeroconf_instance,
-        noise_psk=noise_psk,
-        timezone=hass.config.time_zone,
+    cli = async_create_api_client(
+        hass, entry, zeroconf_instance, noise_psk=entry.data.get(CONF_NOISE_PSK)
     )
 
     domain_data = DomainData.get(hass)
@@ -159,21 +148,10 @@ async def _async_clear_dynamic_encryption_key(
     if await storage.async_get_key(entry.unique_id) is None:
         return
 
-    host: str = entry.data[CONF_HOST]
-    port: int = entry.data[CONF_PORT]
-    password: str | None = entry.data[CONF_PASSWORD]
-    noise_psk: str | None = entry.data.get(CONF_NOISE_PSK)
-
     zeroconf_instance = await zeroconf.async_get_instance(hass)
 
-    cli = APIClient(
-        host,
-        port,
-        password,
-        client_info=CLIENT_INFO,
-        zeroconf_instance=zeroconf_instance,
-        noise_psk=noise_psk,
-        timezone=hass.config.time_zone,
+    cli = async_create_api_client(
+        hass, entry, zeroconf_instance, noise_psk=entry.data.get(CONF_NOISE_PSK)
     )
 
     try:

homeassistant/components/esphome/config_flow.py

@@ -74,7 +74,11 @@ ERROR_INVALID_ENCRYPTION_KEY = "invalid_psk"
 ERROR_INVALID_PASSWORD_AUTH = "invalid_auth"
 _LOGGER = logging.getLogger(__name__)
 
-ZERO_NOISE_PSK = "MDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDA="
+# A deliberately wrong key (base64 of thirty two ASCII zero characters, not
+# zero bytes) used only to elicit the server hello so the device name can be
+# read. Not to be confused with aioesphomeapi.ZERO_NOISE_PSK, the well known
+# all zeros provisioning key.
+PROBE_NOISE_PSK = "MDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDA="
 DEFAULT_NAME = "ESPHome"
 
 _BLUETOOTH_SCANNING_MODE_SELECTOR = SelectSelector(
@@ -271,7 +275,7 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
                 # to get the device name which will allow us to populate
                 # the device name and hopefully get the encryption key
                 # from the dashboard.
-                self._noise_psk = ZERO_NOISE_PSK
+                self._noise_psk = PROBE_NOISE_PSK
                 response = await self.fetch_device_info()
                 self._noise_psk = None
 

homeassistant/components/esphome/const.py

@@ -5,6 +5,7 @@ from typing import TYPE_CHECKING, Final
 from awesomeversion import AwesomeVersion
 
 from homeassistant.components.bluetooth import BluetoothScanningMode
+from homeassistant.const import __version__ as ha_version
 from homeassistant.util.hass_dict import HassKey
 
 if TYPE_CHECKING:
@@ -12,6 +13,8 @@ if TYPE_CHECKING:
 
 DOMAIN = "esphome"
 
+CLIENT_INFO = f"Home Assistant {ha_version}"
+
 ESPHOME_DATA: HassKey[DomainData] = HassKey(DOMAIN)
 
 CONF_ALLOW_SERVICE_CALLS = "allow_service_calls"

homeassistant/components/esphome/manager.py

@@ -9,6 +9,7 @@ import struct
 from typing import TYPE_CHECKING, Any, Final, NamedTuple
 
 from aioesphomeapi import (
+    ZERO_NOISE_PSK,
     APIClient,
     APIConnectionError,
     APIVersion,
@@ -34,7 +35,10 @@ import voluptuous as vol
 from homeassistant.components import bluetooth, tag, zeroconf
 from homeassistant.const import (
     ATTR_DEVICE_ID,
+    CONF_HOST,
     CONF_MODE,
+    CONF_PASSWORD,
+    CONF_PORT,
     EVENT_HOMEASSISTANT_CLOSE,
     EVENT_LOGGING_CHANGED,
     Platform,
@@ -77,6 +81,7 @@ from homeassistant.util.json import json_loads_object
 
 from .bluetooth import async_connect_scanner
 from .const import (
+    CLIENT_INFO,
     CONF_ALLOW_SERVICE_CALLS,
     CONF_BLUETOOTH_MAC_ADDRESS,
     CONF_DEVICE_NAME,
@@ -101,6 +106,26 @@ DEVICE_CONFLICT_ISSUE_FORMAT = "device_conflict-{}"
 UNPACK_UINT32_BE = struct.Struct(">I").unpack_from
 
 
+@callback
+def async_create_api_client(
+    hass: HomeAssistant,
+    entry: ESPHomeConfigEntry,
+    zeroconf_instance: zeroconf.HaZeroconf,
+    *,
+    noise_psk: str | None,
+) -> APIClient:
+    """Create an APIClient for a config entry."""
+    return APIClient(
+        entry.data[CONF_HOST],
+        entry.data[CONF_PORT],
+        entry.data[CONF_PASSWORD],
+        client_info=CLIENT_INFO,
+        zeroconf_instance=zeroconf_instance,
+        noise_psk=noise_psk,
+        timezone=hass.config.time_zone,
+    )
+
+
 if TYPE_CHECKING:
     from aioesphomeapi.api_pb2 import SubscribeLogsResponse  # type: ignore[attr-defined]  # noqa: I001
 
@@ -812,6 +837,51 @@ class ESPHomeManager:
         if self.reconnect_logic:
             await self.reconnect_logic.stop()
 
+    async def _async_provision_key_over_noise(self, new_key: bytes) -> bool:
+        """Send the encryption key over a short lived zero PSK Noise connection.
+
+        The well known all zeros PSK still runs a fresh ephemeral X25519
+        exchange, so the key cannot be read by a passive listener on the
+        network. This protects against sniffing only; it does not
+        authenticate either side against an active man in the middle.
+
+        Returns True if the device accepted the key. On failure the caller
+        simply returns; provisioning runs again on the next connect cycle.
+        """
+        unique_id = self.entry.unique_id
+        cli = async_create_api_client(
+            self.hass, self.entry, self.zeroconf_instance, noise_psk=ZERO_NOISE_PSK
+        )
+        device_name = self.entry.data.get(CONF_DEVICE_NAME, self.host)
+        try:
+            await cli.connect()
+            if await cli.noise_encryption_set_key(new_key):
+                return True
+            _LOGGER.error(
+                "Device %s (%s) rejected the encryption key",
+                device_name,
+                unique_id,
+            )
+        except InvalidEncryptionKeyAPIError:
+            _LOGGER.error(
+                "Device %s (%s) rejected the zero PSK handshake; it appears "
+                "to already have an encryption key set",
+                device_name,
+                unique_id,
+            )
+        except APIConnectionError as ex:
+            # Whatever went wrong, we never downgrade to a plaintext push;
+            # provisioning simply runs again on the next connect cycle
+            _LOGGER.error(
+                "Error provisioning encryption key for device %s (%s): %s",
+                device_name,
+                unique_id,
+                ex,
+            )
+        finally:
+            await cli.disconnect(force=True)
+        return False
+
     async def _handle_dynamic_encryption_key(
         self, device_info: EsphomeDeviceInfo
     ) -> None:
@@ -853,18 +923,24 @@ class ESPHomeManager:
             new_key = base64.b64encode(secrets.token_bytes(32))
             new_key_str = new_key.decode()
 
-        try:
-            # Store the key on the device using the existing connection
-            result = await self.cli.noise_encryption_set_key(new_key)
-        except APIConnectionError as ex:
-            _LOGGER.error(
-                "Connection error while storing encryption key for device %s (%s): %s",
-                self.entry.data.get(CONF_DEVICE_NAME, self.host),
-                self.entry.unique_id,
-                ex,
-            )
-            return
+        if device_info.api_encryption_provisionable:
+            # New firmware: send the key over an encrypted zero PSK Noise
+            # connection so it cannot be sniffed off the network
+            if not await self._async_provision_key_over_noise(new_key):
+                return
         else:
+            # Old firmware only accepts the key over the existing plaintext
+            # connection. Deprecated; will be removed after the usual window.
+            try:
+                result = await self.cli.noise_encryption_set_key(new_key)
+            except APIConnectionError as ex:
+                _LOGGER.error(
+                    "Connection error while storing encryption key for device %s (%s): %s",
+                    self.entry.data.get(CONF_DEVICE_NAME, self.host),
+                    self.entry.unique_id,
+                    ex,
+                )
+                return
             if not result:
                 _LOGGER.error(
                     "Failed to set dynamic encryption key on device %s (%s)",

tests/components/esphome/conftest.py

@@ -210,7 +210,7 @@ def mock_client(mock_device_info) -> Generator[APIClient]:
             "homeassistant.components.esphome.manager.ReconnectLogic",
             BaseMockReconnectLogic,
         ),
-        patch("homeassistant.components.esphome.APIClient", mock_client),
+        patch("homeassistant.components.esphome.manager.APIClient", mock_client),
         patch("homeassistant.components.esphome.config_flow.APIClient", mock_client),
     ):
         yield mock_client

tests/components/esphome/test_manager.py

@@ -2,11 +2,13 @@
 
 import asyncio
 import base64
+from collections.abc import Generator
 import logging
 from typing import Any
 from unittest.mock import AsyncMock, Mock, call, patch
 
 from aioesphomeapi import (
+    ZERO_NOISE_PSK,
     APIClient,
     APIConnectionError,
     APIVersion,
@@ -32,6 +34,7 @@ import pytest
 import voluptuous as vol
 
 from homeassistant import config_entries
+from homeassistant.components.esphome.config_flow import PROBE_NOISE_PSK
 from homeassistant.components.esphome.const import (
     CONF_ALLOW_SERVICE_CALLS,
     CONF_BLUETOOTH_MAC_ADDRESS,
@@ -2724,6 +2727,208 @@ async def test_manager_handle_dynamic_encryption_key_connection_error(
     assert mac_address not in hass_storage[ENCRYPTION_KEY_STORAGE_KEY]["data"]["keys"]
 
 
+@pytest.fixture
+def mock_provisioning_client(mock_client: APIClient) -> Generator[Mock]:
+    """Mock the APIClient built for the zero PSK provisioning connection."""
+    client = Mock(spec=APIClient)
+    client.connect = AsyncMock()
+    client.disconnect = AsyncMock()
+    client.noise_encryption_set_key = AsyncMock(return_value=True)
+
+    def _api_client(*args: Any, **kwargs: Any) -> Mock:
+        if kwargs.get("noise_psk") == ZERO_NOISE_PSK:
+            return client
+        return mock_client(*args, **kwargs)
+
+    with patch(
+        "homeassistant.components.esphome.manager.APIClient", side_effect=_api_client
+    ):
+        yield client
+
+
+def _make_provisionable_entry(hass: HomeAssistant, mac_address: str) -> MockConfigEntry:
+    """Create a config entry without a noise PSK."""
+    entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={
+            CONF_HOST: "192.168.1.100",
+            CONF_PORT: 6053,
+            CONF_PASSWORD: "",
+            CONF_DEVICE_NAME: "test-device",
+        },
+        unique_id=mac_address,
+    )
+    entry.add_to_hass(hass)
+    return entry
+
+
+@patch("homeassistant.components.esphome.manager.secrets.token_bytes")
+async def test_dynamic_encryption_key_provisioned_over_zero_psk(
+    mock_token_bytes: Mock,
+    hass: HomeAssistant,
+    mock_client: APIClient,
+    mock_provisioning_client: Mock,
+    mock_esphome_device: MockESPHomeDeviceType,
+    hass_storage: dict[str, Any],
+) -> None:
+    """Test provisionable firmware gets the key over a zero PSK connection."""
+    mac_address = "11:22:33:44:55:aa"
+    test_key_bytes = b"test_key_32_bytes_long_exactly!"
+    mock_token_bytes.return_value = test_key_bytes
+    expected_key = base64.b64encode(test_key_bytes).decode()
+
+    entry = _make_provisionable_entry(hass, mac_address)
+
+    # The main (plaintext) client must never be used to push the key
+    mock_client.noise_encryption_set_key = AsyncMock(return_value=True)
+
+    device = await mock_esphome_device(
+        mock_client=mock_client,
+        entry=entry,
+        device_info={
+            "uses_password": False,
+            "name": "test-device",
+            "mac_address": mac_address,
+            "esphome_version": "2026.8.0",
+            "api_encryption_supported": True,
+            "api_encryption_provisionable": True,
+        },
+    )
+
+    await device.mock_disconnect(True)
+    await device.mock_connect()
+
+    # The key went over the zero PSK client (the fixture only hands it out
+    # for constructions using ZERO_NOISE_PSK), not the plaintext connection
+    mock_provisioning_client.noise_encryption_set_key.assert_called_once_with(
+        base64.b64encode(test_key_bytes)
+    )
+    mock_client.noise_encryption_set_key.assert_not_called()
+    mock_provisioning_client.disconnect.assert_called_with(force=True)
+
+    # Entry and storage were updated
+    assert entry.data[CONF_NOISE_PSK] == expected_key
+    assert (
+        hass_storage[ENCRYPTION_KEY_STORAGE_KEY]["data"]["keys"][mac_address]
+        == expected_key
+    )
+
+
+async def test_dynamic_encryption_key_provisioned_over_zero_psk_from_storage(
+    hass: HomeAssistant,
+    mock_client: APIClient,
+    mock_provisioning_client: Mock,
+    mock_esphome_device: MockESPHomeDeviceType,
+    hass_storage: dict[str, Any],
+) -> None:
+    """Test a stored key is re-provisioned over the zero PSK connection."""
+    mac_address = "11:22:33:44:55:aa"
+    test_key = base64.b64encode(b"existing_key_32_bytes_long!!!").decode()
+
+    hass_storage[ENCRYPTION_KEY_STORAGE_KEY] = {
+        "version": 1,
+        "minor_version": 1,
+        "key": ENCRYPTION_KEY_STORAGE_KEY,
+        "data": {"keys": {mac_address: test_key}},
+    }
+
+    entry = _make_provisionable_entry(hass, mac_address)
+    mock_client.noise_encryption_set_key = AsyncMock(return_value=True)
+
+    device = await mock_esphome_device(
+        mock_client=mock_client,
+        entry=entry,
+        device_info={
+            "uses_password": False,
+            "name": "test-device",
+            "mac_address": mac_address,
+            "esphome_version": "2026.8.0",
+            "api_encryption_supported": True,
+            "api_encryption_provisionable": True,
+        },
+    )
+
+    await device.mock_disconnect(True)
+    await device.mock_connect()
+
+    mock_provisioning_client.noise_encryption_set_key.assert_called_once_with(
+        test_key.encode()
+    )
+    mock_client.noise_encryption_set_key.assert_not_called()
+    assert entry.data[CONF_NOISE_PSK] == test_key
+
+
+@pytest.mark.parametrize(
+    ("connect_error", "set_key_result"),
+    [
+        # Device already has a key (distinct log branch)
+        (InvalidEncryptionKeyAPIError("already keyed"), True),
+        # Old firmware answering plaintext to the noise hello (generic branch;
+        # all connection errors are APIConnectionError subclasses)
+        (EncryptionPlaintextAPIError("plaintext"), True),
+        # Device accepted the connection but rejected the key
+        (None, False),
+    ],
+)
+@patch("homeassistant.components.esphome.manager.secrets.token_bytes")
+async def test_dynamic_encryption_key_zero_psk_failures_never_use_plaintext(
+    mock_token_bytes: Mock,
+    hass: HomeAssistant,
+    mock_client: APIClient,
+    mock_provisioning_client: Mock,
+    mock_esphome_device: MockESPHomeDeviceType,
+    hass_storage: dict[str, Any],
+    connect_error: Exception | None,
+    set_key_result: bool,
+) -> None:
+    """Test zero PSK provisioning failures do not fall back to plaintext."""
+    mac_address = "11:22:33:44:55:aa"
+    mock_token_bytes.return_value = b"test_key_32_bytes_long_exactly!"
+
+    hass_storage[ENCRYPTION_KEY_STORAGE_KEY] = {
+        "version": 1,
+        "minor_version": 1,
+        "key": ENCRYPTION_KEY_STORAGE_KEY,
+        "data": {"keys": {}},
+    }
+
+    entry = _make_provisionable_entry(hass, mac_address)
+    mock_client.noise_encryption_set_key = AsyncMock(return_value=True)
+
+    # A None side_effect leaves connect behaving normally
+    mock_provisioning_client.connect.side_effect = connect_error
+    mock_provisioning_client.noise_encryption_set_key.return_value = set_key_result
+
+    device = await mock_esphome_device(
+        mock_client=mock_client,
+        entry=entry,
+        device_info={
+            "uses_password": False,
+            "name": "test-device",
+            "mac_address": mac_address,
+            "esphome_version": "2026.8.0",
+            "api_encryption_supported": True,
+            "api_encryption_provisionable": True,
+        },
+    )
+
+    await device.mock_disconnect(True)
+    await device.mock_connect()
+
+    # The plaintext connection was never used to push the key, the entry was
+    # not updated, and no generated key was stored
+    mock_client.noise_encryption_set_key.assert_not_called()
+    assert CONF_NOISE_PSK not in entry.data
+    assert mac_address not in hass_storage[ENCRYPTION_KEY_STORAGE_KEY]["data"]["keys"]
+    mock_provisioning_client.disconnect.assert_called_with(force=True)
+
+
+def test_zero_noise_psk_is_not_the_probe_key() -> None:
+    """Test the provisioning PSK is 32 zero bytes and differs from the probe."""
+    assert base64.b64decode(ZERO_NOISE_PSK) == bytes(32)
+    assert ZERO_NOISE_PSK != PROBE_NOISE_PSK
+
+
 async def test_zwave_proxy_request_home_id_change(
     hass: HomeAssistant,
     mock_client: APIClient,