starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-08-24 10:13:52 -05:00
Code Issues Packages Projects Releases Wiki Activity

Fix a stale encryption key shadowing the working one in the config flow (#178047)

Browse Source
This commit is contained in:
J. Nick Koston
2026-08-04 14:33:56 +02:00
committed by Bram Kragten
parent b76b00d501
commit 28754d012e
2 changed files with 468 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/esphome/config_flow.py
+83 -48
View File
@@ -1,7 +1,7 @@
"""Config flow to configure esphome component."""
from collections import OrderedDict
from collections.abc import Mapping
from collections.abc import AsyncIterator, Mapping
import json
import logging
from typing import Any, cast, override
@@ -209,18 +209,24 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
"""Handle reauthorization flow."""
errors = {}
if (
await self._retrieve_encryption_key_from_storage()
or await self._retrieve_encryption_key_from_dashboard()
):
error = await self.fetch_device_info()
if error is None:
return await self._async_authenticate_or_add()
if user_input is not None:
if user_input is None:
async for candidate in self._async_encryption_key_candidates():
self._noise_psk = candidate
error = await self.fetch_device_info()
if error is None:
await self._async_repair_stored_key(candidate)
return await self._async_authenticate_or_add()
if error != ERROR_INVALID_ENCRYPTION_KEY:
# Not a key problem (device offline) — more keys
# can't help, and the dashboard needn't be asked.
errors["base"] = error
break
self._noise_psk = None
else:
self._noise_psk = user_input[CONF_NOISE_PSK]
error = await self.fetch_device_info()
if error is None:
await self._async_repair_stored_key(self._noise_psk)
return await self._async_authenticate_or_add()
errors["base"] = error
@@ -279,16 +285,21 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
response = await self.fetch_device_info()
self._noise_psk = None
# Try to retrieve an existing key from dashboard or storage.
if (
self._device_name
and await self._retrieve_encryption_key_from_dashboard()
) or (
self._device_mac and await self._retrieve_encryption_key_from_storage()
):
# Try every known key against the device — either source
# can be stale, and the first one must not shadow the other.
async for candidate in self._async_encryption_key_candidates():
self._noise_psk = candidate
response = await self.fetch_device_info()
if response is None:
await self._async_repair_stored_key(candidate)
break
if response != ERROR_INVALID_ENCRYPTION_KEY:
# Not a key problem — don't leave an unproven
# candidate behind for a later retry.
self._noise_psk = None
break
# If the fetched key is invalid, unset it again.
# If no fetched key worked, unset again for the manual step.
if response == ERROR_INVALID_ENCRYPTION_KEY:
self._noise_psk = None
response = ERROR_REQUIRES_ENCRYPTION_KEY
@@ -754,6 +765,7 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
self._noise_psk = user_input[CONF_NOISE_PSK]
error = await self.fetch_device_info()
if error is None:
await self._async_repair_stored_key(self._noise_psk)
return await self._async_authenticate_or_add()
errors["base"] = error
@@ -893,59 +905,82 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
return None
async def _retrieve_encryption_key_from_dashboard(self) -> bool:
"""Try to retrieve the encryption key from the dashboard.
Return boolean if a key was retrieved.
"""
async def _async_get_key_from_dashboard(self) -> str | None:
"""Return the dashboard's key for this device, or None."""
if (
self._device_name is None
or (manager := await async_get_or_create_dashboard_manager(self.hass))
is None
or (dashboard := manager.async_get()) is None
):
return False
return None
await dashboard.async_request_refresh()
if not dashboard.last_update_success:
return False
_LOGGER.debug(
"Dashboard refresh failed; skipping dashboard key for %s",
self._device_name,
)
return None
device = dashboard.data.get(self._device_name)
if device is None:
return False
return None
try:
noise_psk = await dashboard.api.get_encryption_key(device["configuration"])
return await dashboard.api.get_encryption_key(device["configuration"])
except aiohttp.ClientError as err:
_LOGGER.error("Error talking to the dashboard: %s", err)
return False
except json.JSONDecodeError:
_LOGGER.exception("Error parsing response from dashboard")
return False
return None
self._noise_psk = noise_psk
return True
async def _retrieve_encryption_key_from_storage(self) -> bool:
"""Try to retrieve the encryption key from storage.
Return boolean if a key was retrieved.
"""
# Try to get MAC address from current flow state or reauth entry
mac_address = self._device_mac
if mac_address is None and self._reauth_entry is not None:
# In reauth flow, get MAC from the existing entry's unique_id
mac_address = self._reauth_entry.unique_id
assert mac_address is not None
@callback
def _async_get_storage_mac(self) -> str | None:
"""MAC for encryption-key storage, from flow state or the reauth entry."""
if self._device_mac is not None:
return self._device_mac
if self.source == SOURCE_REAUTH:
return self._reauth_entry.unique_id
return None
async def _async_get_key_from_storage(self) -> str | None:
"""Return the stored key for this device, or None."""
if (mac_address := self._async_get_storage_mac()) is None:
return None
storage = await async_get_encryption_key_storage(self.hass)
if stored_key := await storage.async_get_key(mac_address):
self._noise_psk = stored_key
return True
return await storage.async_get_key(mac_address)
return False
async def _async_encryption_key_candidates(self) -> AsyncIterator[str]:
"""Distinct candidate keys from storage then the dashboard, lazily.
The device connect is the only truth test: either source can be
stale, so both are offered instead of the first one shadowing
the other. Lazy so the dashboard isn't consulted when the
stored key already works.
"""
seen: set[str] = set()
for source in (
self._async_get_key_from_storage,
self._async_get_key_from_dashboard,
):
if (key := await source()) and key not in seen:
seen.add(key)
yield key
async def _async_repair_stored_key(self, key: str) -> None:
"""Repair an existing stored key with the one proven to work on the device.
Repair-in-place only, never create: presence in the store means
"HA generated this key" and gates the device-side key wipe on
entry removal, so dashboard/user keys must not be enrolled.
"""
if (mac_address := self._async_get_storage_mac()) is None:
return
storage = await async_get_encryption_key_storage(self.hass)
if (stored := await storage.async_get_key(mac_address)) and stored != key:
await storage.async_store_key(mac_address, key)
@staticmethod
@callback
tests/components/esphome/test_config_flow.py
+385 -1
View File
@@ -20,7 +20,7 @@ import aiohttp
import pytest
from homeassistant import config_entries
from homeassistant.components.esphome import dashboard
from homeassistant.components.esphome import config_flow, dashboard
from homeassistant.components.esphome.const import (
CONF_ALLOW_SERVICE_CALLS,
CONF_BLUETOOTH_SCANNING_MODE,
@@ -33,6 +33,7 @@ from homeassistant.components.esphome.const import (
)
from homeassistant.components.esphome.encryption_key_storage import (
ENCRYPTION_KEY_STORAGE_KEY,
async_get_encryption_key_storage,
)
from homeassistant.config_entries import SOURCE_IGNORE, ConfigFlowResult
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT
@@ -1349,6 +1350,389 @@ async def test_reauth_fixed_via_dashboard(
assert len(mock_get_encryption_key.mock_calls) == 1
@pytest.mark.usefixtures("mock_setup_entry", "mock_zeroconf")
async def test_reauth_stale_storage_key_tries_dashboard_key(
hass: HomeAssistant,
mock_client: APIClient,
mock_dashboard: dict[str, Any],
hass_storage: dict[str, Any],
) -> None:
"""Test a stale stored key does not shadow the dashboard's working key."""
hass_storage[ENCRYPTION_KEY_STORAGE_KEY] = {
"version": 1,
"minor_version": 1,
"key": ENCRYPTION_KEY_STORAGE_KEY,
"data": {"keys": {"11:22:33:44:55:aa": WRONG_NOISE_PSK}},
}
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "127.0.0.1",
CONF_PORT: 6053,
CONF_PASSWORD: "",
CONF_DEVICE_NAME: "test",
},
unique_id="11:22:33:44:55:aa",
)
entry.add_to_hass(hass)
mock_client.device_info.side_effect = [
RequiresEncryptionAPIError, # initial reauth probe with the entry's empty psk
InvalidEncryptionKeyAPIError("Wrong key", "test"), # stale stored key
DeviceInfo(uses_password=False, name="test", mac_address="11:22:33:44:55:aa"),
]
mock_dashboard["configured"].append({"name": "test", "configuration": "test.yaml"})
await dashboard.async_get_dashboard(hass).async_refresh()
with patch(
"homeassistant.components.esphome.coordinator.ESPHomeDashboardAPI.get_encryption_key",
return_value=VALID_NOISE_PSK,
):
result = await entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.ABORT, result
assert result["reason"] == "reauth_successful"
assert entry.data[CONF_NOISE_PSK] == VALID_NOISE_PSK
# The stale stored key is replaced with the one proven on the device.
storage = await async_get_encryption_key_storage(hass)
assert await storage.async_get_key("11:22:33:44:55:aa") == VALID_NOISE_PSK
@pytest.mark.usefixtures("mock_setup_entry", "mock_zeroconf")
async def test_reauth_working_storage_key_never_asks_dashboard(
hass: HomeAssistant,
mock_client: APIClient,
mock_dashboard: dict[str, Any],
hass_storage: dict[str, Any],
) -> None:
"""Test the dashboard is not consulted when the stored key already works."""
hass_storage[ENCRYPTION_KEY_STORAGE_KEY] = {
"version": 1,
"minor_version": 1,
"key": ENCRYPTION_KEY_STORAGE_KEY,
"data": {"keys": {"11:22:33:44:55:aa": VALID_NOISE_PSK}},
}
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "127.0.0.1",
CONF_PORT: 6053,
CONF_PASSWORD: "",
CONF_DEVICE_NAME: "test",
},
unique_id="11:22:33:44:55:aa",
)
entry.add_to_hass(hass)
mock_client.device_info.return_value = DeviceInfo(
uses_password=False, name="test", mac_address="11:22:33:44:55:aa"
)
mock_dashboard["configured"].append({"name": "test", "configuration": "test.yaml"})
await dashboard.async_get_dashboard(hass).async_refresh()
with patch(
"homeassistant.components.esphome.coordinator.ESPHomeDashboardAPI.get_encryption_key",
return_value=WRONG_NOISE_PSK,
) as mock_get_encryption_key:
result = await entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.ABORT, result
assert result["reason"] == "reauth_successful"
assert entry.data[CONF_NOISE_PSK] == VALID_NOISE_PSK
mock_get_encryption_key.assert_not_called()
@pytest.mark.usefixtures("mock_setup_entry", "mock_zeroconf")
async def test_reauth_both_keys_wrong_falls_back_to_manual(
hass: HomeAssistant,
mock_client: APIClient,
mock_dashboard: dict[str, Any],
hass_storage: dict[str, Any],
) -> None:
"""Test both stale sources fall through to manual entry, which repairs storage."""
hass_storage[ENCRYPTION_KEY_STORAGE_KEY] = {
"version": 1,
"minor_version": 1,
"key": ENCRYPTION_KEY_STORAGE_KEY,
"data": {"keys": {"11:22:33:44:55:aa": WRONG_NOISE_PSK}},
}
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "127.0.0.1",
CONF_PORT: 6053,
CONF_PASSWORD: "",
CONF_DEVICE_NAME: "test",
},
unique_id="11:22:33:44:55:aa",
)
entry.add_to_hass(hass)
mock_client.device_info.side_effect = [
RequiresEncryptionAPIError, # initial reauth probe with the entry's empty psk
InvalidEncryptionKeyAPIError("Wrong key", "test"), # stale stored key
InvalidEncryptionKeyAPIError("Wrong key", "test"), # stale dashboard key
DeviceInfo(uses_password=False, name="test", mac_address="11:22:33:44:55:aa"),
]
mock_dashboard["configured"].append({"name": "test", "configuration": "test.yaml"})
await dashboard.async_get_dashboard(hass).async_refresh()
with patch(
"homeassistant.components.esphome.coordinator.ESPHomeDashboardAPI.get_encryption_key",
return_value=INVALID_NOISE_PSK,
):
result = await entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM, result
assert result["step_id"] == "reauth_confirm"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_NOISE_PSK: VALID_NOISE_PSK}
)
assert result["type"] is FlowResultType.ABORT, result
assert result["reason"] == "reauth_successful"
assert entry.data[CONF_NOISE_PSK] == VALID_NOISE_PSK
# The manually entered working key replaces the stale stored one.
storage = await async_get_encryption_key_storage(hass)
assert await storage.async_get_key("11:22:33:44:55:aa") == VALID_NOISE_PSK
@pytest.mark.usefixtures("mock_setup_entry", "mock_zeroconf")
async def test_user_flow_stale_storage_key_falls_back_to_dashboard(
hass: HomeAssistant,
mock_client: APIClient,
mock_dashboard: dict[str, Any],
hass_storage: dict[str, Any],
) -> None:
"""Test a stale stored key does not shadow the dashboard key in the user flow."""
hass_storage[ENCRYPTION_KEY_STORAGE_KEY] = {
"version": 1,
"minor_version": 1,
"key": ENCRYPTION_KEY_STORAGE_KEY,
"data": {"keys": {"11:22:33:44:55:aa": WRONG_NOISE_PSK}},
}
mock_client.device_info.side_effect = [
RequiresEncryptionAPIError,
InvalidEncryptionKeyAPIError("Wrong key", "test", "11:22:33:44:55:AA"),
InvalidEncryptionKeyAPIError("Wrong key", "test", "11:22:33:44:55:AA"),
DeviceInfo(uses_password=False, name="test", mac_address="11:22:33:44:55:AA"),
]
mock_dashboard["configured"].append({"name": "test", "configuration": "test.yaml"})
await dashboard.async_get_dashboard(hass).async_refresh()
with patch(
"homeassistant.components.esphome.coordinator.ESPHomeDashboardAPI.get_encryption_key",
return_value=VALID_NOISE_PSK,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={CONF_HOST: "127.0.0.1", CONF_PORT: 6053},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"][CONF_NOISE_PSK] == VALID_NOISE_PSK
assert mock_client.noise_psk == VALID_NOISE_PSK
# The stale stored key was repaired in place with the working one.
storage = await async_get_encryption_key_storage(hass)
assert await storage.async_get_key("11:22:33:44:55:aa") == VALID_NOISE_PSK
@pytest.mark.usefixtures("mock_setup_entry", "mock_zeroconf")
async def test_reauth_offline_device_stops_candidate_probing(
hass: HomeAssistant,
mock_client: APIClient,
mock_dashboard: dict[str, Any],
hass_storage: dict[str, Any],
) -> None:
"""Test a non-key connection error stops the loop without asking the dashboard."""
hass_storage[ENCRYPTION_KEY_STORAGE_KEY] = {
"version": 1,
"minor_version": 1,
"key": ENCRYPTION_KEY_STORAGE_KEY,
"data": {"keys": {"11:22:33:44:55:aa": WRONG_NOISE_PSK}},
}
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "127.0.0.1",
CONF_PORT: 6053,
CONF_PASSWORD: "",
CONF_DEVICE_NAME: "test",
},
unique_id="11:22:33:44:55:aa",
)
entry.add_to_hass(hass)
mock_client.device_info.side_effect = [
RequiresEncryptionAPIError, # initial reauth probe
APIConnectionError("timeout"), # storage candidate: device went offline
]
mock_dashboard["configured"].append({"name": "test", "configuration": "test.yaml"})
await dashboard.async_get_dashboard(hass).async_refresh()
with patch(
"homeassistant.components.esphome.coordinator.ESPHomeDashboardAPI.get_encryption_key",
return_value=VALID_NOISE_PSK,
) as mock_get_encryption_key:
result = await entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM, result
assert result["step_id"] == "reauth_confirm"
assert result["errors"] == {"base": "connection_error"}
mock_get_encryption_key.assert_not_called()
@pytest.mark.usefixtures("mock_setup_entry", "mock_zeroconf")
async def test_user_flow_offline_device_stops_candidate_probing(
hass: HomeAssistant,
mock_client: APIClient,
mock_dashboard: dict[str, Any],
hass_storage: dict[str, Any],
) -> None:
"""Test a non-key connection error stops the discovery-path probing."""
hass_storage[ENCRYPTION_KEY_STORAGE_KEY] = {
"version": 1,
"minor_version": 1,
"key": ENCRYPTION_KEY_STORAGE_KEY,
"data": {"keys": {"11:22:33:44:55:aa": WRONG_NOISE_PSK}},
}
mock_client.device_info.side_effect = [
RequiresEncryptionAPIError,
InvalidEncryptionKeyAPIError("Wrong key", "test", "11:22:33:44:55:AA"),
APIConnectionError("timeout"), # storage candidate: device went offline
]
mock_dashboard["configured"].append({"name": "test", "configuration": "test.yaml"})
await dashboard.async_get_dashboard(hass).async_refresh()
with patch(
"homeassistant.components.esphome.coordinator.ESPHomeDashboardAPI.get_encryption_key",
return_value=VALID_NOISE_PSK,
) as mock_get_encryption_key:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={CONF_HOST: "127.0.0.1", CONF_PORT: 6053},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "connection_error"}
mock_get_encryption_key.assert_not_called()
@pytest.mark.usefixtures("mock_setup_entry", "mock_zeroconf")
async def test_user_flow_manual_key_repairs_stale_storage(
hass: HomeAssistant,
mock_client: APIClient,
hass_storage: dict[str, Any],
) -> None:
"""Test a manual key in the encryption_key step repairs a stale stored key."""
hass_storage[ENCRYPTION_KEY_STORAGE_KEY] = {
"version": 1,
"minor_version": 1,
"key": ENCRYPTION_KEY_STORAGE_KEY,
"data": {"keys": {"11:22:33:44:55:aa": WRONG_NOISE_PSK}},
}
mock_client.device_info.side_effect = [
RequiresEncryptionAPIError,
InvalidEncryptionKeyAPIError("Wrong key", "test", "11:22:33:44:55:AA"),
InvalidEncryptionKeyAPIError("Wrong key", "test", "11:22:33:44:55:AA"),
DeviceInfo(uses_password=False, name="test", mac_address="11:22:33:44:55:AA"),
]
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={CONF_HOST: "127.0.0.1", CONF_PORT: 6053},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "encryption_key"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_NOISE_PSK: VALID_NOISE_PSK}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"][CONF_NOISE_PSK] == VALID_NOISE_PSK
# The stale stored key was repaired with the manually proven one.
storage = await async_get_encryption_key_storage(hass)
assert await storage.async_get_key("11:22:33:44:55:aa") == VALID_NOISE_PSK
async def test_repair_stored_key_without_mac_is_a_no_op(hass: HomeAssistant) -> None:
"""Test the repair helper does nothing when no MAC is known."""
flow = config_flow.EsphomeFlowHandler()
flow.hass = hass
with patch(
"homeassistant.components.esphome.config_flow.async_get_encryption_key_storage"
) as mock_storage:
await flow._async_repair_stored_key(VALID_NOISE_PSK)
mock_storage.assert_not_called()
@pytest.mark.usefixtures("mock_setup_entry", "mock_zeroconf")
async def test_reauth_manual_key_is_not_enrolled_in_storage(
hass: HomeAssistant,
mock_client: APIClient,
hass_storage: dict[str, Any],
) -> None:
"""Test a manually entered key never creates a storage entry.
Presence in storage means "HA generated this key" and gates the
device-side key wipe on entry removal, so user keys stay out.
"""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "127.0.0.1",
CONF_PORT: 6053,
CONF_PASSWORD: "",
CONF_DEVICE_NAME: "test",
},
unique_id="11:22:33:44:55:aa",
)
entry.add_to_hass(hass)
mock_client.device_info.side_effect = [
RequiresEncryptionAPIError, # initial reauth probe
DeviceInfo(uses_password=False, name="test", mac_address="11:22:33:44:55:aa"),
]
result = await entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM, result
assert result["step_id"] == "reauth_confirm"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_NOISE_PSK: VALID_NOISE_PSK}
)
assert result["type"] is FlowResultType.ABORT, result
assert result["reason"] == "reauth_successful"
assert entry.data[CONF_NOISE_PSK] == VALID_NOISE_PSK
storage = await async_get_encryption_key_storage(hass)
assert await storage.async_get_key("11:22:33:44:55:aa") is None
@pytest.mark.usefixtures("mock_setup_entry", "mock_zeroconf")
async def test_reauth_fixed_via_dashboard_add_encryption_remove_password(
hass: HomeAssistant,