diff --git a/litellm/litellm_core_utils/cli_token_utils.py b/litellm/litellm_core_utils/cli_token_utils.py index 70ae0e292a9..87ea97bb851 100644 --- a/litellm/litellm_core_utils/cli_token_utils.py +++ b/litellm/litellm_core_utils/cli_token_utils.py @@ -146,10 +146,7 @@ def get_cli_token_file_path() -> str: def cli_token_lock() -> FileLock: - """One reentrant process lock for the metadata file and shared keychain entry. - - Keep it beside the credential directory so a read-only directory still permits logout. - """ + """Lock metadata and keychain operations beside the directory so read-only stores still allow logout.""" directory: Final = Path(get_cli_token_file_path()).parent directory.parent.mkdir(parents=True, exist_ok=True) path: Final = directory.with_name(f"{directory.name}-token.lock") @@ -395,13 +392,17 @@ def get_litellm_gateway_api_key( >>> base_url="https://your-proxy.com/v1" >>> ) """ - record: Final = _read_token_file() - if record is None: + try: + with cli_token_lock(): + record: Final = _read_token_file() + if record is None: + return None + if expected_base_url is not None and record.base_url != expected_base_url.rstrip("/"): + return None + resolved: Final = _resolve_secret(record, vault) + return None if resolved is None else resolved.key + except (OSError, Timeout): return None - if expected_base_url is not None and record.base_url != expected_base_url.rstrip("/"): - return None - resolved: Final = _resolve_secret(record, vault) - return None if resolved is None else resolved.key def is_cli_token_fresh( diff --git a/tests/unit/litellm_core_utils/test_cli_token_utils.py b/tests/unit/litellm_core_utils/test_cli_token_utils.py index 1f398017817..11700c9ed66 100644 --- a/tests/unit/litellm_core_utils/test_cli_token_utils.py +++ b/tests/unit/litellm_core_utils/test_cli_token_utils.py @@ -474,6 +474,21 @@ class TestGetLitellmGatewayApiKey: assert get_litellm_gateway_api_key(vault=secret_vault_factory(blob=_blob())) is None +def test_sdk_read_does_not_migrate_credentials_when_the_store_cannot_be_locked( + isolated_home: Path, + secret_vault_factory: type[FakeSecretVault], +) -> None: + path: Final = _write_legacy_file(isolated_home, timestamp=1000.0) + original: Final = path.read_bytes() + vault: Final = secret_vault_factory() + Path(cli_token_lock().lock_file).mkdir() + + assert get_litellm_gateway_api_key(expected_base_url=SERVER, vault=vault) is None + assert path.read_bytes() == original + assert vault.reads == 0 + assert vault.writes == [] + + class TestSaveCliToken: def test_secret_goes_to_the_keychain_and_never_to_the_file(self, isolated_home, secret_vault_factory): vault = secret_vault_factory()