mirror of
https://github.com/BerriAI/litellm.git
synced 2026-08-28 05:25:59 +00:00
fix: scope CLI stored token to base_url to prevent cross-domain credential leakage (#26945)
* fix: add expected_base_url origin check to get_litellm_gateway_api_key * fix: scope get_stored_api_key and save base_url on login * fix: pass base_url to get_stored_api_key in CLI entrypoint * fix: scope ProxyClient stored key to base_url * test: add expected_base_url coverage for get_stored_api_key * fix: initialize self.http with resolved api_key not raw param * fix: black formatting in client.py and test_auth_commands.py
This commit is contained in:
parent
c06cc560e0
commit
231c430200
5 changed files with 83 additions and 15 deletions
|
|
@ -31,15 +31,23 @@ def load_cli_token() -> Optional[dict]:
|
|||
return None
|
||||
|
||||
|
||||
def get_litellm_gateway_api_key() -> Optional[str]:
|
||||
def get_litellm_gateway_api_key(
|
||||
expected_base_url: Optional[str] = None,
|
||||
) -> Optional[str]:
|
||||
"""
|
||||
Get the stored CLI API key for use with LiteLLM SDK.
|
||||
|
||||
This function reads the token file created by `litellm-proxy login`
|
||||
and returns the API key for use in Python scripts.
|
||||
|
||||
Args:
|
||||
expected_base_url: When provided, the key is only returned if it was
|
||||
originally issued for this URL. Pass the target server URL to
|
||||
prevent credential leakage when the client is pointed at a
|
||||
different (possibly malicious) server.
|
||||
|
||||
Returns:
|
||||
str: The API key if found, None otherwise
|
||||
str: The API key if found (and origin matches), None otherwise
|
||||
|
||||
Example:
|
||||
>>> import litellm
|
||||
|
|
@ -53,6 +61,10 @@ def get_litellm_gateway_api_key() -> Optional[str]:
|
|||
>>> )
|
||||
"""
|
||||
token_data = load_cli_token()
|
||||
if token_data and "key" in token_data:
|
||||
return token_data["key"]
|
||||
return None
|
||||
if not token_data or "key" not in token_data:
|
||||
return None
|
||||
if expected_base_url is not None:
|
||||
stored_url = token_data.get("base_url")
|
||||
if stored_url != expected_base_url.rstrip("/"):
|
||||
return None
|
||||
return token_data["key"]
|
||||
|
|
|
|||
|
|
@ -53,12 +53,16 @@ def clear_token() -> None:
|
|||
os.remove(token_file)
|
||||
|
||||
|
||||
def get_stored_api_key() -> Optional[str]:
|
||||
"""Get the stored API key from token file"""
|
||||
# Use the SDK-level utility
|
||||
def get_stored_api_key(expected_base_url: Optional[str] = None) -> Optional[str]:
|
||||
"""Get the stored API key from token file.
|
||||
|
||||
If expected_base_url is provided, the key is only returned when it was
|
||||
originally issued for that URL. This prevents credential leakage when the
|
||||
CLI is pointed at a different (possibly malicious) server.
|
||||
"""
|
||||
from litellm.litellm_core_utils.cli_token_utils import get_litellm_gateway_api_key
|
||||
|
||||
return get_litellm_gateway_api_key()
|
||||
return get_litellm_gateway_api_key(expected_base_url=expected_base_url)
|
||||
|
||||
|
||||
# Team selection utilities
|
||||
|
|
@ -572,9 +576,11 @@ def login(ctx: click.Context):
|
|||
api_key = auth_result["api_key"]
|
||||
user_id = auth_result["user_id"]
|
||||
|
||||
# Save token data (simplified for CLI - we just need the key)
|
||||
# Save token data. base_url is stored so we can verify origin
|
||||
# before reusing the key on a subsequent CLI invocation.
|
||||
save_token(
|
||||
{
|
||||
"base_url": base_url.rstrip("/"),
|
||||
"key": api_key,
|
||||
"user_id": user_id or "cli-user",
|
||||
"user_email": "unknown",
|
||||
|
|
|
|||
|
|
@ -74,9 +74,10 @@ def cli(ctx: click.Context, base_url: str, api_key: Optional[str]) -> None:
|
|||
"""LiteLLM Proxy CLI - Manage your LiteLLM proxy server"""
|
||||
ctx.ensure_object(dict)
|
||||
|
||||
# If no API key provided via flag or environment variable, try to load from saved token
|
||||
# If no API key provided via flag or environment variable, try to load from saved token.
|
||||
# Pass base_url so we only use the stored key when it was issued for this server.
|
||||
if api_key is None:
|
||||
api_key = get_stored_api_key()
|
||||
api_key = get_stored_api_key(expected_base_url=base_url)
|
||||
|
||||
ctx.obj["base_url"] = base_url
|
||||
ctx.obj["api_key"] = api_key
|
||||
|
|
|
|||
|
|
@ -28,12 +28,17 @@ class Client:
|
|||
api_key (Optional[str]): API key for authentication. If provided, it will be sent as a Bearer token.
|
||||
timeout: Request timeout in seconds (default: 30)
|
||||
"""
|
||||
self._base_url = base_url.rstrip("/") # Remove trailing slash if present
|
||||
self._api_key = get_litellm_gateway_api_key() or api_key
|
||||
self._base_url = base_url.rstrip("/")
|
||||
# Only use the stored CLI key when it was issued for this server.
|
||||
self._api_key = api_key or get_litellm_gateway_api_key(
|
||||
expected_base_url=self._base_url
|
||||
)
|
||||
|
||||
# Initialize resource clients
|
||||
|
||||
self.http = HTTPClient(base_url=base_url, api_key=api_key, timeout=timeout)
|
||||
self.http = HTTPClient(
|
||||
base_url=base_url, api_key=self._api_key, timeout=timeout
|
||||
)
|
||||
self.models = ModelsManagementClient(
|
||||
base_url=self._base_url, api_key=self._api_key
|
||||
)
|
||||
|
|
|
|||
|
|
@ -231,6 +231,50 @@ class TestTokenUtilities:
|
|||
result = get_stored_api_key()
|
||||
assert result is None
|
||||
|
||||
def test_get_stored_api_key_base_url_match(self):
|
||||
"""Stored key is returned when expected_base_url matches stored origin"""
|
||||
token_data = {"key": "sk-prod", "base_url": "https://real-proxy.com"}
|
||||
with patch(
|
||||
"litellm.litellm_core_utils.cli_token_utils.load_cli_token",
|
||||
return_value=token_data,
|
||||
):
|
||||
assert (
|
||||
get_stored_api_key(expected_base_url="https://real-proxy.com")
|
||||
== "sk-prod"
|
||||
)
|
||||
|
||||
def test_get_stored_api_key_base_url_match_trailing_slash(self):
|
||||
"""Trailing slash on expected_base_url is normalised before comparison"""
|
||||
token_data = {"key": "sk-prod", "base_url": "https://real-proxy.com"}
|
||||
with patch(
|
||||
"litellm.litellm_core_utils.cli_token_utils.load_cli_token",
|
||||
return_value=token_data,
|
||||
):
|
||||
assert (
|
||||
get_stored_api_key(expected_base_url="https://real-proxy.com/")
|
||||
== "sk-prod"
|
||||
)
|
||||
|
||||
def test_get_stored_api_key_base_url_mismatch(self):
|
||||
"""Stored key is NOT returned when expected_base_url differs from stored origin"""
|
||||
token_data = {"key": "sk-prod", "base_url": "https://real-proxy.com"}
|
||||
with patch(
|
||||
"litellm.litellm_core_utils.cli_token_utils.load_cli_token",
|
||||
return_value=token_data,
|
||||
):
|
||||
assert get_stored_api_key(expected_base_url="https://evil.com") is None
|
||||
|
||||
def test_get_stored_api_key_old_token_no_base_url(self):
|
||||
"""Old tokens without a base_url field are rejected when origin check is requested"""
|
||||
token_data = {"key": "sk-old-token"}
|
||||
with patch(
|
||||
"litellm.litellm_core_utils.cli_token_utils.load_cli_token",
|
||||
return_value=token_data,
|
||||
):
|
||||
assert (
|
||||
get_stored_api_key(expected_base_url="https://real-proxy.com") is None
|
||||
)
|
||||
|
||||
|
||||
class TestLoginCommand:
|
||||
"""Test login CLI command"""
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue