fix(auth): make token refresh and session restore work

kinde_sdk/auth/oauth.py

@@ -120,7 +120,13 @@ def _initialize_framework(self) -> None:
         self._storage = StorageFactory.create_storage({"type": self.framework})
 
         # Initialize storage manager with the framework-specific storage
-        self._storage_manager.initialize(config={"type": self.framework, "device_id": self._framework.get_name()}, storage=self._storage)
+        # device_id must be passed as an argument: inside config it is ignored and a random
+        # per-process id is used, so sessions can't be found after a restart or on another worker
+        self._storage_manager.initialize(
+            config={"type": self.framework},
+            device_id=self._framework.get_name(),
+            storage=self._storage,
+        )
 
     def _initialize_null_framework(self, storage_config: Dict[str, Any]) -> None:
         """

kinde_sdk/auth/token_manager.py

@@ -1,5 +1,6 @@
 import time
 import requests
+from kinde_sdk.core.exceptions import KindeTokenPersistenceException
 import threading
 import logging
 from typing import Any, Dict, Optional
@@ -33,9 +34,13 @@ def __init__(self, user_id, client_id, client_secret, token_url):
         self.client_secret = client_secret
         self.token_url = token_url
         self.tokens = {}  # Store tokens (access/refresh)
-        self.lock = threading.Lock()  # Add a lock for thread safety
+        # Reentrant: get_access_token() holds the lock while a refresh calls set_tokens()
+        self.lock = threading.RLock()
         self.redirect_uri = None  # Initialize the redirect_uri attribute
         self.force_api = False  # Initialize force_api setting
+        self.on_tokens_refreshed = None  # Called after a successful refresh, e.g. to persist tokens
+        self.load_persisted_tokens = None  # Returns the saved tokens, which another process may have refreshed
+        self._save_pending = False  # A refresh succeeded but saving its tokens failed
         self.initialized = True
 
     def set_force_api(self, force_api: bool):
@@ -124,18 +129,84 @@ def get_access_token(self):
         with self.lock:
             if not self.tokens or "access_token" not in self.tokens:
                 raise ValueError("No access token available")
-
+
+            # Save tokens from an earlier refresh before deciding whether to refresh
+            # again. Refresh tokens rotate, so storage must not keep the old one: if
+            # this save fails and a refresh below fails too, a restart could not
+            # recover the session. A failed save keeps the refreshed tokens in memory.
+            save_error = None
+            if self._save_pending:
+                try:
+                    self._save_refreshed_tokens()
+                except KindeTokenPersistenceException as e:
+                    save_error = e
+
             # Check if token is expired
             # if time.time() >= self.tokens["expires_at"]:
             if time.time() >= self.tokens.get("expires_at", 0):
+                # Another worker may already have refreshed and saved newer tokens
+                self._adopt_newer_persisted_tokens()
+                if time.time() < self.tokens.get("expires_at", 0):
+                    return self.tokens["access_token"]
+
                 # Try to refresh token if available
                 if "refresh_token" in self.tokens:
-                    return self.refresh_access_token()
+                    try:
+                        return self.refresh_access_token()
+                    except requests.RequestException as e:
+                        self._log_refresh_failure(e)
+                        raise
                 else:
                     raise ValueError("Access token expired and no refresh token available")
-
+
+            if save_error is not None:
+                raise save_error
             return self.tokens["access_token"]
 
+    def _save_refreshed_tokens(self) -> None:
+        """Save the tokens from a refresh. If saving fails, the refreshed tokens are kept
+        and the save is retried on the next call, without refreshing again."""
+        if self.on_tokens_refreshed:
+            try:
+                self.on_tokens_refreshed()
+            except Exception as e:
+                logging.getLogger(__name__).warning(
+                    "Refreshed tokens could not be saved: %s", type(e).__name__
+                )
+                raise KindeTokenPersistenceException(
+                    "Refreshed tokens could not be saved", access_token=self.tokens["access_token"]
+                ) from e
+        self._save_pending = False
+
+    def _adopt_newer_persisted_tokens(self) -> None:
+        """Use tokens that another process refreshed and saved, so this process
+        doesn't refresh with a refresh token Kinde has already rotated out."""
+        if not self.load_persisted_tokens:
+            return
+        try:
+            persisted = self.load_persisted_tokens()
+        except Exception as e:
+            logging.getLogger(__name__).warning("Saved tokens could not be read: %s", type(e).__name__)
+            return
+        if (persisted and persisted.get("access_token")
+                and persisted.get("expires_at", 0) > self.tokens.get("expires_at", 0)):
+            self.tokens = dict(persisted)
+            self._save_pending = False  # The saved tokens are newer than any unsaved ones
+
+    def _log_refresh_failure(self, error: Exception) -> None:
+        """Log why a refresh failed, without logging any token values."""
+        response = getattr(error, "response", None)
+        if response is None:
+            logging.getLogger(__name__).warning("Token refresh failed: %s", type(error).__name__)
+            return
+        try:
+            error_code = response.json().get("error")
+        except Exception:
+            error_code = None
+        logging.getLogger(__name__).warning(
+            "Token refresh failed: HTTP %s (%s)", response.status_code, error_code or "no error code"
+        )
+
     def refresh_access_token(self):
         """ Use the refresh token to get a new access token. """
         if "refresh_token" not in self.tokens:
