fix(cli): keep credential locks off network homes

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/litellm_core_utils/cli_credential_lock.py

@@ -1,11 +1,11 @@
-import ctypes
+import ctypes.wintypes
 import errno
 import os
+import stat
 import sys
 import time
 from collections.abc import Generator
 from contextlib import contextmanager
-from ctypes import wintypes
 from hashlib import sha256
 from pathlib import Path
 from typing import TYPE_CHECKING, Final, cast
@@ -18,12 +18,12 @@ if TYPE_CHECKING or sys.platform != "win32":
 
 @contextmanager
 def credential_lock(home: Path, timeout: float = 30) -> Generator[None, None, None]:
-    """Serialize credential changes without creating or modifying a lock file."""
+    """Serialize credential changes on this host without writing inside the home directory."""
     if sys.platform == "win32":
         with _windows_mutex(home, timeout):
             yield
         return
-    fd: Final = os.open(home, os.O_RDONLY | os.O_DIRECTORY)
+    fd: Final = _open_posix_lock(home)
     try:
         deadline: Final = time.monotonic() + timeout
         while True:
@@ -37,6 +37,9 @@ def credential_lock(home: Path, timeout: float = 30) -> Generator[None, None, No
                     raise Timeout(str(home)) from None
                 time.sleep(0.05)
         try:
+            os.utime(fd, None)
+            if os.fstat(fd).st_nlink != 1:
+                raise OSError("The CLI lock file was removed while waiting")
             yield
         finally:
             fcntl.flock(fd, fcntl.LOCK_UN)
@@ -44,17 +47,47 @@ def credential_lock(home: Path, timeout: float = 30) -> Generator[None, None, No
         os.close(fd)
 
 
+def _open_posix_lock(home: Path) -> int:
+    if sys.platform == "win32":
+        raise OSError("POSIX lock files are unavailable on Windows")
+    directory: Final = Path("/tmp") / f"litellm-cli-{os.getuid()}"
+    directory.mkdir(mode=0o700, exist_ok=True)
+    directory_fd: Final = os.open(directory, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
+    try:
+        directory_stat: Final = os.fstat(directory_fd)
+        if directory_stat.st_uid != os.getuid() or stat.S_IMODE(directory_stat.st_mode) & 0o077:
+            raise PermissionError("The CLI lock directory must be private and owned by the current user")
+        identity: Final = sha256(str(home.resolve()).encode()).hexdigest()
+        fd: Final = os.open(
+            f"{identity}.lock", os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_NONBLOCK, 0o600, dir_fd=directory_fd
+        )
+        try:
+            file_stat: Final = os.fstat(fd)
+            if not stat.S_ISREG(file_stat.st_mode) or file_stat.st_uid != os.getuid() or file_stat.st_nlink != 1:
+                raise PermissionError("The CLI lock must be a regular file owned only by the current user")
+            return fd
+        except OSError:
+            os.close(fd)
+            raise
+    finally:
+        os.close(directory_fd)
+
+
 @contextmanager
 def _windows_mutex(home: Path, timeout: float) -> Generator[None, None, None]:
     kernel: Final = ctypes.WinDLL("kernel32", use_last_error=True)
     create: Final = ctypes.WINFUNCTYPE(
-        wintypes.HANDLE, ctypes.c_void_p, wintypes.BOOL, wintypes.LPCWSTR, use_last_error=True
+        ctypes.wintypes.HANDLE, ctypes.c_void_p, ctypes.wintypes.BOOL, ctypes.wintypes.LPCWSTR, use_last_error=True
     )(("CreateMutexW", kernel))
-    wait: Final = ctypes.WINFUNCTYPE(wintypes.DWORD, wintypes.HANDLE, wintypes.DWORD, use_last_error=True)(
-        ("WaitForSingleObject", kernel)
+    wait: Final = ctypes.WINFUNCTYPE(
+        ctypes.wintypes.DWORD, ctypes.wintypes.HANDLE, ctypes.wintypes.DWORD, use_last_error=True
+    )(("WaitForSingleObject", kernel))
+    release: Final = ctypes.WINFUNCTYPE(ctypes.wintypes.BOOL, ctypes.wintypes.HANDLE, use_last_error=True)(
+        ("ReleaseMutex", kernel)
+    )
+    close: Final = ctypes.WINFUNCTYPE(ctypes.wintypes.BOOL, ctypes.wintypes.HANDLE, use_last_error=True)(
+        ("CloseHandle", kernel)
     )
-    release: Final = ctypes.WINFUNCTYPE(wintypes.BOOL, wintypes.HANDLE, use_last_error=True)(("ReleaseMutex", kernel))
-    close: Final = ctypes.WINFUNCTYPE(wintypes.BOOL, wintypes.HANDLE, use_last_error=True)(("CloseHandle", kernel))
     identity: Final = sha256(os.path.normcase(str(home.resolve())).encode()).hexdigest()
     handle: Final = cast(int | None, create(None, False, f"Global\\litellm-cli-{identity}"))
     if handle is None:

tests/test_litellm/proxy/client/cli/test_auth_commands.py

@@ -44,6 +44,9 @@ from litellm.proxy.client.cli.commands.auth import (
 from litellm.proxy.client.cli.commands.pkce_login import PkceFailure, RevocationUnavailable
 from tests.test_litellm_rust.support.child_interpreter import run_child_interpreter
 
+if sys.platform != "win32":
+    import fcntl
+
 
 @pytest.fixture
 def isolated_home(monkeypatch, tmp_path):
@@ -2499,6 +2502,52 @@ def windows_mutex_api():
         yield api
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX filesystem locking")
+def test_saved_login_operations_do_not_lock_network_home_directories(isolated_home, secret_vault_factory):
+    vault = secret_vault_factory()
+    save_cli_token(CliTokenRecord(**_pkce_record(team_id="team-a")), vault=vault)
+    flock = fcntl.flock
+
+    def network_flock(fd, operation):
+        if stat.S_ISDIR(os.fstat(fd).st_mode):
+            raise OSError(9, "network filesystem requires a write-open regular file")
+        return flock(fd, operation)
+
+    with (
+        patch("fcntl.flock", side_effect=network_flock),
+        patch("litellm.proxy.client.cli.commands.auth.requests.Session") as session,
+    ):
+        session.return_value.post.return_value = _FakeHttpResponse(200, PKCE_TOKEN_RESPONSE)
+        assert get_stored_api_key(PKCE_BASE_URL, vault=vault) == "sk-cli-rotated"
+        outcome = _replace_stored_token(_pkce_record(team_id="team-b"), _FakeSession(), vault, "team-b")
+        assert isinstance(outcome, SecretStored)
+        assert load_token(vault=vault)["team_id"] == "team-b"
+        result = CliRunner().invoke(logout, obj={"secret_vault": vault})
+        assert result.exit_code == 0, result.output
+        assert load_token(vault=vault) is None
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX filesystem locking")
+@pytest.mark.parametrize("unsafe", ["directory-permissions", "directory-owner", "file-owner", "hardlink"])
+def test_credential_lock_refuses_unsafe_filesystem_state(isolated_home, unsafe):
+    fstat = os.fstat
+
+    def unsafe_stat(fd):
+        original = fstat(fd)
+        directory = stat.S_ISDIR(original.st_mode)
+        if unsafe == "directory-permissions" and directory:
+            return os.stat_result((original.st_mode | 0o020, *original[1:]))
+        if (unsafe == "directory-owner" and directory) or (unsafe == "file-owner" and not directory):
+            return os.stat_result((*original[:4], original.st_uid + 1, *original[5:]))
+        if unsafe == "hardlink" and not directory:
+            return os.stat_result((*original[:3], 2, *original[4:]))
+        return original
+
+    with patch("os.fstat", side_effect=unsafe_stat), pytest.raises(PermissionError, match="CLI lock"):
+        with credential_lock(isolated_home):
+            pytest.fail("unsafe lock entered the credential operation")
+
+
 @pytest.mark.parametrize("wait_result", [0, 0x80])
 @pytest.mark.parametrize("release_result", [True, False])
 def test_windows_mutex_preserves_body_errors_and_closes_handle(