feat: self-healing login — validate tokens & recover from poisoned cache

Makes authentication failure- and future-proof against the #369 class of
problems where a strategy obtains a DI token the API tier later rejects
("Token is not active"). Fully backwards compatible for end-user calls.

In-chain token validation (client.login):
  After a strategy obtains a token, verify it with one lightweight
  authenticated call before accepting it. A token the API rejects (401/403)
  is discarded and the chain continues to the next strategy. Only definitive
  auth rejections fall through — transient 5xx/network errors are treated as
  inconclusive so a flaky network never blocks a working login. This is what
  makes a widget+cffi token that works on one account but is rejected on
  another auto-fall-through to portal+cffi, with no skip_strategies needed.
  Opt out via Garmin(verify_login=False) for legacy "first token wins".

Disk-token self-healing (Garmin.login):
  If cached tokens load but the API rejects them, the cache is discarded and
  a full credential login runs automatically, then the profile fetch retries.
  Previously a poisoned cache silently short-circuited the strategy chain on
  every run (the load-before-chain footgun documented by issue #369).

logout() helper:
  Upgraded the deprecated no-op into a real helper that clears in-memory auth
  state and removes cached tokens on disk. Still callable with no args.

Also:
  - get_morning_training_readiness: restore defensive dict handling (a single
    dict response is passed through), fixing a regression from the #361 fix.
  - typed.get_training_readiness: empty {} / None now normalizes to [].
  - 12 new in-process tests covering all of the above.

Refs #369

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: cyberjunky

garminconnect/__init__.py

@@ -7,6 +7,7 @@
 import os
 import random
 import re
+import shutil
 import time
 from collections.abc import Callable
 from datetime import UTC, date, datetime, timedelta
@@ -282,6 +283,7 @@ def __init__(
         retry_attempts: int = 3,
         retry_min_wait: float = 1.0,
         retry_max_wait: float = 10.0,
+        verify_login: bool = True,
     ) -> None:
         """Create a new class instance.
 
@@ -292,6 +294,11 @@ def __init__(
         :param retry_min_wait: Initial backoff in seconds; grows exponentially
             with jitter between attempts.
         :param retry_max_wait: Upper bound on the backoff in seconds.
+        :param verify_login: When ``True`` (default), each login strategy's
+            token is validated against the API tier before the chain accepts
+            it, so a token the API rejects (401/403) is discarded and the next
+            strategy is tried. Set ``False`` for the legacy "first token wins"
+            behavior.
         """
         # Validate input types
         if email is not None and not isinstance(email, str):
@@ -306,6 +313,8 @@ def __init__(
             raise ValueError("retry_attempts must be an int")
         if retry_attempts < 0:
             raise ValueError("retry_attempts must be non-negative")
+        if not isinstance(verify_login, bool):
+            raise ValueError("verify_login must be a boolean")
 
         self.username = email
         self.password = password
@@ -315,6 +324,7 @@ def __init__(
         self.retry_attempts = retry_attempts
         self.retry_min_wait = float(retry_min_wait)
         self.retry_max_wait = float(retry_max_wait)
+        self.verify_login = verify_login
 
         self.garmin_connect_user_settings_url = (
             "/userprofile-service/userprofile/user-settings"
@@ -505,6 +515,7 @@ def __init__(
             domain="garmin.cn" if is_cn else "garmin.com",
             pool_connections=20,
             pool_maxsize=20,
+            verify_login=verify_login,
         )
 
         self.display_name: str | None = None
@@ -629,49 +640,37 @@ def login(self, /, tokenstore: str | None = None) -> tuple[str | None, str | Non
                         self.client.dump(tokenstore_path)
                 # Continue to load profile/settings below
 
-            # Ensure profile is loaded (tokenstore path may not populate it)
-            prof = None
-            for attempt in range(3):
-                try:
-                    prof = self.client.connectapi("/userprofile-service/socialProfile")
-                    if isinstance(prof, dict):
-                        break
-                except Exception as e:
-                    if attempt == 2:
-                        raise GarminConnectAuthenticationError(
-                            "Failed to retrieve social profile"
-                        ) from e
-                    logger.debug("Retrying social profile fetch: %s", e)
-                    time.sleep(1)
-            else:
-                raise GarminConnectAuthenticationError("Invalid profile data found")
-
-            self.display_name = prof.get("displayName", self.username)
-            self.full_name = prof.get("fullName", "")
-
-            settings = None
-            for attempt in range(3):
-                try:
-                    settings = self.client.connectapi(
-                        self.garmin_connect_user_settings_url
-                    )
-                    if (
-                        settings
-                        and isinstance(settings, dict)
-                        and "userData" in settings
-                    ):
-                        break
-                except Exception as e:
-                    if attempt == 2:
-                        raise GarminConnectAuthenticationError(
-                            "Failed to retrieve user settings"
-                        ) from e
-                    logger.debug("Retrying user settings fetch: %s", e)
-                    time.sleep(1)
-            else:
-                raise GarminConnectAuthenticationError("Invalid user settings found")
-
-            self.unit_system = settings["userData"].get("measurementSystem")
+            # Ensure profile is loaded (tokenstore path may not populate it).
+            try:
+                self._load_profile_and_settings()
+            except GarminConnectAuthenticationError:
+                # If we resumed from cached tokens and the API rejects them,
+                # the cache is stale/poisoned (e.g. a token whose audience the
+                # API tier no longer accepts — see #369). Discard it and run a
+                # full credential login so the strategy chain can find a token
+                # the API accepts. Without this, a poisoned cache silently
+                # short-circuits the chain on every run.
+                username, password = self.username, self.password
+                if not (
+                    tokens_loaded and username and password and not self.return_on_mfa
+                ):
+                    raise
+                logger.warning(
+                    "Cached tokens were rejected by the API; discarding and "
+                    "performing a fresh login."
+                )
+                self.client._clear_auth_state()
+                if tokenstore_path is not None:
+                    self.client._tokenstore_path = tokenstore_path
+                mfa_status, _legacy_token = self.client.login(
+                    username,
+                    password,
+                    prompt_mfa=self.prompt_mfa,
+                )
+                if tokenstore_path is not None:
+                    with contextlib.suppress(Exception):
+                        self.client.dump(tokenstore_path)
+                self._load_profile_and_settings()
 
             return mfa_status, _legacy_token
 
@@ -723,6 +722,48 @@ def login(self, /, tokenstore: str | None = None) -> tuple[str | None, str | Non
             logger.debug("Login failed: %s", e)
             raise GarminConnectConnectionError(f"Login failed: {e}") from e
 
+    def _load_profile_and_settings(self) -> None:
+        """Fetch social profile and user settings, populating display name,
+        full name and unit system. Raises ``GarminConnectAuthenticationError``
+        if either cannot be retrieved (e.g. the token is rejected).
+        """
+        prof = None
+        for attempt in range(3):
+            try:
+                prof = self.client.connectapi("/userprofile-service/socialProfile")
+                if isinstance(prof, dict):
+                    break
+            except Exception as e:
+                if attempt == 2:
+                    raise GarminConnectAuthenticationError(
+                        "Failed to retrieve social profile"
+                    ) from e
+                logger.debug("Retrying social profile fetch: %s", e)
+                time.sleep(1)
+        else:
+            raise GarminConnectAuthenticationError("Invalid profile data found")
+
+        self.display_name = prof.get("displayName", self.username)
+        self.full_name = prof.get("fullName", "")
+
+        settings = None
+        for attempt in range(3):
+            try:
+                settings = self.client.connectapi(self.garmin_connect_user_settings_url)
+                if settings and isinstance(settings, dict) and "userData" in settings:
+                    break
+            except Exception as e:
+                if attempt == 2:
+                    raise GarminConnectAuthenticationError(
+                        "Failed to retrieve user settings"
+                    ) from e
+                logger.debug("Retrying user settings fetch: %s", e)
+                time.sleep(1)
+        else:
+            raise GarminConnectAuthenticationError("Invalid user settings found")
+
+        self.unit_system = settings["userData"].get("measurementSystem")
+
     def resume_login(
         self, client_state: dict[str, Any], mfa_code: str
     ) -> tuple[Any, Any]:
@@ -1719,6 +1760,11 @@ def get_morning_training_readiness(self, cdate: str) -> dict[str, Any] | None:
         if not data:
             return None
 
+        # The endpoint normally returns a list of snapshots, but stay defensive:
+        # some responses (or callers stubbing this) hand back a single dict.
+        if isinstance(data, dict):
+            return data
+
         morning_entry = next(
             (
                 entry
@@ -2996,11 +3042,28 @@ def query_garmin_graphql(self, query: dict[str, Any]) -> dict[str, Any]:
             "connectapi", self.garmin_graphql_endpoint, json=query
         ).json()
 
-    def logout(self) -> None:
-        """Log user out of session."""
-        logger.warning(
-            "Deprecated: Alternative is to delete the login tokens to logout."
-        )
+    def logout(self, tokenstore: str | None = None) -> None:
+        """Clear in-memory auth state and any cached tokens on disk.
+
+        Call this after an authentication failure to guarantee the next
+        ``login()`` runs the full strategy chain instead of resuming
+        stale/poisoned cached tokens. ``login()`` already self-heals from
+        rejected cached tokens, so this is only needed for manual control.
+
+        :param tokenstore: Path to the token directory/file to remove. Falls
+            back to the ``GARMINTOKENS`` environment variable. Token strings
+            passed inline (length > 512) are ignored — nothing to delete.
+        """
+        self.client._clear_auth_state()
+        tokenstore = tokenstore or os.getenv("GARMINTOKENS")
+        if not tokenstore or len(tokenstore) > 512:
+            return
+        path = Path(tokenstore).expanduser()
+        with contextlib.suppress(Exception):
+            if path.is_dir():
+                shutil.rmtree(path)
+            elif path.exists():
+                path.unlink()
 
     def get_training_plans(self) -> dict[str, Any]:
         """Return all available training plans."""

garminconnect/client.py

@@ -208,11 +208,50 @@ def __init__(self, domain: str = "garmin.com", **kwargs: Any) -> None:
         # Valid names: mobile+cffi, mobile+requests, widget+cffi,
         #              portal+cffi, portal+requests
         self.skip_strategies: set[str] = set()
+        # When True (default), each login strategy's token is validated against
+        # the API tier before the chain accepts it; a token the API rejects
+        # (401/403) is discarded and the next strategy is tried. Set False to
+        # restore the legacy "first token wins" behavior.
+        self.verify_login: bool = kwargs.get("verify_login", True)
 
     @property
     def is_authenticated(self) -> bool:
         return bool(self.di_token or self.jwt_web)
 
+    def _clear_auth_state(self) -> None:
+        """Wipe all in-memory auth tokens so the next login starts clean."""
+        self.di_token = None
+        self.di_refresh_token = None
+        self.di_client_id = None
+        self.jwt_web = None
+        self.csrf_token = None
+
+    def _verify_token(self) -> bool:
+        """Check that the current token is actually accepted by the API tier.
+
+        A strategy can obtain a DI token from the auth host (HTTP 200) that the
+        API tier (connectapi) then rejects with 401 "Token is not active" —
+        this is account/region dependent (see issue #369). We confirm the token
+        works with one lightweight authenticated call.
+
+        Returns True if accepted, or if the check is inconclusive. Only a
+        definitive auth rejection (401/403) returns False, so a transient
+        network error or 5xx never blocks an otherwise-working login.
+        """
+        try:
+            self.connectapi("/userprofile-service/socialProfile")
+            return True
+        except GarminConnectConnectionError as e:
+            msg = str(e)
+            if "401" in msg or "403" in msg:
+                _LOGGER.warning("Token rejected by API tier: %s", msg)
+                return False
+            _LOGGER.debug("Token validation inconclusive (kept): %s", msg)
+            return True
+        except Exception as e:
+            _LOGGER.debug("Token validation inconclusive (kept): %s", e)
+            return True
+
     def get_api_headers(self) -> dict[str, str]:
         if not self.is_authenticated:
             raise GarminConnectAuthenticationError("Not authenticated")
@@ -284,7 +323,18 @@ def login(
             try:
                 _LOGGER.debug("Trying login strategy: %s", name)
                 run()
-                # Strategy succeeded — login complete
+                # Strategy got a token — make sure the API tier accepts it
+                # before declaring success (else fall through to the next).
+                if self.verify_login and not self._verify_token():
+                    _LOGGER.warning(
+                        "%s obtained a token the API rejected; trying next strategy",
+                        name,
+                    )
+                    self._clear_auth_state()
+                    last_err = GarminConnectConnectionError(
+                        f"{name}: token rejected by API tier"
+                    )
+                    continue
                 return None, None
             except GarminConnectAuthenticationError:
                 # Wrong credentials — stop immediately, no point trying further
@@ -296,6 +346,17 @@ def login(
                 if prompt_mfa:
                     mfa_code = prompt_mfa()
                     self._complete_mfa(mfa_code)
+                    if self.verify_login and not self._verify_token():
+                        _LOGGER.warning(
+                            "%s obtained a token the API rejected after MFA; "
+                            "trying next strategy",
+                            name,
+                        )
+                        self._clear_auth_state()
+                        last_err = GarminConnectConnectionError(
+                            f"{name}: token rejected by API tier"
+                        )
+                        continue
                     return None, None
                 raise GarminConnectAuthenticationError(  # noqa: B904
                     "MFA Required but no prompt_mfa mechanism supplied"

garminconnect/typed.py

@@ -543,6 +543,9 @@ def get_training_readiness(self, cdate: str) -> list[TrainingReadiness]:
         wrapper normalizes both shapes to ``list[TrainingReadiness]``.
         """
         raw = self._garmin.get_training_readiness(cdate)
+        if not raw:
+            # Empty list / empty dict / None — no snapshots available.
+            return []
         if isinstance(raw, list):
             return [
                 self._validate(TrainingReadiness, item, "get_training_readiness")

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "garminconnect"
-version = "0.3.5"
+version = "0.3.6"
 description = "Python 3 API wrapper for Garmin Connect"
 authors = [
     {name = "Ron Klinkien", email = "ron@cyberjunky.nl"},
@@ -55,6 +55,7 @@ module = [
     "test_garmin_unit",
     "test_retry_decorator",
     "test_workout_constants",
+    "test_login_recovery",
     "conftest",
 ]
 disallow_untyped_defs = false