@@ -154,8 +225,12 @@ def refresh_access_token(self):
         response = requests.post(self.token_url, data=data)
         response.raise_for_status()
         token_data = response.json()
-        
+
         self.set_tokens(token_data)
+        # Save here, not in get_access_token(), so direct callers also persist the rotated refresh token
+        with self.lock:
+            self._save_pending = True
+            self._save_refreshed_tokens()
         return self.tokens["access_token"]
 
     def get_id_token(self):

kinde_sdk/auth/user_session.py

@@ -1,8 +1,12 @@
 from .token_manager import TokenManager
+import logging
 import threading
 import time
 from typing import Dict, Any, Optional
 from kinde_sdk.core.storage.storage_manager import StorageManager
+from kinde_sdk.core.exceptions import KindeTokenPersistenceException
+
+logger = logging.getLogger(__name__)
 
 class UserSession:
     def __init__(self):
@@ -25,14 +29,18 @@ def set_user_data(self, user_id: str, user_info: Dict[str, Any], token_data: Dic
                 # Set redirect URI if available
                 if "redirect_uri" in user_info:
                     token_manager.set_redirect_uri(user_info["redirect_uri"])
-
+                self._bind_token_storage(user_id, token_manager)
+
                 self.user_sessions[user_id] = {
                     "user_info": user_info,
                     "token_manager": token_manager
                 }
             else:
                 # Update existing user info
                 self.user_sessions[user_id]["user_info"] = user_info
+                # The token manager is shared per user, so another UserSession may have
+                # bound it since. Bind it here again so refreshes save this user_info
+                self._bind_token_storage(user_id, self.user_sessions[user_id]["token_manager"])
 
             # Set tokens in token manager
             self.user_sessions[user_id]["token_manager"].set_tokens(token_data)
@@ -54,6 +62,13 @@ def _save_to_storage(self, user_id: str):
             # if you want device-specific sessions, remove the "user:" prefix
             self.storage_manager.setItems(user_id, serialized_data)
 
+    def _bind_token_storage(self, user_id: str, token_manager: TokenManager):
+        """Save the session again whenever the token manager refreshes its tokens, and
+        let it read the saved tokens before refreshing, so neither a restarted process
+        nor another worker reuses a refresh token that Kinde has already rotated out."""
+        token_manager.on_tokens_refreshed = lambda: self._save_to_storage(user_id)
+        token_manager.load_persisted_tokens = lambda: (self.storage_manager.get(user_id) or {}).get("tokens")
+
     def reset(self):
         """Reset all session data - useful for testing"""
         with self.lock:
@@ -94,7 +109,8 @@ def _load_from_storage(self, user_id: str) -> bool:
         # Set redirect URI if available
         if "redirect_uri" in user_info:
             token_manager.set_redirect_uri(user_info["redirect_uri"])
-
+        self._bind_token_storage(user_id, token_manager)
+
         # Set tokens
         token_manager.tokens = tokens
 
@@ -138,11 +154,17 @@ def is_authenticated(self, user_id: str) -> bool:
             # This will handle refreshing if needed
             access_token = token_manager.get_access_token()
             return access_token is not None and len(access_token) > 0
-        except ValueError:
+        except KindeTokenPersistenceException:
+            # The refreshed tokens are valid; only saving them failed (already logged),
+            # and the save is retried on the next call
+            return True
+        except ValueError as e:
             # Token is expired and cannot be refreshed
+            logger.info("User is not authenticated: %s", e)
             return False
-        except Exception:
+        except Exception as e:
             # Any other error means authentication failed
+            logger.warning("User is not authenticated: %s", type(e).__name__)
             return False
 
     def logout(self, user_id: str) -> None:

kinde_sdk/core/exceptions.py

@@ -14,6 +14,13 @@ class KindeTokenException(KindeException):
     """Raised when there is an error with token operations."""
     pass
 
+class KindeTokenPersistenceException(KindeTokenException):
+    """Raised when tokens were refreshed but could not be saved. The refreshed tokens
+    are kept, `access_token` is still valid, and the save is retried on the next call."""
+    def __init__(self, message: str, access_token: str = None):
+        super().__init__(message)
+        self.access_token = access_token
+
 class KindeRetrieveException(KindeException):
     """Raised when there is an error retrieving data."""
     pass 

kinde_sdk/core/storage/storage_manager.py

@@ -8,7 +8,8 @@
 
 class StorageManager:
     _instance = None
-    _lock = threading.Lock()  # Lock for thread safety
+    # Reentrant: reset() calls initialize() and get_device_id() calls get() while holding it
+    _lock = threading.RLock()
 
     def __new__(cls):
         with cls._lock: