Cache token info in Wallbox (home-assistant#154147)

hesselonline · joostlek · frenck · commit ffe524d95a41 · 2025-11-21T16:26:17.000Z
Co-authored-by: Joost Lekkerkerker &lt;joostlek@outlook.com&gt;
diff --git a/homeassistant/components/wallbox/__init__.py b/homeassistant/components/wallbox/__init__.py
@@ -6,15 +6,15 @@
 
 from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, Platform
 from homeassistant.core import HomeAssistant
-from homeassistant.exceptions import ConfigEntryAuthFailed
-
-from .const import UPDATE_INTERVAL
-from .coordinator import (
-    InvalidAuth,
-    WallboxConfigEntry,
-    WallboxCoordinator,
-    async_validate_input,
+
+from .const import (
+    CHARGER_JWT_REFRESH_TOKEN,
+    CHARGER_JWT_REFRESH_TTL,
+    CHARGER_JWT_TOKEN,
+    CHARGER_JWT_TTL,
+    UPDATE_INTERVAL,
 )
+from .coordinator import WallboxConfigEntry, WallboxCoordinator, check_token_validity
 
 PLATFORMS = [
     Platform.LOCK,
@@ -32,10 +32,16 @@ async def async_setup_entry(hass: HomeAssistant, entry: WallboxConfigEntry) -> b
         entry.data[CONF_PASSWORD],
         jwtTokenDrift=UPDATE_INTERVAL,
     )
-    try:
-        await async_validate_input(hass, wallbox)
-    except InvalidAuth as ex:
-        raise ConfigEntryAuthFailed from ex
+
+    if CHARGER_JWT_TOKEN in entry.data and check_token_validity(
+        jwt_token_ttl=entry.data.get(CHARGER_JWT_TTL, 0),
+        jwt_token_drift=UPDATE_INTERVAL,
+    ):
+        wallbox.jwtToken = entry.data.get(CHARGER_JWT_TOKEN)
+        wallbox.jwtRefreshToken = entry.data.get(CHARGER_JWT_REFRESH_TOKEN)
+        wallbox.jwtTokenTtl = entry.data.get(CHARGER_JWT_TTL)
+        wallbox.jwtRefreshTokenTtl = entry.data.get(CHARGER_JWT_REFRESH_TTL)
+        wallbox.headers["Authorization"] = f"Bearer {entry.data.get(CHARGER_JWT_TOKEN)}"
 
     wallbox_coordinator = WallboxCoordinator(hass, entry, wallbox)
     await wallbox_coordinator.async_config_entry_first_refresh()
diff --git a/homeassistant/components/wallbox/config_flow.py b/homeassistant/components/wallbox/config_flow.py
@@ -12,7 +12,15 @@
 from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
 from homeassistant.core import HomeAssistant
 
-from .const import CONF_STATION, DOMAIN
+from .const import (
+    CHARGER_JWT_REFRESH_TOKEN,
+    CHARGER_JWT_REFRESH_TTL,
+    CHARGER_JWT_TOKEN,
+    CHARGER_JWT_TTL,
+    CONF_STATION,
+    DOMAIN,
+    UPDATE_INTERVAL,
+)
 from .coordinator import InvalidAuth, async_validate_input
 
 COMPONENT_DOMAIN = DOMAIN
@@ -26,17 +34,22 @@
 )
 
 
-async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> dict[str, str]:
+async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> dict[str, Any]:
     """Validate the user input allows to connect.
 
     Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user.
     """
-    wallbox = Wallbox(data["username"], data["password"])
+    wallbox = Wallbox(data[CONF_USERNAME], data[CONF_PASSWORD], UPDATE_INTERVAL)
 
     await async_validate_input(hass, wallbox)
 
+    data[CHARGER_JWT_TOKEN] = wallbox.jwtToken
+    data[CHARGER_JWT_REFRESH_TOKEN] = wallbox.jwtRefreshToken
+    data[CHARGER_JWT_TTL] = wallbox.jwtTokenTtl
+    data[CHARGER_JWT_REFRESH_TTL] = wallbox.jwtRefreshTokenTtl
+
     # Return info that you want to store in the config entry.
-    return {"title": "Wallbox Portal"}
+    return {"title": "Wallbox Portal", "data": data}
 
 
 class WallboxConfigFlow(ConfigFlow, domain=COMPONENT_DOMAIN):
@@ -64,8 +77,11 @@ async def async_step_user(
             await self.async_set_unique_id(user_input["station"])
             if self.source != SOURCE_REAUTH:
                 self._abort_if_unique_id_configured()
-                info = await validate_input(self.hass, user_input)
-                return self.async_create_entry(title=info["title"], data=user_input)
+                validation_data = await validate_input(self.hass, user_input)
+                return self.async_create_entry(
+                    title=validation_data["title"],
+                    data=validation_data["data"],
+                )
             reauth_entry = self._get_reauth_entry()
             if user_input["station"] == reauth_entry.data[CONF_STATION]:
                 return self.async_update_reload_and_abort(reauth_entry, data=user_input)
diff --git a/homeassistant/components/wallbox/const.py b/homeassistant/components/wallbox/const.py
@@ -47,6 +47,12 @@
 CHARGER_ECO_SMART_KEY = "ecosmart"
 CHARGER_ECO_SMART_STATUS_KEY = "enabled"
 CHARGER_ECO_SMART_MODE_KEY = "mode"
+CHARGER_WALLBOX_OBJECT_KEY = "wallbox"
+
+CHARGER_JWT_TOKEN = "jwtToken"
+CHARGER_JWT_REFRESH_TOKEN = "jwtRefreshToken"
+CHARGER_JWT_TTL = "jwtTokenTtl"
+CHARGER_JWT_REFRESH_TTL = "jwtRefreshTokenTtl"
 
 
 class ChargerStatus(StrEnum):
diff --git a/homeassistant/components/wallbox/coordinator.py b/homeassistant/components/wallbox/coordinator.py
@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from collections.abc import Callable
-from datetime import timedelta
+from datetime import datetime, timedelta
 from http import HTTPStatus
 import logging
 from typing import Any, Concatenate
@@ -27,6 +27,10 @@
     CHARGER_ECO_SMART_STATUS_KEY,
     CHARGER_ENERGY_PRICE_KEY,
     CHARGER_FEATURES_KEY,
+    CHARGER_JWT_REFRESH_TOKEN,
+    CHARGER_JWT_REFRESH_TTL,
+    CHARGER_JWT_TOKEN,
+    CHARGER_JWT_TTL,
     CHARGER_LOCKED_UNLOCKED_KEY,
     CHARGER_MAX_CHARGING_CURRENT_KEY,
     CHARGER_MAX_CHARGING_CURRENT_POST_KEY,
@@ -86,27 +90,25 @@ def _require_authentication[_WallboxCoordinatorT: WallboxCoordinator, **_P](
 ) -> Callable[Concatenate[_WallboxCoordinatorT, _P], Any]:
     """Authenticate with decorator using Wallbox API."""
 
-    def require_authentication(
+    async def require_authentication(
         self: _WallboxCoordinatorT, *args: _P.args, **kwargs: _P.kwargs
     ) -> Any:
         """Authenticate using Wallbox API."""
-        try:
-            self.authenticate()
-            return func(self, *args, **kwargs)
-        except requests.exceptions.HTTPError as wallbox_connection_error:
-            if wallbox_connection_error.response.status_code == HTTPStatus.FORBIDDEN:
-                raise ConfigEntryAuthFailed(
-                    translation_domain=DOMAIN, translation_key="invalid_auth"
-                ) from wallbox_connection_error
-            raise HomeAssistantError(
-                translation_domain=DOMAIN, translation_key="api_failed"
-            ) from wallbox_connection_error
+        await self.async_authenticate()
+        return await func(self, *args, **kwargs)
 
     return require_authentication
 
 
+def check_token_validity(jwt_token_ttl: int, jwt_token_drift: int) -> bool:
+    """Check if the jwtToken is still valid in order to reuse if possible."""
+    return round((jwt_token_ttl / 1000) - jwt_token_drift, 0) > datetime.timestamp(
+        datetime.now()
+    )
+
+
 def _validate(wallbox: Wallbox) -> None:
-    """Authenticate using Wallbox API."""
+    """Authenticate using Wallbox API to check if the used credentials are valid."""
     try:
         wallbox.authenticate()
     except requests.exceptions.HTTPError as wallbox_connection_error:
@@ -142,11 +144,38 @@ def __init__(
             update_interval=timedelta(seconds=UPDATE_INTERVAL),
         )
 
-    def authenticate(self) -> None:
+    def _authenticate(self) -> dict[str, str]:
+        """Authenticate using Wallbox API. First check token validity."""
+        data = dict(self.config_entry.data)
+        if not check_token_validity(
+            jwt_token_ttl=data.get(CHARGER_JWT_TTL, 0),
+            jwt_token_drift=UPDATE_INTERVAL,
+        ):
+            try:
+                self._wallbox.authenticate()
+            except requests.exceptions.HTTPError as wallbox_connection_error:
+                if (
+                    wallbox_connection_error.response.status_code
+                    == HTTPStatus.FORBIDDEN
+                ):
+                    raise ConfigEntryAuthFailed(
+                        translation_domain=DOMAIN, translation_key="invalid_auth"
+                    ) from wallbox_connection_error
+                raise HomeAssistantError(
+                    translation_domain=DOMAIN, translation_key="api_failed"
+                ) from wallbox_connection_error
+            else:
+                data[CHARGER_JWT_TOKEN] = self._wallbox.jwtToken
+                data[CHARGER_JWT_REFRESH_TOKEN] = self._wallbox.jwtRefreshToken
+                data[CHARGER_JWT_TTL] = self._wallbox.jwtTokenTtl
+                data[CHARGER_JWT_REFRESH_TTL] = self._wallbox.jwtRefreshTokenTtl
+        return data
+
+    async def async_authenticate(self) -> None:
         """Authenticate using Wallbox API."""
-        self._wallbox.authenticate()
+        data = await self.hass.async_add_executor_job(self._authenticate)
+        self.hass.config_entries.async_update_entry(self.config_entry, data=data)
 
-    @_require_authentication
     def _get_data(self) -> dict[str, Any]:
         """Get new sensor data for Wallbox component."""
         try:
@@ -208,6 +237,7 @@ def _get_data(self) -> dict[str, Any]:
                 translation_domain=DOMAIN, translation_key="api_failed"
             ) from wallbox_connection_error
 
+    @_require_authentication
     async def _async_update_data(self) -> dict[str, Any]:
         """Get new sensor data for Wallbox component. Set update interval to be UPDATE_INTERVAL * #wallbox chargers configured, this is necessary due to rate limitations."""
 
@@ -217,7 +247,6 @@ async def _async_update_data(self) -> dict[str, Any]:
         )
         return await self.hass.async_add_executor_job(self._get_data)
 
-    @_require_authentication
     def _set_charging_current(
         self, charging_current: float
     ) -> dict[str, dict[str, dict[str, Any]]]:
@@ -246,14 +275,14 @@ def _set_charging_current(
                 translation_domain=DOMAIN, translation_key="api_failed"
             ) from wallbox_connection_error
 
+    @_require_authentication
     async def async_set_charging_current(self, charging_current: float) -> None:
         """Set maximum charging current for Wallbox."""
         data = await self.hass.async_add_executor_job(
             self._set_charging_current, charging_current
         )
         self.async_set_updated_data(data)
 
-    @_require_authentication
     def _set_icp_current(self, icp_current: float) -> dict[str, Any]:
         """Set maximum icp current for Wallbox."""
         try:
@@ -276,14 +305,14 @@ def _set_icp_current(self, icp_current: float) -> dict[str, Any]:
                 translation_domain=DOMAIN, translation_key="api_failed"
             ) from wallbox_connection_error
 
+    @_require_authentication
     async def async_set_icp_current(self, icp_current: float) -> None:
         """Set maximum icp current for Wallbox."""
         data = await self.hass.async_add_executor_job(
             self._set_icp_current, icp_current
         )
         self.async_set_updated_data(data)
 
-    @_require_authentication
     def _set_energy_cost(self, energy_cost: float) -> dict[str, Any]:
         """Set energy cost for Wallbox."""
         try:
@@ -300,14 +329,14 @@ def _set_energy_cost(self, energy_cost: float) -> dict[str, Any]:
                 translation_domain=DOMAIN, translation_key="api_failed"
             ) from wallbox_connection_error
 
+    @_require_authentication
     async def async_set_energy_cost(self, energy_cost: float) -> None:
         """Set energy cost for Wallbox."""
         data = await self.hass.async_add_executor_job(
             self._set_energy_cost, energy_cost
         )
         self.async_set_updated_data(data)
 
-    @_require_authentication
     def _set_lock_unlock(self, lock: bool) -> dict[str, dict[str, dict[str, Any]]]:
         """Set wallbox to locked or unlocked."""
         try:
@@ -335,12 +364,12 @@ def _set_lock_unlock(self, lock: bool) -> dict[str, dict[str, dict[str, Any]]]:
                 translation_domain=DOMAIN, translation_key="api_failed"
             ) from wallbox_connection_error
 
