diff --git a/homeassistant/components/freebox/config_flow.py b/homeassistant/components/freebox/config_flow.py index 3a4a20c25039..6779c44248e3 100644 --- a/homeassistant/components/freebox/config_flow.py +++ b/homeassistant/components/freebox/config_flow.py @@ -11,7 +11,12 @@ from homeassistant.const import CONF_HOST, CONF_PORT from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo from .const import DOMAIN -from .router import get_api, get_hosts_list_if_supported +from .router import ( + async_forget_registration, + get_api, + get_hosts_list_if_supported, + is_invalid_token_error, +) _LOGGER = logging.getLogger(__name__) @@ -85,6 +90,12 @@ class FreeboxFlowHandler(ConfigFlow, domain=DOMAIN): except AuthorizationError as error: _LOGGER.error(error) errors["base"] = "register_failed" + if is_invalid_token_error(error): + # The stored application token was rejected by the + # Freebox. Clear it so resubmitting this form performs a + # fresh pairing instead of retrying with the same + # rejected token forever. + await async_forget_registration(self.hass, self._data[CONF_HOST]) except HttpRequestError: _LOGGER.error( diff --git a/homeassistant/components/freebox/router.py b/homeassistant/components/freebox/router.py index caec6e3cc157..ca92e7d9da66 100644 --- a/homeassistant/components/freebox/router.py +++ b/homeassistant/components/freebox/router.py @@ -54,6 +54,17 @@ def is_json(json_str: str) -> bool: return True +def _get_token_file(hass: HomeAssistant, host: str) -> Path: + """Return the path of the stored application token file for a host.""" + freebox_path = Store(hass, STORAGE_VERSION, STORAGE_KEY).path + return Path(f"{freebox_path}/{slugify(host)}.conf") + + +def _unlink_if_exists(path: Path) -> None: + """Remove a file if it exists.""" + path.unlink(missing_ok=True) + + async def get_api(hass: HomeAssistant, host: str) -> Freepybox: """Get the Freebox API.""" freebox_path = Store(hass, STORAGE_VERSION, STORAGE_KEY).path @@ -61,11 +72,35 @@ async def get_api(hass: HomeAssistant, host: str) -> Freepybox: if not os.path.exists(freebox_path): await hass.async_add_executor_job(os.makedirs, freebox_path) - token_file = Path(f"{freebox_path}/{slugify(host)}.conf") + token_file = _get_token_file(hass, host) return Freepybox(APP_DESC, token_file, API_VERSION) +def is_invalid_token_error(error: Exception) -> bool: + """Return whether an authorization error reports a rejected app token.""" + return bool( + (matcher := re.search(r"\(APIResponse: (.+)\)$", str(error))) + and is_json(json_str := matcher.group(1)) + and json.loads(json_str).get("error_code") == "invalid_token" + ) + + +async def async_forget_registration(hass: HomeAssistant, host: str) -> None: + """Remove a stored application token for a host, if any. + + The Freebox can reject an application token that Home Assistant still + considers valid, for example after the router was replaced, factory + reset, or the application authorization was revoked from the Freebox + OS settings. When that happens, Home Assistant keeps retrying with the + same rejected token forever. Removing the stored token forces a fresh + pairing (and a new "please confirm on your Freebox" prompt) on the + next attempt. + """ + token_file = _get_token_file(hass, host) + await hass.async_add_executor_job(_unlink_if_exists, token_file) + + async def get_hosts_list_if_supported( fbx_api: Freepybox, ) -> tuple[bool, list[dict[str, Any]]]: diff --git a/tests/components/freebox/test_config_flow.py b/tests/components/freebox/test_config_flow.py index 31a5b063e00b..9478a151251a 100644 --- a/tests/components/freebox/test_config_flow.py +++ b/tests/components/freebox/test_config_flow.py @@ -167,6 +167,73 @@ async def test_on_link_failed(hass: HomeAssistant) -> None: assert result["errors"] == {"base": "unknown"} +async def test_on_link_failed_forgets_registration_on_invalid_token( + hass: HomeAssistant, +) -> None: + """Test that an invalid app token clears the stored registration.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_HOST: MOCK_HOST, CONF_PORT: MOCK_PORT} + ) + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "link" + + error = AuthorizationError( + 'Starting session failed (APIResponse: {"success": false, ' + '"msg": "Erreur d\'authentification de l\'application", ' + '"error_code": "invalid_token"})' + ) + with ( + patch( + "homeassistant.components.freebox.router.Freepybox.open", + side_effect=error, + ), + patch( + "homeassistant.components.freebox.config_flow.async_forget_registration" + ) as mock_forget_registration, + ): + result = await hass.config_entries.flow.async_configure(result["flow_id"], {}) + assert result["type"] is FlowResultType.FORM + assert result["errors"] == {"base": "register_failed"} + mock_forget_registration.assert_awaited_once_with(hass, MOCK_HOST) + + +async def test_on_link_failed_keeps_registration_on_other_authorization_error( + hass: HomeAssistant, +) -> None: + """An AuthorizationError unrelated to the app token must not clear it. + + freebox-api also raises AuthorizationError for a denied or timed out + pairing request, and for transient failures of the challenge/session + calls. None of those mean the stored token itself is bad, so the + registration must be left untouched for them. + """ + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_HOST: MOCK_HOST, CONF_PORT: MOCK_PORT} + ) + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "link" + + with ( + patch( + "homeassistant.components.freebox.router.Freepybox.open", + side_effect=AuthorizationError("Authorization timed out"), + ), + patch( + "homeassistant.components.freebox.config_flow.async_forget_registration" + ) as mock_forget_registration, + ): + result = await hass.config_entries.flow.async_configure(result["flow_id"], {}) + assert result["type"] is FlowResultType.FORM + assert result["errors"] == {"base": "register_failed"} + mock_forget_registration.assert_not_awaited() + + async def test_zeroconf_missing_api_domain( hass: HomeAssistant, ) -> None: diff --git a/tests/components/freebox/test_router.py b/tests/components/freebox/test_router.py index eaada7ed95f7..64d7877b3ae1 100644 --- a/tests/components/freebox/test_router.py +++ b/tests/components/freebox/test_router.py @@ -1,14 +1,48 @@ """Tests for the Freebox utility methods.""" import json +from pathlib import Path from unittest.mock import Mock -from freebox_api.exceptions import HttpRequestError +from freebox_api.exceptions import AuthorizationError, HttpRequestError import pytest -from homeassistant.components.freebox.router import get_hosts_list_if_supported, is_json +from homeassistant.components.freebox.const import STORAGE_KEY, STORAGE_VERSION +from homeassistant.components.freebox.router import ( + async_forget_registration, + get_hosts_list_if_supported, + is_invalid_token_error, + is_json, +) +from homeassistant.core import HomeAssistant +from homeassistant.helpers.storage import Store +from homeassistant.util import slugify -from .const import DATA_LAN_GET_HOSTS_LIST_MODE_BRIDGE, DATA_WIFI_GET_GLOBAL_CONFIG +from .const import ( + DATA_LAN_GET_HOSTS_LIST_MODE_BRIDGE, + DATA_WIFI_GET_GLOBAL_CONFIG, + MOCK_HOST, +) + + +@pytest.fixture(autouse=True) +def mock_path(): + """Use the real pathlib.Path in this module so file removal can be tested. + + This overrides the autouse fixture of the same name in conftest.py, + which stubs out Path for the config flow / setup tests in this package. + """ + return + + +@pytest.fixture +def hass_config_dir(hass_tmp_config_dir: str) -> str: + """Use a temporary config directory so the app token file is isolated. + + This lets async_forget_registration's own Store lookup be exercised + for real, instead of mocking Store to point at a test-controlled path. + """ + return hass_tmp_config_dir async def test_is_json() -> None: @@ -60,3 +94,56 @@ async def test_get_hosts_list_if_supported_bridge_error( """Other exceptions must be propagated.""" with pytest.raises(HttpRequestError): await get_hosts_list_if_supported(mock_router_bridge_mode_error()) + + +@pytest.mark.parametrize( + ("error", "expected"), + [ + pytest.param( + AuthorizationError( + 'Starting session failed (APIResponse: {"success": false, ' + '"error_code": "invalid_token"})' + ), + True, + id="invalid_token", + ), + pytest.param( + AuthorizationError( + 'Starting session failed (APIResponse: {"success": false, ' + '"error_code": "internal_error"})' + ), + False, + id="other_error_code", + ), + pytest.param( + AuthorizationError("Authorization timed out"), + False, + id="no_api_response", + ), + pytest.param( + AuthorizationError("The app token is invalid or has been revoked"), + False, + id="denied_pairing_mentions_invalid_but_has_no_error_code", + ), + ], +) +def test_is_invalid_token_error(error: AuthorizationError, expected: bool) -> None: + """Only a genuine invalid_token APIResponse must be treated as such.""" + assert is_invalid_token_error(error) is expected + + +async def test_async_forget_registration(hass: HomeAssistant) -> None: + """async_forget_registration must remove the stored app token, if any.""" + token_file = ( + Path(Store(hass, STORAGE_VERSION, STORAGE_KEY).path) + / f"{slugify(MOCK_HOST)}.conf" + ) + token_file.parent.mkdir(parents=True, exist_ok=True) + token_file.write_text('{"app_token": "stale"}') + assert token_file.exists() + + await async_forget_registration(hass, MOCK_HOST) + assert not token_file.exists() + + # Calling again with no stored file left must not raise. + await async_forget_registration(hass, MOCK_HOST)