mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-11 08:38:24 +00:00
fix(cli): end the local login when lite logout cannot check the OS keychain (#50098)
* fix(cli): end the local login when logout cannot check the OS keychain With no usable keyring, lite logout kept a secret-free token.json as a note that the keychain went unchecked. Every reader took that note for a live keychain-backed login, so lite whoami, lite auth print-token, and the bare lite shell all reported a login the user had just ended, and a keychain that answered again could pair the note with an earlier login's entry and sign the machine back in. Logout now removes token.json whatever the keychain answers. When the keychain could not be checked and there was a login to end, it records that in a separate owner-only keychain_unchecked file beside it, written after the removal so a full disk still has room for it. A repeat logout still warns while that file is there, and a logout that confirms the keychain is clean removes it. * fix(cli): keep warning after a logout the keychain refused to clear A keychain that will not give up its entry gets the same note as one that could not be checked, so a later logout made while the keychain is out of reach does not report a clean logout while the credential is still there. * fix(cli): revoke the login a logout left in an unreachable keychain once it answers A logout made while the keychain was locked ended the local login and dropped the metadata that pairs with the refresh token still in the keychain, so the next logout had nothing to revoke. The not-cleared note now keeps that record without its secret, and a later logout pairs it with the keychain entry and revokes it before erasing. * fix(cli): revoke a locked-keychain login on the next login, and only once after a refused erase A login made after a logout that could not reach the keychain now revokes the login it replaces, read from the keychain_not_cleared note, as the merge base did through the kept token.json. A keychain that refused the erase had already handed that logout the refresh token, so its note no longer pairs with the entry and the next logout does not revoke the same token again. * fix(cli): keep the first login revocable after a second locked-keychain logout A logout whose login kept its secret in token.json never reached the keychain, so it now leaves the note an earlier logout wrote in place, or writes an empty one, instead of overwriting it. The first login's refresh token stays revocable once the keychain answers, and a keychain-less machine no longer keeps user and team ids in the note. Also documents the limits that remain in the CLI README and drops the docstrings from the tests this PR added. * fix(cli): revoke the keychain's login after a login made while the keychain was locked * fix(cli): stop lite from reading the stored login before login and logout The cli group resolved the stored API key before every command, and that read moved a locked-keychain login's secret over the keychain entry the logout note names, so logout and login never saw the first login to revoke it. * fix(cli): keep the login a locked logout left in the keychain revocable While the keychain still holds the login an earlier logout could not end, reading the stored login now serves the secret token.json kept and leaves that keychain entry alone, whatever the login times say. A clock that stepped back no longer serves the ended login, and commands other than logout and login no longer move the newer secret over it. A silent renewal that writes over the entry revokes that login on the proxy. * fix(cli): keep a login revocable when its note cannot be written, and never block logout on an earlier login's proxy A locked logout whose note cannot take the login's record now puts token.json back, so the next logout that reaches the keychain still revokes that login, as the base did. A logout that ends both a current login and an earlier one left in the keychain warns when the earlier login's proxy cannot record the revocation, as lite login does, instead of refusing to log out of the current proxy. * fix(cli): move a locked keychain login's token file into its note instead of putting it back * fix(cli): never tell a logout to delete the record of a keychain login it could not move aside A read-only ~/.litellm made the locked-keychain logout say to delete token.json, which would leave the keychain's login unrevoked. It now says the logout did not happen and asks to make the directory writable. The README also says an older lite cannot finish a logout this release started. --------- Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
parent
43112547fb
commit
cfddaafe18
7 changed files with 686 additions and 103 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
|
||||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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`.
|
||||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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"}),
|
||||
|
|
|
|||
|
|
@ -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