mirror of
https://github.com/BerriAI/litellm.git
synced 2026-08-28 05:25:59 +00:00
fix(cli): verify every credential store transition before reporting it done
A keyring backend can accept a write and keep nothing. That is exactly what `keyring --disable` and PYTHON_KEYRING_BACKEND=keyring.backends.null.Keyring select, and it raises nothing to distinguish itself, so `lite login` was handing the credential to a black hole, scrubbing its own copy from token.json, and printing a success message over a login that no longer worked. Reading the value back is the only way to tell that backend apart from a keychain that really stored the secret. The same rule closes the rest of the gaps. A credential the token file will not record is taken back out of the keychain instead of being left live on a machine with no record of it, and is reported rather than raised. The migration stages its scrubbed file before the keychain is handed anything, so a directory that will not accept the rewrite stops the move rather than leaving the secret in two places. Logout no longer reads a key in the file as proof that the keychain is clear, which was never sound across two separate runs, and only draws that conclusion when the `keyring` package is missing outright, where nothing could have reached a keychain at all.
This commit is contained in:
mateo-berri
parent
424e74ba9f
commit
3c73a39877
6 changed files with 356 additions and 62 deletions
|
|
@ -6,8 +6,12 @@ Linux Secret Service) that holds the credential minted by `lite login`.
|
|||
|
||||
The `keyring` package is optional and imported lazily, so importing this module
|
||||
never pulls it in. Every failure is returned as a value, naming which of the
|
||||
three ways the keychain can be out of reach applies, so callers can degrade to
|
||||
the token file and tell the user what to do about it.
|
||||
ways the keychain can be out of reach applies, so callers can degrade to the
|
||||
token file and tell the user what to do about it.
|
||||
|
||||
A write is only reported as stored once it has been read back, because keyring's
|
||||
null backend, which `keyring --disable` and headless CI images both select,
|
||||
accepts every write and keeps nothing.
|
||||
"""
|
||||
|
||||
import os
|
||||
|
|
@ -61,7 +65,12 @@ class KeyringUnreachable:
|
|||
pass
|
||||
|
||||
|
||||
KeyringUnusable: TypeAlias = KeyringNotInstalled | KeyringDisabled | KeyringUnreachable
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class KeyringDiscardsWrites:
|
||||
pass
|
||||
|
||||
|
||||
KeyringUnusable: TypeAlias = KeyringNotInstalled | KeyringDisabled | KeyringUnreachable | KeyringDiscardsWrites
|
||||
SecretRead: TypeAlias = SecretFound | SecretMissing | KeyringUnusable
|
||||
SecretWrite: TypeAlias = SecretStored | KeyringUnusable
|
||||
SecretErase: TypeAlias = SecretErased | SecretStranded | KeyringUnusable
|
||||
|
|
@ -119,6 +128,13 @@ class KeyringVault:
|
|||
return SecretMissing() if blob is None else SecretFound(blob)
|
||||
|
||||
def write(self, blob: str) -> SecretWrite:
|
||||
"""Store the secret, reporting stored only once the keychain hands the same bytes back.
|
||||
|
||||
A backend that accepts writes and keeps nothing, which is exactly what `keyring --disable`
|
||||
and `PYTHON_KEYRING_BACKEND=keyring.backends.null.Keyring` select, raises nothing to
|
||||
distinguish itself. Reading the value back is the only way to tell it apart from a keychain
|
||||
that really stored the credential, and the caller is about to drop its own copy on our word.
|
||||
"""
|
||||
api: Final = _keyring_api()
|
||||
if isinstance(api, (KeyringNotInstalled, KeyringDisabled)):
|
||||
return api
|
||||
|
|
@ -126,7 +142,7 @@ class KeyringVault:
|
|||
api.set_password(KEYRING_SERVICE, KEYRING_ACCOUNT, blob)
|
||||
except Exception: # noqa: BLE001 # a keychain that refuses the write falls back to the token file
|
||||
return KeyringUnreachable()
|
||||
return SecretStored()
|
||||
return SecretStored() if self.read() == SecretFound(blob) else KeyringDiscardsWrites()
|
||||
|
||||
def erase(self) -> SecretErase:
|
||||
"""Remove our entry, reporting whether the keychain is guaranteed to be free of it.
|
||||
|
|
@ -137,7 +153,7 @@ class KeyringVault:
|
|||
the caller knows whether this machine ever put a secret in a keychain.
|
||||
"""
|
||||
match self.read():
|
||||
case KeyringNotInstalled() | KeyringDisabled() | KeyringUnreachable() as unusable:
|
||||
case KeyringNotInstalled() | KeyringDisabled() | KeyringUnreachable() | KeyringDiscardsWrites() as unusable:
|
||||
return unusable
|
||||
case SecretMissing():
|
||||
return SecretErased()
|
||||
|
|
|
|||
|
|
@ -13,15 +13,17 @@ This module has no dependencies on proxy code and can be safely imported at the
|
|||
"""
|
||||
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from types import MappingProxyType
|
||||
from typing import Final
|
||||
from typing import Final, TypeAlias
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, ValidationError
|
||||
|
||||
from litellm.litellm_core_utils.cli_keyring import (
|
||||
SYSTEM_KEYRING,
|
||||
KeyringDisabled,
|
||||
KeyringDiscardsWrites,
|
||||
KeyringNotInstalled,
|
||||
KeyringUnreachable,
|
||||
SecretErase,
|
||||
|
|
@ -33,7 +35,23 @@ from litellm.litellm_core_utils.cli_keyring import (
|
|||
SecretVault,
|
||||
SecretWrite,
|
||||
)
|
||||
from litellm.litellm_core_utils.private_json import ensure_private_dir, write_private_json
|
||||
from litellm.litellm_core_utils.private_json import (
|
||||
commit_staged_json,
|
||||
discard_staged_json,
|
||||
ensure_private_dir,
|
||||
stage_private_json,
|
||||
write_private_json,
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class CredentialNotSaved:
|
||||
"""The credential was minted but no store would keep it, so this machine has none."""
|
||||
|
||||
detail: str
|
||||
|
||||
|
||||
SecretSave: TypeAlias = SecretWrite | CredentialNotSaved
|
||||
|
||||
|
||||
class CliTokenRecord(BaseModel):
|
||||
|
|
@ -85,14 +103,24 @@ def load_cli_token(*, vault: SecretVault = SYSTEM_KEYRING) -> CliTokenRecord | N
|
|||
return _resolve_secret(record, vault)
|
||||
|
||||
|
||||
def save_cli_token(record: CliTokenRecord, *, vault: SecretVault = SYSTEM_KEYRING) -> SecretWrite:
|
||||
"""Store a freshly minted credential. Reports where its secret material ended up, and why"""
|
||||
def save_cli_token(record: CliTokenRecord, *, vault: SecretVault = SYSTEM_KEYRING) -> SecretSave:
|
||||
"""Store a freshly minted credential. Reports where its secret material ended up, and why.
|
||||
|
||||
The token file is what makes a keychain-backed credential findable again, so a file that will
|
||||
not be written takes the keychain copy down with it rather than leaving a live credential
|
||||
stored under a machine that has no record of it.
|
||||
"""
|
||||
outcome: Final = (
|
||||
SecretStored()
|
||||
if record.key is None
|
||||
else vault.write(_encode_secret(record.base_url, record.key, record.jwt_token))
|
||||
)
|
||||
_write_token_file(_without_secret(record) if isinstance(outcome, SecretStored) else record)
|
||||
try:
|
||||
_write_token_file(_without_secret(record) if isinstance(outcome, SecretStored) else record)
|
||||
except OSError as error:
|
||||
if record.key is not None and isinstance(outcome, SecretStored):
|
||||
vault.erase()
|
||||
return CredentialNotSaved(str(error))
|
||||
return outcome
|
||||
|
||||
|
||||
|
|
@ -105,24 +133,30 @@ def clear_cli_token(*, vault: SecretVault = SYSTEM_KEYRING) -> SecretErase:
|
|||
|
||||
|
||||
def _nothing_left_behind(outcome: SecretErase) -> bool:
|
||||
"""Whether the keychain can be trusted to hold no credential of ours once the file is gone"""
|
||||
"""Whether the keychain can be trusted to hold no credential of ours once the file is gone.
|
||||
|
||||
A keychain that exists but is out of reach right now is never trusted, whatever the token file
|
||||
looks like: the login that stored a secret there and the logout that cannot remove it are
|
||||
separate runs, free to differ in whether the keychain was usable at the time.
|
||||
"""
|
||||
match outcome:
|
||||
case SecretErased():
|
||||
return True
|
||||
case SecretStranded():
|
||||
case SecretStranded() | KeyringDisabled() | KeyringUnreachable() | KeyringDiscardsWrites():
|
||||
return False
|
||||
case KeyringNotInstalled() | KeyringDisabled() | KeyringUnreachable():
|
||||
return not _secret_lives_in_keychain()
|
||||
case KeyringNotInstalled():
|
||||
return _file_holds_its_own_secret()
|
||||
|
||||
|
||||
def _secret_lives_in_keychain() -> bool:
|
||||
"""Whether the token file is the metadata half of a pair whose secret half went to a keychain.
|
||||
def _file_holds_its_own_secret() -> bool:
|
||||
"""Whether the stored login keeps its secret in the token file, ruling out a keychain entry.
|
||||
|
||||
A file that still carries its own secret rules one out, which keeps `lite logout` quiet on the
|
||||
machines that never had a keychain to begin with.
|
||||
Sound only against a missing `keyring` package, the one way to lose the keychain that had to
|
||||
hold at storage time too, since nothing here can reach a keychain without it. A file whose
|
||||
secret half is absent went to a keychain by definition, and so rules nothing out.
|
||||
"""
|
||||
record: Final = _read_token_file()
|
||||
return record is not None and record.key is None and not record.jwt_token
|
||||
return record is not None and record.key is not None
|
||||
|
||||
|
||||
def get_litellm_gateway_api_key(
|
||||
|
|
@ -179,7 +213,7 @@ def is_cli_token_fresh(token_data: CliTokenRecord, buffer_hours: float = 0.1) ->
|
|||
def _read_token_file() -> CliTokenRecord | None:
|
||||
try:
|
||||
raw: Final = Path(get_cli_token_file_path()).read_text()
|
||||
except OSError:
|
||||
except (OSError, ValueError):
|
||||
return None
|
||||
try:
|
||||
return CliTokenRecord.model_validate_json(raw)
|
||||
|
|
@ -193,7 +227,7 @@ def _resolve_secret(record: CliTokenRecord, vault: SecretVault) -> CliTokenRecor
|
|||
return _apply_vault_secret(record, blob, vault)
|
||||
case SecretMissing():
|
||||
return _migrate_file_secret(record, vault)
|
||||
case KeyringNotInstalled() | KeyringDisabled() | KeyringUnreachable():
|
||||
case KeyringNotInstalled() | KeyringDisabled() | KeyringUnreachable() | KeyringDiscardsWrites():
|
||||
return record
|
||||
|
||||
|
||||
|
|
@ -216,38 +250,46 @@ def _apply_vault_secret(record: CliTokenRecord, blob: str, vault: SecretVault) -
|
|||
|
||||
|
||||
def _migrate_file_secret(record: CliTokenRecord, vault: SecretVault) -> CliTokenRecord | None:
|
||||
"""Move a file-held secret into the vault, but only if the file's copy can be taken away.
|
||||
"""Move a file-held secret into the vault, but only once the file's copy can be taken away.
|
||||
|
||||
Migrating without scrubbing would leave the credential live in two stores instead of one, so a
|
||||
file that will not give its copy up rolls the vault write back rather than widening exposure.
|
||||
The scrubbed file is staged first so a directory that will not accept it stops the migration
|
||||
before the keychain is handed anything. Copying the credential into a second store and only
|
||||
then discovering the first one cannot be cleaned would widen exposure instead of narrowing it,
|
||||
which is the opposite of what moving it into the keychain is for.
|
||||
"""
|
||||
if record.key is None:
|
||||
return None
|
||||
if not isinstance(vault.write(_encode_secret(record.base_url, record.key, record.jwt_token)), SecretStored):
|
||||
staged: Final = _stage_scrubbed_file(record)
|
||||
if staged is None:
|
||||
return record
|
||||
if not _scrub_file_secret(record):
|
||||
if not isinstance(vault.write(_encode_secret(record.base_url, record.key, record.jwt_token)), SecretStored):
|
||||
discard_staged_json(staged)
|
||||
return record
|
||||
if not _commit_scrubbed_file(staged):
|
||||
vault.erase()
|
||||
return record
|
||||
|
||||
|
||||
def _scrub_file_secret(record: CliTokenRecord) -> bool:
|
||||
"""Leave no secret material in the token file once the vault holds it.
|
||||
|
||||
A file that cannot be rewritten without the secret is removed instead. Signing in again costs
|
||||
the user one command; a live credential left behind in cleartext costs them the credential.
|
||||
"""
|
||||
"""Leave no secret material in the token file once the vault holds it"""
|
||||
if record.key is None and not record.jwt_token:
|
||||
return True
|
||||
staged: Final = _stage_scrubbed_file(record)
|
||||
return staged is not None and _commit_scrubbed_file(staged)
|
||||
|
||||
|
||||
def _stage_scrubbed_file(record: CliTokenRecord) -> str | None:
|
||||
path: Final = Path(get_cli_token_file_path())
|
||||
try:
|
||||
_write_token_file(_without_secret(record))
|
||||
ensure_private_dir(path.parent)
|
||||
return stage_private_json(str(path), _without_secret(record).model_dump(exclude_none=True))
|
||||
except OSError:
|
||||
return _discard_token_file()
|
||||
return True
|
||||
return None
|
||||
|
||||
|
||||
def _discard_token_file() -> bool:
|
||||
def _commit_scrubbed_file(staged: str) -> bool:
|
||||
try:
|
||||
Path(get_cli_token_file_path()).unlink(missing_ok=True)
|
||||
commit_staged_json(staged, get_cli_token_file_path())
|
||||
except OSError:
|
||||
return False
|
||||
return True
|
||||
|
|
|
|||
|
|
@ -16,8 +16,12 @@ def ensure_private_dir(directory: Path) -> None:
|
|||
directory.chmod(PRIVATE_DIR_MODE)
|
||||
|
||||
|
||||
def write_private_json(path: str, data: Mapping[str, object]) -> None:
|
||||
"""Atomically write JSON to path with owner-only permissions (0600)"""
|
||||
def stage_private_json(path: str, data: Mapping[str, object]) -> str:
|
||||
"""Write JSON to a private temp file beside `path`, ready for `commit_staged_json`.
|
||||
|
||||
Staging is the half that can fail on a read-only or full directory, so callers with something
|
||||
to lose can find that out before they act on the assumption that the rewrite will land.
|
||||
"""
|
||||
parent: Final = Path(path).parent
|
||||
parent.mkdir(parents=True, exist_ok=True)
|
||||
fd, tmp_path = tempfile.mkstemp(dir=str(parent), prefix=".tmp-", suffix=".json")
|
||||
|
|
@ -26,6 +30,26 @@ def write_private_json(path: str, data: Mapping[str, object]) -> None:
|
|||
json.dump(data, f, indent=2)
|
||||
f.flush()
|
||||
os.fsync(f.fileno())
|
||||
os.replace(tmp_path, path)
|
||||
finally:
|
||||
except BaseException:
|
||||
Path(tmp_path).unlink(missing_ok=True)
|
||||
raise
|
||||
return tmp_path
|
||||
|
||||
|
||||
def commit_staged_json(staged: str, path: str) -> None:
|
||||
"""Move a staged file into place, replacing whatever is there in one step"""
|
||||
try:
|
||||
os.replace(staged, path)
|
||||
except OSError:
|
||||
Path(staged).unlink(missing_ok=True)
|
||||
raise
|
||||
|
||||
|
||||
def discard_staged_json(staged: str) -> None:
|
||||
"""Throw a staged file away when the change it was part of is abandoned"""
|
||||
Path(staged).unlink(missing_ok=True)
|
||||
|
||||
|
||||
def write_private_json(path: str, data: Mapping[str, object]) -> None:
|
||||
"""Atomically write JSON to path with owner-only permissions (0600)"""
|
||||
commit_staged_json(stage_private_json(path, data), path)
|
||||
|
|
|
|||
|
|
@ -15,16 +15,20 @@ from litellm.litellm_core_utils.cli_keyring import (
|
|||
DISABLE_KEYRING_ENV_VAR,
|
||||
SYSTEM_KEYRING,
|
||||
KeyringDisabled,
|
||||
KeyringDiscardsWrites,
|
||||
KeyringNotInstalled,
|
||||
KeyringUnreachable,
|
||||
SecretErased,
|
||||
SecretFound,
|
||||
SecretMissing,
|
||||
SecretStored,
|
||||
SecretStranded,
|
||||
SecretVault,
|
||||
SecretWrite,
|
||||
)
|
||||
from litellm.litellm_core_utils.cli_token_utils import (
|
||||
CliTokenRecord,
|
||||
CredentialNotSaved,
|
||||
SecretSave,
|
||||
clear_cli_token,
|
||||
get_cli_token_file_path,
|
||||
get_litellm_gateway_api_key,
|
||||
|
|
@ -84,17 +88,19 @@ class CliAuthResult(TypedDict):
|
|||
|
||||
KEYRING_INSTALL_HINT: Final = "pip install 'litellm[cli]'"
|
||||
|
||||
KEYRING_ENABLE_HINT: Final = "keyring --enable (or unset PYTHON_KEYRING_BACKEND)"
|
||||
|
||||
STRANDED_CREDENTIAL_MESSAGE: Final = (
|
||||
"Logged out locally, but your credential is still in the OS keychain and could not be removed."
|
||||
)
|
||||
|
||||
KEYCHAIN_UNREACHABLE_MESSAGE: Final = (
|
||||
"Your credential is stored in your OS keychain, which could not be read. Unlock it, or install "
|
||||
f"the keyring package with: {KEYRING_INSTALL_HINT}. Run 'lite login' to start over."
|
||||
UNCHECKED_KEYCHAIN_MESSAGE: Final = (
|
||||
"Logged out locally, but your OS keychain could not be checked, so a credential stored there by "
|
||||
"an earlier login may still be usable."
|
||||
)
|
||||
|
||||
|
||||
def storage_notice(outcome: SecretWrite) -> str:
|
||||
def storage_notice(outcome: SecretSave) -> str:
|
||||
"""Tell the user where the credential ended up, and how to get keychain storage if it did not."""
|
||||
path: Final = get_cli_token_file_path()
|
||||
match outcome:
|
||||
|
|
@ -109,6 +115,38 @@ def storage_notice(outcome: SecretWrite) -> str:
|
|||
return f"Keychain storage is off ({DISABLE_KEYRING_ENV_VAR}). Credential stored in {path} (owner-only)."
|
||||
case KeyringUnreachable():
|
||||
return f"No OS keychain available. Credential stored in {path} (owner-only)."
|
||||
case KeyringDiscardsWrites():
|
||||
return (
|
||||
f"Your keyring backend keeps nothing it is given, so the credential was stored in {path} "
|
||||
f"(owner-only) instead. For OS keychain storage, run: {KEYRING_ENABLE_HINT}"
|
||||
)
|
||||
case CredentialNotSaved(detail=detail):
|
||||
return (
|
||||
f"Signed in, but the credential could not be saved to {path}: {detail}. "
|
||||
"Nothing was kept, so run 'lite login' again once that path is writable."
|
||||
)
|
||||
|
||||
|
||||
def keychain_unreadable_notice(vault: SecretVault) -> str:
|
||||
"""Explain why the secret half of a stored login cannot be produced, and what fixes it"""
|
||||
match vault.read():
|
||||
case KeyringNotInstalled():
|
||||
return (
|
||||
"Your credential is in your OS keychain, which this install cannot read without the "
|
||||
f"keyring package. Install it with: {KEYRING_INSTALL_HINT}, or run 'lite login' to start over."
|
||||
)
|
||||
case KeyringDisabled():
|
||||
return (
|
||||
f"Your credential is in your OS keychain, which {DISABLE_KEYRING_ENV_VAR} is blocking. "
|
||||
"Unset it, or run 'lite login' to start over."
|
||||
)
|
||||
case KeyringUnreachable() | KeyringDiscardsWrites():
|
||||
return (
|
||||
"Your credential is in your OS keychain, which could not be read. Unlock it, or run "
|
||||
"'lite login' to start over."
|
||||
)
|
||||
case SecretFound() | SecretMissing():
|
||||
return "Your credential could not be read from your OS keychain. Run 'lite login' to start over."
|
||||
|
||||
|
||||
def context_secret_vault(ctx: click.Context) -> SecretVault:
|
||||
|
|
@ -715,6 +753,8 @@ def login(ctx: click.Context, config_claude: bool):
|
|||
click.echo("\nLogin successful!")
|
||||
click.echo(f"JWT Token: {api_key[:20]}...")
|
||||
click.echo(storage_notice(stored))
|
||||
if isinstance(stored, CredentialNotSaved):
|
||||
return
|
||||
click.echo("You can now use the CLI without specifying --api-key")
|
||||
|
||||
if config_claude:
|
||||
|
|
@ -751,14 +791,17 @@ def logout(ctx: click.Context):
|
|||
match clear_cli_token(vault=context_secret_vault(ctx)):
|
||||
case SecretErased():
|
||||
click.echo("Logged out successfully. Authentication token cleared.")
|
||||
case SecretStranded():
|
||||
click.echo(STRANDED_CREDENTIAL_MESSAGE)
|
||||
click.echo("Unlock your keychain and run 'lite logout' again to clear it.")
|
||||
case KeyringNotInstalled():
|
||||
click.echo(STRANDED_CREDENTIAL_MESSAGE)
|
||||
click.echo(f"Install the keyring package with: {KEYRING_INSTALL_HINT}, then run 'lite logout' again.")
|
||||
case KeyringDisabled():
|
||||
click.echo(STRANDED_CREDENTIAL_MESSAGE)
|
||||
click.echo(UNCHECKED_KEYCHAIN_MESSAGE)
|
||||
click.echo(f"Unset {DISABLE_KEYRING_ENV_VAR} and run 'lite logout' again to clear it.")
|
||||
case SecretStranded() | KeyringUnreachable():
|
||||
click.echo(STRANDED_CREDENTIAL_MESSAGE)
|
||||
case KeyringUnreachable() | KeyringDiscardsWrites():
|
||||
click.echo(UNCHECKED_KEYCHAIN_MESSAGE)
|
||||
click.echo("Unlock your keychain and run 'lite logout' again to clear it.")
|
||||
|
||||
|
||||
|
|
@ -795,7 +838,7 @@ def print_token(ctx: click.Context):
|
|||
|
||||
api_key: Final = token_data.key
|
||||
if not api_key:
|
||||
click.echo(KEYCHAIN_UNREACHABLE_MESSAGE, err=True)
|
||||
click.echo(keychain_unreadable_notice(context_secret_vault(ctx)), err=True)
|
||||
sys.exit(1)
|
||||
|
||||
click.echo(api_key)
|
||||
|
|
@ -821,7 +864,7 @@ def whoami(ctx: click.Context):
|
|||
click.echo(f"Token age: {age_hours:.1f} hours")
|
||||
|
||||
if token_data.key is None:
|
||||
click.echo(KEYCHAIN_UNREACHABLE_MESSAGE)
|
||||
click.echo(keychain_unreadable_notice(context_secret_vault(ctx)))
|
||||
|
||||
if age_hours > CLI_JWT_EXPIRATION_HOURS:
|
||||
click.echo(f"Warning: Token is more than {CLI_JWT_EXPIRATION_HOURS} hours old and may have expired.")
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@ from litellm.litellm_core_utils.cli_keyring import (
|
|||
SecretFound,
|
||||
SecretMissing,
|
||||
KeyringDisabled,
|
||||
KeyringDiscardsWrites,
|
||||
KeyringNotInstalled,
|
||||
KeyringUnreachable,
|
||||
SecretErased,
|
||||
|
|
@ -23,6 +24,7 @@ from litellm.litellm_core_utils.cli_keyring import (
|
|||
)
|
||||
from litellm.litellm_core_utils.cli_token_utils import (
|
||||
CliTokenRecord,
|
||||
CredentialNotSaved,
|
||||
clear_cli_token,
|
||||
get_cli_token_file_path,
|
||||
get_litellm_gateway_api_key,
|
||||
|
|
@ -225,6 +227,15 @@ class TestLoadCliToken:
|
|||
|
||||
assert record.key == "sk-legacy"
|
||||
|
||||
def test_a_token_file_that_is_not_text_is_not_a_login(self, isolated_home, secret_vault_factory):
|
||||
"""A truncated write or a half-synced backup can leave bytes that are not UTF-8 at all.
|
||||
Reading them must fail the way an absent file does, not crash every `lite` command."""
|
||||
path = _token_file(isolated_home)
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_bytes(b"\xff\xfe not utf-8 at all")
|
||||
|
||||
assert load_cli_token(vault=secret_vault_factory()) is None
|
||||
|
||||
def test_corrupt_token_file_is_not_a_login(self, isolated_home, secret_vault_factory):
|
||||
_token_file(isolated_home).parent.mkdir()
|
||||
_token_file(isolated_home).write_text("not json at all {{{")
|
||||
|
|
@ -301,6 +312,40 @@ class TestSaveCliToken:
|
|||
|
||||
assert stat.S_IMODE(config_dir.stat().st_mode) == 0o700
|
||||
|
||||
def test_a_credential_no_store_would_keep_is_reported_rather_than_raised(
|
||||
self, isolated_home, secret_vault_factory, monkeypatch
|
||||
):
|
||||
"""`lite login` catches whatever escapes here and calls it an authentication failure, which
|
||||
is the one thing that did not happen: the proxy minted a real credential. Saying so lets the
|
||||
user act on the actual problem instead of retrying a sign-in that already worked."""
|
||||
|
||||
def _explode(*args, **kwargs):
|
||||
raise OSError("read-only file system")
|
||||
|
||||
monkeypatch.setattr("litellm.litellm_core_utils.private_json.json.dump", _explode)
|
||||
|
||||
outcome = save_cli_token(CliTokenRecord(base_url=SERVER, key="sk-new"), vault=secret_vault_factory())
|
||||
|
||||
assert isinstance(outcome, CredentialNotSaved)
|
||||
assert "read-only file system" in outcome.detail
|
||||
|
||||
def test_a_credential_the_file_will_not_record_is_taken_back_out_of_the_keychain(
|
||||
self, isolated_home, secret_vault_factory, monkeypatch
|
||||
):
|
||||
"""The token file is what makes a keychain entry findable again. Leaving the secret in the
|
||||
keychain with nothing pointing at it strands a live credential under a machine that has no
|
||||
idea it is there, and no `lite logout` would ever go looking for it."""
|
||||
vault = secret_vault_factory()
|
||||
|
||||
def _explode(*args, **kwargs):
|
||||
raise OSError("read-only file system")
|
||||
|
||||
monkeypatch.setattr("litellm.litellm_core_utils.private_json.json.dump", _explode)
|
||||
|
||||
save_cli_token(CliTokenRecord(base_url=SERVER, key="sk-new"), vault=vault)
|
||||
|
||||
assert vault.blob is None
|
||||
|
||||
def test_a_failed_write_leaves_the_previous_credential_intact(self, isolated_home, secret_vault_factory, monkeypatch):
|
||||
path = _write_legacy_file(isolated_home)
|
||||
before = path.read_text()
|
||||
|
|
@ -340,9 +385,11 @@ class TestScrubFailure:
|
|||
assert json.loads(path.read_text())["key"] == "sk-legacy"
|
||||
assert vault.blob is None
|
||||
|
||||
def test_a_file_that_cannot_be_rewritten_is_removed_instead(
|
||||
def test_a_full_disk_stops_the_migration_before_the_keychain_is_handed_anything(
|
||||
self, isolated_home, secret_vault_factory, monkeypatch
|
||||
):
|
||||
"""The scrubbed file is staged first precisely so this is knowable in advance. A disk that
|
||||
cannot take the rewrite leaves the credential where it already was, in one store."""
|
||||
path = _write_legacy_file(isolated_home)
|
||||
vault = secret_vault_factory()
|
||||
|
||||
|
|
@ -354,8 +401,8 @@ class TestScrubFailure:
|
|||
record = load_cli_token(vault=vault)
|
||||
|
||||
assert record.key == "sk-legacy"
|
||||
assert json.loads(vault.blob)["key"] == "sk-legacy"
|
||||
assert not path.exists()
|
||||
assert vault.blob is None
|
||||
assert json.loads(path.read_text())["key"] == "sk-legacy"
|
||||
assert list(path.parent.glob(".tmp-*")) == []
|
||||
|
||||
|
||||
|
|
@ -376,12 +423,39 @@ class TestClearCliToken:
|
|||
assert clear_cli_token(vault=vault) == SecretStranded()
|
||||
assert not _token_file(isolated_home).exists()
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"failure", [KeyringDisabled(), KeyringUnreachable(), KeyringDiscardsWrites()]
|
||||
)
|
||||
def test_a_secret_in_the_file_is_no_evidence_about_a_keychain_that_exists(
|
||||
self, isolated_home, secret_vault_factory, failure
|
||||
):
|
||||
"""Store a secret in the keychain, sign in again while the keychain is unusable so the new
|
||||
secret lands in the file, then log out while it is still unusable. The file now carries its
|
||||
own secret and the first login's entry is still there, so reading the file as proof of a
|
||||
clean keychain reports a logout that did not happen."""
|
||||
_write_legacy_file(isolated_home)
|
||||
vault = secret_vault_factory(available=False, failure=failure)
|
||||
|
||||
assert clear_cli_token(vault=vault) == failure
|
||||
assert not _token_file(isolated_home).exists()
|
||||
|
||||
def test_a_second_logout_still_reports_the_keychain_it_could_not_clear(
|
||||
self, isolated_home, secret_vault_factory
|
||||
):
|
||||
"""The first logout deletes the file and tells the user to run it again once the keychain is
|
||||
reachable. If the second run reads that missing file as proof of a clean keychain, the advice
|
||||
turns into the very false all-clear it was issued to prevent."""
|
||||
_write_metadata_only_file(isolated_home)
|
||||
vault = secret_vault_factory(available=False, failure=KeyringUnreachable())
|
||||
|
||||
assert clear_cli_token(vault=vault) == KeyringUnreachable()
|
||||
assert clear_cli_token(vault=vault) == KeyringUnreachable()
|
||||
|
||||
def test_logout_from_an_install_without_keyring_does_not_claim_the_keychain_is_clear(
|
||||
self, isolated_home, secret_vault_factory
|
||||
):
|
||||
"""Log in where `litellm[cli]` is installed and the secret goes to the OS keychain; log out
|
||||
from a venv without it and the entry survives, because it belongs to the OS rather than to
|
||||
the package. Reporting a clean logout there leaves a live credential the user thinks is gone."""
|
||||
"""A file holding only metadata put its secret in a keychain by definition. Losing the
|
||||
package that reaches it does not take the entry with it, so this cannot report success."""
|
||||
_write_metadata_only_file(isolated_home)
|
||||
vault = secret_vault_factory(available=False, failure=KeyringNotInstalled())
|
||||
|
||||
|
|
@ -389,8 +463,9 @@ class TestClearCliToken:
|
|||
assert not _token_file(isolated_home).exists()
|
||||
|
||||
def test_a_file_backed_login_logs_out_quietly_without_keyring(self, isolated_home, secret_vault_factory):
|
||||
"""The complement: a user who never had a keychain keeps their whole credential in the file,
|
||||
so removing it is a complete logout and must not warn about an entry that cannot exist."""
|
||||
"""The complement, and the one inference the file does support: nothing here can reach a
|
||||
keychain without the package, so an install that lacks it and a file that still holds its
|
||||
own secret between them account for the whole credential."""
|
||||
_write_legacy_file(isolated_home)
|
||||
vault = secret_vault_factory(available=False, failure=KeyringNotInstalled())
|
||||
|
||||
|
|
@ -417,11 +492,12 @@ class TestIsCliTokenFresh:
|
|||
|
||||
|
||||
class _FakeKeyringModule:
|
||||
def __init__(self, stored=None, *, get_error=None, set_error=None, delete_error=None):
|
||||
def __init__(self, stored=None, *, get_error=None, set_error=None, delete_error=None, discard=False):
|
||||
self.stored = stored
|
||||
self.get_error = get_error
|
||||
self.set_error = set_error
|
||||
self.delete_error = delete_error
|
||||
self.discard = discard
|
||||
self.calls = []
|
||||
|
||||
def get_password(self, service_name, username):
|
||||
|
|
@ -434,6 +510,8 @@ class _FakeKeyringModule:
|
|||
self.calls.append(("set", service_name, username))
|
||||
if self.set_error is not None:
|
||||
raise self.set_error
|
||||
if self.discard:
|
||||
return
|
||||
self.stored = password
|
||||
|
||||
def delete_password(self, service_name, username):
|
||||
|
|
@ -503,6 +581,41 @@ class TestKeyringVault:
|
|||
|
||||
assert KeyringVault().erase() == SecretStranded()
|
||||
|
||||
def test_a_backend_that_keeps_nothing_is_not_a_successful_write(self, install_fake_keyring):
|
||||
"""keyring's null backend accepts every write, stores nothing, and raises nothing to say so.
|
||||
Taking its silence for success is how a credential gets deleted: the caller drops its own
|
||||
copy on our word. Only reading the value back tells the two apart."""
|
||||
fake = install_fake_keyring(_FakeKeyringModule(discard=True))
|
||||
|
||||
assert KeyringVault().write("blob-1") == KeyringDiscardsWrites()
|
||||
assert fake.stored is None
|
||||
|
||||
def test_the_real_null_backend_is_rejected(self, monkeypatch):
|
||||
"""Pinned against the actual library rather than the double above, because the whole risk is
|
||||
that upstream's no-op write looks exactly like a successful one."""
|
||||
keyring = pytest.importorskip("keyring")
|
||||
null_backend = pytest.importorskip("keyring.backends.null")
|
||||
monkeypatch.delenv(DISABLE_KEYRING_ENV_VAR, raising=False)
|
||||
previous = keyring.get_keyring()
|
||||
keyring.set_keyring(null_backend.Keyring())
|
||||
try:
|
||||
assert KeyringVault().write("blob-1") == KeyringDiscardsWrites()
|
||||
finally:
|
||||
keyring.set_keyring(previous)
|
||||
|
||||
def test_a_credential_survives_a_backend_that_keeps_nothing(
|
||||
self, isolated_home, install_fake_keyring
|
||||
):
|
||||
"""The end of the same story: the credential must still be usable afterwards. Reporting the
|
||||
discard is only worth anything if the token file then keeps the copy the keychain refused."""
|
||||
install_fake_keyring(_FakeKeyringModule(discard=True))
|
||||
|
||||
outcome = save_cli_token(CliTokenRecord(base_url=SERVER, key="sk-only-copy"))
|
||||
|
||||
assert outcome == KeyringDiscardsWrites()
|
||||
assert json.loads(_token_file(isolated_home).read_text())["key"] == "sk-only-copy"
|
||||
assert load_cli_token().key == "sk-only-copy"
|
||||
|
||||
def test_erasing_a_locked_keychain_is_a_failure(self, install_fake_keyring):
|
||||
install_fake_keyring(_FakeKeyringModule(get_error=RuntimeError("locked")))
|
||||
|
||||
|
|
|
|||
|
|
@ -16,12 +16,13 @@ from litellm.constants import CLI_JWT_EXPIRATION_HOURS
|
|||
from litellm.litellm_core_utils.cli_keyring import (
|
||||
DISABLE_KEYRING_ENV_VAR,
|
||||
KeyringDisabled,
|
||||
KeyringDiscardsWrites,
|
||||
KeyringNotInstalled,
|
||||
)
|
||||
from litellm.litellm_core_utils.cli_token_utils import CliTokenRecord, save_cli_token
|
||||
from litellm.proxy.client.cli import cli
|
||||
from litellm.proxy.client.cli.commands.auth import (
|
||||
KEYCHAIN_UNREACHABLE_MESSAGE,
|
||||
DISABLE_KEYRING_ENV_VAR,
|
||||
get_stored_api_key,
|
||||
login,
|
||||
logout,
|
||||
|
|
@ -461,6 +462,20 @@ class TestLogoutCommand:
|
|||
assert "still in the OS keychain" in result.output
|
||||
assert "pip install 'litellm[cli]'" in result.output
|
||||
|
||||
def test_logout_does_not_call_an_unusable_keychain_clean(self, isolated_home, secret_vault_factory):
|
||||
"""A keychain-backed login, then a login that fell back to the file because the keychain had
|
||||
become unusable, leaves the first entry live. The file's own secret says nothing about it,
|
||||
so a clean bill of health here is the one answer that cannot be justified."""
|
||||
_write_token_file(isolated_home, key="sk-in-file")
|
||||
vault = secret_vault_factory(available=False, failure=KeyringDisabled())
|
||||
|
||||
result = self.runner.invoke(logout, obj={"secret_vault": vault})
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "Logged out successfully" not in result.output
|
||||
assert "could not be checked" in result.output
|
||||
assert DISABLE_KEYRING_ENV_VAR in result.output
|
||||
|
||||
def test_logout_warns_when_the_keychain_refuses_to_release_the_entry(
|
||||
self, isolated_home, secret_vault_factory
|
||||
):
|
||||
|
|
@ -1003,6 +1018,19 @@ class TestKeychainBackedCommands:
|
|||
assert "No OS keychain available" not in result.output
|
||||
assert json.loads(token_file.read_text())["key"] == "sk-minted"
|
||||
|
||||
def test_login_keeps_the_credential_when_the_backend_keeps_nothing(
|
||||
self, isolated_home, secret_vault_factory
|
||||
):
|
||||
"""A backend that accepts writes and stores nothing must not be reported as keychain
|
||||
storage, because the file is then told to drop the only remaining copy."""
|
||||
result = self._login(secret_vault_factory(available=False, failure=KeyringDiscardsWrites()))
|
||||
|
||||
token_file = isolated_home / ".litellm" / "token.json"
|
||||
assert result.exit_code == 0
|
||||
assert "Credential stored in your OS keychain." not in result.output
|
||||
assert "keyring --enable" in result.output
|
||||
assert json.loads(token_file.read_text())["key"] == "sk-minted"
|
||||
|
||||
def test_login_names_the_kill_switch_instead_of_blaming_the_machine(
|
||||
self, isolated_home, secret_vault_factory
|
||||
):
|
||||
|
|
@ -1061,7 +1089,8 @@ class TestKeychainBackedCommands:
|
|||
result = self.runner.invoke(print_token, obj=obj)
|
||||
|
||||
assert result.exit_code == 1
|
||||
assert KEYCHAIN_UNREACHABLE_MESSAGE in result.output
|
||||
assert "could not be read" in result.output
|
||||
assert "lite login" in result.output
|
||||
|
||||
def test_whoami_flags_a_locked_keychain(self, isolated_home, secret_vault_factory):
|
||||
_write_home_json(
|
||||
|
|
@ -1074,7 +1103,34 @@ class TestKeychainBackedCommands:
|
|||
result = self.runner.invoke(whoami, obj=obj)
|
||||
|
||||
assert "Authenticated" in result.output
|
||||
assert KEYCHAIN_UNREACHABLE_MESSAGE in result.output
|
||||
assert "could not be read" in result.output
|
||||
|
||||
def test_whoami_names_the_kill_switch_rather_than_a_missing_package(
|
||||
self, isolated_home, secret_vault_factory
|
||||
):
|
||||
"""Every unreachable keychain used to be described as a locked one needing the keyring
|
||||
package installed. Someone who set the kill switch has the package and an unlocked keychain,
|
||||
so that advice sends them to fix two things that were never wrong."""
|
||||
_write_token_file(isolated_home, key=None)
|
||||
vault = secret_vault_factory(available=False, failure=KeyringDisabled())
|
||||
|
||||
result = self.runner.invoke(whoami, obj={"base_url": "https://test.example.com", "secret_vault": vault})
|
||||
|
||||
assert DISABLE_KEYRING_ENV_VAR in result.output
|
||||
assert "pip install" not in result.output
|
||||
|
||||
def test_print_token_points_an_install_without_keyring_at_the_package(
|
||||
self, isolated_home, secret_vault_factory
|
||||
):
|
||||
_write_token_file(isolated_home, key=None)
|
||||
vault = secret_vault_factory(available=False, failure=KeyringNotInstalled())
|
||||
obj = {"base_url": "https://test.example.com", "secret_vault": vault}
|
||||
|
||||
result = self.runner.invoke(print_token, obj=obj)
|
||||
|
||||
assert result.exit_code == 1
|
||||
assert "pip install 'litellm[cli]'" in result.output
|
||||
assert DISABLE_KEYRING_ENV_VAR not in result.output
|
||||
|
||||
|
||||
class TestApiKeyPrecedence:
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue