diff --git a/litellm/llms/github_copilot/authenticator.py b/litellm/llms/github_copilot/authenticator.py index adca402601c..dafaf913327 100644 --- a/litellm/llms/github_copilot/authenticator.py +++ b/litellm/llms/github_copilot/authenticator.py @@ -38,6 +38,7 @@ class Authenticator: os.getenv("GITHUB_COPILOT_ACCESS_TOKEN_FILE", "access-token"), ) self.api_key_file = os.path.join(self.token_dir, os.getenv("GITHUB_COPILOT_API_KEY_FILE", "api-key.json")) + self._session_cache: dict[str, dict[str, Any]] = {} def get_access_token(self, access_token: str | None = None) -> str: """ @@ -96,6 +97,32 @@ class Authenticator: Raises: GetAPIKeyError: If unable to obtain an API key. """ + # When an explicit access_token is provided, isolate in memory to prevent sharing a cached key across callers + if access_token: + cached_info = self._session_cache.get(access_token) + if cached_info and cached_info.get("expires_at", 0) > time.time(): + token = cached_info.get("token") + if token: + return token + + try: + api_key_info = self._refresh_api_key(access_token) + self._session_cache[access_token] = api_key_info + token: Final = api_key_info.get("token") + if token: + return token + else: + raise GetAPIKeyError( + message="API key response missing token", + status_code=401, + ) + except RefreshAPIKeyError as e: + raise GetAPIKeyError( + message=f"Failed to refresh API key: {e}", + status_code=401, + ) + + # Fallback to single-user local file storage when no explicit token is passed try: with open(self.api_key_file, "r") as f: api_key_info = json.load(f) @@ -115,7 +142,7 @@ class Authenticator: pass # Already logged in the try block try: - api_key_info = self._refresh_api_key(access_token) + api_key_info = self._refresh_api_key() try: self._ensure_token_dir() with open(self.api_key_file, "w") as f: @@ -194,7 +221,7 @@ class Authenticator: ) def _ensure_token_dir(self) -> None: - """Ensure the token directory exists, falling back to temp directory on permission errors.""" + """Ensure the token directory exists, falling back to secure user-isolated temp directory on permission errors.""" try: if not os.path.exists(self.token_dir): os.makedirs(self.token_dir, mode=0o700, exist_ok=True) @@ -204,7 +231,8 @@ class Authenticator: self.token_dir, e, ) - self.token_dir = os.path.join(tempfile.gettempdir(), "litellm", "github_copilot") + uid = os.getuid() if hasattr(os, "getuid") else "user" + self.token_dir = os.path.join(tempfile.gettempdir(), f"litellm_{uid}", "github_copilot") try: os.makedirs(self.token_dir, mode=0o700, exist_ok=True) except OSError: diff --git a/tests/test_litellm/llms/github_copilot/test_github_copilot_authenticator.py b/tests/test_litellm/llms/github_copilot/test_github_copilot_authenticator.py index 4055ec13bbf..db68eb416c6 100644 --- a/tests/test_litellm/llms/github_copilot/test_github_copilot_authenticator.py +++ b/tests/test_litellm/llms/github_copilot/test_github_copilot_authenticator.py @@ -35,7 +35,6 @@ class TestGitHubCopilotAuthenticator: assert os.path.basename(auth.token_dir) == "github_copilot" assert os.path.basename(auth.access_token_file) == "access-token" assert os.path.basename(auth.api_key_file) == "api-key.json" - # Lazy init: directory is NOT created eagerly on __init__ mock_makedirs.assert_not_called() def test_ensure_token_dir(self): @@ -60,6 +59,18 @@ class TestGitHubCopilotAuthenticator: assert auth.token_dir != original_dir assert "litellm" in auth.token_dir + def test_get_api_key_with_explicit_token_isolation(self, authenticator): + """Test that explicit tokens use isolated in-memory caching and do not read stale disk files.""" + mock_data = {"token": "token-b-session-key", "expires_at": (datetime.now() + timedelta(hours=1)).timestamp()} + with ( + patch.object(authenticator, "_refresh_api_key", return_value=mock_data) as mock_refresh, + patch("builtins.open") as mock_file_open, + ): + api_key = authenticator.get_api_key(access_token="user-b-custom-token") + assert api_key == "token-b-session-key" + mock_refresh.assert_called_once_with("user-b-custom-token") + mock_file_open.assert_not_called() + def test_get_github_headers(self, authenticator): """Test that GitHub headers are correctly generated.""" headers = authenticator._get_github_headers()