Force one token refresh after an API 401, bound HTTP timeouts, and keep reauth reachable

custom_components/generac/api.py

@@ -8,6 +8,7 @@
 returning identical payloads for the endpoints we use. The iOS app uses
 `/api/v5`; we follow suit for futureproofing.
 """
+import asyncio
 import json
 import logging
 
@@ -16,14 +17,15 @@
 
 from .auth import GeneracAuth
 from .auth import InvalidGrantError
+from .auth import REQUEST_TIMEOUT
 from .auth import USER_AGENT_API
 from .const import ALLOWED_DEVICES
 from .const import API_BASE
 from .models import Apparatus
 from .models import ApparatusDetail
 from .models import Item
 
-TIMEOUT = 10
+REQUEST_RETRY_DELAY = 1
 
 _LOGGER: logging.Logger = logging.getLogger(__package__)
 
@@ -112,6 +114,62 @@ async def get_endpoint(self, endpoint: str):
         except InvalidGrantError as ex:
             raise InvalidCredentialsException(str(ex)) from ex
 
+        url = API_BASE + endpoint
+        refreshed = False
+        retried = False
+        while True:
+            try:
+                status, data = await self._get(url, access_token)
+            except (asyncio.TimeoutError, aiohttp.ClientConnectionError) as ex:
+                # GET is idempotent: retry a transient transport failure once.
+                if retried:
+                    raise IOError(
+                        f"GET {url} failed after retry: {type(ex).__name__}: {ex}"
+                    ) from ex
+                retried = True
+                _LOGGER.warning(
+                    "Mobile Link request for %s failed (%s); retrying once",
+                    endpoint,
+                    type(ex).__name__,
+                )
+                await asyncio.sleep(REQUEST_RETRY_DELAY)
+                continue
+            except Exception as ex:
+                raise IOError(f"GET {url} failed: {type(ex).__name__}: {ex}") from ex
+
+            if status == 204:
+                return None
+
+            if status == 401:
+                if refreshed:
+                    raise SessionExpiredException(
+                        f"API returned 401 for {endpoint} after token refresh"
+                    )
+                refreshed = True
+                _LOGGER.warning(
+                    "Mobile Link rejected the access token for %s; "
+                    "forcing one token refresh",
+                    endpoint,
+                )
+                # Outside the transport try/except above on purpose: an
+                # InvalidGrantError here must reach the coordinator as
+                # InvalidCredentialsException (-> reauth), not as IOError.
+                try:
+                    access_token = await self._auth.force_refresh(access_token)
+                except InvalidGrantError as ex:
+                    raise InvalidCredentialsException(str(ex)) from ex
+                continue
+
+            if status != 200:
+                raise SessionExpiredException(
+                    f"API returned status code {status} for {endpoint}: {data}"
+                )
+
+            _LOGGER.debug("getEndpoint %s", json.dumps(data))
+            return data
+
+    async def _get(self, url: str, access_token: str) -> tuple[int, object]:
+        """GET `url`; return (status, JSON body for 200, text excerpt otherwise)."""
         # Bearer is intentional here, even though this access token is
         # DPoP-*bound* (authorize sends dpop_jkt, so the token carries a
         # cnf.jkt claim). Probed 2026-06-25 (probe_rs_dpop.py): the MobileLink
@@ -129,31 +187,16 @@ async def get_endpoint(self, endpoint: str):
             "Accept": "application/json",
             "User-Agent": USER_AGENT_API,
         }
-
-        url = API_BASE + endpoint
-        try:
-            async with self._session.get(url, headers=headers) as response:
-                if response.status == 204:
-                    return None
-
-                if response.status == 401:
-                    raise SessionExpiredException(f"API returned 401 for {endpoint}")
-
-                if response.status != 200:
-                    body = ""
-                    try:
-                        body = (await response.text())[:200]
-                    except Exception:
-                        pass
-                    raise SessionExpiredException(
-                        f"API returned status code {response.status} for "
-                        f"{endpoint}: {body}"
-                    )
-
-                data = await response.json()
-                _LOGGER.debug("getEndpoint %s", json.dumps(data))
-                return data
-        except SessionExpiredException:
-            raise
-        except Exception as ex:
-            raise IOError(f"GET {url} failed: {type(ex).__name__}: {ex}") from ex
+        async with self._session.get(
+            url, headers=headers, timeout=REQUEST_TIMEOUT
+        ) as response:
+            if response.status == 200:
+                return response.status, await response.json()
+            if response.status in (204, 401):
+                return response.status, None
+            body = ""
+            try:
+                body = (await response.text())[:200]
+            except Exception:
+                pass
+            return response.status, body

custom_components/generac/auth.py

@@ -40,6 +40,7 @@
 from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature
 
 _LOGGER = logging.getLogger(__name__)
+REQUEST_TIMEOUT = aiohttp.ClientTimeout(total=45, connect=10, sock_read=30)
 
 AUTH0_DOMAIN = "auth.ecobee.com"
 AUTHORIZE_URL = f"https://{AUTH0_DOMAIN}/authorize"
@@ -686,7 +687,7 @@ def __init__(
         # GeneracAuth.login's docstring for why this must stay isolated
         # from the shared session. We own it and close it in aclose().
         self._login_session: Optional[aiohttp.ClientSession] = aiohttp.ClientSession(
-            cookie_jar=aiohttp.CookieJar(unsafe=True)
+            cookie_jar=aiohttp.CookieJar(unsafe=True), timeout=REQUEST_TIMEOUT
         )
         # Populated when a code challenge pauses the flow.
         self._mfa_type: Optional[str] = None
@@ -995,6 +996,20 @@ async def ensure_access_token(self) -> str:
             assert self._access_token is not None
             return self._access_token
 
+    async def force_refresh(self, rejected_token: str) -> str:
+        """Refresh once after the resource server rejects an access token.
+
+        Callers pass the token that was rejected. If another caller already
+        replaced it while we waited for the lock, reuse that token instead of
+        spending a second refresh on the same rejection.
+        """
+        async with self._refresh_lock:
+            if self._access_token is not None and self._access_token != rejected_token:
+                return self._access_token
+            await self._refresh()
+            assert self._access_token is not None
+            return self._access_token
+
     async def _refresh(self) -> None:
         body = {
             "grant_type": "refresh_token",
@@ -1011,7 +1026,7 @@ async def _post(nonce: str | None) -> tuple[int, dict, str | None]:
                 "User-Agent": USER_AGENT_API,
             }
             async with self._session.post(
-                TOKEN_URL, json=body, headers=headers
+                TOKEN_URL, json=body, headers=headers, timeout=REQUEST_TIMEOUT
             ) as resp:
                 text = await resp.text()
                 try:

custom_components/generac/config_flow.py

@@ -56,6 +56,17 @@
     "otp": "your authenticator app",
     "email": "email",
 }
+# Credentials from the retired cookie / password flows. Reauth and
+# reconfigure replace them instead of merging them into the new entry data.
+_DISCARDED_AUTH_KEYS = frozenset({CONF_PASSWORD, "session_cookie"})
+
+
+def _replace_auth_data(existing: dict, new_data: dict) -> dict:
+    """Replace discarded credentials instead of retaining stale secrets."""
+    retained = {
+        key: value for key, value in existing.items() if key not in _DISCARDED_AUTH_KEYS
+    }
+    return {**retained, **new_data}
 
 
 def _map_runtime_error(msg: str) -> str:
@@ -260,7 +271,7 @@ async def _finish_reconfigure(self, entry_data: dict):
         # unload, surfacing as "failed to unload".
         self.hass.config_entries.async_update_entry(
             entry,
-            data={**entry.data, **entry_data},
+            data=_replace_auth_data(entry.data, entry_data),
             options=new_options,
         )
         self._reload_if_setup_failed(entry)
@@ -325,7 +336,7 @@ async def _finish_reauth(self, entry_data: dict):
         # reload here would race the listener-driven one in the loaded case and
         # surface as "failed to unload".
         self.hass.config_entries.async_update_entry(
-            entry, data={**entry.data, **entry_data}
+            entry, data=_replace_auth_data(entry.data, entry_data)
         )
         self._reload_if_setup_failed(entry)
         return self.async_abort(reason="reauth_successful")
@@ -402,13 +413,18 @@ async def _discard_login_flow(self) -> None:
         if flow is not None:
             await flow.aclose()
 
-    async def async_remove(self) -> None:
-        """Close the login session if the flow is abandoned mid-MFA."""
-        if self._login_flow is not None:
-            try:
-                await self._login_flow.aclose()
-            except Exception:  # noqa: BLE001
-                _LOGGER.debug("Error closing login flow on remove", exc_info=True)
+    @callback
+    def async_remove(self) -> None:
+        """Close the login session if the flow is abandoned mid-MFA.
+
+        HA calls this synchronously when the flow is removed, so an async
+        override would never run. Schedule the close instead; aclose() is
+        idempotent and never raises.
+        """
+        flow = self._login_flow
+        self._login_flow = None
+        if flow is not None:
+            self.hass.async_create_task(flow.aclose())
 
     @staticmethod
     @callback

custom_components/generac/coordinator.py

@@ -60,6 +60,12 @@ async def _async_update_data(self):
             _LOGGER.warning("Generac API session error: %s", ex)
             self.is_online = False
             raise UpdateFailed(f"API session error: {ex}") from ex
+        except OSError as ex:
+            # Network timeouts and connection failures are transient and do
+            # not indicate that the saved credentials need reauthentication.
+            _LOGGER.warning("Generac cloud request temporarily failed: %s", ex)
+            self.is_online = False
+            raise UpdateFailed(str(ex)) from ex
         except Exception as exception:
             _LOGGER.exception("Unexpected error refreshing Generac data")
             self.is_online = False