+    @_require_authentication
     async def async_set_lock_unlock(self, lock: bool) -> None:
         """Set wallbox to locked or unlocked."""
         data = await self.hass.async_add_executor_job(self._set_lock_unlock, lock)
         self.async_set_updated_data(data)
 
-    @_require_authentication
     def _pause_charger(self, pause: bool) -> None:
         """Set wallbox to pause or resume."""
         try:
@@ -357,12 +386,12 @@ def _pause_charger(self, pause: bool) -> None:
                 translation_domain=DOMAIN, translation_key="api_failed"
             ) from wallbox_connection_error
 
+    @_require_authentication
     async def async_pause_charger(self, pause: bool) -> None:
         """Set wallbox to pause or resume."""
         await self.hass.async_add_executor_job(self._pause_charger, pause)
         await self.async_request_refresh()
 
-    @_require_authentication
     def _set_eco_smart(self, option: str) -> None:
         """Set wallbox solar charging mode."""
         try:
@@ -381,6 +410,7 @@ def _set_eco_smart(self, option: str) -> None:
                 translation_domain=DOMAIN, translation_key="api_failed"
             ) from wallbox_connection_error
 
+    @_require_authentication
     async def async_set_eco_smart(self, option: str) -> None:
         """Set wallbox solar charging mode."""
 
