Clear stale Freebox app token when registration fails (fixes #153988) (#180699)

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

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(

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]]]:

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:

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)