fix(redis): authenticate sync clusters with IAM credential providers (#40204)

* fix(redis): authenticate sync clusters with IAM credential providers Signed-off-by: Silu Panda <31051721+SiluPanda@users.noreply.github.com> * test(redis): exercise IAM cluster authentication over TCP Run Azure and GCP regressions against a real local cluster with only cloud token issuance stubbed. Build a checksum-verified Redis server in the compatibility workflow and report its coverage. Signed-off-by: Silu Panda <31051721+SiluPanda@users.noreply.github.com> * test(redis): separate unit and cluster integration coverage Keep the mapped test tree mock-only. Run the live cluster cases from the existing local caching integration file, selected by explicit node IDs in the Redis compatibility workflow. Signed-off-by: Silu Panda <31051721+SiluPanda@users.noreply.github.com> * test(ci): isolate workflow coverage audit fixtures Replace the stale unrun caching-file assumption with isolated workflow fixtures for file and node-ID selectors. Keep the unnamed-file negative check and clarify which live caching cases remain outside CI. Signed-off-by: Silu Panda <31051721+SiluPanda@users.noreply.github.com> --------- Signed-off-by: Silu Panda <31051721+SiluPanda@users.noreply.github.com> Co-authored-by: yassin <yassin@berri.ai> Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-09-26 01:12:21 +00:00 · 2026-09-23 15:23:55 -07:00 · 2026-09-23 15:23:55 -07:00 · 2b87b3c873
commit 2b87b3c873
parent fecc8c8f74
6 changed files with 219 additions and 32 deletions
--- a/.github/ci-coverage-allowlist.yml
+++ b/.github/ci-coverage-allowlist.yml
@ -10,12 +10,13 @@ test_paths:
    paths:
      - tests/rust-python-harness
  - reason: >-
-      What is left of the caching suite in tests/local_testing that runs nowhere. Every job that
-      globs that directory either deselects it (local_testing_part1 and part2 carry `-k "... and
-      not caching and not cache"`) or keeps only another keyword (langfuse, router, assistants),
-      and no job names these files the way redis_caching_unit_tests names test_dual_cache.py.
-      The gap was eight files and 118 tests when measured 2026-08-20; the five keyless ones now
-      run in the caching-local shard, leaving these three. Measured 2026-08-21 with no provider
+      Live-provider caching cases in tests/local_testing that remain outside CI. Jobs that
+      glob that directory either deselect them (local_testing_part1 and part2 carry `-k "... and
+      not caching and not cache"`) or keep only another keyword (langfuse, router, assistants).
+      Separately, test-redis-compat.yml selects two IAM cluster authentication tests in
+      test_caching.py by node ID. It does not run that file's other tests.
+      The gap was eight files and 118 tests when measured 2026-08-20; the five keyless files now
+      run in the caching-local shard, leaving live cases in these three. Measured 2026-08-21 with no provider
      credentials and no Redis: test_caching.py needs both (37 of 65 fail without them),
      test_disk_cache_unit_tests.py needs OPENAI_API_KEY for 2 of its 4, and
      test_gcs_cache_unit_tests.py needs GCS credentials for all 4. They want the keyless/live
--- a/.github/workflows/test-redis-compat.yml
+++ b/.github/workflows/test-redis-compat.yml
@ -9,6 +9,7 @@ on:
      - "litellm/_redis.py"
      - "litellm/_redis_credential_provider.py"
      - "tests/test_litellm/test_redis.py"
+      - "tests/local_testing/test_caching.py"
      - "tests/test_litellm/caching/test_redis_connection_pool.py"
      - ".github/workflows/test-redis-compat.yml"
      - "pyproject.toml"
@ -26,6 +27,9 @@ jobs:
    name: "redis-py ${{ matrix.redis-version }}"
    runs-on: ubuntu-latest
    timeout-minutes: 15
+    permissions:
+      contents: read
+      id-token: write

    strategy:
      fail-fast: false
@ -55,7 +59,7 @@ jobs:

      - name: Install dependencies
        run: |
-          .github/scripts/uv_sync_with_retries.sh --frozen --group ci --group proxy-dev --extra google --extra proxy --extra semantic-router
+          .github/scripts/uv_sync_with_retries.sh --frozen --group ci --group proxy-dev --extra google --extra proxy --extra extra_proxy --extra semantic-router

      - name: Pin redis-py to the matrix version
        env:
@ -64,12 +68,33 @@ jobs:
          uv pip install "redis==${REDIS_VERSION:?}"
          uv run --no-sync python -c "import redis; assert redis.__version__ == '${REDIS_VERSION:?}', redis.__version__; print('redis-py', redis.__version__)"

+      - name: Build Redis for cluster authentication tests
+        run: |
+          curl --fail --location --retry 3 https://download.redis.io/releases/redis-7.2.16.tar.gz -o "$RUNNER_TEMP/redis-7.2.16.tar.gz"
+          echo "960a8ec15e34ff40e57ff16837b26b33bd81f2da6d24497bb63de532a323a18e  $RUNNER_TEMP/redis-7.2.16.tar.gz" | sha256sum --check
+          tar -xzf "$RUNNER_TEMP/redis-7.2.16.tar.gz" -C "$RUNNER_TEMP"
+          make -C "$RUNNER_TEMP/redis-7.2.16" -j2 MALLOC=libc OPTIMIZATION=-O1 redis-server
+          echo "$RUNNER_TEMP/redis-7.2.16/src" >> "$GITHUB_PATH"
+
      - name: Run redis unit tests
        run: |
+          redis-server --version
          uv run --no-sync pytest \
            tests/test_litellm/test_redis.py \
            tests/test_litellm/caching/test_redis_connection_pool.py \
+            tests/local_testing/test_caching.py::test_sync_cluster_authenticates_with_azure_credentials \
+            tests/local_testing/test_caching.py::test_sync_cluster_authenticates_with_gcp_credentials \
            --tb=short -vv \
            --reruns 2 \
            --reruns-delay 1 \
-            --durations=20
+            --durations=20 \
+            --cov=./litellm --cov-report=xml:coverage-redis.xml
+
+      - name: Upload Redis coverage
+        if: matrix.redis-version == '5.3.1'
+        uses: codecov/codecov-action@75cd11691c0faa626561e295848008c8a7dddffe # v5.5.4
+        with:
+          use_oidc: true
+          files: coverage-redis.xml
+          flags: redis-compat
+          fail_ci_if_error: false
--- a/litellm/_redis.py
+++ b/litellm/_redis.py
@ -689,11 +689,12 @@ def init_redis_cluster(redis_kwargs) -> redis.RedisCluster:
    verbose_logger.debug("init_redis_cluster: startup nodes are being initialized.")
    from redis.cluster import ClusterNode

+    auth_kwargs: Final = _credential_provider_auth_kwargs(redis_kwargs)
    args: Final = _get_redis_cluster_kwargs()
    cluster_kwargs: Final = {}
-    for arg in redis_kwargs:
+    for arg in auth_kwargs:
        if arg in args:
-            cluster_kwargs[arg] = redis_kwargs[arg]
+            cluster_kwargs[arg] = auth_kwargs[arg]

    new_startup_nodes: Final[list[ClusterNode]] = []

@ -771,13 +772,13 @@ def _init_async_redis_sentinel(redis_kwargs) -> async_redis.Redis:
    return sentinel.master_for(service_name, **connection_kwargs)


-def _async_credential_provider(redis_connect_func: object | None) -> CredentialProvider | None:
-    """The Azure AD and GCP IAM connect funcs run their AUTH exchange with the blocking client
-    API, so on an async connection their ``send_command``/``read_response`` calls return
-    coroutines nobody awaits and every connect fails. Async paths authenticate through a
-    ``CredentialProvider`` instead, which redis-py consults per connection so the token stays
-    fresh. Any other ``redis_connect_func`` is left where it is, since redis-py awaits it
-    itself when it is a coroutine function."""
+def _credential_provider_from_connect_func(redis_connect_func: object | None) -> CredentialProvider | None:
+    """Translate IAM callbacks for paths that need credentials during the standard handshake.
+
+    Async connections cannot run blocking AUTH callbacks. Sync clusters authenticate before
+    invoking the callback, so they also need the provider during the initial handshake.
+    redis-py consults the provider for each connection, keeping token refresh intact.
+    """
    gcp_service_account: Final = getattr(redis_connect_func, "_gcp_service_account", None)
    if gcp_service_account is not None:
        return GCPIAMCredentialProvider(gcp_service_account)
@ -789,14 +790,13 @@ def _async_credential_provider(redis_connect_func: object | None) -> CredentialP
    return None


-def _async_auth_kwargs(redis_kwargs: dict) -> dict:
-    """Swaps a connect func an async path cannot run for the equivalent credential provider,
-    which supersedes any static username or password redis-py would otherwise reject it with."""
+def _credential_provider_auth_kwargs(redis_kwargs: dict) -> dict:
+    """Use a credential provider instead of an IAM callback and conflicting static credentials."""
    explicit_provider: Final = redis_kwargs.get("credential_provider")
    credential_provider: Final = (
        explicit_provider
        if explicit_provider is not None
-        else _async_credential_provider(redis_kwargs.get("redis_connect_func"))
+        else _credential_provider_from_connect_func(redis_kwargs.get("redis_connect_func"))
    )
    if credential_provider is None:
        return redis_kwargs
@ -834,7 +834,7 @@ def get_redis_async_client(
    connection_pool: async_redis.BlockingConnectionPool | None = None,
    **env_overrides,
 ) -> async_redis.Redis | async_redis.RedisCluster:
-    redis_kwargs: Final = _async_auth_kwargs(_get_redis_client_logic(**env_overrides))
+    redis_kwargs: Final = _credential_provider_auth_kwargs(_get_redis_client_logic(**env_overrides))

    if "startup_nodes" in redis_kwargs:
        from redis.cluster import ClusterNode
@ -906,7 +906,7 @@ def get_redis_async_client(
 def get_redis_connection_pool(
    **env_overrides,
 ) -> async_redis.BlockingConnectionPool | None:
-    redis_kwargs: Final = _async_auth_kwargs(_get_redis_client_logic(**env_overrides))
+    redis_kwargs: Final = _credential_provider_auth_kwargs(_get_redis_client_logic(**env_overrides))
    verbose_logger.debug("get_redis_connection_pool: redis_kwargs", redis_kwargs)

    if "startup_nodes" in redis_kwargs:
--- a/tests/local_testing/test_caching.py
+++ b/tests/local_testing/test_caching.py
@ -1,6 +1,17 @@
 import os
 import time
 import traceback
+import shutil
+import subprocess
+from collections.abc import Callable, Iterator
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Final
+
+import redis
+
+from litellm._redis import _get_redis_env_kwarg_mapping, get_redis_client
+from litellm._redis_credential_provider import _token_cache
 from litellm._uuid import uuid

 from dotenv import load_dotenv
@ -1032,6 +1043,102 @@ def test_redis_cache_completion_stream():
 # test_redis_cache_completion_stream()


+@pytest.fixture
+def clean_cluster_iam_environment(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
+    for var in ("REDIS_URL", "REDIS_CLUSTER_NODES", "REDIS_SENTINEL_NODES", *_get_redis_env_kwarg_mapping()):
+        monkeypatch.delenv(var, raising=False)
+    _token_cache.clear()
+    yield
+    _token_cache.clear()
+
+
+@pytest.fixture
+def authenticated_redis_cluster(tmp_path: Path, unused_tcp_port_factory: Callable[[], int]) -> Iterator[int]:
+    server: Final = shutil.which("redis-server")
+    if server is None:
+        pytest.skip("redis-server is required for the cluster authentication regression tests")
+    port: Final = unused_tcp_port_factory()
+    bus_port: Final = unused_tcp_port_factory()
+    log_path: Final = tmp_path / "redis.log"
+    config: Final = tmp_path / "redis.conf"
+    config.write_text(
+        f"bind 127.0.0.1\nport {port}\ncluster-port {bus_port}\n"
+        f'cluster-enabled yes\ncluster-config-file "{tmp_path / "nodes.conf"}"\n'
+        f'dir "{tmp_path}"\nsave ""\nappendonly no\n'
+    )
+    with log_path.open("w") as log:
+        process: Final = subprocess.Popen((server, str(config)), stdout=log, stderr=subprocess.STDOUT)
+        try:
+            with redis.Redis(host="127.0.0.1", port=port, socket_timeout=1, socket_connect_timeout=1) as admin:
+                for _ in range(100):
+                    try:
+                        admin.ping()
+                        break
+                    except redis.ConnectionError:
+                        time.sleep(0.1)
+                else:
+                    pytest.fail(f"Redis did not start: {log_path.read_text()}")
+                admin.execute_command("CLUSTER", "ADDSLOTS", *range(16384))
+                for _ in range(100):
+                    if admin.cluster("INFO")["cluster_state"] == "ok":
+                        break
+                    time.sleep(0.1)
+                else:
+                    pytest.fail(f"Redis cluster did not become ready: {log_path.read_text()}")
+                admin.execute_command(
+                    "ACL", "SETUSER", "identity-object-id", "on", ">local-fixture-token", "allcommands", "allkeys"
+                )
+                admin.execute_command("ACL", "SETUSER", "default", "resetpass", ">local-fixture-token")
+            yield port
+        finally:
+            process.terminate()
+            try:
+                process.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                process.kill()
+                process.wait()
+
+
+def test_sync_cluster_authenticates_with_azure_credentials(
+    clean_cluster_iam_environment: None, monkeypatch: pytest.MonkeyPatch, authenticated_redis_cluster: int
+) -> None:
+    monkeypatch.setenv("REDIS_USERNAME", "identity-object-id")
+    credential: Final = MagicMock()
+    credential.get_token.return_value = SimpleNamespace(token="local-fixture-token")
+
+    with patch("azure.identity.DefaultAzureCredential", return_value=credential):
+        with get_redis_client(
+            startup_nodes=[{"host": "127.0.0.1", "port": authenticated_redis_cluster}],
+            azure_redis_ad_token=True,
+            password="stale-password",
+            socket_timeout=1,
+            socket_connect_timeout=1,
+        ) as client:
+            assert client.ping() is True
+            assert client.set("iam-regression", "success") is True
+            assert client.get("iam-regression") == b"success"
+
+
+def test_sync_cluster_authenticates_with_gcp_credentials(
+    clean_cluster_iam_environment: None, authenticated_redis_cluster: int
+) -> None:
+    iam_client: Final = MagicMock()
+    iam_client.generate_access_token.return_value = SimpleNamespace(access_token="local-fixture-token")
+
+    with patch("google.cloud.iam_credentials_v1.IAMCredentialsClient", return_value=iam_client):
+        with get_redis_client(
+            startup_nodes=[{"host": "127.0.0.1", "port": authenticated_redis_cluster}],
+            gcp_service_account="projects/-/serviceAccounts/sa@project.iam.gserviceaccount.com",
+            username="stale-user",
+            password="stale-password",
+            socket_timeout=1,
+            socket_connect_timeout=1,
+        ) as client:
+            assert client.ping() is True
+            assert client.set("iam-regression", "success") is True
+            assert client.get("iam-regression") == b"success"
+
+
@pytest.mark.skip(reason="Local test. Requires running redis cluster locally.")
@pytest.mark.asyncio
 async def test_redis_cache_cluster_init_unit_test():
--- a/tests/test_litellm/test_assert_ci_coverage.py
+++ b/tests/test_litellm/test_assert_ci_coverage.py
@ -13,6 +13,7 @@ import sys
 from pathlib import Path
 from typing import Final

+import pytest
 import yaml

 _REPO_ROOT = Path(__file__).resolve().parents[2]
@ -350,8 +351,20 @@ def test_the_slice_check_credits_only_workflows_never_the_circleci_config():
    )


-def test_a_file_no_workflow_names_is_still_reported_when_every_slice_drops_it():
-    named = coverage._workflow_named_tokens()
-    assert not any(coverage._token_covers(token, "tests/local_testing/test_caching.py") for token in named), (
-        "test_caching.py is allowlisted, not run; crediting it would hide a real gap"
+@pytest.mark.parametrize("selector", ("test_selected.py", "test_selected.py::test_redis_auth"))
+def test_a_workflow_does_not_credit_a_file_it_never_names(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, selector: str
+) -> None:
+    workflows: Final = tmp_path / "workflows"
+    workflows.mkdir()
+    (workflows / "test.yml").write_text(
+        f"jobs:\n  test:\n    steps:\n      - run: uv run pytest tests/local_testing/{selector}\n"
+    )
+    monkeypatch.setattr(coverage, "WORKFLOW_DIR", workflows)
+    monkeypatch.setattr(coverage, "CIRCLECI_CONFIG", tmp_path / "circleci.yml")
+
+    named: Final = coverage._workflow_named_tokens()
+    assert named == frozenset({"tests/local_testing/test_selected.py"})
+    assert not any(
+        coverage._token_covers(token, "tests/local_testing/test_unrun.py") for token in named
    )
--- a/tests/test_litellm/test_redis.py
+++ b/tests/test_litellm/test_redis.py
@ -11,8 +11,8 @@ from redis.credentials import CredentialProvider
 import litellm
 from litellm._redis import (
    _AWS_IAM_KWARG_NAMES,
-    _async_auth_kwargs,
    _coerce_redis_kwargs_types,
+    _credential_provider_auth_kwargs,
    _get_redis_client_logic,
    _get_redis_cluster_kwargs,
    _get_redis_env_kwarg_mapping,
@ -273,6 +273,47 @@ def test_sync_cluster_preserves_credential_provider_identity(clean_redis_environ
    assert [(node.host, node.port) for node in cluster_kwargs["startup_nodes"]] == [("cluster-node", 6379)]


+def test_sync_cluster_authenticates_with_azure_credentials(clean_redis_environment, monkeypatch):
+    monkeypatch.setenv("REDIS_USERNAME", "identity-object-id")
+    credential = MagicMock()
+    credential.get_token.return_value = SimpleNamespace(token="azure-access-token")
+
+    with (
+        patch("azure.identity.DefaultAzureCredential", return_value=credential),
+        patch("redis.RedisCluster", autospec=True) as cluster,
+    ):
+        get_redis_client(
+            startup_nodes=[{"host": "cluster-node", "port": 6379}],
+            azure_redis_ad_token=True,
+            password="stale-password",
+        )
+
+    kwargs = cluster.call_args.kwargs
+    provider = kwargs.get("credential_provider")
+    assert isinstance(provider, AzureADCredentialProvider)
+    assert provider.get_credentials() == ("identity-object-id", "azure-access-token")
+    assert "username" not in kwargs
+    assert "password" not in kwargs
+    assert "redis_connect_func" not in kwargs
+    credential.get_token.assert_called_once_with("https://redis.azure.com/.default")
+
+
+def test_sync_cluster_authenticates_with_gcp_credentials(clean_redis_environment):
+    with patch("redis.RedisCluster", autospec=True) as cluster:
+        get_redis_client(
+            startup_nodes=[{"host": "cluster-node", "port": 6379}],
+            redis_connect_func=_gcp_marker_callback(),
+            username="stale-user",
+            password="stale-password",
+        )
+
+    kwargs = cluster.call_args.kwargs
+    assert isinstance(kwargs.get("credential_provider"), GCPIAMCredentialProvider)
+    assert "username" not in kwargs
+    assert "password" not in kwargs
+    assert "redis_connect_func" not in kwargs
+
+
 def test_async_cluster_preserves_credential_provider_identity(clean_redis_environment):
    provider = _StubCredentialProvider()
    startup_nodes = [{"host": "cluster-node", "port": 6379}]
@ -676,10 +717,10 @@ def test_provider_free_url_is_left_untouched(clean_redis_environment):
    assert redis_kwargs["url"] == url


-def test_async_auth_kwargs_supersedes_credentials_an_explicit_provider_replaces():
+def test_credential_provider_auth_kwargs_supersedes_credentials_an_explicit_provider_replaces():
    provider = _StubCredentialProvider()

-    auth_kwargs = _async_auth_kwargs(
+    auth_kwargs = _credential_provider_auth_kwargs(
        {
            "host": "redis-host",
            "port": 6379,
@ -698,10 +739,10 @@ def test_async_auth_kwargs_supersedes_credentials_an_explicit_provider_replaces(
    assert "password" not in auth_kwargs


-def test_async_auth_kwargs_leaves_provider_free_kwargs_alone():
+def test_credential_provider_auth_kwargs_leaves_provider_free_kwargs_alone():
    redis_kwargs = {"host": "redis-host", "port": 6379, "username": "url-user", "password": "url-pass"}

-    assert _async_auth_kwargs(redis_kwargs) == redis_kwargs
+    assert _credential_provider_auth_kwargs(redis_kwargs) == redis_kwargs


@pytest.mark.asyncio