Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/client/cli/commands/auth.py: 0%
597 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1import os
2import sys
3import time
4import webbrowser
5from collections.abc import Callable, Mapping, Sequence
6from typing import Any, Final, TypeVar
7from urllib.parse import urlencode
9import click
10import requests
11from rich.console import Console
12from rich.table import Table
13from typing_extensions import NotRequired, ReadOnly, TypedDict, assert_never
15from litellm.constants import CLI_JWT_EXPIRATION_HOURS
16from litellm.litellm_core_utils.cli_keyring import (
17 DISABLE_KEYRING_ENV_VAR,
18 SYSTEM_KEYRING,
19 KeyringDisabled,
20 KeyringDiscardsWrites,
21 KeyringNotInstalled,
22 KeyringUnreachable,
23 SecretErased,
24 SecretFound,
25 SecretMissing,
26 SecretStored,
27 SecretStranded,
28 SecretVault,
29)
30from litellm.litellm_core_utils.cli_token_utils import (
31 CliTokenRecord,
32 CredentialNotCleared,
33 CredentialNotRecorded,
34 CredentialNotSaved,
35 SecretSave,
36 clear_cli_token,
37 get_cli_token_file_path,
38 is_cli_token_fresh,
39 load_cli_token,
40 save_cli_token,
41)
43from .claude_settings import (
44 STARTING_MODEL_ROLE,
45 ClaudeSettingsError,
46 KeepModel,
47 StaticToken,
48 claude_settings_path,
49 configure_claude_settings,
50 configure_state_path,
51 refuse_while_owned,
52 settings_file_owners,
53)
54from .pkce_login import (
55 Http,
56 PkceFailure,
57 RevocationUnavailable,
58 fresh_api_key,
59 pkce_token_record,
60 revoke_stored_credential,
61 run_pkce_login,
62)
65class CliTokenData(TypedDict):
66 base_url: str
67 key: str
68 user_id: str
69 user_email: str
70 user_role: str
71 auth_header_name: str
72 jwt_token: str
73 timestamp: float
74 expires_at: ReadOnly[NotRequired[float]]
75 refresh_token: ReadOnly[NotRequired[str]]
76 client_id: ReadOnly[NotRequired[str]]
77 token_endpoint: ReadOnly[NotRequired[str]]
78 revocation_endpoint: ReadOnly[NotRequired[str]]
79 resource: ReadOnly[NotRequired[str]]
80 team_id: ReadOnly[NotRequired[str | None]]
83class CliTeam(TypedDict, total=False):
84 team_id: str | None
85 team_alias: str | None
86 models: list[str]
87 max_budget: float | None
90class CliContextObj(TypedDict):
91 base_url: str
92 base_url_explicit: NotRequired[bool]
93 secret_vault: NotRequired[ReadOnly[SecretVault]]
94 api_key: ReadOnly[NotRequired[str | None]]
95 api_key_from_token_file: ReadOnly[NotRequired[bool]]
98class CliPollData(TypedDict, total=False):
99 status: str
100 key: str
101 user_id: str
102 teams: list[str]
103 team_details: object
104 requires_team_selection: bool
105 team_id: str
108class CliSsoStartData(TypedDict):
109 login_id: ReadOnly[str]
110 poll_secret: ReadOnly[str]
111 user_code: ReadOnly[str]
112 verification_uri_complete: ReadOnly[NotRequired[str]]
115class CliAuthResult(TypedDict):
116 api_key: str
117 user_id: str | None
118 teams: list[str]
119 team_id: str | None
122_TeamMapping: Final = TypeVar("_TeamMapping", bound=Mapping[str, object])
124KEYRING_INSTALL_HINT: Final = "pip install 'litellm[cli]'"
126KEYRING_ENABLE_HINT: Final = "keyring --enable (or unset PYTHON_KEYRING_BACKEND)"
128STRANDED_CREDENTIAL_MESSAGE: Final = (
129 "Logged out locally, but your credential is still in the OS keychain and could not be removed."
130)
132UNCHECKED_KEYCHAIN_MESSAGE: Final = (
133 "Logged out locally, but your OS keychain could not be checked, so a credential stored there by "
134 "an earlier login may still be usable."
135)
138def storage_notice(outcome: SecretSave) -> str:
139 """Tell the user where the credential ended up, and how to get keychain storage if it did not."""
140 path: Final = get_cli_token_file_path()
141 match outcome:
142 case SecretStored():
143 return "Credential stored in your OS keychain."
144 case KeyringNotInstalled():
145 return (
146 f"Credential stored in {path} (owner-only). "
147 f"For OS keychain storage, install the keyring package with: {KEYRING_INSTALL_HINT}"
148 )
149 case KeyringDisabled():
150 return f"Keychain storage is off ({DISABLE_KEYRING_ENV_VAR}). Credential stored in {path} (owner-only)."
151 case KeyringUnreachable():
152 return f"No OS keychain available. Credential stored in {path} (owner-only)."
153 case KeyringDiscardsWrites():
154 return (
155 f"Your keyring backend keeps nothing it is given, so the credential was stored in {path} "
156 f"(owner-only) instead. For OS keychain storage, run: {KEYRING_ENABLE_HINT}"
157 )
158 case CredentialNotSaved(detail=detail):
159 return (
160 f"Signed in, but the credential could not be saved to {path}: {detail}. "
161 "Any login you already had is untouched. Run 'lite login' again once that path is "
162 "writable, or 'lite logout' to clear whatever is stored now."
163 )
164 case CredentialNotRecorded():
165 return (
166 f"Signed in, and the credential is in your OS keychain, but {path} could not be "
167 "replaced, so it still describes your previous login and may still hold its "
168 "credential. Run 'lite login' again once that path is writable, or 'lite logout' "
169 "to clear both."
170 )
173def keychain_unreadable_notice(vault: SecretVault) -> str:
174 """Explain why the secret half of a stored login cannot be produced, and what fixes it"""
175 match vault.read():
176 case KeyringNotInstalled():
177 return (
178 "Your credential is in your OS keychain, which this install cannot read without the "
179 f"keyring package. Install it with: {KEYRING_INSTALL_HINT}, or run 'lite login' to start over."
180 )
181 case KeyringDisabled():
182 return (
183 f"Your credential is in your OS keychain, which {DISABLE_KEYRING_ENV_VAR} is blocking. "
184 "Unset it, or run 'lite login' to start over."
185 )
186 case KeyringUnreachable():
187 return (
188 "Your credential is in your OS keychain, which could not be read. Unlock it, or run "
189 "'lite login' to start over."
190 )
191 case SecretFound() | SecretMissing():
192 return "Your credential could not be read from your OS keychain. Run 'lite login' to start over."
195def context_secret_vault(ctx: click.Context) -> SecretVault:
196 """Where this invocation reads and writes secret material; injectable through ctx.obj for tests"""
197 ctx_obj: Final[CliContextObj | None] = ctx.obj
198 if ctx_obj is None:
199 return SYSTEM_KEYRING
200 return ctx_obj.get("secret_vault") or SYSTEM_KEYRING
203def load_token(*, vault: SecretVault = SYSTEM_KEYRING) -> Mapping[str, object] | None:
204 """The stored credential as a plain mapping, with the secret resolved out of the vault.
206 The PKCE renewal and revocation helpers read records by field name, so this is the
207 shape they get; the keychain split lives underneath, in `load_cli_token`.
208 """
209 record: Final = load_cli_token(vault=vault)
210 return None if record is None else record.model_dump(exclude_none=True)
213def save_token(record: CliTokenData, *, vault: SecretVault = SYSTEM_KEYRING) -> SecretSave:
214 """Store a credential the PKCE layer produced, secret in the vault and the rest on disk"""
215 return save_cli_token(CliTokenRecord(**record), vault=vault)
218def _renewal_saver(vault: SecretVault) -> Callable[[CliTokenData], None]:
219 """Persist a silently renewed credential, and say on stderr when no store would keep it.
221 A renewal rotates the refresh token, so a rotation that is never stored logs this
222 machine out on the next command; the user hears about it rather than guessing.
223 """
225 def save(record: CliTokenData) -> None:
226 outcome: Final = save_token(record, vault=vault)
227 if isinstance(outcome, (CredentialNotSaved, CredentialNotRecorded)):
228 _warn(storage_notice(outcome))
230 return save
233def _renewal_reader(vault: SecretVault) -> Callable[[], Mapping[str, object] | None]:
234 """Re-read the record mid-renewal, so a rotation a sibling `lite` process saved is seen"""
236 def reload() -> Mapping[str, object] | None:
237 return load_token(vault=vault)
239 return reload
242def get_stored_api_key(
243 expected_base_url: str | None = None,
244 *,
245 vault: SecretVault = SYSTEM_KEYRING,
246) -> str | None:
247 """Get the stored API key.
249 If expected_base_url is provided, the key is only returned when it was
250 originally issued for that URL. This prevents credential leakage when the
251 CLI is pointed at a different (possibly malicious) server. A key obtained by
252 ``lite login --pkce`` is refreshed here once it nears expiry.
253 """
254 token_data: Final = load_token(vault=vault)
255 if token_data is None:
256 return None
257 if expected_base_url is not None and token_data.get("base_url") != expected_base_url.rstrip("/"):
258 return None
259 return fresh_api_key(
260 token_data,
261 _renewal_saver(vault),
262 requests.Session(),
263 reload=_renewal_reader(vault),
264 warn=_warn,
265 )
268def _warn(message: str) -> None:
269 click.echo(message, err=True)
272def _login_command(renews: bool) -> str:
273 return "lite login --pkce" if renews else "lite login"
276# Team selection utilities
277def display_teams_table(teams: list[CliTeam]) -> None:
278 """Display teams in a formatted table"""
279 console: Final = Console()
281 if not teams:
282 console.print("No teams found for your user.")
283 return
285 table: Final = Table(title="Available Teams")
286 table.add_column("Index", style="cyan", no_wrap=True)
287 table.add_column("Team Alias", style="magenta")
288 table.add_column("Team ID", style="green")
289 table.add_column("Models", style="yellow")
290 table.add_column("Max Budget", style="blue")
292 for i, team in enumerate(teams):
293 team_alias = team.get("team_alias") or "N/A"
294 team_id = team.get("team_id", "N/A")
295 models = team.get("models", [])
296 max_budget = team.get("max_budget")
298 # Format models list
299 if models:
300 if len(models) > 3:
301 models_str = ", ".join(models[:3]) + f" (+{len(models) - 3} more)"
302 else:
303 models_str = ", ".join(models)
304 else:
305 models_str = "All models"
307 # Format budget
308 budget_str = f"${max_budget}" if max_budget else "Unlimited"
310 table.add_row(str(i + 1), team_alias, team_id, models_str, budget_str)
312 console.print(table)
315def get_key_input():
316 """Get a single key input from the user (cross-platform)"""
317 try:
318 if sys.platform == "win32":
319 import msvcrt
321 key = msvcrt.getch()
322 if key == b"\xe0": # Arrow keys on Windows
323 key = msvcrt.getch()
324 if key == b"H": # Up arrow
325 return "up"
326 elif key == b"P": # Down arrow
327 return "down"
328 elif key == b"\r": # Enter key
329 return "enter"
330 elif key == b"\x1b": # Escape key
331 return "escape"
332 elif key == b"q":
333 return "quit"
334 return None
335 else:
336 import termios
337 import tty
339 fd: Final = sys.stdin.fileno()
340 old_settings: Final = termios.tcgetattr(fd)
341 try:
342 tty.setraw(sys.stdin.fileno())
343 key = sys.stdin.read(1)
345 if key == "\x1b": # Escape sequence
346 key += sys.stdin.read(2)
347 if key == "\x1b[A": # Up arrow
348 return "up"
349 elif key == "\x1b[B": # Down arrow
350 return "down"
351 elif key == "\x1b": # Just escape
352 return "escape"
353 elif key == "\r" or key == "\n": # Enter key
354 return "enter"
355 elif key == "q":
356 return "quit"
357 return None
358 finally:
359 termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
360 except ImportError:
361 # Fallback to simple input if termios/msvcrt not available
362 return None
365def display_interactive_team_selection(teams: Sequence[Mapping[str, Any]], selected_index: int = 0) -> None:
366 """Display teams with one highlighted for selection"""
367 console: Final = Console()
369 # Clear the screen using Rich's method
370 console.clear()
372 console.print("Select a Team (Use up/down arrows, Enter to select, 'q' to skip):\n")
374 for i, team in enumerate(teams):
375 team_alias = team.get("team_alias") or "N/A"
376 team_id = team.get("team_id", "N/A")
377 models: list[str] = team.get("models", [])
378 max_budget = team.get("max_budget")
380 # Format models list
381 if models:
382 if len(models) > 3:
383 models_str = ", ".join(models[:3]) + f" (+{len(models) - 3} more)"
384 else:
385 models_str = ", ".join(models)
386 else:
387 models_str = "All models"
389 # Format budget
390 budget_str = f"${max_budget}" if max_budget else "Unlimited"
392 # Highlight the selected item
393 if i == selected_index:
394 console.print(f"> [bold cyan]{team_alias}[/bold cyan] ({team_id})")
395 console.print(f" Models: [yellow]{models_str}[/yellow]")
396 console.print(f" Budget: [blue]{budget_str}[/blue]\n")
397 else:
398 console.print(f" [dim]{team_alias}[/dim] ({team_id})")
399 console.print(f" Models: [dim]{models_str}[/dim]")
400 console.print(f" Budget: [dim]{budget_str}[/dim]\n")
403def prompt_team_selection(teams: Sequence[_TeamMapping]) -> _TeamMapping | None:
404 """Interactive team selection with arrow keys"""
405 if not teams:
406 return None
408 selected_index = 0
410 try:
411 # Check if we can use interactive mode
412 if not sys.stdin.isatty():
413 # Fallback to simple selection for non-interactive environments
414 return prompt_team_selection_fallback(teams)
416 while True:
417 display_interactive_team_selection(teams, selected_index)
419 key = get_key_input()
421 if key == "up":
422 selected_index = (selected_index - 1) % len(teams)
423 elif key == "down":
424 selected_index = (selected_index + 1) % len(teams)
425 elif key == "enter":
426 selected_team = teams[selected_index]
427 # Clear screen and show selection
428 console = Console()
429 console.clear()
430 click.echo(f"Selected team: {selected_team.get('team_alias', 'N/A')} ({selected_team.get('team_id')})")
431 return selected_team
432 elif key == "quit" or key == "escape":
433 # Clear screen
434 console = Console()
435 console.clear()
436 click.echo("Team selection skipped.")
437 return None
438 elif key is None:
439 # If we can't get key input, fall back to simple selection
440 return prompt_team_selection_fallback(teams)
442 except KeyboardInterrupt:
443 console = Console()
444 console.clear()
445 click.echo("\nTeam selection cancelled.")
446 return None
447 except Exception:
448 # If interactive mode fails, fall back to simple selection
449 return prompt_team_selection_fallback(teams)
452def prompt_team_selection_fallback(
453 teams: Sequence[_TeamMapping],
454) -> _TeamMapping | None:
455 """Fallback team selection for non-interactive environments"""
456 if not teams:
457 return None
459 while True:
460 try:
461 prompt_response: str = click.prompt(
462 "\nSelect a team by entering the index number (or 'skip' to continue without a team)",
463 type=str,
464 )
465 choice = prompt_response.strip()
467 if choice.lower() == "skip":
468 return None
470 index = int(choice) - 1
471 if 0 <= index < len(teams):
472 selected_team = teams[index]
473 click.echo(
474 f"\nSelected team: {selected_team.get('team_alias', 'N/A')} ({selected_team.get('team_id')})"
475 )
476 return selected_team
477 else:
478 click.echo(f"Invalid selection. Please enter a number between 1 and {len(teams)}")
479 except ValueError:
480 click.echo("Invalid input. Please enter a number or 'skip'")
481 except KeyboardInterrupt:
482 click.echo("\nTeam selection cancelled.")
483 return None
486def _response_error_detail(response: requests.Response) -> str | None:
487 try:
488 body: Final[dict[str, object] | list[object] | str | int | float | bool | None] = response.json()
489 except ValueError:
490 return None
491 detail: Final = body.get("detail") if isinstance(body, dict) else None
492 if isinstance(detail, str) and detail:
493 return detail
494 return None
497def _polling_error_message(response: requests.Response) -> str:
498 detail: Final = _response_error_detail(response)
499 if detail:
500 return f"Polling error: HTTP {response.status_code}: {detail}"
501 return f"Polling error: HTTP {response.status_code}"
504def _is_permanent_polling_error(status_code: int) -> bool:
505 return 400 <= status_code < 500 and status_code != 429
508# Polling-based authentication - no local server needed
509def _poll_for_ready_data(
510 url: str,
511 *,
512 headers: dict[str, str] | None = None,
513 total_timeout: int = 300,
514 poll_interval: int = 2,
515 request_timeout: int = 10,
516 pending_message: str | None = None,
517 pending_log_every: int = 10,
518 other_status_message: str | None = None,
519 other_status_log_every: int = 10,
520 http_error_log_every: int = 10,
521 connection_error_log_every: int = 10,
522) -> CliPollData | None:
523 for attempt in range(total_timeout // poll_interval):
524 try:
525 response = requests.get(url, headers=headers, timeout=request_timeout)
526 if response.status_code == 200:
527 data: CliPollData = response.json()
528 status = data.get("status")
529 if status == "ready":
530 return data
531 if status == "pending":
532 if pending_message and pending_log_every > 0 and attempt % pending_log_every == 0:
533 click.echo(pending_message)
534 elif other_status_message and other_status_log_every > 0 and attempt % other_status_log_every == 0:
535 click.echo(other_status_message)
536 elif _is_permanent_polling_error(response.status_code):
537 detail = _response_error_detail(response)
538 raise ValueError(
539 f"The proxy rejected the login session with HTTP {response.status_code}"
540 + (f": {detail}" if detail else f" and no error detail (from {url})")
541 )
542 elif http_error_log_every > 0 and attempt % http_error_log_every == 0:
543 click.echo(_polling_error_message(response))
544 except requests.RequestException as e:
545 if connection_error_log_every > 0 and attempt % connection_error_log_every == 0:
546 click.echo(f"Connection error (will retry): {e}")
547 time.sleep(poll_interval)
548 return None
551def _normalize_teams(teams: object, team_details: object) -> list[CliTeam]:
552 """If team_details are a
554 Args:
555 teams (_type_): _description_
556 team_details (_type_): _description_
558 Returns:
559 _type_: _description_
560 """
561 if isinstance(team_details, list) and team_details:
562 return [
563 {
564 "team_id": i.get("team_id") or i.get("id"),
565 "team_alias": i.get("team_alias"),
566 }
567 for i in team_details
568 if isinstance(i, dict) and (i.get("team_id") or i.get("id"))
569 ]
570 if isinstance(teams, list):
571 return [{"team_id": str(t), "team_alias": None} for t in teams]
572 return []
575def _start_cli_sso_flow(base_url: str) -> CliSsoStartData:
576 start_url: Final = f"{base_url}/sso/cli/start"
577 try:
578 response: Final = requests.post(start_url, timeout=10)
579 except requests.RequestException as e:
580 raise ValueError(
581 f"Could not reach the proxy at {start_url}: {e}. "
582 "Check that the proxy is running and that --base-url points at it."
583 ) from e
585 if response.status_code in (404, 405):
586 raise ValueError(
587 f"POST {start_url} returned HTTP {response.status_code}. "
588 "Either --base-url is wrong, or the proxy is older than this CLI and does not support "
589 "the CLI SSO login flow; upgrade the proxy or use a CLI version that matches it."
590 )
591 if response.status_code != 200:
592 detail: Final = _response_error_detail(response)
593 raise ValueError(
594 f"Starting CLI login failed: HTTP {response.status_code} from {start_url}"
595 + (f": {detail}" if detail else "")
596 )
598 try:
599 data: Final[CliSsoStartData] = response.json()
600 except ValueError:
601 content_type: Final = response.headers.get("content-type", "unknown")
602 raise ValueError(
603 f"The proxy returned a non-JSON response from {start_url} (content-type: {content_type}). "
604 "A proxy, load balancer, or auth gateway in front of LiteLLM may be intercepting the request. "
605 f"Response starts with: {response.text[:200]!r}"
606 )
608 required_fields: Final[tuple[str, ...]] = ("login_id", "poll_secret", "user_code")
609 missing_fields: Final = tuple(field for field in required_fields if not isinstance(data.get(field), str))
610 if missing_fields:
611 raise ValueError(
612 f"The response from {start_url} is missing required field(s): {', '.join(missing_fields)}. "
613 "The proxy version may not match this CLI; upgrade whichever is older."
614 )
615 return data
618def _get_cli_sso_poll_headers(poll_secret: str) -> dict[str, str]:
619 return {"x-litellm-cli-poll-secret": poll_secret}
622def _poll_for_authentication(base_url: str, key_id: str, poll_secret: str) -> CliAuthResult | None:
623 """
624 Poll the server for authentication completion and handle team selection.
626 Returns:
627 Dictionary with authentication data if successful, None otherwise
628 """
629 poll_url: Final = f"{base_url}/sso/cli/poll/{key_id}"
630 data: Final = _poll_for_ready_data(
631 poll_url,
632 headers=_get_cli_sso_poll_headers(poll_secret),
633 pending_message="Still waiting for authentication...",
634 )
635 if not data:
636 return None
637 if data.get("requires_team_selection"):
638 teams = data.get("teams", [])
639 team_details: Final = data.get("team_details")
640 user_id = data.get("user_id")
641 normalized_teams: Final[list[CliTeam]] = _normalize_teams(teams, team_details)
642 if not normalized_teams:
643 click.echo("Warning: No teams available for selection.")
644 return None
646 # User has multiple teams - let them select
647 jwt_with_team: Final = _handle_team_selection_during_polling(
648 base_url=base_url,
649 key_id=key_id,
650 poll_secret=poll_secret,
651 teams=normalized_teams,
652 )
654 # Use the team-specific JWT if selection succeeded
655 if jwt_with_team:
656 return {
657 "api_key": jwt_with_team,
658 "user_id": user_id,
659 "teams": teams,
660 "team_id": None, # Set by server in JWT
661 }
663 click.echo("Team selection cancelled or JWT generation failed.")
664 return None
666 # JWT is ready (single team or team already selected)
667 api_key: Final = data.get("key")
668 user_id = data.get("user_id")
669 teams = data.get("teams", [])
670 team_id: Final = data.get("team_id")
672 # Show which team was assigned
673 if team_id and len(teams) == 1:
674 click.echo(f"\nAutomatically assigned to team: {team_id}")
676 if api_key:
677 return {
678 "api_key": api_key,
679 "user_id": user_id,
680 "teams": teams,
681 "team_id": team_id,
682 }
684 return None
687def _handle_team_selection_during_polling(
688 base_url: str, key_id: str, poll_secret: str, teams: list[CliTeam]
689) -> str | None:
690 """
691 Handle team selection and re-poll with selected team_id.
693 Args:
694 teams: List of team IDs (strings)
696 Returns:
697 The JWT token with the selected team, or None if selection was skipped
698 """
699 if not teams:
700 click.echo("No teams found. You can create or join teams using the web interface.")
701 return None
703 click.echo("\n" + "=" * 60)
704 click.echo("Select a team for your CLI session...")
706 team_id: Final = _render_and_prompt_for_team_selection(teams)
708 if not team_id:
709 click.echo("No team selected.")
710 return None
712 click.echo(f"\nGenerating JWT for team: {team_id}")
714 poll_url: Final = f"{base_url}/sso/cli/poll/{key_id}?team_id={team_id}"
715 data: Final = _poll_for_ready_data(
716 poll_url,
717 headers=_get_cli_sso_poll_headers(poll_secret),
718 pending_message="Still waiting for team authentication...",
719 other_status_message="Waiting for team authentication to complete...",
720 http_error_log_every=10,
721 )
722 if not data:
723 return None
724 jwt_token: Final = data.get("key")
725 if jwt_token:
726 click.echo(f"Successfully generated JWT for team: {team_id}")
727 return jwt_token
729 return None
732def _render_and_prompt_for_team_selection(teams: list[CliTeam]) -> str | None:
733 """Render teams table and prompt user for a team selection.
735 Returns the selected team_id as a string, or None if selection was
736 cancelled or skipped without any teams available.
737 """
738 # Display teams as a simple list, but prefer showing aliases where
739 # available while still keeping the underlying IDs intact.
740 console: Final = Console()
741 table: Final = Table(title="Available Teams")
742 table.add_column("Index", style="cyan", no_wrap=True)
743 table.add_column("Team Name", style="magenta")
744 table.add_column("Team ID", style="green")
746 for i, team in enumerate(teams):
747 team_id = str(team.get("team_id"))
748 team_alias = team.get("team_alias") or team_id
749 table.add_row(str(i + 1), team_alias, team_id)
751 console.print(table)
753 # Simple selection
754 while True:
755 try:
756 prompt_response: str = click.prompt(
757 "\nSelect a team by entering the index number (or 'skip' to use first team)",
758 type=str,
759 )
760 choice = prompt_response.strip()
762 if choice.lower() == "skip":
763 # Default to the first team's ID if the user skips an
764 # explicit selection.
765 if teams:
766 first_team = teams[0]
767 return str(first_team.get("team_id"))
768 return None
770 index = int(choice) - 1
771 if 0 <= index < len(teams):
772 selected_team = teams[index]
773 team_id = str(selected_team.get("team_id"))
774 team_alias = selected_team.get("team_alias") or team_id
775 click.echo(f"\nSelected team: {team_alias} ({team_id})")
776 return team_id
778 click.echo(f"Invalid selection. Please enter a number between 1 and {len(teams)}")
779 except ValueError:
780 click.echo("Invalid input. Please enter a number or 'skip'")
781 except KeyboardInterrupt:
782 click.echo("\nTeam selection cancelled.")
783 return None
786def _configure_claude_code(base_url: str, api_key: str) -> None:
787 """Write the key this login just minted into Claude Code's settings.json as a static token, undoable with
788 `lite unconfigure claude`. The key expires with the login, so the flag is the re-wire step of each login
789 rather than a one-time setup: no apiKeyHelper is written, since Claude Code would spawn `lite` (and its
790 keychain probe) on every credential refresh to keep one fresh."""
791 settings_path: Final = claude_settings_path(os.environ)
792 try:
793 configure_claude_settings(
794 base_url,
795 StaticToken(api_key),
796 KeepModel(),
797 settings_path,
798 configure_state_path(settings_path),
799 settings_file_owners(settings_path),
800 )
801 except ClaudeSettingsError as e:
802 raise click.ClickException(f"Logged in, but could not configure Claude Code: {e}")
803 click.echo(f"\nConfigured Claude Code: {settings_path} now routes through {base_url.rstrip('/')}.")
804 click.echo(
805 "This login's key is stored in the file, so run `lite login --config-claude` again after it expires. "
806 "Your other Claude Code settings were left untouched. Restart Claude Code to pick this up. "
807 f"Undo with `lite unconfigure claude`; `lite configure claude --model` sets {STARTING_MODEL_ROLE}."
808 )
811def _finish_login(base_url: str, api_key: str, config_claude: bool, stored: SecretSave) -> None:
812 """Claude Code is configured from the key in hand, so it does not wait on the CLI's own store: a login whose
813 token file or keychain refused it still has a usable key, and `--config-claude` asked for exactly that
814 key to be written into settings.json."""
815 from litellm.proxy.client.cli.interface import show_commands
817 click.echo("\nLogin successful!")
818 click.echo(f"JWT Token: {api_key[:20]}...")
819 click.echo(storage_notice(stored))
820 if config_claude:
821 _configure_claude_code(base_url, api_key)
822 if isinstance(stored, (CredentialNotSaved, CredentialNotRecorded)):
823 if config_claude:
824 click.echo("Claude Code was configured with this key even though the CLI itself could not keep it.")
825 return
826 click.echo("You can now use the CLI without specifying --api-key")
827 click.echo("\n" + "=" * 60)
828 show_commands()
831def _replace_stored_token(record: CliTokenData, http: Http, vault: SecretVault) -> SecretSave:
832 previous: Final = load_token(vault=vault)
833 stored: Final = save_token(record, vault=vault)
834 if previous is None or isinstance(stored, CredentialNotSaved):
835 return stored
836 revocation: Final = revoke_stored_credential(previous, http)
837 if revocation is not None:
838 click.echo(
839 f"Could not revoke the previous login's refresh token on the proxy ({revocation.reason}); "
840 "it expires on its own."
841 )
842 return stored
845def _pkce_login(base_url: str, config_claude: bool, vault: SecretVault) -> None:
846 http: Final = requests.Session()
847 credential: Final = run_pkce_login(base_url, http, echo=click.echo)
848 if isinstance(credential, PkceFailure):
849 click.echo(f"Authentication failed: {credential.reason}")
850 return
851 stored: Final = _replace_stored_token(pkce_token_record(base_url, credential), http, vault)
852 _finish_login(base_url, credential.access_token, config_claude, stored)
855@click.command(name="login")
856@click.option(
857 "--config-claude",
858 is_flag=True,
859 default=False,
860 help=(
861 "After logging in, write this login's key into ~/.claude/settings.json so Claude Code routes through "
862 "this proxy; run it again after the key expires. Unrelated settings are preserved."
863 ),
864)
865@click.option(
866 "--pkce",
867 is_flag=True,
868 default=False,
869 help=(
870 "Sign in with OAuth authorization code + PKCE through your system browser (loopback redirect), "
871 "with a refresh token that renews the key automatically. Requires a proxy that serves "
872 "/.well-known/litellm-cli-auth."
873 ),
874)
875@click.pass_context
876def login(ctx: click.Context, config_claude: bool, pkce: bool) -> None:
877 """Login to LiteLLM proxy using SSO authentication"""
878 from litellm.constants import LITELLM_CLI_SOURCE_IDENTIFIER
880 ctx_obj: Final[CliContextObj] = ctx.obj
881 base_url: Final = ctx_obj["base_url"]
882 if config_claude:
883 settings_path: Final = claude_settings_path(os.environ)
884 try:
885 refuse_while_owned(settings_path, settings_file_owners(settings_path))
886 except ClaudeSettingsError as e:
887 raise click.ClickException(f"Cannot configure Claude Code, so not logging in: {e}")
889 try:
890 if pkce:
891 _pkce_login(base_url, config_claude, context_secret_vault(ctx))
892 return
893 cli_sso_flow: Final = _start_cli_sso_flow(base_url=base_url)
894 key_id: Final = cli_sso_flow["login_id"]
895 poll_secret: Final = cli_sso_flow["poll_secret"]
896 user_code: Final = cli_sso_flow["user_code"]
898 browser_prefills_code: Final = isinstance(cli_sso_flow.get("verification_uri_complete"), str)
899 sso_url: Final = f"{base_url}/sso/key/generate?" + urlencode(
900 (
901 ("source", LITELLM_CLI_SOURCE_IDENTIFIER),
902 ("key", key_id),
903 *((("user_code", user_code),) if browser_prefills_code else ()),
904 )
905 )
907 click.echo(f"Opening browser to: {sso_url}")
908 click.echo("Please complete the SSO authentication in your browser...")
909 click.echo(
910 f"Verification code: {user_code} (pre-filled in the browser, check it matches)"
911 if browser_prefills_code
912 else f"Verification code: {user_code}"
913 )
914 click.echo(f"Session ID: {key_id}")
916 # Open browser
917 webbrowser.open(sso_url)
919 # Poll for authentication completion
920 click.echo("Waiting for authentication...")
922 auth_result: Final = _poll_for_authentication(base_url=base_url, key_id=key_id, poll_secret=poll_secret)
924 if auth_result:
925 api_key: Final = auth_result["api_key"]
926 user_id: Final = auth_result["user_id"]
928 # Save token data. base_url is stored so we can verify origin
929 # before reusing the key on a subsequent CLI invocation.
930 stored: Final = _replace_stored_token(
931 {
932 "base_url": base_url.rstrip("/"),
933 "key": api_key,
934 "user_id": user_id or "cli-user",
935 "user_email": "unknown",
936 "user_role": "cli",
937 "auth_header_name": "Authorization",
938 "jwt_token": "",
939 "timestamp": time.time(),
940 },
941 requests.Session(),
942 context_secret_vault(ctx),
943 )
945 _finish_login(base_url, api_key, config_claude, stored)
946 return
947 else:
948 click.echo("Authentication timed out. Please try again.")
949 click.echo(
950 "The proxy never reported the browser sign-in as finished. If you did complete it, "
951 "check the proxy logs for /sso/callback errors and confirm SSO is configured on the proxy."
952 )
953 return
955 except KeyboardInterrupt:
956 click.echo("\nAuthentication cancelled by user.")
957 return
958 except click.ClickException:
959 # Login itself already succeeded; only the post-login step failed, so this
960 # must not be relabelled as an authentication failure by the handler below.
961 raise
962 except Exception as e:
963 click.echo(f"Authentication failed: {e}")
964 return
967@click.command(name="logout")
968@click.pass_context
969def logout(ctx: click.Context):
970 """Logout and clear stored authentication"""
971 vault: Final = context_secret_vault(ctx)
972 token_data: Final = load_token(vault=vault)
973 revocation: Final = revoke_stored_credential(token_data, requests.Session()) if token_data is not None else None
974 match revocation:
975 case RevocationUnavailable(reason=reason):
976 raise click.ClickException(
977 f"The proxy could not record the revocation ({reason}). Nothing was cleared; "
978 "run `lite logout` again shortly."
979 )
980 case PkceFailure(reason=reason):
981 click.echo(f"Could not revoke the refresh token on the proxy ({reason}); it expires on its own.")
982 case None:
983 pass
984 case _:
985 assert_never(revocation)
987 path: Final = get_cli_token_file_path()
988 match clear_cli_token(vault=vault):
989 case SecretErased():
990 click.echo("Logged out successfully. Authentication token cleared.")
991 case CredentialNotCleared(detail=detail):
992 click.echo(f"Your credential is still in {path}, which could not be removed: {detail}.")
993 click.echo("Delete that file, or make the directory writable and run 'lite logout' again.")
994 case SecretStranded():
995 click.echo(STRANDED_CREDENTIAL_MESSAGE)
996 click.echo("Unlock your keychain and run 'lite logout' again to clear it.")
997 case KeyringNotInstalled():
998 click.echo(UNCHECKED_KEYCHAIN_MESSAGE)
999 click.echo(f"Install the keyring package with: {KEYRING_INSTALL_HINT}, then run 'lite logout' again.")
1000 case KeyringDisabled():
1001 click.echo(UNCHECKED_KEYCHAIN_MESSAGE)
1002 click.echo(f"Unset {DISABLE_KEYRING_ENV_VAR} and run 'lite logout' again to clear it.")
1003 case KeyringUnreachable():
1004 click.echo(UNCHECKED_KEYCHAIN_MESSAGE)
1005 click.echo("Unlock your keychain and run 'lite logout' again to clear it.")
1008@click.command(name="print-token")
1009@click.pass_context
1010def print_token(ctx: click.Context):
1011 """Print a valid API token for this proxy.
1013 Designed to be used as Claude Code's `apiKeyHelper`
1014 (https://docs.claude.com/en/docs/claude-code/settings): stdout must
1015 contain only the token, so all diagnostics go to stderr. The token
1016 expires after `LITELLM_CLI_JWT_EXPIRATION_HOURS` (default 24h); a
1017 `lite login --pkce` token renews itself here first, and once a token
1018 has expired for good, run the same `lite login` command again.
1019 """
1020 vault: Final = context_secret_vault(ctx)
1021 token_data: Final = load_token(vault=vault)
1022 if not token_data:
1023 click.echo("Not authenticated. Run 'lite login'.", err=True)
1024 sys.exit(1)
1026 # apiKeyHelper is invoked bare (no --base-url), so unless the caller
1027 # explicitly pointed us at a server, trust whichever one `lite login`
1028 # actually issued this token for -- that's the whole point of not
1029 # needing a wrapper command.
1030 ctx_obj: Final[CliContextObj] = ctx.obj
1031 issued_for_this_server: Final = token_data.get("base_url") == ctx_obj.get("base_url", "").rstrip("/")
1032 if ctx_obj.get("base_url_explicit") and not issued_for_this_server:
1033 click.echo("Not authenticated for this server. Run 'lite login'.", err=True)
1034 sys.exit(1)
1036 renews: Final = "refresh_token" in token_data
1037 if not is_cli_token_fresh(token_data) and not renews:
1038 click.echo("Token expired. Run 'lite login' again.", err=True)
1039 sys.exit(1)
1041 if token_data.get("key") is None:
1042 click.echo(keychain_unreadable_notice(vault), err=True)
1043 sys.exit(1)
1045 api_key: Final = (
1046 ctx_obj.get("api_key")
1047 if issued_for_this_server and ctx_obj.get("api_key_from_token_file")
1048 else fresh_api_key(
1049 token_data,
1050 _renewal_saver(vault),
1051 requests.Session(),
1052 reload=_renewal_reader(vault),
1053 warn=_warn,
1054 )
1055 )
1056 if not api_key:
1057 click.echo(f"Key expired. Run '{_login_command(renews)}' again.", err=True)
1058 sys.exit(1)
1060 click.echo(api_key)
1063@click.command(name="whoami")
1064@click.pass_context
1065def whoami(ctx: click.Context):
1066 """Show current authentication status"""
1067 vault: Final = context_secret_vault(ctx)
1068 token_data: Final = load_token(vault=vault)
1070 if not token_data:
1071 click.echo("Not authenticated. Run 'lite login' to authenticate.")
1072 return
1074 key_readable: Final = token_data.get("key") is not None
1075 click.echo("Authenticated" if key_readable else "Signed in, but the credential cannot be read")
1076 click.echo(f"User Email: {token_data.get('user_email') or 'Unknown'}")
1077 click.echo(f"User ID: {token_data.get('user_id') or 'Unknown'}")
1078 click.echo(f"User Role: {token_data.get('user_role') or 'Unknown'}")
1079 team_id: Final = token_data.get("team_id")
1080 if team_id:
1081 click.echo(f"Team ID: {team_id}")
1083 stamped: Final = token_data.get("timestamp")
1084 age_hours: Final = (time.time() - (stamped if isinstance(stamped, (int, float)) else 0.0)) / 3600
1085 click.echo(f"Token age: {age_hours:.1f} hours")
1087 if not key_readable:
1088 click.echo(keychain_unreadable_notice(vault))
1090 expires_at: Final = token_data.get("expires_at")
1091 if isinstance(expires_at, (int, float)):
1092 click.echo(_key_expiry_line(expires_at, renews="refresh_token" in token_data))
1093 elif age_hours > CLI_JWT_EXPIRATION_HOURS:
1094 click.echo(f"Warning: Token is more than {CLI_JWT_EXPIRATION_HOURS} hours old and may have expired.")
1097def _key_expiry_line(expires_at: float, renews: bool) -> str:
1098 remaining_hours: Final = (expires_at - time.time()) / 3600
1099 if remaining_hours <= 0:
1100 return f"Key expired. Run '{_login_command(renews)}' again"
1101 status: Final = f"Key expires in: {remaining_hours:.1f} hours"
1102 return f"{status}, renewed on next use" if renews else status
1105@click.group(name="auth")
1106def auth_group():
1107 """Manage CLI authentication (apiKeyHelper support, etc.)"""
1110auth_group.add_command(print_token)
1113# Export functions for use by other CLI commands
1114__all__ = ["auth_group", "login", "logout", "print_token", "prompt_team_selection", "whoami"]
1116# Export individual commands instead of grouping them
1117# login, logout, and whoami will be added as top-level commands