diff --git a/kinde_sdk/auth/oauth.py b/kinde_sdk/auth/oauth.py index 0c2d19d1..90d5f0d5 100644 --- a/kinde_sdk/auth/oauth.py +++ b/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: """ diff --git a/kinde_sdk/auth/token_manager.py b/kinde_sdk/auth/token_manager.py index 3411fc39..82f2d91e 100644 --- a/kinde_sdk/auth/token_manager.py +++ b/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): diff --git a/kinde_sdk/auth/user_session.py b/kinde_sdk/auth/user_session.py index 19d9ca07..50c4e78e 100644 --- a/kinde_sdk/auth/user_session.py +++ b/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,7 +29,8 @@ 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 @@ -54,6 +59,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 +106,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 +151,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: diff --git a/kinde_sdk/core/exceptions.py b/kinde_sdk/core/exceptions.py index 7d562ea2..03e5fb96 100644 --- a/kinde_sdk/core/exceptions.py +++ b/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 diff --git a/kinde_sdk/core/storage/storage_manager.py b/kinde_sdk/core/storage/storage_manager.py index c40d4f8b..d9eccb9f 100644 --- a/kinde_sdk/core/storage/storage_manager.py +++ b/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: diff --git a/testv2/testv2_auth/test_token_refresh.py b/testv2/testv2_auth/test_token_refresh.py new file mode 100644 index 00000000..79bd06f2 --- /dev/null +++ b/testv2/testv2_auth/test_token_refresh.py @@ -0,0 +1,290 @@ +import copy +import threading +import time +import unittest +from unittest.mock import MagicMock, patch + +import jwt +import requests + +from kinde_sdk.auth.oauth import OAuth +from kinde_sdk.auth.token_manager import TokenManager +from kinde_sdk.auth.user_session import UserSession +from kinde_sdk.core.exceptions import KindeTokenPersistenceException +from kinde_sdk.core.storage.memory_storage import MemoryStorage +from kinde_sdk.core.storage.storage_manager import StorageManager + +TOKEN_URL = "https://example.kinde.com/oauth2/token" +USER_INFO = {"client_id": "refresh_test_client", "token_url": TOKEN_URL} + + +def _jwt(sub="user_1"): + return jwt.encode({"sub": sub, "iat": time.time()}, "key", algorithm="HS256") + + +class CopyingMemoryStorage(MemoryStorage): + """Copies values on write and read, like a serialising session store (e.g. Flask-Session + files), so tests can't pass by sharing a dict with the in-memory token manager.""" + + def set(self, key, value): + super().set(key, copy.deepcopy(value)) + + def get(self, key): + return copy.deepcopy(super().get(key)) + + +def _token_response(status, body): + response = requests.Response() + response.status_code = status + response._content = body.encode() + return response + + +class TestTokenRefresh(unittest.TestCase): + def setUp(self): + StorageManager().reset() + StorageManager().initialize({"type": "memory"}, storage=CopyingMemoryStorage()) + TokenManager.reset_instances() + self.old_access_token = _jwt() + self.new_access_token = _jwt() + # Expired access token with a refresh token, as stored after a login with the offline scope + self.expired_tokens = { + "access_token": self.old_access_token, + "refresh_token": "old_refresh_token", + "expires_in": -10, + } + self.refresh_ok = _token_response( + 200, + '{"access_token": "%s", "refresh_token": "new_refresh_token", "expires_in": 3600}' + % self.new_access_token, + ) + + def tearDown(self): + StorageManager().reset() + TokenManager.reset_instances() + + def test_refresh_does_not_deadlock(self): + token_manager = TokenManager("user_1", "refresh_test_client", "secret", TOKEN_URL) + token_manager.set_tokens(self.expired_tokens) + result = {} + + def get_token(): + result["token"] = token_manager.get_access_token() + + with patch("kinde_sdk.auth.token_manager.requests.post", return_value=self.refresh_ok): + thread = threading.Thread(target=get_token, daemon=True) + thread.start() + thread.join(5) + + self.assertFalse(thread.is_alive(), "get_access_token() blocked while refreshing") + self.assertEqual(result["token"], self.new_access_token) + + def test_refreshed_tokens_are_saved_to_session_storage(self): + UserSession().set_user_data("user_1", dict(USER_INFO), self.expired_tokens) + + with patch("kinde_sdk.auth.token_manager.requests.post", return_value=self.refresh_ok): + self.assertTrue(UserSession().is_authenticated("user_1")) + + stored_tokens = StorageManager().get("user_1")["tokens"] + self.assertEqual(stored_tokens["access_token"], self.new_access_token) + self.assertEqual(stored_tokens["refresh_token"], "new_refresh_token") + + # A restarted process restores the refreshed tokens, not the used refresh token + TokenManager.reset_instances() + restored = UserSession().get_token_manager("user_1") + self.assertEqual(restored.tokens["refresh_token"], "new_refresh_token") + + def _saved_by_other_worker(self, tokens): + """Overwrite the stored session as another worker's refresh would.""" + StorageManager().setItems("user_1", {"user_info": dict(USER_INFO), "tokens": tokens}) + + def test_direct_refresh_access_token_call_saves_rotated_tokens(self): + session = UserSession() + session.set_user_data("user_1", dict(USER_INFO), self.expired_tokens) + + with patch("kinde_sdk.auth.token_manager.requests.post", return_value=self.refresh_ok): + session.get_token_manager("user_1").refresh_access_token() + + self.assertEqual(StorageManager().get("user_1")["tokens"]["refresh_token"], "new_refresh_token") + + def test_get_access_token_saves_once_per_refresh(self): + session = UserSession() + session.set_user_data("user_1", dict(USER_INFO), self.expired_tokens) + token_manager = session.get_token_manager("user_1") + save = MagicMock(wraps=token_manager.on_tokens_refreshed) + token_manager.on_tokens_refreshed = save + + with patch("kinde_sdk.auth.token_manager.requests.post", return_value=self.refresh_ok): + token_manager.get_access_token() + token_manager.get_access_token() # Still valid: no refresh, no save + + self.assertEqual(save.call_count, 1) + + def test_uses_tokens_another_worker_already_refreshed(self): + # This worker has the session cached with an expired access token + session = UserSession() + session.set_user_data("user_1", dict(USER_INFO), self.expired_tokens) + # Another worker refreshed and saved new tokens, rotating the refresh token + self._saved_by_other_worker({ + "access_token": self.new_access_token, + "refresh_token": "rotated_by_other_worker", + "expires_at": time.time() + 3600, + }) + + with patch("kinde_sdk.auth.token_manager.requests.post") as post: + self.assertEqual(session.get_token_manager("user_1").get_access_token(), self.new_access_token) + post.assert_not_called() + + def test_refreshes_with_the_latest_saved_refresh_token(self): + session = UserSession() + session.set_user_data("user_1", dict(USER_INFO), self.expired_tokens) + # Another worker refreshed earlier; its access token has since expired too + self._saved_by_other_worker({ + "access_token": _jwt(), + "refresh_token": "rotated_by_other_worker", + "expires_at": time.time() - 1, + }) + + with patch("kinde_sdk.auth.token_manager.requests.post", return_value=self.refresh_ok) as post: + self.assertEqual(session.get_token_manager("user_1").get_access_token(), self.new_access_token) + self.assertEqual(post.call_args.kwargs["data"]["refresh_token"], "rotated_by_other_worker") + + def test_failed_save_keeps_tokens_raises_and_is_retried_without_refreshing(self): + session = UserSession() + session.set_user_data("user_1", dict(USER_INFO), self.expired_tokens) + token_manager = session.get_token_manager("user_1") + + with patch("kinde_sdk.auth.token_manager.requests.post", return_value=self.refresh_ok) as post: + with patch.object(StorageManager, "setItems", side_effect=OSError("disk full")), \ + self.assertRaises(KindeTokenPersistenceException) as raised: + token_manager.get_access_token() + # The refreshed tokens are kept and handed to the caller + self.assertEqual(raised.exception.access_token, self.new_access_token) + self.assertEqual(token_manager.tokens["refresh_token"], "new_refresh_token") + self.assertEqual(StorageManager().get("user_1")["tokens"]["refresh_token"], "old_refresh_token") + + # Next call: storage works again, the save is retried and Kinde isn't called again + self.assertEqual(token_manager.get_access_token(), self.new_access_token) + self.assertEqual(post.call_count, 1) + self.assertEqual(StorageManager().get("user_1")["tokens"]["refresh_token"], "new_refresh_token") + + def test_pending_save_runs_before_another_refresh_even_if_that_refresh_fails(self): + session = UserSession() + session.set_user_data("user_1", dict(USER_INFO), self.expired_tokens) + token_manager = session.get_token_manager("user_1") + + with patch("kinde_sdk.auth.token_manager.requests.post", return_value=self.refresh_ok), \ + patch.object(StorageManager, "setItems", side_effect=OSError("disk full")), \ + self.assertRaises(KindeTokenPersistenceException): + token_manager.get_access_token() + + # The refreshed access token expires before the next call, and Kinde is unreachable + token_manager.tokens["expires_at"] = time.time() - 10 + with patch("kinde_sdk.auth.token_manager.requests.post", + side_effect=requests.ConnectionError("offline")), \ + self.assertRaises(requests.RequestException): + token_manager.get_access_token() + + # The rotated refresh token was saved before the refresh was attempted + self.assertEqual(StorageManager().get("user_1")["tokens"]["refresh_token"], "new_refresh_token") + + def test_user_stays_authenticated_when_saving_refreshed_tokens_fails(self): + session = UserSession() + session.set_user_data("user_1", dict(USER_INFO), self.expired_tokens) + + with patch("kinde_sdk.auth.token_manager.requests.post", return_value=self.refresh_ok), \ + patch.object(StorageManager, "setItems", side_effect=OSError("disk full")): + self.assertTrue(session.is_authenticated("user_1")) + + def test_unreadable_saved_tokens_are_skipped_and_refresh_still_happens(self): + session = UserSession() + session.set_user_data("user_1", dict(USER_INFO), self.expired_tokens) + token_manager = session.get_token_manager("user_1") + token_manager.load_persisted_tokens = MagicMock(side_effect=OSError("storage down")) + + with patch("kinde_sdk.auth.token_manager.requests.post", return_value=self.refresh_ok), \ + self.assertLogs("kinde_sdk.auth.token_manager", level="WARNING") as logs: + self.assertEqual(token_manager.get_access_token(), self.new_access_token) + + self.assertIn("Saved tokens could not be read: OSError", "\n".join(logs.output)) + + def test_refresh_without_a_response_logs_the_error_type(self): + UserSession().set_user_data("user_1", dict(USER_INFO), self.expired_tokens) + + with patch("kinde_sdk.auth.token_manager.requests.post", + side_effect=requests.ConnectionError("network down")), \ + self.assertLogs("kinde_sdk.auth.token_manager", level="WARNING") as logs, \ + self.assertRaises(requests.ConnectionError): + UserSession().get_token_manager("user_1").get_access_token() + + self.assertIn("Token refresh failed: ConnectionError", "\n".join(logs.output)) + + def test_refresh_error_without_a_json_body_is_logged(self): + UserSession().set_user_data("user_1", dict(USER_INFO), self.expired_tokens) + bad_gateway = _token_response(502, "Bad gateway") + + with patch("kinde_sdk.auth.token_manager.requests.post", return_value=bad_gateway), \ + self.assertLogs("kinde_sdk.auth.token_manager", level="WARNING") as logs, \ + self.assertRaises(requests.HTTPError): + UserSession().get_token_manager("user_1").get_access_token() + + self.assertIn("Token refresh failed: HTTP 502 (no error code)", "\n".join(logs.output)) + + def test_refresh_failure_is_logged_without_token_values(self): + UserSession().set_user_data("user_1", dict(USER_INFO), self.expired_tokens) + rejected = _token_response(400, '{"error": "invalid_grant", "error_description": "expired"}') + + with patch("kinde_sdk.auth.token_manager.requests.post", return_value=rejected), \ + self.assertLogs("kinde_sdk.auth", level="WARNING") as logs: + self.assertFalse(UserSession().is_authenticated("user_1")) + + output = "\n".join(logs.output) + self.assertIn("Token refresh failed: HTTP 400 (invalid_grant)", output) + self.assertNotIn("old_refresh_token", output) + self.assertNotIn(self.old_access_token, output) + + +class TestFrameworkSessionSurvivesRestart(unittest.TestCase): + def setUp(self): + StorageManager().reset() + TokenManager.reset_instances() + # Stands in for the framework's session store (e.g. Flask-Session files), + # which outlives the process + self.session_store = MemoryStorage() + + def tearDown(self): + StorageManager().reset() + TokenManager.reset_instances() + + def _start_process(self): + """Simulate an app process starting: a new random device id, then OAuth set-up.""" + StorageManager()._device_id = None + TokenManager.reset_instances() + framework = MagicMock() + framework.get_name.return_value = "flask" + with patch("kinde_sdk.auth.oauth.FrameworkFactory.create_framework", return_value=framework), \ + patch("kinde_sdk.auth.oauth.StorageFactory.create_storage", return_value=self.session_store), \ + patch("requests.get", side_effect=Exception("offline")): + OAuth( + client_id="refresh_test_client", + redirect_uri="http://localhost/callback", + host="https://example.kinde.com", + framework="flask", + app=MagicMock(), + ) + + def test_session_is_found_after_restart(self): + self._start_process() + self.assertEqual(StorageManager().get_device_id(), "flask") + UserSession().set_user_data( + "user_1", dict(USER_INFO), {"access_token": _jwt(), "expires_in": 3600} + ) + + self._start_process() + + self.assertEqual(StorageManager().get_device_id(), "flask") + self.assertIsNotNone(UserSession().get_token_manager("user_1")) + + +if __name__ == "__main__": + unittest.main()