diff --git a/litellm/litellm_core_utils/cli_token_utils.py b/litellm/litellm_core_utils/cli_token_utils.py index 37ed4efa0ca..5f45f942e4e 100644 --- a/litellm/litellm_core_utils/cli_token_utils.py +++ b/litellm/litellm_core_utils/cli_token_utils.py @@ -13,8 +13,10 @@ This module has no dependencies on proxy code and can be safely imported at the """ import math +import os import time from collections.abc import Mapping +from contextlib import suppress from dataclasses import dataclass from pathlib import Path from types import MappingProxyType @@ -42,6 +44,7 @@ from litellm.litellm_core_utils.private_json import ( ensure_private_dir, overwrite_private_json, stage_private_json, + write_private_bytes, write_private_json, ) from litellm.types.llms.base_model import LiteLLMBaseModel @@ -79,9 +82,21 @@ class CredentialNotCleared: detail: str +@dataclass(frozen=True, slots=True) +class LoginRecordNotMoved: + """The token file of a login the keychain kept could not be moved aside as the note. + + The file holds no secret, but it is the record a later logout pairs with the keychain entry to + revoke that login, so deleting it would leave the login live. It stays until its directory + takes the rename. + """ + + detail: str + + SecretSave: TypeAlias = SecretWrite | CredentialNotSaved | CredentialNotRecorded -SecretClear: TypeAlias = SecretErase | CredentialNotCleared +SecretClear: TypeAlias = SecretErase | CredentialNotCleared | LoginRecordNotMoved class CliTokenRecord(LiteLLMBaseModel): @@ -159,6 +174,7 @@ def save_cli_token(record: CliTokenRecord, *, vault: SecretVault = SYSTEM_KEYRIN keychain has already taken the new secret, and it reports itself as such rather than claiming the previous login survived. """ + replaced: Final = _read_token_file() stamped: Final = _stamped_past_every_stored_login(record, vault) staged: Final = _stage_token_file(_without_secret(stamped)) if isinstance(staged, CredentialNotSaved): @@ -167,7 +183,22 @@ def save_cli_token(record: CliTokenRecord, *, vault: SecretVault = SYSTEM_KEYRIN if isinstance(outcome, SecretStored): return outcome if _commit_token_file(staged) else CredentialNotRecorded() discard_staged_json(staged) - return _keep_the_secret_in_the_file(stamped, outcome) + kept: Final = _keep_the_secret_in_the_file(stamped, outcome) + if not isinstance(kept, CredentialNotSaved): + _note_the_login_left_in_the_keychain(replaced) + return kept + + +def _note_the_login_left_in_the_keychain(replaced: CliTokenRecord | None) -> None: + """Keep the login a save could not take out of the keychain revocable once the keychain answers. + + Its token file record is about to be overwritten by one that holds its own secret, so the note + is the only record left of the login whose refresh token the keychain still holds. + """ + if replaced is None or _holds_a_secret(replaced): + return + with suppress(OSError): + _write_keychain_not_cleared_note(_keychain_not_cleared_note_path(), replaced) def _stamped_past_every_stored_login(record: CliTokenRecord, vault: SecretVault) -> CliTokenRecord: @@ -220,27 +251,62 @@ def _keep_the_secret_in_the_file(record: CliTokenRecord, outcome: SecretWrite) - def clear_cli_token(*, vault: SecretVault = SYSTEM_KEYRING) -> SecretClear: """Remove the credential from both stores. Reports whether the keychain is now free of it. - A logout the keychain never answered keeps the token file, with its secret taken out, because - that file is the only remaining record that something may still be in there to remove. It is - what lets a later run tell a machine with a credential it cannot reach apart from one that never - had a login at all, and taking it away would leave the next logout answering the warning this - one just issued with a false all-clear. The secret goes either way, and a file that will give up - neither its copy nor itself is removed rather than kept, with the note written again afterwards - so the warning still outlives this run. + The token file goes whatever the keychain answers, so nothing left on disk reads as a login. A + keychain that could not be checked, or that would not give up its entry, may still hold a + credential of ours, so a logout that had something to end leaves a note beside the file, and the + next logout warns again instead of reading the missing file as a clean keychain. Only a keychain + that confirms the erase takes the note away. A file that will not go has its secret taken out + where it lies instead. """ outcome: Final = vault.erase() record: Final = _read_token_file() settled: Final = _nothing_left_behind(outcome, record) - if not settled and _keep_the_unchecked_keychain_on_record(outcome, record): - return outcome - removal: Final = _remove_token_file() - if removal is not None and record is not None and not _scrub_file_secret(record): - return removal - if removal is None and record is not None and _the_keychain_went_unchecked(outcome): - _write_the_note_the_removal_took_with_it(record) + left_on_disk: Final = _end_the_local_login(outcome, record) + if left_on_disk is not None: + return left_on_disk return SecretErased() if settled else outcome +def _end_the_local_login( + outcome: SecretErase, record: CliTokenRecord | None +) -> CredentialNotCleared | LoginRecordNotMoved | None: + if _names_a_login_the_keychain_kept(outcome, record): + return _move_token_file_to_note() + left_on_disk: Final = _remove_token_file_or_its_secret(record) + _note_what_the_keychain_answered(outcome, record) + return left_on_disk + + +def _names_a_login_the_keychain_kept(outcome: SecretErase, record: CliTokenRecord | None) -> bool: + match outcome: + case KeyringDisabled() | KeyringNotInstalled() | KeyringUnreachable(): + return record is not None and not _holds_a_secret(record) + case SecretErased() | SecretStranded(): + return False + + +def _move_token_file_to_note() -> LoginRecordNotMoved | None: + """Turn the token file of a login the keychain could not be asked to end into the note that ends it later. + + The file holds no secret, so it is already the record a later logout or login pairs with the + keychain entry to revoke the login, and renaming it needs no free space on a full disk. The file + never stays behind as a login that signs the user back in once the keychain unlocks: a note path + that will not take it leaves the file where it is, and the logout says so. + """ + try: + os.replace(get_cli_token_file_path(), _keychain_not_cleared_note_path()) + except OSError as error: + return LoginRecordNotMoved(str(error)) + return None + + +def _remove_token_file_or_its_secret(record: CliTokenRecord | None) -> CredentialNotCleared | None: + removal: Final = _remove_token_file() + if removal is None or record is None or _scrub_file_secret(record): + return None + return removal + + def _remove_token_file() -> CredentialNotCleared | None: try: Path(get_cli_token_file_path()).unlink(missing_ok=True) @@ -249,54 +315,87 @@ def _remove_token_file() -> CredentialNotCleared | None: return None -def _write_the_note_the_removal_took_with_it(record: CliTokenRecord) -> None: - """Put the secret-free note back after the file carrying it had to go to get the secret off disk. +def _keychain_not_cleared_note_path() -> Path: + return Path(get_cli_token_file_path()).with_name("keychain_not_cleared") - Reaching here means neither rewrite would take, so the file went instead, and its absence is - what the next logout would read as a keychain already known to be clean. Removing it is also - what frees the room the rewrite was refused for, so the note usually lands on this second try. - When it does not, the warning this logout printed is the only one the user gets. + +def load_cli_token_left_in_keychain(*, vault: SecretVault = SYSTEM_KEYRING) -> CliTokenRecord | None: + """The login an earlier logout could not erase from the keychain, once the keychain answers again. + + The token file went with that logout, so its note is the only record of the login, and the + refresh token that revokes it is only in the keychain. A keychain entry from any other sign-in + is never paired with the note, since the endpoints it names belong to that one login. """ - staged: Final = _stage_scrubbed_file(record) - if staged is not None: - _commit_token_file(staged) + note: Final = _read_keychain_not_cleared_note() + if note is None: + return None + secret: Final = _stored_secret(note.base_url, vault) + if secret is None or secret.timestamp != note.timestamp: + return None + return note.model_copy( + update=MappingProxyType( + {"key": secret.key, "jwt_token": secret.jwt_token, "refresh_token": secret.refresh_token} + ) + ) -def _keep_the_unchecked_keychain_on_record(outcome: SecretErase, record: CliTokenRecord | None) -> bool: - """Whether the token file, stripped of its secret, is worth keeping as the note that says so. - - Only a keychain that could not be reached leaves the question open. One that answered for itself - is remembered without any help from the file, and a file it can still pair a live entry with - would leave the machine signed in to the login that was just ended. A copy that will give up - its secret neither to a staged replacement nor to an overwrite is not kept either, because the - secret goes first. - """ - if record is None or not _the_keychain_went_unchecked(outcome): +def _holds_the_noted_login(blob: str) -> bool: + """Whether the keychain entry is the login an earlier logout could not end, which only it can revoke""" + note: Final = _read_keychain_not_cleared_note() + if note is None: return False - return _scrub_file_secret(record) + secret: Final = _decode_secret(blob, note.base_url) + return secret is not None and secret.timestamp == note.timestamp -def _the_keychain_went_unchecked(outcome: SecretErase) -> bool: - """Whether the keychain neither confirmed the erase nor answered that it still holds the secret""" - match outcome: - case SecretErased() | SecretStranded(): - return False - case KeyringDisabled() | KeyringNotInstalled() | KeyringUnreachable(): - return True +def _read_keychain_not_cleared_note() -> CliTokenRecord | None: + try: + return CliTokenRecord.model_validate_json(_keychain_not_cleared_note_path().read_bytes()) + except (OSError, ValidationError): + return None + + +def _note_what_the_keychain_answered(outcome: SecretErase, record: CliTokenRecord | None) -> None: + """Keep the note that a logout left the keychain uncleared for exactly as long as that is true. + + A keychain that would not give up its entry had already handed this logout the refresh token, + so its note holds nothing to revoke again. A login that kept its secret in the file never + reached the keychain, so the note an earlier logout left still names the login the keychain + holds and stays, and a logout with no file to end keeps it too. + + The secret is off disk by the time this runs, so a note that will not be written or removed is + no reason to fail the logout. The warning printed now is then the only one the user gets. + """ + note: Final = _keychain_not_cleared_note_path() + with suppress(OSError): + match outcome: + case SecretErased(): + note.unlink(missing_ok=True) + case SecretStranded(): + _write_keychain_not_cleared_note(note, None) + case KeyringDisabled() | KeyringNotInstalled() | KeyringUnreachable(): + if record is not None and not note.exists(): + _write_keychain_not_cleared_note(note, None) + + +def _write_keychain_not_cleared_note(note: Path, record: CliTokenRecord | None) -> None: + """Record the login a later command can still revoke, or only that the keychain was left uncleared""" + ensure_private_dir(note.parent) + if record is None: + write_private_bytes(str(note), b"") + return + write_private_json(str(note), _without_secret(record).model_dump(exclude_none=True)) def _nothing_left_behind(outcome: SecretErase, record: CliTokenRecord | None) -> bool: """Whether the keychain can be trusted to hold no credential of ours once the file is gone. - A machine with no token file has no stored login to end, and `clear_cli_token` keeps one behind - whenever the keychain is left unconfirmed, taking the secret out in place when it cannot stage a - replacement and writing the note again when the file holding it had to go, so a missing file is - real evidence rather than the absence of it. Past that, a - keychain that could not be reached is never trusted, whatever the file looks like. Even a file - holding its own secret says only that the login which wrote it had no keychain to write to, and - the login before it may well have had one: the entry that login left outlives both the - uninstalled package and the file that replaced it. `SecretStranded` is - the keychain answering for itself and outranks the file. + A keychain that could not be reached is trusted only on a machine with nothing to end: no token + file, and no note from an earlier logout that left the keychain uncleared. Even a file holding + its own secret says only that the login which wrote it had no keychain to write to, and the + login before it may well have had one: the entry that login left outlives both the uninstalled + package and the file that replaced it. `SecretStranded` is the keychain answering for itself and + outranks the file. """ match outcome: case SecretErased(): @@ -304,7 +403,7 @@ def _nothing_left_behind(outcome: SecretErase, record: CliTokenRecord | None) -> case SecretStranded(): return False case KeyringDisabled() | KeyringNotInstalled() | KeyringUnreachable(): - return record is None + return record is None and not os.path.exists(_keychain_not_cleared_note_path()) def get_litellm_gateway_api_key( @@ -407,7 +506,14 @@ def _apply_vault_secret(record: CliTokenRecord, blob: str, vault: SecretVault) - than counting as a clean sweep. Rolling the vault back the way a migration does is not the answer here, because the two stores hold different credentials and the rollback would hand the superseded one back out. + + A keychain entry an earlier logout could not end is left alone whatever the stamps say. Its note + pairs only with that entry, so the next login or logout can still revoke it, while serving it + would sign an ended login back in and moving the file's secret over it would leave its refresh + token live with nothing on this machine able to revoke it. """ + if _holds_a_secret(record) and _holds_the_noted_login(blob): + return record secret: Final = _decode_secret(blob, record.base_url) if secret is None or (_holds_a_secret(record) and secret.timestamp <= record.timestamp): return _migrate_file_secret(_rejoined(record, secret), vault, replacing=secret) diff --git a/litellm/proxy/client/README.md b/litellm/proxy/client/README.md index ceec636567f..9a7e067c792 100644 --- a/litellm/proxy/client/README.md +++ b/litellm/proxy/client/README.md @@ -380,7 +380,7 @@ The key itself, together with the refresh token that renews a `--pkce` credentia Keychain storage needs the `keyring` package, which ships with `pip install 'litellm[cli]'`. Headless boxes and CI runners usually have no keychain either. In all of those cases the key and the refresh token stay in the same `0600` file alongside the metadata, exactly as they did before, and `lite login` names which one applies: the package is missing, the machine has no keychain, or you set `LITELLM_CLI_DISABLE_KEYRING=1` to force the file even where a keychain exists. A `token.json` written by an older `lite` keeps working and is moved into the keychain, and scrubbed from the file, the first time a keychain-capable `lite` reads it. That includes a refresh token left behind by the release that moved only the key. -`lite logout` clears both stores. If the keychain is locked at that moment it says so, and re-running it once the keychain is unlocked finishes the job. +`lite logout` clears both stores. If the keychain is locked at that moment it says so, removes `token.json`, and leaves `~/.litellm/keychain_not_cleared` holding that login's metadata without its secret. Re-running it once the keychain is unlocked revokes a `--pkce` refresh token on the proxy and finishes the job. A `lite login` made once the keychain answers revokes that login too, since it replaces the entry. A login made while the keychain is still locked keeps its secret in `token.json` and the note names the keychain's login, so the first `lite logout` or `lite login` after the unlock revokes both. Other commands leave the noted entry where it is, and a silent renewal that takes over the keychain entry revokes that login too. On a machine with no keychain at all, every later `lite logout` repeats the warning until you delete `~/.litellm/keychain_not_cleared`. If `~/.litellm` will not take the note, `lite logout` says it did not log out and leaves `token.json` where it is, so keep that file, make the directory writable, and run it again. An older `lite` does not read the note, so finish that logout with this release or a later one: an older `lite logout` erases the keychain entry without revoking its `--pkce` refresh token, which then stays valid until it expires. A machine that an older `lite` logged out while its keychain was locked still has its `token.json`, and still reads as signed in, until you run `lite logout` once more. The stored credential is a short-lived, per-session agent token, not a managed virtual key. It is scoped to the user and team you logged in as and inherits their models and budgets; spend is tracked against the shared team and user budgets rather than a separate per-session cap, so multiple logins or several concurrent agents all draw down the same allowance. It is short-lived by design (default 24h, configurable via `LITELLM_CLI_JWT_EXPIRATION_HOURS`); re-run `lite login` to refresh it and pick up your latest team and user settings. `lite auth print-token` (usable as Claude Code's `apiKeyHelper`) prints it, and once it is past 24 hours (or the client's `LITELLM_CLI_JWT_EXPIRATION_HOURS`) asks the proxy first and fails if the proxy rejects it. There is no silent renewal. It is accepted on a default deployment without `EXPERIMENTAL_UI_LOGIN`, does not appear in the Keys UI, and cannot be rotated or revoked mid-session. A credential from `lite login --pkce` is the exception: it carries a refresh token, so the CLI renews the key shortly before it expires and `lite logout` revokes the refresh token on the proxy (see [Browser sign-in with PKCE](https://docs.litellm.ai/docs/proxy/cli_sso#browser-sign-in-with-pkce)). Only the holder can end a `--pkce` session early, with `lite logout`; an admin has no button for it, but every renewal re-reads the user on the proxy, so deactivating the user or removing them from the team makes the next renewal fail and the key runs out within `LITELLM_CLI_JWT_EXPIRATION_HOURS`. On a proxy with more than one worker or replica, configure Redis (`litellm_settings.cache` with Redis `cache_params`, or `general_settings.coordination_redis`) so a refresh token stays single-use and `lite logout` holds on every worker; without Redis each worker keeps its own record. For a long-lived, rotatable, Keys-UI-visible credential, create a dedicated virtual key in the dashboard and pass it via `--api-key` or `LITELLM_PROXY_API_KEY`. diff --git a/litellm/proxy/client/cli/commands/auth.py b/litellm/proxy/client/cli/commands/auth.py index e38f54adc4f..7bd209140ac 100644 --- a/litellm/proxy/client/cli/commands/auth.py +++ b/litellm/proxy/client/cli/commands/auth.py @@ -32,11 +32,13 @@ from litellm.litellm_core_utils.cli_token_utils import ( CredentialNotCleared, CredentialNotRecorded, CredentialNotSaved, + LoginRecordNotMoved, SecretSave, clear_cli_token, get_cli_token_file_path, is_cli_token_fresh, load_cli_token, + load_cli_token_left_in_keychain, save_cli_token, ) @@ -234,22 +236,45 @@ def load_token(*, vault: SecretVault = SYSTEM_KEYRING) -> Mapping[str, object] | return None if record is None else record.model_dump(exclude_none=True) +def _login_left_in_keychain(vault: SecretVault) -> Mapping[str, object] | None: + """The login an earlier logout left in a keychain it could not reach""" + left: Final = load_cli_token_left_in_keychain(vault=vault) + return None if left is None else left.model_dump(exclude_none=True) + + +def _same_refresh_token(left: Mapping[str, object] | None, current: Mapping[str, object] | None) -> bool: + return left is not None and current is not None and left.get("refresh_token") == current.get("refresh_token") + + +def _credentials_to_revoke(vault: SecretVault) -> tuple[Mapping[str, object], ...]: + left: Final = _login_left_in_keychain(vault) + current: Final = load_token(vault=vault) + if current is None or _same_refresh_token(left, current): + return () if left is None else (left,) + return (current,) if left is None else (left, current) + + def save_token(record: CliTokenData, *, vault: SecretVault = SYSTEM_KEYRING) -> SecretSave: """Store a credential the PKCE layer produced, secret in the vault and the rest on disk""" return save_cli_token(CliTokenRecord(**record), vault=vault) -def _renewal_saver(vault: SecretVault) -> Callable[[CliTokenData], None]: +def _renewal_saver(vault: SecretVault, http: Http) -> Callable[[CliTokenData], None]: """Persist a silently renewed credential, and say on stderr when no store would keep it. A renewal rotates the refresh token, so a rotation that is never stored logs this - machine out on the next command; the user hears about it rather than guessing. + machine out on the next command; the user hears about it rather than guessing. The save + can take the keychain entry of a login an earlier logout could not end, which leaves no + other record of that login's refresh token, so it is revoked the way a login revokes it. """ def save(record: CliTokenData) -> None: + left: Final = _login_left_in_keychain(vault) outcome: Final = save_token(record, vault=vault) if isinstance(outcome, (CredentialNotSaved, CredentialNotRecorded)): _warn(storage_notice(outcome)) + if left is not None: + _revoke_the_replaced_login(left, http, _warn) return save @@ -280,10 +305,11 @@ def get_stored_api_key( return None if expected_base_url is not None and token_data.get("base_url") != expected_base_url.rstrip("/"): return None + http: Final = requests.Session() return fresh_api_key( token_data, - _renewal_saver(vault), - requests.Session(), + _renewal_saver(vault, http), + http, reload=_renewal_reader(vault), warn=_warn, ) @@ -853,17 +879,22 @@ def _finish_login(base_url: str, api_key: str, config_claude: bool, stored: Secr def _replace_stored_token(record: CliTokenData, http: Http, vault: SecretVault) -> SecretSave: - previous: Final = load_token(vault=vault) + previous: Final = _credentials_to_revoke(vault) stored: Final = save_token(record, vault=vault) - if previous is None or isinstance(stored, CredentialNotSaved): + if isinstance(stored, CredentialNotSaved): return stored - revocation: Final = revoke_stored_credential(previous, http) + for login in previous: + _revoke_the_replaced_login(login, http, click.echo) + return stored + + +def _revoke_the_replaced_login(login: Mapping[str, object], http: Http, report: Callable[[str], None]) -> None: + revocation: Final = revoke_stored_credential(login, http) if revocation is not None: - click.echo( + report( f"Could not revoke the previous login's refresh token on the proxy ({revocation.reason}); " "it expires on its own." ) - return stored def _pkce_login(base_url: str, config_claude: bool, vault: SecretVault) -> None: @@ -987,13 +1018,22 @@ def login(ctx: click.Context, config_claude: bool, pkce: bool | None) -> None: return -@click.command(name="logout") -@click.pass_context -def logout(ctx: click.Context): - """Logout and clear stored authentication""" - vault: Final = context_secret_vault(ctx) - token_data: Final = load_token(vault=vault) - revocation: Final = revoke_stored_credential(token_data, requests.Session()) if token_data is not None else None +def _revoke_before_logout(left: Mapping[str, object] | None, current: Mapping[str, object] | None, http: Http) -> None: + """Revoke every login this logout ends, holding the logout back only for the one it signs out. + + An earlier login left in the keychain is the previous login once another one is current, so a + proxy that cannot revoke it only earns the warning `lite login` gives, never a logout that the + current proxy is ready to finish. + """ + if left is None or current is None or _same_refresh_token(left, current): + _revoke_the_login_or_keep_it(current if left is None else left, http) + return + _revoke_the_replaced_login(left, http, click.echo) + _revoke_the_login_or_keep_it(current, http) + + +def _revoke_the_login_or_keep_it(token_data: Mapping[str, object] | None, http: Http) -> None: + revocation: Final = None if token_data is None else revoke_stored_credential(token_data, http) match revocation: case RevocationUnavailable(reason=reason): raise click.ClickException( @@ -1007,6 +1047,17 @@ def logout(ctx: click.Context): case _: assert_never(revocation) + +@click.command(name="logout") +@click.pass_context +def logout(ctx: click.Context): + """Logout and clear stored authentication""" + vault: Final = context_secret_vault(ctx) + http: Final = requests.Session() + left: Final = _login_left_in_keychain(vault) + current: Final = load_token(vault=vault) + _revoke_before_logout(left, current, http) + path: Final = get_cli_token_file_path() match clear_cli_token(vault=vault): case SecretErased(): @@ -1014,6 +1065,12 @@ def logout(ctx: click.Context): case CredentialNotCleared(detail=detail): click.echo(f"Your credential is still in {path}, which could not be removed: {detail}.") click.echo("Delete that file, or make the directory writable and run 'lite logout' again.") + case LoginRecordNotMoved(detail=detail): + click.echo(f"Not logged out. {path} could not be moved aside: {detail}.") + click.echo( + "Keep that file, since 'lite logout' needs it to revoke a login your OS keychain may still hold. " + "Make its directory writable and run 'lite logout' again." + ) case SecretStranded(): click.echo(STRANDED_CREDENTIAL_MESSAGE) click.echo("Unlock your keychain and run 'lite logout' again to clear it.") @@ -1067,13 +1124,14 @@ def print_token(ctx: click.Context): click.echo(keychain_unreadable_notice(vault), err=True) sys.exit(1) + http: Final = requests.Session() api_key: Final = ( ctx_obj.get("api_key") if issued_for_this_server and ctx_obj.get("api_key_from_token_file") else fresh_api_key( token_data, - _renewal_saver(vault), - requests.Session(), + _renewal_saver(vault, http), + http, reload=_renewal_reader(vault), warn=_warn, ) diff --git a/litellm/proxy/client/cli/main.py b/litellm/proxy/client/cli/main.py index a1964d8a1bb..e0bcaaf5a66 100644 --- a/litellm/proxy/client/cli/main.py +++ b/litellm/proxy/client/cli/main.py @@ -199,7 +199,8 @@ def cli(ctx: click.Context, show_version: bool, base_url: str | None, api_key: s # 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. setup_command: Final = ctx.invoked_subcommand in ("configure", "reconfigure", "unconfigure") - api_key_from_token_file: Final = api_key is None and not setup_command + reads_the_stored_login_itself: Final = ctx.invoked_subcommand in ("login", "logout") + api_key_from_token_file: Final = api_key is None and not setup_command and not reads_the_stored_login_itself resolved_api_key: Final = ( get_stored_api_key(expected_base_url=base_url, vault=context_secret_vault(ctx)) if api_key_from_token_file diff --git a/tests/unit/litellm_core_utils/test_cli_token_utils.py b/tests/unit/litellm_core_utils/test_cli_token_utils.py index 7e7eee5373f..68d3098cc91 100644 --- a/tests/unit/litellm_core_utils/test_cli_token_utils.py +++ b/tests/unit/litellm_core_utils/test_cli_token_utils.py @@ -31,11 +31,13 @@ from litellm.litellm_core_utils.cli_token_utils import ( CredentialNotCleared, CredentialNotRecorded, CredentialNotSaved, + LoginRecordNotMoved, clear_cli_token, get_cli_token_file_path, get_litellm_gateway_api_key, is_cli_token_fresh, load_cli_token, + load_cli_token_left_in_keychain, save_cli_token, ) @@ -54,6 +56,10 @@ def _token_file(home): return home / ".litellm" / "token.json" +def _everything_on_disk(home): + return "".join(path.read_text() for path in (home / ".litellm").rglob("*") if path.is_file()) + + def _write_legacy_file(home, **overrides): payload = { "base_url": SERVER, @@ -123,6 +129,10 @@ class _MkstempThatNeedsTheOldFileGone: return _REAL_MKSTEMP(*args, **kwargs) +def _full_disk_mkstemp(*args, **kwargs): + raise OSError(errno.ENOSPC, "No space left on device") + + _REAL_REPLACE = os.replace @@ -777,6 +787,17 @@ class TestClearCliToken: assert clear_cli_token(vault=vault) == SecretStranded() assert load_cli_token(vault=vault) is None + def test_a_keychain_that_kept_the_secret_is_still_warned_about_once_it_is_out_of_reach( + self, isolated_home, secret_vault_factory + ): + _write_metadata_only_file(isolated_home) + vault = secret_vault_factory(blob=_blob(), erasable=False) + assert clear_cli_token(vault=vault) == SecretStranded() + + vault.available = False + + assert clear_cli_token(vault=vault) == KeyringUnreachable() + @pytest.mark.parametrize( "failure", [KeyringDisabled(), KeyringUnreachable(), KeyringNotInstalled()] ) @@ -794,7 +815,8 @@ class TestClearCliToken: vault = secret_vault_factory(available=False, failure=failure) assert clear_cli_token(vault=vault) == failure - assert json.loads(_token_file(isolated_home).read_text()).get("key") is None + assert not _token_file(isolated_home).exists() + assert "sk-legacy" not in _everything_on_disk(isolated_home) def test_a_second_logout_still_reports_the_keychain_it_could_not_clear( self, isolated_home, secret_vault_factory @@ -817,20 +839,20 @@ class TestClearCliToken: vault = secret_vault_factory(available=False, failure=KeyringNotInstalled()) assert clear_cli_token(vault=vault) == KeyringNotInstalled() - assert json.loads(_token_file(isolated_home).read_text()).get("key") is None + assert load_cli_token(vault=vault) is None - def test_a_logout_that_cannot_clear_the_keychain_keeps_the_record_that_it_has_to( - self, isolated_home, secret_vault_factory + @pytest.mark.parametrize("failure", [KeyringDisabled(), KeyringUnreachable(), KeyringNotInstalled()]) + def test_a_logout_that_cannot_check_the_keychain_still_ends_the_local_login( + self, isolated_home, secret_vault_factory, failure ): - """The file left behind holds no secret. It is what a later run reads to tell a machine with - a credential it cannot reach apart from one that never had a login, which is the difference - between warning the user and inventing a credential for them to worry about.""" - _write_metadata_only_file(isolated_home) - vault = secret_vault_factory(available=False, failure=KeyringUnreachable()) + vault = secret_vault_factory(available=False, failure=failure) + save_cli_token(CliTokenRecord(base_url=SERVER, key="sk-in-file", user_id="u-1"), vault=vault) - clear_cli_token(vault=vault) + assert clear_cli_token(vault=vault) == failure - assert json.loads(_token_file(isolated_home).read_text()).get("key") is None + assert load_cli_token(vault=vault) is None + assert get_litellm_gateway_api_key(vault=vault) is None + assert not _token_file(isolated_home).exists() def test_a_logout_that_cannot_clear_the_keychain_still_takes_the_file_secret_away( self, isolated_home, secret_vault_factory @@ -842,7 +864,7 @@ class TestClearCliToken: clear_cli_token(vault=vault) - left_on_disk = _token_file(isolated_home).read_text() + left_on_disk = _everything_on_disk(isolated_home) assert "sk-legacy" not in left_on_disk assert "rt-legacy" not in left_on_disk @@ -862,7 +884,91 @@ class TestClearCliToken: assert clear_cli_token(vault=vault) == KeyringUnreachable() assert clear_cli_token(vault=vault) == KeyringUnreachable() assert vault.blob is not None - assert "sk-second" not in _token_file(isolated_home).read_text() + assert load_cli_token(vault=vault) is None + assert "sk-second" not in _everything_on_disk(isolated_home) + + def test_a_keychain_that_answers_again_never_signs_an_ended_login_back_in( + self, isolated_home, secret_vault_factory + ): + vault = secret_vault_factory() + save_cli_token(CliTokenRecord(base_url=SERVER, key="sk-first"), vault=vault) + vault.available = False + save_cli_token(CliTokenRecord(base_url=SERVER, key="sk-second"), vault=vault) + assert clear_cli_token(vault=vault) == KeyringUnreachable() + + vault.available = True + + assert load_cli_token(vault=vault) is None + assert get_litellm_gateway_api_key(vault=vault) is None + assert clear_cli_token(vault=vault) == SecretErased() + assert vault.blob is None + + def test_a_keychain_confirmed_clean_is_not_warned_about_again(self, isolated_home, secret_vault_factory): + vault = secret_vault_factory() + save_cli_token(CliTokenRecord(base_url=SERVER, key="sk-first"), vault=vault) + vault.available = False + save_cli_token(CliTokenRecord(base_url=SERVER, key="sk-second"), vault=vault) + assert clear_cli_token(vault=vault) == KeyringUnreachable() + vault.available = True + assert clear_cli_token(vault=vault) == SecretErased() + + vault.available = False + + assert clear_cli_token(vault=vault) == SecretErased() + + def test_a_second_logout_that_misses_the_keychain_keeps_the_first_login_revocable( + self, isolated_home, secret_vault_factory + ): + vault = secret_vault_factory() + save_cli_token( + CliTokenRecord(base_url=SERVER, key="sk-first", refresh_token="rt-first", timestamp=1000.0), vault=vault + ) + vault.available = False + assert clear_cli_token(vault=vault) == KeyringUnreachable() + save_cli_token( + CliTokenRecord(base_url=SERVER, key="sk-second", refresh_token="rt-second", timestamp=2000.0), vault=vault + ) + assert clear_cli_token(vault=vault) == KeyringUnreachable() + + vault.available = True + left = load_cli_token_left_in_keychain(vault=vault) + + assert left is not None + assert (left.key, left.refresh_token) == ("sk-first", "rt-first") + + def test_a_locked_login_over_a_keychain_login_keeps_the_keychain_login_revocable( + self, isolated_home, secret_vault_factory + ): + vault = secret_vault_factory() + save_cli_token( + CliTokenRecord(base_url=SERVER, key="sk-first", refresh_token="rt-first", timestamp=1000.0), vault=vault + ) + vault.available = False + save_cli_token( + CliTokenRecord(base_url=SERVER, key="sk-second", refresh_token="rt-second", timestamp=2000.0), vault=vault + ) + + vault.available = True + left = load_cli_token_left_in_keychain(vault=vault) + + assert left is not None + assert (left.key, left.refresh_token) == ("sk-first", "rt-first") + + def test_the_note_never_pairs_with_a_keychain_entry_a_later_login_wrote( + self, isolated_home, secret_vault_factory + ): + vault = secret_vault_factory() + save_cli_token( + CliTokenRecord(base_url=SERVER, key="sk-first", refresh_token="rt-first", timestamp=1000.0), vault=vault + ) + vault.available = False + assert clear_cli_token(vault=vault) == KeyringUnreachable() + vault.available = True + save_cli_token( + CliTokenRecord(base_url=SERVER, key="sk-second", refresh_token="rt-second", timestamp=2000.0), vault=vault + ) + + assert load_cli_token_left_in_keychain(vault=vault) is None @pytest.mark.parametrize("failure", [KeyringNotInstalled(), KeyringDisabled(), KeyringUnreachable()]) def test_logging_out_of_a_machine_that_never_logged_in_invents_nothing_to_warn_about( @@ -874,6 +980,7 @@ class TestClearCliToken: vault = secret_vault_factory(available=False, failure=failure) assert clear_cli_token(vault=vault) == SecretErased() + assert clear_cli_token(vault=vault) == SecretErased() def test_a_file_backed_login_cannot_vouch_for_a_keychain_no_package_can_reach( self, isolated_home, secret_vault_factory @@ -890,7 +997,8 @@ class TestClearCliToken: assert clear_cli_token(vault=vault) == KeyringNotInstalled() assert vault.blob is not None - assert "sk-in-file" not in _token_file(isolated_home).read_text() + assert not _token_file(isolated_home).exists() + assert "sk-in-file" not in _everything_on_disk(isolated_home) @pytest.mark.skipif(os.geteuid() == 0, reason="root ignores file permissions") def test_a_file_that_gives_up_neither_its_secret_nor_itself_is_reported_not_raised( @@ -917,10 +1025,10 @@ class TestClearCliToken: def test_a_directory_that_takes_no_new_file_still_gives_up_the_secret_in_the_old_one( self, isolated_home, secret_vault_factory ): - """A read-only `~/.litellm` accepts no replacement token file and no removal of the one it - has, and still lets that one be shortened. The secret goes, the file stays as the note that - the keychain went unchecked, and the logout after it warns again instead of reading the gap - the removal would have left as a clean keychain. + """A read-only `~/.litellm` accepts no removal of the token file and no new file beside it, + and still lets the one it has be shortened. The secret goes, the file stays because nothing + else can, and the logout after it says the file is still there, since a keychain that answers + again would sign it back in, rather than calling the machine logged out. The key is a realistic length so the file genuinely shrinks: a rewrite in place that leaves the tail of the old contents behind hands the next run a file it cannot parse.""" @@ -929,25 +1037,16 @@ class TestClearCliToken: path.parent.chmod(0o500) try: assert clear_cli_token(vault=vault) == KeyringUnreachable() - assert clear_cli_token(vault=vault) == KeyringUnreachable() + assert isinstance(clear_cli_token(vault=vault), LoginRecordNotMoved) finally: path.parent.chmod(0o700) assert json.loads(path.read_text()).get("key") is None - @pytest.mark.skipif(os.geteuid() == 0, reason="root ignores file permissions") - def test_a_note_the_logout_had_to_remove_is_written_again_for_the_next_one( + def test_a_full_disk_still_keeps_the_unchecked_keychain_on_record( self, isolated_home, secret_vault_factory, monkeypatch ): - """A full disk refuses the replacement file and a read-only token file refuses the rewrite - in place, so the only way left to get the secret off disk is to remove the file carrying it. - That file was also the note saying the keychain went unchecked, and its absence is what the - next logout would read as a keychain already known to be clean. - - Removing it is what frees the room the replacement was refused for, so the note is written - again on the way out and the logout after this one still warns.""" path = _write_legacy_file(isolated_home) - path.chmod(0o400) monkeypatch.setattr( "litellm.litellm_core_utils.private_json.tempfile.mkstemp", _MkstempThatNeedsTheOldFileGone(path), @@ -957,7 +1056,26 @@ class TestClearCliToken: assert clear_cli_token(vault=vault) == KeyringUnreachable() assert clear_cli_token(vault=vault) == KeyringUnreachable() - assert json.loads(path.read_text()).get("key") is None + assert not path.exists() + assert "sk-legacy" not in _everything_on_disk(isolated_home) + + def test_a_full_disk_keeps_a_locked_keychain_login_revocable_without_signing_it_back_in( + self, isolated_home, secret_vault_factory, monkeypatch + ): + vault = secret_vault_factory() + save_cli_token( + CliTokenRecord(base_url=SERVER, key="sk-first", refresh_token="rt-first", timestamp=1000.0), vault=vault + ) + monkeypatch.setattr("litellm.litellm_core_utils.private_json.tempfile.mkstemp", _full_disk_mkstemp) + vault.available = False + + assert clear_cli_token(vault=vault) == KeyringUnreachable() + + vault.available = True + left = load_cli_token_left_in_keychain(vault=vault) + assert left is not None + assert left.refresh_token == "rt-first" + assert load_cli_token(vault=vault) is None @pytest.mark.skipif(os.geteuid() == 0, reason="root ignores directory permissions") def test_a_metadata_file_that_will_not_go_is_not_worth_alarming_the_user_over( diff --git a/tests/unit/proxy/client/cli/test_auth_commands.py b/tests/unit/proxy/client/cli/test_auth_commands.py index 5d5ebcf5dfd..18c2c6ce8af 100644 --- a/tests/unit/proxy/client/cli/test_auth_commands.py +++ b/tests/unit/proxy/client/cli/test_auth_commands.py @@ -1,3 +1,4 @@ +import dataclasses import json import os import stat @@ -16,6 +17,7 @@ from litellm.litellm_core_utils.cli_keyring import ( DISABLE_KEYRING_ENV_VAR, KeyringDisabled, KeyringNotInstalled, + KeyringUnreachable, SecretErased, SecretStored, ) @@ -1344,6 +1346,22 @@ class TestKeychainBackedCommands: assert "could not be removed" in result.output assert not (isolated_home / ".litellm" / "token.json").exists() + @pytest.mark.parametrize("failure", [KeyringDisabled(), KeyringUnreachable(), KeyringNotInstalled()]) + def test_logout_without_a_usable_keychain_leaves_every_command_signed_out( + self, isolated_home, secret_vault_factory, failure + ): + vault = secret_vault_factory(available=False, failure=failure) + self._login(vault) + obj = {"base_url": "https://test.example.com", "secret_vault": vault} + self.runner.invoke(logout, obj=obj) + + whoami_result = self.runner.invoke(whoami, obj=obj) + print_result = self.runner.invoke(print_token, obj=obj) + + assert whoami_result.output == "Not authenticated. Run 'lite login' to authenticate.\n" + assert print_result.exit_code == 1 + assert print_result.output == "Not authenticated. Run 'lite login'.\n" + def test_print_token_explains_a_locked_keychain_instead_of_printing_nothing( self, isolated_home, secret_vault_factory ): @@ -1933,6 +1951,287 @@ class TestPkceLogoutCommand: assert "Logged out successfully" not in result.output clear.assert_not_called() + def test_a_logout_that_reaches_the_keychain_late_still_revokes_the_login( + self, isolated_home, secret_vault_factory + ): + unlocked = secret_vault_factory() + save_cli_token(CliTokenRecord(**_pkce_record()), vault=unlocked) + locked = secret_vault_factory(blob=unlocked.blob, available=False) + + with patch("litellm.proxy.client.cli.commands.auth.requests.Session", _FakeSession): + self.runner.invoke(logout, obj={"secret_vault": locked}) + result = self.runner.invoke(logout, obj={"secret_vault": unlocked}) + + assert result.output == "Logged out successfully. Authentication token cleared.\n" + assert unlocked.blob is None + assert _FakeSession.instances[-1].posts == [ + ( + f"{PKCE_BASE_URL}/revoke", + {"token": "llm_srefresh_old", "token_type_hint": "refresh_token", "client_id": "llm_dcrc_abc"}, + ) + ] + + def test_a_logout_that_cannot_note_the_keychain_login_leaves_it_for_the_next_logout_to_revoke( + self, isolated_home, secret_vault_factory + ): + unlocked = secret_vault_factory() + save_cli_token(CliTokenRecord(**_pkce_record()), vault=unlocked) + locked = secret_vault_factory(blob=unlocked.blob, available=False) + (isolated_home / ".litellm" / "keychain_not_cleared").mkdir() + + with patch("litellm.proxy.client.cli.commands.auth.requests.Session", _FakeSession): + locked_logout = self.runner.invoke(logout, obj={"secret_vault": locked}) + result = self.runner.invoke(logout, obj={"secret_vault": unlocked}) + + assert locked_logout.exit_code == 0, locked_logout.output + assert "Not logged out" in locked_logout.output + assert "Delete that file" not in locked_logout.output + assert result.output == "Logged out successfully. Authentication token cleared.\n" + assert unlocked.blob is None + assert [post for session in _FakeSession.instances for post in session.posts] == [ + ( + f"{PKCE_BASE_URL}/revoke", + {"token": "llm_srefresh_old", "token_type_hint": "refresh_token", "client_id": "llm_dcrc_abc"}, + ) + ] + + def test_a_login_after_a_logout_that_missed_the_keychain_revokes_the_login_it_replaces( + self, isolated_home, secret_vault_factory + ): + unlocked = secret_vault_factory() + save_cli_token(CliTokenRecord(**_pkce_record()), vault=unlocked) + locked = secret_vault_factory(blob=unlocked.blob, available=False) + + with ( + patch("litellm.proxy.client.cli.commands.auth.requests.Session", _FakeSession), + patch("litellm.proxy.client.cli.commands.auth.run_pkce_login", return_value=_pkce_credential()), + patch("litellm.proxy.client.cli.interface.show_commands"), + ): + self.runner.invoke(logout, obj={"secret_vault": locked}) + result = self.runner.invoke( + login, ["--pkce"], obj={"base_url": PKCE_BASE_URL, "secret_vault": unlocked} + ) + + assert result.exit_code == 0, result.output + assert "Could not revoke" not in result.output + assert [post for session in _FakeSession.instances for post in session.posts] == [ + ( + f"{PKCE_BASE_URL}/revoke", + {"token": "llm_srefresh_old", "token_type_hint": "refresh_token", "client_id": "llm_dcrc_abc"}, + ) + ] + + def test_a_second_logout_that_misses_the_keychain_still_lets_the_first_login_be_revoked( + self, isolated_home, secret_vault_factory + ): + unlocked = secret_vault_factory() + save_cli_token(CliTokenRecord(**_pkce_record()), vault=unlocked) + locked = secret_vault_factory(blob=unlocked.blob, available=False) + + with ( + patch("litellm.proxy.client.cli.commands.auth.requests.Session", _FakeSession), + patch("litellm.proxy.client.cli.commands.auth.run_pkce_login", return_value=_pkce_credential()), + patch("litellm.proxy.client.cli.interface.show_commands"), + ): + self.runner.invoke(logout, obj={"secret_vault": locked}) + self.runner.invoke(login, ["--pkce"], obj={"base_url": PKCE_BASE_URL, "secret_vault": locked}) + self.runner.invoke(logout, obj={"secret_vault": locked}) + result = self.runner.invoke(logout, obj={"secret_vault": unlocked}) + + assert result.output == "Logged out successfully. Authentication token cleared.\n" + assert unlocked.blob is None + assert [post["token"] for session in _FakeSession.instances for _, post in session.posts] == [ + "llm_srefresh_fresh", + "llm_srefresh_old", + ] + + @pytest.mark.parametrize("logout_while_locked", [True, False]) + def test_a_logout_after_a_locked_keychain_login_revokes_the_login_the_keychain_kept( + self, isolated_home, secret_vault_factory, logout_while_locked + ): + unlocked = secret_vault_factory() + save_cli_token(CliTokenRecord(**_pkce_record()), vault=unlocked) + locked = secret_vault_factory(blob=unlocked.blob, available=False) + + with ( + patch("litellm.proxy.client.cli.commands.auth.requests.Session", _FakeSession), + patch("litellm.proxy.client.cli.commands.auth.run_pkce_login", return_value=_pkce_credential()), + patch("litellm.proxy.client.cli.interface.show_commands"), + ): + if logout_while_locked: + self.runner.invoke(cli, ["logout"], obj={"secret_vault": locked}) + self.runner.invoke(cli, ["--base-url", PKCE_BASE_URL, "login", "--pkce"], obj={"secret_vault": locked}) + result = self.runner.invoke(cli, ["logout"], obj={"secret_vault": unlocked}) + + assert result.output == "Logged out successfully. Authentication token cleared.\n" + assert unlocked.blob is None + assert list((isolated_home / ".litellm").iterdir()) == [] + assert [post for session in _FakeSession.instances for post in session.posts] == [ + ( + f"{PKCE_BASE_URL}/revoke", + {"token": token, "token_type_hint": "refresh_token", "client_id": "llm_dcrc_abc"}, + ) + for token in ("llm_srefresh_old", "llm_srefresh_fresh") + ] + + def test_a_proxy_that_cannot_revoke_the_earlier_login_never_holds_back_the_current_logout( + self, isolated_home, secret_vault_factory + ): + class _EarlierProxyUnavailableSession(_FakeSession): + def post(self, url, *, data=None, json=None, timeout, allow_redirects): + self.posts.append((url, data)) + if data["token"] == "llm_srefresh_old": + return _FakeHttpResponse(503, {"error": "temporarily_unavailable"}) + return _FakeHttpResponse(200, {}) + + unlocked = secret_vault_factory() + save_cli_token(CliTokenRecord(**_pkce_record()), vault=unlocked) + locked = secret_vault_factory(blob=unlocked.blob, available=False) + + with ( + patch("litellm.proxy.client.cli.commands.auth.requests.Session", _EarlierProxyUnavailableSession), + patch("litellm.proxy.client.cli.commands.auth.run_pkce_login", return_value=_pkce_credential()), + patch("litellm.proxy.client.cli.interface.show_commands"), + ): + self.runner.invoke(cli, ["logout"], obj={"secret_vault": locked}) + self.runner.invoke(cli, ["--base-url", PKCE_BASE_URL, "login", "--pkce"], obj={"secret_vault": locked}) + result = self.runner.invoke(cli, ["logout"], obj={"secret_vault": unlocked}) + + assert result.exit_code == 0, result.output + assert ( + "Could not revoke the previous login's refresh token on the proxy (revocation failed with 503" + in result.output + ) + assert result.output.endswith("Logged out successfully. Authentication token cleared.\n") + assert unlocked.blob is None + assert list((isolated_home / ".litellm").iterdir()) == [] + assert [post["token"] for session in _FakeSession.instances for _, post in session.posts] == [ + "llm_srefresh_old", + "llm_srefresh_fresh", + ] + + def test_a_logout_whose_only_login_is_the_one_left_in_the_keychain_waits_for_its_revocation( + self, isolated_home, secret_vault_factory + ): + class _UnavailableSession(_FakeSession): + def __init__(self): + super().__init__() + self.response = _FakeHttpResponse(503, {"error": "temporarily_unavailable"}) + + unlocked = secret_vault_factory() + save_cli_token(CliTokenRecord(**_pkce_record()), vault=unlocked) + locked = secret_vault_factory(blob=unlocked.blob, available=False) + + with patch("litellm.proxy.client.cli.commands.auth.requests.Session", _UnavailableSession): + self.runner.invoke(cli, ["logout"], obj={"secret_vault": locked}) + result = self.runner.invoke(cli, ["logout"], obj={"secret_vault": unlocked}) + + assert result.exit_code == 1 + assert "Nothing was cleared; run `lite logout` again shortly." in result.output + assert unlocked.blob is not None + + def test_a_login_after_a_locked_keychain_login_revokes_the_login_the_keychain_kept( + self, isolated_home, secret_vault_factory + ): + unlocked = secret_vault_factory() + save_cli_token(CliTokenRecord(**_pkce_record()), vault=unlocked) + locked = secret_vault_factory(blob=unlocked.blob, available=False) + + with ( + patch("litellm.proxy.client.cli.commands.auth.requests.Session", _FakeSession), + patch("litellm.proxy.client.cli.commands.auth.run_pkce_login", return_value=_pkce_credential()), + patch("litellm.proxy.client.cli.interface.show_commands"), + ): + self.runner.invoke(cli, ["--base-url", PKCE_BASE_URL, "login", "--pkce"], obj={"secret_vault": locked}) + result = self.runner.invoke( + cli, ["--base-url", PKCE_BASE_URL, "login", "--pkce"], obj={"secret_vault": unlocked} + ) + + assert result.exit_code == 0, result.output + assert "Could not revoke" not in result.output + assert [post["token"] for session in _FakeSession.instances for _, post in session.posts] == [ + "llm_srefresh_old", + "llm_srefresh_fresh", + ] + + @pytest.mark.parametrize("seconds_the_clock_stepped_back", [0, 600]) + def test_a_command_between_the_unlock_and_the_logout_keeps_the_login_the_keychain_kept_revocable( + self, isolated_home, secret_vault_factory, seconds_the_clock_stepped_back + ): + unlocked = secret_vault_factory() + first_login = _pkce_record(timestamp=time.time() + seconds_the_clock_stepped_back) + save_cli_token(CliTokenRecord(**first_login), vault=unlocked) + locked = secret_vault_factory(blob=unlocked.blob, available=False) + + with ( + patch("litellm.proxy.client.cli.commands.auth.requests.Session", _FakeSession), + patch("litellm.proxy.client.cli.commands.auth.run_pkce_login", return_value=_pkce_credential()), + patch("litellm.proxy.client.cli.interface.show_commands"), + ): + self.runner.invoke(cli, ["logout"], obj={"secret_vault": locked}) + self.runner.invoke(cli, ["--base-url", PKCE_BASE_URL, "login", "--pkce"], obj={"secret_vault": locked}) + printed = self.runner.invoke(cli, ["auth", "print-token"], obj={"secret_vault": unlocked}) + result = self.runner.invoke(cli, ["logout"], obj={"secret_vault": unlocked}) + + assert printed.stdout == "sk-cli-fresh\n" + assert result.output == "Logged out successfully. Authentication token cleared.\n" + assert [post["token"] for session in _FakeSession.instances for _, post in session.posts] == [ + "llm_srefresh_old", + "llm_srefresh_fresh", + ] + + @pytest.mark.parametrize("revocation_status", [200, 503]) + def test_a_renewal_that_takes_the_keychain_entry_revokes_the_login_left_in_it( + self, isolated_home, secret_vault_factory, revocation_status + ): + class _RenewingSession(_FakeSession): + def post(self, url, *, data=None, json=None, timeout, allow_redirects): + self.posts.append((url, data)) + if url.endswith("/token"): + return _FakeHttpResponse(200, PKCE_TOKEN_RESPONSE) + return _FakeHttpResponse(revocation_status, {}) + + unlocked = secret_vault_factory() + save_cli_token(CliTokenRecord(**_pkce_record()), vault=unlocked) + locked = secret_vault_factory(blob=unlocked.blob, available=False) + expiring_login = dataclasses.replace(_pkce_credential(), expires_at=time.time() + 30) + + with ( + patch("litellm.proxy.client.cli.commands.auth.requests.Session", _RenewingSession), + patch("litellm.proxy.client.cli.commands.auth.run_pkce_login", return_value=expiring_login), + patch("litellm.proxy.client.cli.interface.show_commands"), + ): + self.runner.invoke(cli, ["logout"], obj={"secret_vault": locked}) + self.runner.invoke(cli, ["--base-url", PKCE_BASE_URL, "login", "--pkce"], obj={"secret_vault": locked}) + printed = self.runner.invoke(cli, ["auth", "print-token"], obj={"secret_vault": unlocked}) + self.runner.invoke(cli, ["logout"], obj={"secret_vault": unlocked}) + + assert printed.stdout == "sk-cli-rotated\n" + assert ("Could not revoke" in printed.stderr) is (revocation_status != 200) + assert [ + post["token"] for session in _FakeSession.instances for url, post in session.posts if url.endswith("/revoke") + ] == ["llm_srefresh_old", "llm_srefresh_rotated"] + + def test_a_logout_after_the_keychain_refused_the_erase_does_not_revoke_the_login_again( + self, isolated_home, secret_vault_factory + ): + vault = secret_vault_factory(erasable=False) + save_cli_token(CliTokenRecord(**_pkce_record()), vault=vault) + + with patch("litellm.proxy.client.cli.commands.auth.requests.Session", _FakeSession): + first = self.runner.invoke(logout, obj={"secret_vault": vault}) + second = self.runner.invoke(logout, obj={"secret_vault": vault}) + + assert first.exit_code == 0, first.output + assert second.exit_code == 0, second.output + assert "still in the OS keychain" in second.output + assert [post for session in _FakeSession.instances for post in session.posts] == [ + ( + f"{PKCE_BASE_URL}/revoke", + {"token": "llm_srefresh_old", "token_type_hint": "refresh_token", "client_id": "llm_dcrc_abc"}, + ) + ] + def test_logout_of_a_classic_token_makes_no_request(self): with ( patch("litellm.proxy.client.cli.commands.auth.load_token", return_value={"key": "sk-classic"}), diff --git a/tests/unit/proxy/client/cli/test_global_options.py b/tests/unit/proxy/client/cli/test_global_options.py index 8a1e9a3d141..2acffb704ca 100644 --- a/tests/unit/proxy/client/cli/test_global_options.py +++ b/tests/unit/proxy/client/cli/test_global_options.py @@ -253,6 +253,7 @@ def test_lite_goes_back_to_localhost_after_logout_without_a_keychain(cli_runner, result = cli_runner.invoke(cli, [], obj={"secret_vault": no_keychain}, input="quit\n") assert "Connected to LiteLLM server: http://localhost:4000" in result.output + assert "keychain" not in result.output def test_corrupt_config_file_falls_back_to_default(cli_runner, isolated_home):