diff --git a/tests/components/wallbox/conftest.py b/tests/components/wallbox/conftest.py
@@ -1,5 +1,6 @@
 """Test fixtures for the Wallbox integration."""
 
+from datetime import datetime, timedelta
 from http import HTTPStatus
 from unittest.mock import MagicMock, Mock, patch
 
@@ -10,6 +11,10 @@
     CHARGER_DATA_POST_L1_KEY,
     CHARGER_DATA_POST_L2_KEY,
     CHARGER_ENERGY_PRICE_KEY,
+    CHARGER_JWT_REFRESH_TOKEN,
+    CHARGER_JWT_REFRESH_TTL,
+    CHARGER_JWT_TOKEN,
+    CHARGER_JWT_TTL,
     CHARGER_LOCKED_UNLOCKED_KEY,
     CHARGER_MAX_CHARGING_CURRENT_POST_KEY,
     CHARGER_MAX_ICP_CURRENT_KEY,
@@ -43,6 +48,14 @@ def entry(hass: HomeAssistant) -> MockConfigEntry:
             CONF_USERNAME: "test_username",
             CONF_PASSWORD: "test_password",
             CONF_STATION: "12345",
+            CHARGER_JWT_TOKEN: "test_token",
+            CHARGER_JWT_REFRESH_TOKEN: "test_refresh_token",
+            CHARGER_JWT_TTL: (
+                datetime.timestamp(datetime.now() + timedelta(hours=1)) * 1000
+            ),
+            CHARGER_JWT_REFRESH_TTL: (
+                datetime.timestamp(datetime.now() + timedelta(hours=1)) * 1000
+            ),
         },
         entry_id="testEntry",
     )
@@ -82,6 +95,14 @@ def mock_wallbox():
         )
         wallbox.setIcpMaxCurrent = Mock(return_value={CHARGER_MAX_ICP_CURRENT_KEY: 25})
         wallbox.getChargerStatus = Mock(return_value=WALLBOX_STATUS_RESPONSE)
+        wallbox.jwtToken = "test_token"
+        wallbox.jwtRefreshToken = "test_refresh_token"
+        wallbox.jwtTokenTtl = (
+            datetime.timestamp(datetime.now() + timedelta(hours=1)) * 1000
+        )
+        wallbox.jwtRefreshTokenTtl = (
+            datetime.timestamp(datetime.now() + timedelta(hours=1)) * 1000
+        )
         mock.return_value = wallbox
         yield wallbox
 
diff --git a/tests/components/wallbox/test_config_flow.py b/tests/components/wallbox/test_config_flow.py
diff --git a/tests/components/wallbox/test_init.py b/tests/components/wallbox/test_init.py
