mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-14 09:03:16 +00:00
Merge 693d4217ec into 08f2a98250
This commit is contained in:
zoveki
GitHub
commit
53fffd71ec
7 changed files with 662 additions and 102 deletions
|
|
@ -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
|
||||
|
|
@ -159,6 +162,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 +171,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 +239,60 @@ 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 | 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() -> CredentialNotCleared | 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 CredentialNotCleared(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 +301,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 +389,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 +492,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)
|
||||
|
|
|
|||
|
|
@ -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`. 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`.
|
||||
|
||||
|
|
|
|||
|
|
@ -37,6 +37,7 @@ from litellm.litellm_core_utils.cli_token_utils import (
|
|||
get_cli_token_file_path,
|
||||
is_cli_token_fresh,
|
||||
load_cli_token,
|
||||
load_cli_token_left_in_keychain,
|
||||
save_cli_token,
|
||||
)
|
||||
|
||||
|
|
@ -234,22 +235,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 +304,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 +878,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 +1017,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 +1046,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():
|
||||
|
|
@ -1067,13 +1117,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,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -36,6 +36,7 @@ from litellm.litellm_core_utils.cli_token_utils import (
|
|||
get_litellm_gateway_api_key,
|
||||
is_cli_token_fresh,
|
||||
load_cli_token,
|
||||
load_cli_token_left_in_keychain,
|
||||
save_cli_token,
|
||||
)
|
||||
|
||||
|
|
@ -54,6 +55,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 +128,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 +786,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 +814,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 +838,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 +863,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 +883,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 +979,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 +996,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 +1024,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 +1036,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), CredentialNotCleared)
|
||||
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 +1055,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(
|
||||
|
|
|
|||
|
|
@ -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,286 @@ 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 "could not be removed" 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"}),
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue