From 7afd657c462eb988eb85ab46ad2eecc9a347a512 Mon Sep 17 00:00:00 2001
From: Sameer Kankute <sameer@berri.ai>
Date: Thu, 28 May 2026 11:45:17 +0530
Subject: [PATCH 01/15] fix(batches): skip unnecessary batch input file reads

Skip expensive pre-read of batch input files when no batch limits apply and model allowlist checks are not required, and decode model-embedded file IDs before file-content fetches to prevent upstream 404s.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 litellm/proxy/hooks/batch_rate_limiter.py     | 214 +++++++++++++++++-
 .../proxy/hooks/test_batch_file_validation.py | 106 +++++++++
 2 files changed, 315 insertions(+), 5 deletions(-)

diff --git a/litellm/proxy/hooks/batch_rate_limiter.py b/litellm/proxy/hooks/batch_rate_limiter.py
index f740d5dd40c4..cc5a1400a9d3 100644
--- a/litellm/proxy/hooks/batch_rate_limiter.py
+++ b/litellm/proxy/hooks/batch_rate_limiter.py
@@ -17,7 +17,7 @@
 - async_log_success_event() fires on GET /v1/batches/{id} (batch completion)
 """
 
-from typing import TYPE_CHECKING, Any, Dict, List, Literal, Optional, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Literal, Optional, Tuple, Union
 
 from fastapi import HTTPException
 from pydantic import BaseModel
@@ -25,12 +25,13 @@
 import litellm
 from litellm._logging import verbose_proxy_logger
 from litellm.batches.batch_utils import (
+    _extract_file_access_credentials,
     _get_batch_job_input_file_usage,
     _get_file_content_as_dictionary,
     _get_models_from_batch_input_file_content,
 )
 from litellm.integrations.custom_logger import CustomLogger
-from litellm.proxy._types import UserAPIKeyAuth
+from litellm.proxy._types import SpecialModelNames, UserAPIKeyAuth
 
 if TYPE_CHECKING:
     from opentelemetry.trace import Span as _Span
@@ -98,6 +99,189 @@ def __init__(
         self.internal_usage_cache = internal_usage_cache
         self.parallel_request_limiter = parallel_request_limiter
 
+    def _get_batch_routing_model(self, data: Dict) -> Optional[str]:
+        """Resolve the deployment/model used for this batch from request data."""
+        model = data.get("model")
+        if isinstance(model, str) and model:
+            return model
+
+        input_file_id = data.get("input_file_id")
+        if not isinstance(input_file_id, str) or not input_file_id:
+            return None
+
+        from litellm.proxy.openai_files_endpoints.common_utils import (
+            _is_base64_encoded_unified_file_id,
+            decode_model_from_file_id,
+            get_models_from_unified_file_id,
+        )
+
+        model_from_file_id = decode_model_from_file_id(input_file_id)
+        if model_from_file_id:
+            return model_from_file_id
+
+        unified_file_id = _is_base64_encoded_unified_file_id(input_file_id)
+        if unified_file_id:
+            target_model_names = get_models_from_unified_file_id(unified_file_id)
+            if target_model_names:
+                return target_model_names[0]
+
+        return None
+
+    def _matches_skip_list(self, value: str, skip_list: List[str]) -> bool:
+        if not skip_list:
+            return False
+        for entry in skip_list:
+            if not isinstance(entry, str) or not entry:
+                continue
+            if value == entry or value.startswith(f"{entry}/"):
+                return True
+        return False
+
+    def _should_skip_batch_input_file_processing(
+        self,
+        data: Dict,
+        user_api_key_dict: UserAPIKeyAuth,
+    ) -> bool:
+        """
+        Skip downloading batch input files when configured or when there is
+        nothing to enforce (no applicable rate limits and no model allowlist).
+        """
+        from litellm.proxy.proxy_server import general_settings
+
+        if general_settings.get("disable_batch_input_file_rate_limiting") is True:
+            return True
+
+        litellm_metadata = data.get("litellm_metadata") or {}
+        if litellm_metadata.get("skip_batch_input_file_rate_limiting") is True:
+            return True
+
+        batch_model = self._get_batch_routing_model(data)
+        skip_models = (
+            general_settings.get("skip_batch_input_file_rate_limiting_for_models") or []
+        )
+        if batch_model and self._matches_skip_list(batch_model, skip_models):
+            verbose_proxy_logger.debug(
+                f"Skipping batch input file processing for model={batch_model}"
+            )
+            return True
+
+        skip_providers = (
+            general_settings.get("skip_batch_input_file_rate_limiting_for_providers")
+            or []
+        )
+        custom_llm_provider = data.get("custom_llm_provider")
+        if (
+            isinstance(custom_llm_provider, str)
+            and custom_llm_provider in skip_providers
+        ):
+            verbose_proxy_logger.debug(
+                "Skipping batch input file processing for "
+                f"custom_llm_provider={custom_llm_provider}"
+            )
+            return True
+
+        if not self._key_requires_batch_model_access_check(user_api_key_dict):
+            descriptors = self.parallel_request_limiter._create_rate_limit_descriptors(
+                user_api_key_dict=user_api_key_dict,
+                data=data,
+                rpm_limit_type=None,
+                tpm_limit_type=None,
+                model_has_failures=False,
+            )
+            if not self._has_applicable_batch_rate_limits(descriptors):
+                verbose_proxy_logger.debug(
+                    "Skipping batch input file processing: no rate limits configured"
+                )
+                return True
+
+        return False
+
+    @staticmethod
+    def _key_requires_batch_model_access_check(
+        user_api_key_dict: UserAPIKeyAuth,
+    ) -> bool:
+        """True when the key may only call a subset of models (JSONL must be checked)."""
+        models = user_api_key_dict.models or []
+        if user_api_key_dict.access_group_ids:
+            return True
+        if not models:
+            return False
+        if "*" in models:
+            return False
+        if SpecialModelNames.all_proxy_models.value in models:
+            return False
+        return True
+
+    @staticmethod
+    def _has_applicable_batch_rate_limits(
+        descriptors: List["RateLimitDescriptor"],
+    ) -> bool:
+        for descriptor in descriptors:
+            rate_limit = descriptor.get("rate_limit") or {}
+            if (
+                rate_limit.get("requests_per_unit") is not None
+                or rate_limit.get("tokens_per_unit") is not None
+                or rate_limit.get("max_parallel_requests") is not None
+            ):
+                return True
+        return False
+
+    def _resolve_batch_input_file_fetch_params(
+        self,
+        file_id: str,
+        custom_llm_provider: str,
+        data: Dict,
+    ) -> Tuple[str, Dict[str, Any]]:
+        """
+        Map proxy-facing file IDs to provider file IDs and credentials.
+
+        Model-embedded IDs (``file-<base64>``) are not unified managed-file IDs;
+        without decoding them, ``afile_content`` is called with the encoded ID
+        and the upstream provider returns 404.
+        """
+        from litellm.proxy.openai_files_endpoints.common_utils import (
+            decode_model_from_file_id,
+            get_credentials_for_model,
+            get_original_file_id,
+        )
+        from litellm.proxy.proxy_server import llm_router
+
+        fetch_kwargs: Dict[str, Any] = {
+            "custom_llm_provider": custom_llm_provider,
+        }
+
+        model_from_file_id = decode_model_from_file_id(file_id)
+        if model_from_file_id:
+            credentials = get_credentials_for_model(
+                llm_router=llm_router,
+                model_id=model_from_file_id,
+                operation_context="batch input file read (rate limiting)",
+            )
+            fetch_kwargs.update(_extract_file_access_credentials(credentials))
+            fetch_kwargs["model"] = model_from_file_id
+            provider = credentials.get("custom_llm_provider")
+            if provider:
+                fetch_kwargs["custom_llm_provider"] = provider
+            return get_original_file_id(file_id), fetch_kwargs
+
+        request_model = data.get("model")
+        if isinstance(request_model, str) and request_model and llm_router is not None:
+            try:
+                credentials = get_credentials_for_model(
+                    llm_router=llm_router,
+                    model_id=request_model,
+                    operation_context="batch input file read (rate limiting)",
+                )
+                fetch_kwargs.update(_extract_file_access_credentials(credentials))
+                fetch_kwargs["model"] = request_model
+                provider = credentials.get("custom_llm_provider")
+                if provider:
+                    fetch_kwargs["custom_llm_provider"] = provider
+            except HTTPException:
+                pass
+
+        return file_id, fetch_kwargs
+
     def _raise_rate_limit_error(
         self,
         status: "RateLimitStatus",
@@ -211,6 +395,7 @@ async def count_input_file_usage(
         file_id: str,
         custom_llm_provider: Literal["openai", "azure", "vertex_ai"] = "openai",
         user_api_key_dict: Optional[UserAPIKeyAuth] = None,
+        data: Optional[Dict] = None,
     ) -> BatchFileUsage:
         """
         Count number of requests and tokens in a batch input file.
@@ -238,14 +423,27 @@ async def count_input_file_usage(
                     user_api_key_dict=user_api_key_dict,
                 )
             else:
+                provider_file_id, fetch_kwargs = (
+                    self._resolve_batch_input_file_fetch_params(
+                        file_id=file_id,
+                        custom_llm_provider=custom_llm_provider,
+                        data=data or {},
+                    )
+                )
                 # For non-managed files, use the standard litellm.afile_content
                 file_content = await litellm.afile_content(
-                    file_id=file_id,
-                    custom_llm_provider=custom_llm_provider,
+                    file_id=provider_file_id,
                     user_api_key_dict=user_api_key_dict,
+                    **fetch_kwargs,
                 )
 
-            file_content_as_dict = _get_file_content_as_dictionary(file_content.content)
+            file_content_bytes = getattr(file_content, "content", None)
+            if not isinstance(file_content_bytes, bytes):
+                raise ValueError(
+                    f"Expected bytes content from file retrieval for {file_id}, "
+                    f"got {type(file_content_bytes)}"
+                )
+            file_content_as_dict = _get_file_content_as_dictionary(file_content_bytes)
 
             # Validate every model named in the batch JSONL against the
             # caller's per-key model allowlist. Without this, a caller
@@ -435,6 +633,11 @@ async def async_pre_call_hook(
                 )
                 return data
 
+            if self._should_skip_batch_input_file_processing(
+                data=data, user_api_key_dict=user_api_key_dict
+            ):
+                return data
+
             # Get custom_llm_provider for token counting
             custom_llm_provider = data.get("custom_llm_provider", "openai")
 
@@ -446,6 +649,7 @@ async def async_pre_call_hook(
                 file_id=input_file_id,
                 custom_llm_provider=custom_llm_provider,
                 user_api_key_dict=user_api_key_dict,
+                data=data,
             )
 
             verbose_proxy_logger.debug(
diff --git a/tests/test_litellm/proxy/hooks/test_batch_file_validation.py b/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
index 7f1006543bbb..4a1fe1a37dd0 100644
--- a/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
+++ b/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
@@ -260,6 +260,112 @@ async def test_pre_call_allows_authorized_model_in_batch_file():
         )
 
 
+@pytest.mark.asyncio
+async def test_pre_call_skips_file_fetch_when_disabled_in_general_settings():
+    from litellm.proxy.hooks.batch_rate_limiter import _PROXY_BatchRateLimiter
+
+    rate_limiter = _PROXY_BatchRateLimiter(
+        internal_usage_cache=MagicMock(),
+        parallel_request_limiter=MagicMock(),
+    )
+    user = UserAPIKeyAuth(api_key="sk-ok", user_id="alice", models=["gpt-4o"])
+
+    with patch(
+        "litellm.proxy.proxy_server.general_settings",
+        {"disable_batch_input_file_rate_limiting": True},
+    ):
+        result = await rate_limiter.async_pre_call_hook(
+            user_api_key_dict=user,
+            cache=MagicMock(),
+            data={"input_file_id": "file-abc123"},
+            call_type="acreate_batch",
+        )
+
+    assert result == {"input_file_id": "file-abc123"}
+    rate_limiter.parallel_request_limiter._create_rate_limit_descriptors.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_pre_call_skips_file_fetch_for_configured_provider():
+    from litellm.proxy.hooks.batch_rate_limiter import _PROXY_BatchRateLimiter
+
+    rate_limiter = _PROXY_BatchRateLimiter(
+        internal_usage_cache=MagicMock(),
+        parallel_request_limiter=MagicMock(),
+    )
+    user = UserAPIKeyAuth(api_key="sk-ok", user_id="alice", models=["gpt-4o"])
+
+    with patch(
+        "litellm.proxy.proxy_server.general_settings",
+        {"skip_batch_input_file_rate_limiting_for_providers": ["hosted_vllm"]},
+    ):
+        result = await rate_limiter.async_pre_call_hook(
+            user_api_key_dict=user,
+            cache=MagicMock(),
+            data={
+                "input_file_id": "file-abc123",
+                "custom_llm_provider": "hosted_vllm",
+            },
+            call_type="acreate_batch",
+        )
+
+    assert result["custom_llm_provider"] == "hosted_vllm"
+
+
+@pytest.mark.asyncio
+async def test_count_input_file_usage_decodes_model_embedded_file_id():
+    import base64
+
+    from litellm.proxy.hooks.batch_rate_limiter import _PROXY_BatchRateLimiter
+
+    original_file_id = "file-provider-xyz"
+    encoded_payload = (
+        base64.urlsafe_b64encode(
+            f"litellm:{original_file_id};model,my-vllm-batch".encode()
+        )
+        .decode()
+        .rstrip("=")
+    )
+    encoded_file_id = f"file-{encoded_payload}"
+
+    rate_limiter = _PROXY_BatchRateLimiter(
+        internal_usage_cache=MagicMock(),
+        parallel_request_limiter=MagicMock(),
+    )
+
+    mock_content = MagicMock()
+    mock_content.content = b'{"custom_id": "1", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "my-vllm-batch", "messages": [{"role": "user", "content": "hi"}]}}\n'
+
+    with (
+        patch(
+            "litellm.afile_content",
+            new=AsyncMock(return_value=mock_content),
+        ) as mock_afile_content,
+        patch(
+            "litellm.proxy.proxy_server.llm_router",
+            MagicMock(),
+        ),
+        patch(
+            "litellm.proxy.openai_files_endpoints.common_utils.get_credentials_for_model",
+            return_value={
+                "api_key": "test-key",
+                "api_base": "http://vllm:8000/v1",
+                "custom_llm_provider": "hosted_vllm",
+            },
+        ),
+    ):
+        await rate_limiter.count_input_file_usage(
+            file_id=encoded_file_id,
+            custom_llm_provider="openai",
+            user_api_key_dict=UserAPIKeyAuth(api_key="sk-ok", user_id="alice"),
+            data={},
+        )
+
+    mock_afile_content.assert_awaited_once()
+    assert mock_afile_content.await_args.kwargs["file_id"] == original_file_id
+    assert mock_afile_content.await_args.kwargs["custom_llm_provider"] == "hosted_vllm"
+
+
 @pytest.mark.asyncio
 async def test_pre_call_skips_check_when_no_models_present():
     """Files without any `body.model` (corrupt or empty) must not 500;

From b391f772ade4fb7f930e6ed2cfd7425c579fd016 Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Thu, 28 May 2026 06:23:21 +0000
Subject: [PATCH 02/15] fix(batch-rate-limiter): prevent user metadata flag
 from bypassing model allowlist

The skip_batch_input_file_rate_limiting flag in litellm_metadata is
user-controllable for batch requests (request-body metadata lands in
litellm_metadata via LITELLM_METADATA_ROUTES). Honoring it
unconditionally also skipped _enforce_batch_file_model_access, letting
a restricted key submit a JSONL referencing models outside its
allowlist. Only honor the metadata-based skip when the key has no
model allowlist to enforce.

Co-authored-by: Yassin Kortam <yassin@berri.ai>
---
 litellm/proxy/hooks/batch_rate_limiter.py | 11 ++++++++++-
 1 file changed, 10 insertions(+), 1 deletion(-)

diff --git a/litellm/proxy/hooks/batch_rate_limiter.py b/litellm/proxy/hooks/batch_rate_limiter.py
index cc5a1400a9d3..6b8dca3e8c4c 100644
--- a/litellm/proxy/hooks/batch_rate_limiter.py
+++ b/litellm/proxy/hooks/batch_rate_limiter.py
@@ -151,8 +151,17 @@ def _should_skip_batch_input_file_processing(
         if general_settings.get("disable_batch_input_file_rate_limiting") is True:
             return True
 
+        # Only honor the metadata-based skip when the key has no model
+        # allowlist to enforce. Otherwise a caller could set this flag in
+        # the request body (it lands in ``litellm_metadata`` for batch
+        # routes) and skip ``_enforce_batch_file_model_access``, smuggling
+        # restricted models into the JSONL.
         litellm_metadata = data.get("litellm_metadata") or {}
-        if litellm_metadata.get("skip_batch_input_file_rate_limiting") is True:
+        if litellm_metadata.get(
+            "skip_batch_input_file_rate_limiting"
+        ) is True and not self._key_requires_batch_model_access_check(
+            user_api_key_dict
+        ):
             return True
 
         batch_model = self._get_batch_routing_model(data)

From 4f9ceddd5210e22474e3a919c4f62be04bdca60d Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Thu, 28 May 2026 06:35:11 +0000
Subject: [PATCH 03/15] fix(batch_rate_limiter): enforce model access check
 before honoring skip paths

Admin-configured skips (disable_batch_input_file_rate_limiting,
skip_batch_input_file_rate_limiting_for_models/_for_providers) and the
no-applicable-rate-limits short-circuit previously bypassed
_enforce_batch_file_model_access. A key with a restricted model
allowlist could therefore submit a batch JSONL referencing models
outside its allowlist whenever any of these skip paths fired, and the
provider-skip path was attacker-controllable via the request body's
custom_llm_provider field. Hoist the model-access guard to the top so
restricted keys always have their JSONL validated regardless of which
skip would otherwise apply.

Co-authored-by: Yassin Kortam <yassin@berri.ai>
---
 litellm/proxy/hooks/batch_rate_limiter.py     | 45 ++++++++++---------
 .../proxy/hooks/test_batch_file_validation.py |  5 +--
 2 files changed, 25 insertions(+), 25 deletions(-)

diff --git a/litellm/proxy/hooks/batch_rate_limiter.py b/litellm/proxy/hooks/batch_rate_limiter.py
index 6b8dca3e8c4c..38d4209499ee 100644
--- a/litellm/proxy/hooks/batch_rate_limiter.py
+++ b/litellm/proxy/hooks/batch_rate_limiter.py
@@ -145,23 +145,25 @@ def _should_skip_batch_input_file_processing(
         """
         Skip downloading batch input files when configured or when there is
         nothing to enforce (no applicable rate limits and no model allowlist).
+
+        When the caller's key has a model allowlist to enforce, no skip path
+        is honored: the JSONL must still be downloaded so
+        ``_enforce_batch_file_model_access`` can validate every ``body.model``
+        entry. Otherwise a restricted key could smuggle unauthorized models
+        into the file via an admin-configured skip (global disable,
+        per-model, per-provider) or via the user-controlled
+        ``custom_llm_provider`` field.
         """
+        if self._key_requires_batch_model_access_check(user_api_key_dict):
+            return False
+
         from litellm.proxy.proxy_server import general_settings
 
         if general_settings.get("disable_batch_input_file_rate_limiting") is True:
             return True
 
-        # Only honor the metadata-based skip when the key has no model
-        # allowlist to enforce. Otherwise a caller could set this flag in
-        # the request body (it lands in ``litellm_metadata`` for batch
-        # routes) and skip ``_enforce_batch_file_model_access``, smuggling
-        # restricted models into the JSONL.
         litellm_metadata = data.get("litellm_metadata") or {}
-        if litellm_metadata.get(
-            "skip_batch_input_file_rate_limiting"
-        ) is True and not self._key_requires_batch_model_access_check(
-            user_api_key_dict
-        ):
+        if litellm_metadata.get("skip_batch_input_file_rate_limiting") is True:
             return True
 
         batch_model = self._get_batch_routing_model(data)
@@ -189,19 +191,18 @@ def _should_skip_batch_input_file_processing(
             )
             return True
 
-        if not self._key_requires_batch_model_access_check(user_api_key_dict):
-            descriptors = self.parallel_request_limiter._create_rate_limit_descriptors(
-                user_api_key_dict=user_api_key_dict,
-                data=data,
-                rpm_limit_type=None,
-                tpm_limit_type=None,
-                model_has_failures=False,
+        descriptors = self.parallel_request_limiter._create_rate_limit_descriptors(
+            user_api_key_dict=user_api_key_dict,
+            data=data,
+            rpm_limit_type=None,
+            tpm_limit_type=None,
+            model_has_failures=False,
+        )
+        if not self._has_applicable_batch_rate_limits(descriptors):
+            verbose_proxy_logger.debug(
+                "Skipping batch input file processing: no rate limits configured"
             )
-            if not self._has_applicable_batch_rate_limits(descriptors):
-                verbose_proxy_logger.debug(
-                    "Skipping batch input file processing: no rate limits configured"
-                )
-                return True
+            return True
 
         return False
 
diff --git a/tests/test_litellm/proxy/hooks/test_batch_file_validation.py b/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
index 4a1fe1a37dd0..fdaaee381f7e 100644
--- a/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
+++ b/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
@@ -14,7 +14,6 @@
 
 from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
 
-
 # ---------------------------------------------------------------------------
 # Token counter — covers all three batch payload shapes
 # ---------------------------------------------------------------------------
@@ -268,7 +267,7 @@ async def test_pre_call_skips_file_fetch_when_disabled_in_general_settings():
         internal_usage_cache=MagicMock(),
         parallel_request_limiter=MagicMock(),
     )
-    user = UserAPIKeyAuth(api_key="sk-ok", user_id="alice", models=["gpt-4o"])
+    user = UserAPIKeyAuth(api_key="sk-ok", user_id="alice", models=["*"])
 
     with patch(
         "litellm.proxy.proxy_server.general_settings",
@@ -293,7 +292,7 @@ async def test_pre_call_skips_file_fetch_for_configured_provider():
         internal_usage_cache=MagicMock(),
         parallel_request_limiter=MagicMock(),
     )
-    user = UserAPIKeyAuth(api_key="sk-ok", user_id="alice", models=["gpt-4o"])
+    user = UserAPIKeyAuth(api_key="sk-ok", user_id="alice", models=["*"])
 
     with patch(
         "litellm.proxy.proxy_server.general_settings",

From f41ec3e2ab3f6347aaf84898e872469100ae4cf7 Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Thu, 28 May 2026 06:44:13 +0000
Subject: [PATCH 04/15] fix(batch_rate_limiter): wildcard model bypass +
 fail-open embedded model creds

- _key_requires_batch_model_access_check: check '*' / all-proxy-models
  before access_group_ids so wildcard keys skip the JSONL download.
- _resolve_batch_input_file_fetch_params: wrap embedded-model
  get_credentials_for_model in try/except HTTPException, mirroring the
  request-model fallback path, and always decode the file id.

Co-authored-by: Yassin Kortam <yassin@berri.ai>
---
 litellm/proxy/hooks/batch_rate_limiter.py | 31 +++++++++++++----------
 1 file changed, 17 insertions(+), 14 deletions(-)

diff --git a/litellm/proxy/hooks/batch_rate_limiter.py b/litellm/proxy/hooks/batch_rate_limiter.py
index 38d4209499ee..2ab42e5d5170 100644
--- a/litellm/proxy/hooks/batch_rate_limiter.py
+++ b/litellm/proxy/hooks/batch_rate_limiter.py
@@ -212,14 +212,14 @@ def _key_requires_batch_model_access_check(
     ) -> bool:
         """True when the key may only call a subset of models (JSONL must be checked)."""
         models = user_api_key_dict.models or []
-        if user_api_key_dict.access_group_ids:
-            return True
-        if not models:
-            return False
         if "*" in models:
             return False
         if SpecialModelNames.all_proxy_models.value in models:
             return False
+        if user_api_key_dict.access_group_ids:
+            return True
+        if not models:
+            return False
         return True
 
     @staticmethod
@@ -262,16 +262,19 @@ def _resolve_batch_input_file_fetch_params(
 
         model_from_file_id = decode_model_from_file_id(file_id)
         if model_from_file_id:
-            credentials = get_credentials_for_model(
-                llm_router=llm_router,
-                model_id=model_from_file_id,
-                operation_context="batch input file read (rate limiting)",
-            )
-            fetch_kwargs.update(_extract_file_access_credentials(credentials))
-            fetch_kwargs["model"] = model_from_file_id
-            provider = credentials.get("custom_llm_provider")
-            if provider:
-                fetch_kwargs["custom_llm_provider"] = provider
+            try:
+                credentials = get_credentials_for_model(
+                    llm_router=llm_router,
+                    model_id=model_from_file_id,
+                    operation_context="batch input file read (rate limiting)",
+                )
+                fetch_kwargs.update(_extract_file_access_credentials(credentials))
+                fetch_kwargs["model"] = model_from_file_id
+                provider = credentials.get("custom_llm_provider")
+                if provider:
+                    fetch_kwargs["custom_llm_provider"] = provider
+            except HTTPException:
+                pass
             return get_original_file_id(file_id), fetch_kwargs
 
         request_model = data.get("model")

From c976d844bd4195a3dd87570f1e16283bf837511e Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Sat, 30 May 2026 02:44:40 +0000
Subject: [PATCH 05/15] perf(batch_rate_limiter): reuse rate-limit descriptors
 across skip check and counter increment

---
 litellm/proxy/hooks/batch_rate_limiter.py | 64 +++++++++++++++--------
 1 file changed, 42 insertions(+), 22 deletions(-)

diff --git a/litellm/proxy/hooks/batch_rate_limiter.py b/litellm/proxy/hooks/batch_rate_limiter.py
index 2ab42e5d5170..40c7f3dfa556 100644
--- a/litellm/proxy/hooks/batch_rate_limiter.py
+++ b/litellm/proxy/hooks/batch_rate_limiter.py
@@ -137,11 +137,24 @@ def _matches_skip_list(self, value: str, skip_list: List[str]) -> bool:
                 return True
         return False
 
+    def _create_batch_rate_limit_descriptors(
+        self,
+        user_api_key_dict: UserAPIKeyAuth,
+        data: Dict,
+    ) -> List["RateLimitDescriptor"]:
+        return self.parallel_request_limiter._create_rate_limit_descriptors(
+            user_api_key_dict=user_api_key_dict,
+            data=data,
+            rpm_limit_type=None,
+            tpm_limit_type=None,
+            model_has_failures=False,
+        )
+
     def _should_skip_batch_input_file_processing(
         self,
         data: Dict,
         user_api_key_dict: UserAPIKeyAuth,
-    ) -> bool:
+    ) -> Tuple[bool, Optional[List["RateLimitDescriptor"]]]:
         """
         Skip downloading batch input files when configured or when there is
         nothing to enforce (no applicable rate limits and no model allowlist).
@@ -153,18 +166,22 @@ def _should_skip_batch_input_file_processing(
         into the file via an admin-configured skip (global disable,
         per-model, per-provider) or via the user-controlled
         ``custom_llm_provider`` field.
+
+        Returns ``(should_skip, descriptors)`` where ``descriptors`` is the
+        rate-limit descriptor list computed for the no-limits check, so the
+        caller can reuse it for counter enforcement without recomputing.
         """
         if self._key_requires_batch_model_access_check(user_api_key_dict):
-            return False
+            return False, None
 
         from litellm.proxy.proxy_server import general_settings
 
         if general_settings.get("disable_batch_input_file_rate_limiting") is True:
-            return True
+            return True, None
 
         litellm_metadata = data.get("litellm_metadata") or {}
         if litellm_metadata.get("skip_batch_input_file_rate_limiting") is True:
-            return True
+            return True, None
 
         batch_model = self._get_batch_routing_model(data)
         skip_models = (
@@ -174,7 +191,7 @@ def _should_skip_batch_input_file_processing(
             verbose_proxy_logger.debug(
                 f"Skipping batch input file processing for model={batch_model}"
             )
-            return True
+            return True, None
 
         skip_providers = (
             general_settings.get("skip_batch_input_file_rate_limiting_for_providers")
@@ -189,22 +206,19 @@ def _should_skip_batch_input_file_processing(
                 "Skipping batch input file processing for "
                 f"custom_llm_provider={custom_llm_provider}"
             )
-            return True
+            return True, None
 
-        descriptors = self.parallel_request_limiter._create_rate_limit_descriptors(
+        descriptors = self._create_batch_rate_limit_descriptors(
             user_api_key_dict=user_api_key_dict,
             data=data,
-            rpm_limit_type=None,
-            tpm_limit_type=None,
-            model_has_failures=False,
         )
         if not self._has_applicable_batch_rate_limits(descriptors):
             verbose_proxy_logger.debug(
                 "Skipping batch input file processing: no rate limits configured"
             )
-            return True
+            return True, None
 
-        return False
+        return False, descriptors
 
     @staticmethod
     def _key_requires_batch_model_access_check(
@@ -360,6 +374,7 @@ async def _check_and_increment_batch_counters(
         user_api_key_dict: UserAPIKeyAuth,
         data: Dict,
         batch_usage: BatchFileUsage,
+        descriptors: Optional[List["RateLimitDescriptor"]] = None,
     ) -> None:
         """
         Atomically check + increment rate-limit counters by the batch amounts.
@@ -368,14 +383,15 @@ async def _check_and_increment_batch_counters(
         case no counter is modified. Backed by `atomic_check_and_increment_by_n`
         which uses a Redis Lua script when available (multi-process atomic) and
         falls back to a per-process asyncio.Lock + in-memory operation.
+
+        ``descriptors`` may be passed in by the pre-call hook to reuse the list
+        already computed when deciding whether to skip file processing.
         """
-        descriptors = self.parallel_request_limiter._create_rate_limit_descriptors(
-            user_api_key_dict=user_api_key_dict,
-            data=data,
-            rpm_limit_type=None,
-            tpm_limit_type=None,
-            model_has_failures=False,
-        )
+        if descriptors is None:
+            descriptors = self._create_batch_rate_limit_descriptors(
+                user_api_key_dict=user_api_key_dict,
+                data=data,
+            )
 
         increment: Dict[Literal["requests", "tokens"], int] = {
             "requests": batch_usage.request_count,
@@ -646,9 +662,12 @@ async def async_pre_call_hook(
                 )
                 return data
 
-            if self._should_skip_batch_input_file_processing(
-                data=data, user_api_key_dict=user_api_key_dict
-            ):
+            should_skip, batch_rate_limit_descriptors = (
+                self._should_skip_batch_input_file_processing(
+                    data=data, user_api_key_dict=user_api_key_dict
+                )
+            )
+            if should_skip:
                 return data
 
             # Get custom_llm_provider for token counting
@@ -680,6 +699,7 @@ async def async_pre_call_hook(
                 user_api_key_dict=user_api_key_dict,
                 data=data,
                 batch_usage=batch_usage,
+                descriptors=batch_rate_limit_descriptors,
             )
 
             verbose_proxy_logger.debug(

From 82c1a362b93d6e270ee0b98da0eb62cec7711ea1 Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Sat, 30 May 2026 03:07:00 +0000
Subject: [PATCH 06/15] test(batch_rate_limiter): cover skip-path and
 file-fetch helpers

Add unit tests for the batch rate limiter's new skip/routing helpers so
the diff's patch coverage no longer depends on the CircleCI batches job,
whose coverage upload is blocked when an unrelated Bedrock integration
test aborts the run. Covers _get_batch_routing_model, _matches_skip_list,
_key_requires_batch_model_access_check, _has_applicable_batch_rate_limits,
_should_skip_batch_input_file_processing, _resolve_batch_input_file_fetch_params,
the descriptor-reuse path of _check_and_increment_batch_counters, and the
non-bytes file content guard in count_input_file_usage.
---
 .../proxy/hooks/test_batch_file_validation.py | 295 ++++++++++++++++++
 1 file changed, 295 insertions(+)

diff --git a/tests/test_litellm/proxy/hooks/test_batch_file_validation.py b/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
index 9c38e074e63c..515be9df87c2 100644
--- a/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
+++ b/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
@@ -429,3 +429,298 @@ async def test_pre_call_skips_check_when_no_models_present():
         user_api_key_dict=user,
         file_content_as_dict=[{"body": {}}],
     )
+
+
+# ---------------------------------------------------------------------------
+# Skip-path helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_rate_limiter():
+    from litellm.proxy.hooks.batch_rate_limiter import _PROXY_BatchRateLimiter
+
+    return _PROXY_BatchRateLimiter(
+        internal_usage_cache=MagicMock(),
+        parallel_request_limiter=MagicMock(),
+    )
+
+
+def test_get_batch_routing_model_prefers_request_model():
+    rate_limiter = _make_rate_limiter()
+    assert (
+        rate_limiter._get_batch_routing_model({"model": "gpt-4o-mini"}) == "gpt-4o-mini"
+    )
+
+
+def test_get_batch_routing_model_returns_none_without_model_or_file():
+    rate_limiter = _make_rate_limiter()
+    assert rate_limiter._get_batch_routing_model({}) is None
+    assert rate_limiter._get_batch_routing_model({"input_file_id": ""}) is None
+
+
+def test_get_batch_routing_model_decodes_model_embedded_file_id():
+    import base64
+
+    rate_limiter = _make_rate_limiter()
+    encoded = (
+        base64.urlsafe_b64encode(b"litellm:file-xyz;model,vllm-batch")
+        .decode()
+        .rstrip("=")
+    )
+    assert (
+        rate_limiter._get_batch_routing_model({"input_file_id": f"file-{encoded}"})
+        == "vllm-batch"
+    )
+
+
+def test_get_batch_routing_model_uses_unified_file_id_target():
+    rate_limiter = _make_rate_limiter()
+    with (
+        patch(
+            "litellm.proxy.openai_files_endpoints.common_utils.decode_model_from_file_id",
+            return_value=None,
+        ),
+        patch(
+            "litellm.proxy.openai_files_endpoints.common_utils._is_base64_encoded_unified_file_id",
+            return_value="unified-id",
+        ),
+        patch(
+            "litellm.proxy.openai_files_endpoints.common_utils.get_models_from_unified_file_id",
+            return_value=["model-a", "model-b"],
+        ),
+    ):
+        assert (
+            rate_limiter._get_batch_routing_model({"input_file_id": "file-managed"})
+            == "model-a"
+        )
+
+
+def test_matches_skip_list_handles_empty_and_entry_shapes():
+    rate_limiter = _make_rate_limiter()
+    assert rate_limiter._matches_skip_list("gpt-4o", []) is False
+    assert rate_limiter._matches_skip_list("gpt-4o", ["gpt-4o"]) is True
+    assert rate_limiter._matches_skip_list("vertex_ai/gemini", ["vertex_ai"]) is True
+    assert rate_limiter._matches_skip_list("gpt-4o", [None, "", "claude"]) is False
+
+
+def test_key_requires_batch_model_access_check_branches():
+    from litellm.proxy.hooks.batch_rate_limiter import _PROXY_BatchRateLimiter
+
+    check = _PROXY_BatchRateLimiter._key_requires_batch_model_access_check
+    assert check(UserAPIKeyAuth(api_key="sk", models=["*"])) is False
+    assert check(UserAPIKeyAuth(api_key="sk", models=["all-proxy-models"])) is False
+    assert (
+        check(UserAPIKeyAuth(api_key="sk", models=[], access_group_ids=["grp"])) is True
+    )
+    assert check(UserAPIKeyAuth(api_key="sk", models=[])) is False
+    assert check(UserAPIKeyAuth(api_key="sk", models=["gpt-4o-mini"])) is True
+
+
+def test_has_applicable_batch_rate_limits():
+    from litellm.proxy.hooks.batch_rate_limiter import _PROXY_BatchRateLimiter
+
+    has_limits = _PROXY_BatchRateLimiter._has_applicable_batch_rate_limits
+    assert has_limits([{"rate_limit": {"tokens_per_unit": 100}}]) is True
+    assert has_limits([{"rate_limit": {"requests_per_unit": 5}}]) is True
+    assert has_limits([{"rate_limit": {"max_parallel_requests": 2}}]) is True
+    assert has_limits([{"rate_limit": {}}, {}]) is False
+
+
+def test_should_skip_returns_false_when_key_needs_model_access_check():
+    rate_limiter = _make_rate_limiter()
+    user = UserAPIKeyAuth(api_key="sk", models=["gpt-4o-mini"])
+    should_skip, descriptors = rate_limiter._should_skip_batch_input_file_processing(
+        data={"input_file_id": "file-abc"}, user_api_key_dict=user
+    )
+    assert should_skip is False
+    assert descriptors is None
+
+
+def test_should_skip_honors_litellm_metadata_flag():
+    rate_limiter = _make_rate_limiter()
+    user = UserAPIKeyAuth(api_key="sk", models=["*"])
+    with patch("litellm.proxy.proxy_server.general_settings", {}):
+        should_skip, descriptors = (
+            rate_limiter._should_skip_batch_input_file_processing(
+                data={
+                    "input_file_id": "file-abc",
+                    "litellm_metadata": {"skip_batch_input_file_rate_limiting": True},
+                },
+                user_api_key_dict=user,
+            )
+        )
+    assert should_skip is True
+    assert descriptors is None
+
+
+def test_should_skip_honors_per_model_skip_list():
+    rate_limiter = _make_rate_limiter()
+    user = UserAPIKeyAuth(api_key="sk", models=["*"])
+    with patch(
+        "litellm.proxy.proxy_server.general_settings",
+        {"skip_batch_input_file_rate_limiting_for_models": ["gpt-4o-mini"]},
+    ):
+        should_skip, descriptors = (
+            rate_limiter._should_skip_batch_input_file_processing(
+                data={"model": "gpt-4o-mini", "input_file_id": "file-abc"},
+                user_api_key_dict=user,
+            )
+        )
+    assert should_skip is True
+    assert descriptors is None
+
+
+def test_should_skip_when_no_rate_limits_configured():
+    rate_limiter = _make_rate_limiter()
+    rate_limiter.parallel_request_limiter._create_rate_limit_descriptors.return_value = [
+        {"rate_limit": {}}
+    ]
+    user = UserAPIKeyAuth(api_key="sk", models=["*"])
+    with patch("litellm.proxy.proxy_server.general_settings", {}):
+        should_skip, descriptors = (
+            rate_limiter._should_skip_batch_input_file_processing(
+                data={"model": "gpt-4o-mini", "input_file_id": "file-abc"},
+                user_api_key_dict=user,
+            )
+        )
+    assert should_skip is True
+    assert descriptors is None
+
+
+def test_should_not_skip_and_reuses_descriptors_when_limits_present():
+    rate_limiter = _make_rate_limiter()
+    descriptors = [{"rate_limit": {"tokens_per_unit": 100}}]
+    rate_limiter.parallel_request_limiter._create_rate_limit_descriptors.return_value = (
+        descriptors
+    )
+    user = UserAPIKeyAuth(api_key="sk", models=["*"])
+    with patch("litellm.proxy.proxy_server.general_settings", {}):
+        should_skip, returned = rate_limiter._should_skip_batch_input_file_processing(
+            data={"model": "gpt-4o-mini", "input_file_id": "file-abc"},
+            user_api_key_dict=user,
+        )
+    assert should_skip is False
+    assert returned is descriptors
+
+
+def test_resolve_fetch_params_uses_request_model_credentials():
+    rate_limiter = _make_rate_limiter()
+    with (
+        patch("litellm.proxy.proxy_server.llm_router", MagicMock()),
+        patch(
+            "litellm.proxy.openai_files_endpoints.common_utils.get_credentials_for_model",
+            return_value={
+                "api_key": "k",
+                "api_base": "http://vllm:8000/v1",
+                "custom_llm_provider": "hosted_vllm",
+            },
+        ),
+    ):
+        provider_file_id, fetch_kwargs = (
+            rate_limiter._resolve_batch_input_file_fetch_params(
+                file_id="file-plain-openai",
+                custom_llm_provider="openai",
+                data={"model": "my-vllm-batch"},
+            )
+        )
+    assert provider_file_id == "file-plain-openai"
+    assert fetch_kwargs["model"] == "my-vllm-batch"
+    assert fetch_kwargs["custom_llm_provider"] == "hosted_vllm"
+    assert fetch_kwargs["api_base"] == "http://vllm:8000/v1"
+
+
+def test_resolve_fetch_params_fails_open_on_credential_lookup_error():
+    rate_limiter = _make_rate_limiter()
+    with (
+        patch("litellm.proxy.proxy_server.llm_router", MagicMock()),
+        patch(
+            "litellm.proxy.openai_files_endpoints.common_utils.get_credentials_for_model",
+            side_effect=HTTPException(status_code=404, detail="no creds"),
+        ),
+    ):
+        provider_file_id, fetch_kwargs = (
+            rate_limiter._resolve_batch_input_file_fetch_params(
+                file_id="file-plain-openai",
+                custom_llm_provider="openai",
+                data={"model": "my-vllm-batch"},
+            )
+        )
+    assert provider_file_id == "file-plain-openai"
+    assert fetch_kwargs == {"custom_llm_provider": "openai"}
+
+
+def test_resolve_fetch_params_model_embedded_fails_open_on_credential_error():
+    import base64
+
+    rate_limiter = _make_rate_limiter()
+    encoded = (
+        base64.urlsafe_b64encode(b"litellm:file-orig;model,vllm-batch")
+        .decode()
+        .rstrip("=")
+    )
+    encoded_file_id = f"file-{encoded}"
+
+    with patch(
+        "litellm.proxy.openai_files_endpoints.common_utils.get_credentials_for_model",
+        side_effect=HTTPException(status_code=404, detail="no creds"),
+    ):
+        provider_file_id, fetch_kwargs = (
+            rate_limiter._resolve_batch_input_file_fetch_params(
+                file_id=encoded_file_id,
+                custom_llm_provider="openai",
+                data={},
+            )
+        )
+    assert provider_file_id == "file-orig"
+    assert fetch_kwargs == {"custom_llm_provider": "openai"}
+
+
+@pytest.mark.asyncio
+async def test_check_and_increment_computes_descriptors_when_not_passed():
+    from litellm.proxy.hooks.batch_rate_limiter import (
+        BatchFileUsage,
+        _PROXY_BatchRateLimiter,
+    )
+
+    parallel_request_limiter = MagicMock()
+    parallel_request_limiter._create_rate_limit_descriptors.return_value = [
+        {"rate_limit": {"tokens_per_unit": 100}}
+    ]
+    parallel_request_limiter.atomic_check_and_increment_by_n = AsyncMock(
+        return_value={"overall_code": "OK", "statuses": []}
+    )
+    rate_limiter = _PROXY_BatchRateLimiter(
+        internal_usage_cache=MagicMock(),
+        parallel_request_limiter=parallel_request_limiter,
+    )
+
+    await rate_limiter._check_and_increment_batch_counters(
+        user_api_key_dict=UserAPIKeyAuth(api_key="sk", models=["*"]),
+        data={"model": "gpt-4o-mini"},
+        batch_usage=BatchFileUsage(total_tokens=10, request_count=1),
+        descriptors=None,
+    )
+
+    parallel_request_limiter._create_rate_limit_descriptors.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_count_input_file_usage_raises_on_non_bytes_content():
+    from litellm.proxy.hooks.batch_rate_limiter import _PROXY_BatchRateLimiter
+
+    rate_limiter = _PROXY_BatchRateLimiter(
+        internal_usage_cache=MagicMock(),
+        parallel_request_limiter=MagicMock(),
+    )
+
+    bad_content = MagicMock()
+    bad_content.content = "not-bytes"
+
+    with patch("litellm.afile_content", new=AsyncMock(return_value=bad_content)):
+        with pytest.raises(ValueError, match="Expected bytes content"):
+            await rate_limiter.count_input_file_usage(
+                file_id="file-plain",
+                custom_llm_provider="openai",
+                user_api_key_dict=UserAPIKeyAuth(api_key="sk", models=["*"]),
+                data={},
+            )

From 523a984f0d40797829613a8cc1e096676f8324c1 Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Sat, 30 May 2026 03:05:40 +0000
Subject: [PATCH 07/15] fix(batch_rate_limiter): resolve provider skip from
 trusted deployment creds

Resolve the batch provider from router deployment credentials instead of
the user-supplied custom_llm_provider request field, so an unrestricted
key cannot spoof a skip-listed provider to bypass batch rate limiting.

Strengthen the provider-skip test to assert the file download and
descriptor work were short-circuited, and add a test that a spoofed
provider still falls through to rate-limit evaluation.
---
 litellm/proxy/hooks/batch_rate_limiter.py     | 50 +++++++++++----
 .../proxy/hooks/test_batch_file_validation.py | 62 +++++++++++++++++--
 2 files changed, 96 insertions(+), 16 deletions(-)

diff --git a/litellm/proxy/hooks/batch_rate_limiter.py b/litellm/proxy/hooks/batch_rate_limiter.py
index 787592f4e30a..baeadc98c332 100644
--- a/litellm/proxy/hooks/batch_rate_limiter.py
+++ b/litellm/proxy/hooks/batch_rate_limiter.py
@@ -127,6 +127,36 @@ def _get_batch_routing_model(self, data: Dict) -> Optional[str]:
 
         return None
 
+    def _resolve_batch_provider(self, batch_model: Optional[str]) -> Optional[str]:
+        """Resolve the provider from the deployment that serves ``batch_model``.
+
+        The provider is read from trusted router credentials rather than the
+        user-supplied ``custom_llm_provider`` request field, so a caller cannot
+        spoof a skip-listed provider to bypass batch rate limiting.
+        """
+        if not batch_model:
+            return None
+
+        from litellm.proxy.openai_files_endpoints.common_utils import (
+            get_credentials_for_model,
+        )
+        from litellm.proxy.proxy_server import llm_router
+
+        if llm_router is None:
+            return None
+
+        try:
+            credentials = get_credentials_for_model(
+                llm_router=llm_router,
+                model_id=batch_model,
+                operation_context="batch input file read (rate limiting)",
+            )
+        except HTTPException:
+            return None
+
+        provider = credentials.get("custom_llm_provider")
+        return provider if isinstance(provider, str) and provider else None
+
     def _matches_skip_list(self, value: str, skip_list: List[str]) -> bool:
         if not skip_list:
             return False
@@ -164,8 +194,7 @@ def _should_skip_batch_input_file_processing(
         ``_enforce_batch_file_model_access`` can validate every ``body.model``
         entry. Otherwise a restricted key could smuggle unauthorized models
         into the file via an admin-configured skip (global disable,
-        per-model, per-provider) or via the user-controlled
-        ``custom_llm_provider`` field.
+        per-model, per-provider).
 
         Returns ``(should_skip, descriptors)`` where ``descriptors`` is the
         rate-limit descriptor list computed for the no-limits check, so the
@@ -197,16 +226,13 @@ def _should_skip_batch_input_file_processing(
             general_settings.get("skip_batch_input_file_rate_limiting_for_providers")
             or []
         )
-        custom_llm_provider = data.get("custom_llm_provider")
-        if (
-            isinstance(custom_llm_provider, str)
-            and custom_llm_provider in skip_providers
-        ):
-            verbose_proxy_logger.debug(
-                "Skipping batch input file processing for "
-                f"custom_llm_provider={custom_llm_provider}"
-            )
-            return True, None
+        if skip_providers:
+            batch_provider = self._resolve_batch_provider(batch_model)
+            if batch_provider and batch_provider in skip_providers:
+                verbose_proxy_logger.debug(
+                    f"Skipping batch input file processing for provider={batch_provider}"
+                )
+                return True, None
 
         descriptors = self._create_batch_rate_limit_descriptors(
             user_api_key_dict=user_api_key_dict,
diff --git a/tests/test_litellm/proxy/hooks/test_batch_file_validation.py b/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
index 515be9df87c2..f7e6660f5f0e 100644
--- a/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
+++ b/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
@@ -293,22 +293,76 @@ async def test_pre_call_skips_file_fetch_for_configured_provider():
         parallel_request_limiter=MagicMock(),
     )
     user = UserAPIKeyAuth(api_key="sk-ok", user_id="alice", models=["*"])
+    data = {"input_file_id": "file-abc123", "model": "my-vllm-model"}
 
-    with patch(
-        "litellm.proxy.proxy_server.general_settings",
-        {"skip_batch_input_file_rate_limiting_for_providers": ["hosted_vllm"]},
+    with (
+        patch(
+            "litellm.proxy.proxy_server.general_settings",
+            {"skip_batch_input_file_rate_limiting_for_providers": ["hosted_vllm"]},
+        ),
+        patch("litellm.proxy.proxy_server.llm_router", MagicMock()),
+        patch(
+            "litellm.proxy.openai_files_endpoints.common_utils.get_credentials_for_model",
+            return_value={"custom_llm_provider": "hosted_vllm"},
+        ),
+        patch("litellm.afile_content", new=AsyncMock()) as mock_afile_content,
     ):
         result = await rate_limiter.async_pre_call_hook(
+            user_api_key_dict=user,
+            cache=MagicMock(),
+            data=data,
+            call_type="acreate_batch",
+        )
+
+    assert result == data
+    # A real skip must short-circuit before any file download or rate-limit
+    # work — assert the skip happened rather than the hook's error-recovery
+    # path (which also returns data unchanged).
+    mock_afile_content.assert_not_awaited()
+    rate_limiter.parallel_request_limiter._create_rate_limit_descriptors.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_pre_call_does_not_skip_for_spoofed_provider():
+    """The provider skip is resolved from trusted deployment credentials, so a
+    user-supplied ``custom_llm_provider`` that is not backed by the routing
+    deployment must not trigger a skip."""
+    from litellm.proxy.hooks.batch_rate_limiter import _PROXY_BatchRateLimiter
+
+    rate_limiter = _PROXY_BatchRateLimiter(
+        internal_usage_cache=MagicMock(),
+        parallel_request_limiter=MagicMock(),
+    )
+    rate_limiter.parallel_request_limiter._create_rate_limit_descriptors.return_value = (
+        []
+    )
+    user = UserAPIKeyAuth(api_key="sk-ok", user_id="alice", models=["*"])
+
+    with (
+        patch(
+            "litellm.proxy.proxy_server.general_settings",
+            {"skip_batch_input_file_rate_limiting_for_providers": ["hosted_vllm"]},
+        ),
+        patch("litellm.proxy.proxy_server.llm_router", MagicMock()),
+        patch(
+            "litellm.proxy.openai_files_endpoints.common_utils.get_credentials_for_model",
+            return_value={"custom_llm_provider": "openai"},
+        ),
+    ):
+        await rate_limiter.async_pre_call_hook(
             user_api_key_dict=user,
             cache=MagicMock(),
             data={
                 "input_file_id": "file-abc123",
+                "model": "my-openai-model",
                 "custom_llm_provider": "hosted_vllm",
             },
             call_type="acreate_batch",
         )
 
-    assert result["custom_llm_provider"] == "hosted_vllm"
+    # Reaching descriptor evaluation proves the spoofed provider did not
+    # short-circuit the skip decision via the provider allow-list.
+    rate_limiter.parallel_request_limiter._create_rate_limit_descriptors.assert_called_once()
 
 
 @pytest.mark.asyncio

From a115df8ea3685a08c0ab53cbdae7eed475d89972 Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Sat, 30 May 2026 03:31:14 +0000
Subject: [PATCH 08/15] fix(batch_rate_limiter): guard model-embedded
 credential lookup on llm_router presence

---
 litellm/proxy/hooks/batch_rate_limiter.py | 27 ++++++++++++-----------
 1 file changed, 14 insertions(+), 13 deletions(-)

diff --git a/litellm/proxy/hooks/batch_rate_limiter.py b/litellm/proxy/hooks/batch_rate_limiter.py
index baeadc98c332..3e0f5a523f4b 100644
--- a/litellm/proxy/hooks/batch_rate_limiter.py
+++ b/litellm/proxy/hooks/batch_rate_limiter.py
@@ -302,19 +302,20 @@ def _resolve_batch_input_file_fetch_params(
 
         model_from_file_id = decode_model_from_file_id(file_id)
         if model_from_file_id:
-            try:
-                credentials = get_credentials_for_model(
-                    llm_router=llm_router,
-                    model_id=model_from_file_id,
-                    operation_context="batch input file read (rate limiting)",
-                )
-                fetch_kwargs.update(_extract_file_access_credentials(credentials))
-                fetch_kwargs["model"] = model_from_file_id
-                provider = credentials.get("custom_llm_provider")
-                if provider:
-                    fetch_kwargs["custom_llm_provider"] = provider
-            except HTTPException:
-                pass
+            if llm_router is not None:
+                try:
+                    credentials = get_credentials_for_model(
+                        llm_router=llm_router,
+                        model_id=model_from_file_id,
+                        operation_context="batch input file read (rate limiting)",
+                    )
+                    fetch_kwargs.update(_extract_file_access_credentials(credentials))
+                    fetch_kwargs["model"] = model_from_file_id
+                    provider = credentials.get("custom_llm_provider")
+                    if provider:
+                        fetch_kwargs["custom_llm_provider"] = provider
+                except HTTPException:
+                    pass
             return get_original_file_id(file_id), fetch_kwargs
 
         request_model = data.get("model")

From 1e71bdd6e704a2824e995c827a1d8a36b4ac273c Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Sat, 30 May 2026 03:57:31 +0000
Subject: [PATCH 09/15] test(batch_rate_limiter): drive real no-skip fetch path
 and pin wildcard+access-group predicate

The spoofed-provider test configured empty descriptors, so the no-limits
shortcut skipped the file fetch and the assertion only proved the provider
allow-list did not short-circuit before descriptor evaluation. Give the key an
applicable rate limit so the only thing that can prevent the fetch is the
provider skip, then assert afile_content is awaited and the counters are
incremented; the spoofed custom_llm_provider must not skip processing.

Also cover the wildcard / all-proxy-models plus access_group_ids combination in
the model-access predicate so the wildcard-wins behavior is locked down.
---
 .../proxy/hooks/test_batch_file_validation.py | 61 ++++++++++++++++---
 1 file changed, 54 insertions(+), 7 deletions(-)

diff --git a/tests/test_litellm/proxy/hooks/test_batch_file_validation.py b/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
index f7e6660f5f0e..f6249efb4668 100644
--- a/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
+++ b/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
@@ -326,28 +326,49 @@ async def test_pre_call_skips_file_fetch_for_configured_provider():
 async def test_pre_call_does_not_skip_for_spoofed_provider():
     """The provider skip is resolved from trusted deployment credentials, so a
     user-supplied ``custom_llm_provider`` that is not backed by the routing
-    deployment must not trigger a skip."""
+    deployment must not trigger a skip: the input file must still be fetched
+    and the rate-limit counters incremented."""
     from litellm.proxy.hooks.batch_rate_limiter import _PROXY_BatchRateLimiter
 
     rate_limiter = _PROXY_BatchRateLimiter(
         internal_usage_cache=MagicMock(),
         parallel_request_limiter=MagicMock(),
     )
-    rate_limiter.parallel_request_limiter._create_rate_limit_descriptors.return_value = (
-        []
+    # An applicable rate limit keeps the no-limits shortcut from firing, so the
+    # only thing that could prevent the fetch below is the provider skip. If the
+    # spoofed ``custom_llm_provider`` were honored, afile_content would never be
+    # awaited.
+    rate_limiter.parallel_request_limiter._create_rate_limit_descriptors.return_value = [
+        {"rate_limit": {"requests_per_unit": 100}}
+    ]
+    rate_limiter.parallel_request_limiter.atomic_check_and_increment_by_n = AsyncMock(
+        return_value={"overall_code": "OK", "statuses": []}
     )
     user = UserAPIKeyAuth(api_key="sk-ok", user_id="alice", models=["*"])
 
+    mock_router = MagicMock()
+    mock_router.model_list = []
+    mock_router.resolve_model_name_from_model_id.return_value = "my-openai-model"
+
+    mock_content = MagicMock()
+    mock_content.content = (
+        b'{"body": {"model": "my-openai-model", '
+        b'"messages": [{"role": "user", "content": "hi"}]}}\n'
+    )
+
     with (
         patch(
             "litellm.proxy.proxy_server.general_settings",
             {"skip_batch_input_file_rate_limiting_for_providers": ["hosted_vllm"]},
         ),
-        patch("litellm.proxy.proxy_server.llm_router", MagicMock()),
+        patch("litellm.proxy.proxy_server.llm_router", mock_router),
         patch(
             "litellm.proxy.openai_files_endpoints.common_utils.get_credentials_for_model",
             return_value={"custom_llm_provider": "openai"},
         ),
+        patch(
+            "litellm.afile_content", new=AsyncMock(return_value=mock_content)
+        ) as mock_afile_content,
     ):
         await rate_limiter.async_pre_call_hook(
             user_api_key_dict=user,
@@ -360,9 +381,10 @@ async def test_pre_call_does_not_skip_for_spoofed_provider():
             call_type="acreate_batch",
         )
 
-    # Reaching descriptor evaluation proves the spoofed provider did not
-    # short-circuit the skip decision via the provider allow-list.
-    rate_limiter.parallel_request_limiter._create_rate_limit_descriptors.assert_called_once()
+    # The spoofed provider did not short-circuit the skip decision: the file was
+    # fetched and the counters were incremented.
+    mock_afile_content.assert_awaited_once()
+    rate_limiter.parallel_request_limiter.atomic_check_and_increment_by_n.assert_awaited_once()
 
 
 @pytest.mark.asyncio
@@ -568,6 +590,31 @@ def test_key_requires_batch_model_access_check_branches():
     )
     assert check(UserAPIKeyAuth(api_key="sk", models=[])) is False
     assert check(UserAPIKeyAuth(api_key="sk", models=["gpt-4o-mini"])) is True
+    # Wildcard / all-proxy-models grant access to every model, so
+    # can_key_call_model passes any model regardless of access groups (which
+    # only ever widen access). Such keys must not be forced to download and
+    # validate the JSONL even when access_group_ids are also present.
+    assert (
+        check(UserAPIKeyAuth(api_key="sk", models=["*"], access_group_ids=["grp"]))
+        is False
+    )
+    assert (
+        check(
+            UserAPIKeyAuth(
+                api_key="sk", models=["all-proxy-models"], access_group_ids=["grp"]
+            )
+        )
+        is False
+    )
+    # A concrete model allowlist is still a subset even with access groups.
+    assert (
+        check(
+            UserAPIKeyAuth(
+                api_key="sk", models=["gpt-4o-mini"], access_group_ids=["grp"]
+            )
+        )
+        is True
+    )
 
 
 def test_has_applicable_batch_rate_limits():

From 9d5ea6f3fb6d480912126c91e3f0c88ca140673b Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Sat, 30 May 2026 04:00:05 +0000
Subject: [PATCH 10/15] fix(batch_rate_limiter): drop client-controlled skip
 flag to close quota bypass

The litellm_metadata.skip_batch_input_file_rate_limiting flag was read
straight from the request body, so any caller whose key had unrestricted
model access could send it and skip the input-file download, token count,
and RPM/TPM reservation, bypassing their batch rate limits. Skip decisions
now derive only from server-controlled general_settings.
---
 litellm/proxy/hooks/batch_rate_limiter.py            |  4 ----
 .../proxy/hooks/test_batch_file_validation.py        | 12 +++++++++---
 2 files changed, 9 insertions(+), 7 deletions(-)

diff --git a/litellm/proxy/hooks/batch_rate_limiter.py b/litellm/proxy/hooks/batch_rate_limiter.py
index 3e0f5a523f4b..8f5126b907cb 100644
--- a/litellm/proxy/hooks/batch_rate_limiter.py
+++ b/litellm/proxy/hooks/batch_rate_limiter.py
@@ -208,10 +208,6 @@ def _should_skip_batch_input_file_processing(
         if general_settings.get("disable_batch_input_file_rate_limiting") is True:
             return True, None
 
-        litellm_metadata = data.get("litellm_metadata") or {}
-        if litellm_metadata.get("skip_batch_input_file_rate_limiting") is True:
-            return True, None
-
         batch_model = self._get_batch_routing_model(data)
         skip_models = (
             general_settings.get("skip_batch_input_file_rate_limiting_for_models") or []
diff --git a/tests/test_litellm/proxy/hooks/test_batch_file_validation.py b/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
index f6249efb4668..504202d64fad 100644
--- a/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
+++ b/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
@@ -637,8 +637,15 @@ def test_should_skip_returns_false_when_key_needs_model_access_check():
     assert descriptors is None
 
 
-def test_should_skip_honors_litellm_metadata_flag():
+def test_should_skip_ignores_client_supplied_metadata_flag():
+    """A caller must not be able to bypass batch rate limits by setting
+    ``litellm_metadata.skip_batch_input_file_rate_limiting`` in the request
+    body. The skip decision is server-controlled only, so with applicable rate
+    limits the JSONL is still processed despite the client flag."""
     rate_limiter = _make_rate_limiter()
+    rate_limiter.parallel_request_limiter._create_rate_limit_descriptors.return_value = [
+        {"rate_limit": {"requests_per_unit": 5}}
+    ]
     user = UserAPIKeyAuth(api_key="sk", models=["*"])
     with patch("litellm.proxy.proxy_server.general_settings", {}):
         should_skip, descriptors = (
@@ -650,8 +657,7 @@ def test_should_skip_honors_litellm_metadata_flag():
                 user_api_key_dict=user,
             )
         )
-    assert should_skip is True
-    assert descriptors is None
+    assert should_skip is False
 
 
 def test_should_skip_honors_per_model_skip_list():

From 00cd3ed5cbb92e334b40937bec0d5042cf644f73 Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Sat, 30 May 2026 04:51:33 +0000
Subject: [PATCH 11/15] fix(batch_rate_limiter): match per-model skip on
 file-bound model only

The per-model skip resolved its model from _get_batch_routing_model, which
prefers the client-supplied top-level model field. That field only selects
routing credentials; the models a batch actually runs are the body.model
entries in the input JSONL. An unrestricted key could therefore name a
skip-listed deployment at the top level while routing a different,
same-provider model through the file, skipping the download, token count
and rate-limit reservation to bypass batch RPM/TPM limits.

Match the per-model skip against the file-bound model only (model-embedded
file id or unified managed file target), which is fixed when the file is
created and reflects the model the batch runs. The provider skip keeps using
the routing model since an admin opting out of a whole provider already
accepts any of that provider's models.
---
 litellm/proxy/hooks/batch_rate_limiter.py     | 39 ++++++++++++++-----
 .../proxy/hooks/test_batch_file_validation.py | 34 +++++++++++++++-
 2 files changed, 62 insertions(+), 11 deletions(-)

diff --git a/litellm/proxy/hooks/batch_rate_limiter.py b/litellm/proxy/hooks/batch_rate_limiter.py
index 8f5126b907cb..ec59f2e5e998 100644
--- a/litellm/proxy/hooks/batch_rate_limiter.py
+++ b/litellm/proxy/hooks/batch_rate_limiter.py
@@ -99,12 +99,15 @@ def __init__(
         self.internal_usage_cache = internal_usage_cache
         self.parallel_request_limiter = parallel_request_limiter
 
-    def _get_batch_routing_model(self, data: Dict) -> Optional[str]:
-        """Resolve the deployment/model used for this batch from request data."""
-        model = data.get("model")
-        if isinstance(model, str) and model:
-            return model
-
+    def _get_file_bound_batch_model(self, data: Dict) -> Optional[str]:
+        """Resolve the model bound to the batch input file ID.
+
+        The model embedded in a ``file-<base64>`` ID or a unified managed
+        file's target model is fixed when the file is created, so it reflects
+        the model the batch will actually run. Unlike the client-supplied
+        top-level ``model`` field, it cannot be swapped per request to point a
+        skip decision at a deployment the JSONL never routes to.
+        """
         input_file_id = data.get("input_file_id")
         if not isinstance(input_file_id, str) or not input_file_id:
             return None
@@ -127,6 +130,14 @@ def _get_batch_routing_model(self, data: Dict) -> Optional[str]:
 
         return None
 
+    def _get_batch_routing_model(self, data: Dict) -> Optional[str]:
+        """Resolve the deployment/model used for this batch from request data."""
+        model = data.get("model")
+        if isinstance(model, str) and model:
+            return model
+
+        return self._get_file_bound_batch_model(data)
+
     def _resolve_batch_provider(self, batch_model: Optional[str]) -> Optional[str]:
         """Resolve the provider from the deployment that serves ``batch_model``.
 
@@ -196,6 +207,12 @@ def _should_skip_batch_input_file_processing(
         into the file via an admin-configured skip (global disable,
         per-model, per-provider).
 
+        The per-model skip is matched against the file-bound model only, never
+        the client-supplied top-level ``model``. The latter selects routing
+        credentials but not the models the batch runs (those are the JSONL
+        ``body.model`` entries), so honoring it would let a caller name a
+        skip-listed deployment while routing a different, rate-limited model.
+
         Returns ``(should_skip, descriptors)`` where ``descriptors`` is the
         rate-limit descriptor list computed for the no-limits check, so the
         caller can reuse it for counter enforcement without recomputing.
@@ -208,13 +225,13 @@ def _should_skip_batch_input_file_processing(
         if general_settings.get("disable_batch_input_file_rate_limiting") is True:
             return True, None
 
-        batch_model = self._get_batch_routing_model(data)
         skip_models = (
             general_settings.get("skip_batch_input_file_rate_limiting_for_models") or []
         )
-        if batch_model and self._matches_skip_list(batch_model, skip_models):
+        file_bound_model = self._get_file_bound_batch_model(data)
+        if file_bound_model and self._matches_skip_list(file_bound_model, skip_models):
             verbose_proxy_logger.debug(
-                f"Skipping batch input file processing for model={batch_model}"
+                f"Skipping batch input file processing for model={file_bound_model}"
             )
             return True, None
 
@@ -223,7 +240,9 @@ def _should_skip_batch_input_file_processing(
             or []
         )
         if skip_providers:
-            batch_provider = self._resolve_batch_provider(batch_model)
+            batch_provider = self._resolve_batch_provider(
+                self._get_batch_routing_model(data)
+            )
             if batch_provider and batch_provider in skip_providers:
                 verbose_proxy_logger.debug(
                     f"Skipping batch input file processing for provider={batch_provider}"
diff --git a/tests/test_litellm/proxy/hooks/test_batch_file_validation.py b/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
index 504202d64fad..386f7373b118 100644
--- a/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
+++ b/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
@@ -661,15 +661,24 @@ def test_should_skip_ignores_client_supplied_metadata_flag():
 
 
 def test_should_skip_honors_per_model_skip_list():
+    """The per-model skip fires for a model bound to the input file ID (here a
+    model-embedded ``file-<base64>``), which reflects the model the batch runs."""
+    import base64
+
     rate_limiter = _make_rate_limiter()
     user = UserAPIKeyAuth(api_key="sk", models=["*"])
+    encoded = (
+        base64.urlsafe_b64encode(b"litellm:file-xyz;model,gpt-4o-mini")
+        .decode()
+        .rstrip("=")
+    )
     with patch(
         "litellm.proxy.proxy_server.general_settings",
         {"skip_batch_input_file_rate_limiting_for_models": ["gpt-4o-mini"]},
     ):
         should_skip, descriptors = (
             rate_limiter._should_skip_batch_input_file_processing(
-                data={"model": "gpt-4o-mini", "input_file_id": "file-abc"},
+                data={"input_file_id": f"file-{encoded}"},
                 user_api_key_dict=user,
             )
         )
@@ -677,6 +686,29 @@ def test_should_skip_honors_per_model_skip_list():
     assert descriptors is None
 
 
+def test_should_not_skip_per_model_for_spoofed_top_level_model():
+    """A caller must not bypass batch rate limits by naming a skip-listed model
+    in the top-level ``model`` while routing a different model through the JSONL
+    ``body.model`` entries. The per-model skip only trusts the file-bound model,
+    so a skip-listed top-level model over a plain file still gets processed."""
+    rate_limiter = _make_rate_limiter()
+    rate_limiter.parallel_request_limiter._create_rate_limit_descriptors.return_value = [
+        {"rate_limit": {"requests_per_unit": 5}}
+    ]
+    user = UserAPIKeyAuth(api_key="sk", models=["*"])
+    with patch(
+        "litellm.proxy.proxy_server.general_settings",
+        {"skip_batch_input_file_rate_limiting_for_models": ["gpt-4o-mini"]},
+    ):
+        should_skip, descriptors = (
+            rate_limiter._should_skip_batch_input_file_processing(
+                data={"model": "gpt-4o-mini", "input_file_id": "file-abc"},
+                user_api_key_dict=user,
+            )
+        )
+    assert should_skip is False
+
+
 def test_should_skip_when_no_rate_limits_configured():
     rate_limiter = _make_rate_limiter()
     rate_limiter.parallel_request_limiter._create_rate_limit_descriptors.return_value = [

From 7a7e24b120a8c884bde08f8991d9f64af279d403 Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Sat, 30 May 2026 05:25:21 +0000
Subject: [PATCH 12/15] fix(batch_rate_limiter): drop forgeable per-model skip
 to close quota bypass

The per-model skip matched skip_batch_input_file_rate_limiting_for_models
against the model bound to the input file id. That model comes from
decode_model_from_file_id / the unified file id, both unsigned base64 the
caller fully controls, so a caller could re-encode an accessible provider
file id with a skip-listed model while the JSONL still routes rate-limited
body.model entries and bypass the batch RPM/TPM counters. The models a batch
actually runs are its JSONL body.model entries, which cannot be known without
reading the file, so no caller-influenced model identifier can safely gate a
skip.

Remove the per-model skip entirely. The provider skip stays because the
provider is resolved from trusted deployment credentials and the batch is
constrained to run on that provider; the global disable and
no-applicable-limits skips stay because they do not depend on caller input.
---
 litellm/proxy/hooks/batch_rate_limiter.py     | 58 +++++++------------
 .../proxy/hooks/test_batch_file_validation.py | 31 +++++-----
 2 files changed, 35 insertions(+), 54 deletions(-)

diff --git a/litellm/proxy/hooks/batch_rate_limiter.py b/litellm/proxy/hooks/batch_rate_limiter.py
index ec59f2e5e998..c782d9373e94 100644
--- a/litellm/proxy/hooks/batch_rate_limiter.py
+++ b/litellm/proxy/hooks/batch_rate_limiter.py
@@ -102,11 +102,9 @@ def __init__(
     def _get_file_bound_batch_model(self, data: Dict) -> Optional[str]:
         """Resolve the model bound to the batch input file ID.
 
-        The model embedded in a ``file-<base64>`` ID or a unified managed
-        file's target model is fixed when the file is created, so it reflects
-        the model the batch will actually run. Unlike the client-supplied
-        top-level ``model`` field, it cannot be swapped per request to point a
-        skip decision at a deployment the JSONL never routes to.
+        Used only as a fallback routing model when the request omits a
+        top-level ``model``; the provider is then read from that deployment's
+        trusted credentials for the provider-level skip decision.
         """
         input_file_id = data.get("input_file_id")
         if not isinstance(input_file_id, str) or not input_file_id:
@@ -168,16 +166,6 @@ def _resolve_batch_provider(self, batch_model: Optional[str]) -> Optional[str]:
         provider = credentials.get("custom_llm_provider")
         return provider if isinstance(provider, str) and provider else None
 
-    def _matches_skip_list(self, value: str, skip_list: List[str]) -> bool:
-        if not skip_list:
-            return False
-        for entry in skip_list:
-            if not isinstance(entry, str) or not entry:
-                continue
-            if value == entry or value.startswith(f"{entry}/"):
-                return True
-        return False
-
     def _create_batch_rate_limit_descriptors(
         self,
         user_api_key_dict: UserAPIKeyAuth,
@@ -197,21 +185,25 @@ def _should_skip_batch_input_file_processing(
         user_api_key_dict: UserAPIKeyAuth,
     ) -> Tuple[bool, Optional[List["RateLimitDescriptor"]]]:
         """
-        Skip downloading batch input files when configured or when there is
-        nothing to enforce (no applicable rate limits and no model allowlist).
+        Skip downloading batch input files when the operator disabled batch
+        input-file rate limiting, when the batch runs entirely on a skip-listed
+        provider, or when there is nothing to enforce (no applicable rate
+        limits).
 
-        When the caller's key has a model allowlist to enforce, no skip path
-        is honored: the JSONL must still be downloaded so
+        A skip is only honored for keys with unrestricted model access. When
+        the key has a model allowlist, the JSONL must still be downloaded so
         ``_enforce_batch_file_model_access`` can validate every ``body.model``
-        entry. Otherwise a restricted key could smuggle unauthorized models
-        into the file via an admin-configured skip (global disable,
-        per-model, per-provider).
-
-        The per-model skip is matched against the file-bound model only, never
-        the client-supplied top-level ``model``. The latter selects routing
-        credentials but not the models the batch runs (those are the JSONL
-        ``body.model`` entries), so honoring it would let a caller name a
-        skip-listed deployment while routing a different, rate-limited model.
+        entry, otherwise a restricted key could smuggle unauthorized models
+        into the file via an admin-configured skip.
+
+        The skip is never keyed on a specific model name. The models a batch
+        actually runs are its JSONL ``body.model`` entries, and any model
+        identifier the caller can influence (the top-level ``model`` or the
+        unsigned model embedded in a ``file-...`` id) can be pointed at a
+        skip-listed deployment while the file routes a different, rate-limited
+        model. The provider skip is safe because the provider is read from the
+        routing deployment's trusted credentials and the batch is constrained
+        to run on that provider.
 
         Returns ``(should_skip, descriptors)`` where ``descriptors`` is the
         rate-limit descriptor list computed for the no-limits check, so the
@@ -225,16 +217,6 @@ def _should_skip_batch_input_file_processing(
         if general_settings.get("disable_batch_input_file_rate_limiting") is True:
             return True, None
 
-        skip_models = (
-            general_settings.get("skip_batch_input_file_rate_limiting_for_models") or []
-        )
-        file_bound_model = self._get_file_bound_batch_model(data)
-        if file_bound_model and self._matches_skip_list(file_bound_model, skip_models):
-            verbose_proxy_logger.debug(
-                f"Skipping batch input file processing for model={file_bound_model}"
-            )
-            return True, None
-
         skip_providers = (
             general_settings.get("skip_batch_input_file_rate_limiting_for_providers")
             or []
diff --git a/tests/test_litellm/proxy/hooks/test_batch_file_validation.py b/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
index 386f7373b118..96da4e66325d 100644
--- a/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
+++ b/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
@@ -571,14 +571,6 @@ def test_get_batch_routing_model_uses_unified_file_id_target():
         )
 
 
-def test_matches_skip_list_handles_empty_and_entry_shapes():
-    rate_limiter = _make_rate_limiter()
-    assert rate_limiter._matches_skip_list("gpt-4o", []) is False
-    assert rate_limiter._matches_skip_list("gpt-4o", ["gpt-4o"]) is True
-    assert rate_limiter._matches_skip_list("vertex_ai/gemini", ["vertex_ai"]) is True
-    assert rate_limiter._matches_skip_list("gpt-4o", [None, "", "claude"]) is False
-
-
 def test_key_requires_batch_model_access_check_branches():
     from litellm.proxy.hooks.batch_rate_limiter import _PROXY_BatchRateLimiter
 
@@ -660,12 +652,19 @@ def test_should_skip_ignores_client_supplied_metadata_flag():
     assert should_skip is False
 
 
-def test_should_skip_honors_per_model_skip_list():
-    """The per-model skip fires for a model bound to the input file ID (here a
-    model-embedded ``file-<base64>``), which reflects the model the batch runs."""
+def test_should_not_skip_for_forged_model_embedded_file_id():
+    """A ``file-<base64>`` id embeds an unsigned model name the caller fully
+    controls, so a caller can re-encode any accessible provider file id with a
+    skip-listed model while the JSONL still routes rate-limited ``body.model``
+    entries. The per-model skip must therefore never fire: with applicable rate
+    limits, a forged skip-listed file-bound model still falls through to file
+    processing and counter enforcement."""
     import base64
 
     rate_limiter = _make_rate_limiter()
+    rate_limiter.parallel_request_limiter._create_rate_limit_descriptors.return_value = [
+        {"rate_limit": {"requests_per_unit": 5}}
+    ]
     user = UserAPIKeyAuth(api_key="sk", models=["*"])
     encoded = (
         base64.urlsafe_b64encode(b"litellm:file-xyz;model,gpt-4o-mini")
@@ -682,15 +681,15 @@ def test_should_skip_honors_per_model_skip_list():
                 user_api_key_dict=user,
             )
         )
-    assert should_skip is True
-    assert descriptors is None
+    assert should_skip is False
+    assert descriptors is not None
 
 
-def test_should_not_skip_per_model_for_spoofed_top_level_model():
+def test_should_not_skip_for_skip_listed_top_level_model():
     """A caller must not bypass batch rate limits by naming a skip-listed model
     in the top-level ``model`` while routing a different model through the JSONL
-    ``body.model`` entries. The per-model skip only trusts the file-bound model,
-    so a skip-listed top-level model over a plain file still gets processed."""
+    ``body.model`` entries. No per-model skip exists, so a skip-listed model over
+    a plain file still gets processed."""
     rate_limiter = _make_rate_limiter()
     rate_limiter.parallel_request_limiter._create_rate_limit_descriptors.return_value = [
         {"rate_limit": {"requests_per_unit": 5}}

From f0e7c45d6fdf70241ae3c60cd583533e1b0846bd Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Sat, 30 May 2026 05:42:38 +0000
Subject: [PATCH 13/15] fix(batch_rate_limiter): warn when no-op per-model skip
 key is configured

---
 litellm/proxy/hooks/batch_rate_limiter.py     | 27 +++++++++-
 .../proxy/hooks/test_batch_file_validation.py | 51 +++++++++++++++++++
 2 files changed, 76 insertions(+), 2 deletions(-)

diff --git a/litellm/proxy/hooks/batch_rate_limiter.py b/litellm/proxy/hooks/batch_rate_limiter.py
index c782d9373e94..b4abd86102a5 100644
--- a/litellm/proxy/hooks/batch_rate_limiter.py
+++ b/litellm/proxy/hooks/batch_rate_limiter.py
@@ -98,6 +98,7 @@ def __init__(
         """
         self.internal_usage_cache = internal_usage_cache
         self.parallel_request_limiter = parallel_request_limiter
+        self._warned_unsupported_model_skip = False
 
     def _get_file_bound_batch_model(self, data: Dict) -> Optional[str]:
         """Resolve the model bound to the batch input file ID.
@@ -209,11 +210,13 @@ def _should_skip_batch_input_file_processing(
         rate-limit descriptor list computed for the no-limits check, so the
         caller can reuse it for counter enforcement without recomputing.
         """
+        from litellm.proxy.proxy_server import general_settings
+
+        self._warn_if_unsupported_model_skip_configured(general_settings)
+
         if self._key_requires_batch_model_access_check(user_api_key_dict):
             return False, None
 
-        from litellm.proxy.proxy_server import general_settings
-
         if general_settings.get("disable_batch_input_file_rate_limiting") is True:
             return True, None
 
@@ -243,6 +246,26 @@ def _should_skip_batch_input_file_processing(
 
         return False, descriptors
 
+    def _warn_if_unsupported_model_skip_configured(
+        self, general_settings: Dict
+    ) -> None:
+        """Warn once that ``skip_batch_input_file_rate_limiting_for_models`` is a no-op.
+
+        A per-model skip is intentionally not honored because the model a batch
+        runs on is caller-influenced and can be pointed at a skip-listed
+        deployment while the JSONL routes a different, rate-limited model.
+        """
+        if self._warned_unsupported_model_skip:
+            return
+        if general_settings.get("skip_batch_input_file_rate_limiting_for_models"):
+            self._warned_unsupported_model_skip = True
+            verbose_proxy_logger.warning(
+                "general_settings.skip_batch_input_file_rate_limiting_for_models is not "
+                "supported and has no effect. Use "
+                "skip_batch_input_file_rate_limiting_for_providers or "
+                "disable_batch_input_file_rate_limiting instead."
+            )
+
     @staticmethod
     def _key_requires_batch_model_access_check(
         user_api_key_dict: UserAPIKeyAuth,
diff --git a/tests/test_litellm/proxy/hooks/test_batch_file_validation.py b/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
index 96da4e66325d..fbe2cec1eff4 100644
--- a/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
+++ b/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
@@ -708,6 +708,57 @@ def test_should_not_skip_for_skip_listed_top_level_model():
     assert should_skip is False
 
 
+def test_warns_once_for_unsupported_model_skip_setting():
+    """Operators who set the no-op per-model skip key get a single warning so a
+    misconfigured deployment does not silently leave batch limits unenforced."""
+    rate_limiter = _make_rate_limiter()
+    rate_limiter.parallel_request_limiter._create_rate_limit_descriptors.return_value = [
+        {"rate_limit": {"requests_per_unit": 5}}
+    ]
+    user = UserAPIKeyAuth(api_key="sk", models=["*"])
+    with (
+        patch(
+            "litellm.proxy.proxy_server.general_settings",
+            {"skip_batch_input_file_rate_limiting_for_models": ["gpt-4o-mini"]},
+        ),
+        patch(
+            "litellm.proxy.hooks.batch_rate_limiter.verbose_proxy_logger"
+        ) as mock_logger,
+    ):
+        for _ in range(3):
+            rate_limiter._should_skip_batch_input_file_processing(
+                data={"model": "gpt-4o-mini", "input_file_id": "file-abc"},
+                user_api_key_dict=user,
+            )
+    assert mock_logger.warning.call_count == 1
+    assert (
+        "skip_batch_input_file_rate_limiting_for_models"
+        in mock_logger.warning.call_args[0][0]
+    )
+
+
+def test_no_warning_when_model_skip_setting_absent():
+    rate_limiter = _make_rate_limiter()
+    rate_limiter.parallel_request_limiter._create_rate_limit_descriptors.return_value = [
+        {"rate_limit": {"requests_per_unit": 5}}
+    ]
+    user = UserAPIKeyAuth(api_key="sk", models=["*"])
+    with (
+        patch(
+            "litellm.proxy.proxy_server.general_settings",
+            {"skip_batch_input_file_rate_limiting_for_providers": ["openai"]},
+        ),
+        patch(
+            "litellm.proxy.hooks.batch_rate_limiter.verbose_proxy_logger"
+        ) as mock_logger,
+    ):
+        rate_limiter._should_skip_batch_input_file_processing(
+            data={"model": "gpt-4o-mini", "input_file_id": "file-abc"},
+            user_api_key_dict=user,
+        )
+    mock_logger.warning.assert_not_called()
+
+
 def test_should_skip_when_no_rate_limits_configured():
     rate_limiter = _make_rate_limiter()
     rate_limiter.parallel_request_limiter._create_rate_limit_descriptors.return_value = [

From 56d7069a8e4dd8a6defc600d0af8a57905682d8f Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Sat, 30 May 2026 16:28:30 +0000
Subject: [PATCH 14/15] test(batch_rate_limiter): patch llm_router so
 model-embedded credential-error test hits fallback

---
 .../proxy/hooks/test_batch_file_validation.py       | 13 ++++++++++---
 1 file changed, 10 insertions(+), 3 deletions(-)

diff --git a/tests/test_litellm/proxy/hooks/test_batch_file_validation.py b/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
index fbe2cec1eff4..e9bfaf7205e1 100644
--- a/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
+++ b/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
@@ -849,9 +849,15 @@ def test_resolve_fetch_params_model_embedded_fails_open_on_credential_error():
     )
     encoded_file_id = f"file-{encoded}"
 
-    with patch(
-        "litellm.proxy.openai_files_endpoints.common_utils.get_credentials_for_model",
-        side_effect=HTTPException(status_code=404, detail="no creds"),
+    get_credentials = MagicMock(
+        side_effect=HTTPException(status_code=404, detail="no creds")
+    )
+    with (
+        patch("litellm.proxy.proxy_server.llm_router", MagicMock()),
+        patch(
+            "litellm.proxy.openai_files_endpoints.common_utils.get_credentials_for_model",
+            get_credentials,
+        ),
     ):
         provider_file_id, fetch_kwargs = (
             rate_limiter._resolve_batch_input_file_fetch_params(
@@ -860,6 +866,7 @@ def test_resolve_fetch_params_model_embedded_fails_open_on_credential_error():
                 data={},
             )
         )
+    get_credentials.assert_called_once()
     assert provider_file_id == "file-orig"
     assert fetch_kwargs == {"custom_llm_provider": "openai"}
 

From c44fafaa246a6ba07c0412101148e1bd15812e95 Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Mon, 1 Jun 2026 21:38:42 +0000
Subject: [PATCH 15/15] fix(batch_rate_limiter): resolve provider skip from
 file-bound model

create_batch routes a model-embedded or unified file id on the model
bound to that file and ignores the top-level model, so deriving the
provider skip from the top-level model first let a caller point model at
a skip-listed provider while the file routed a rate-limited one, skipping
counter enforcement. Resolve the routing model from the file binding
first, matching the batch endpoint.
---
 litellm/proxy/hooks/batch_rate_limiter.py     |  23 +++-
 .../proxy/hooks/test_batch_file_validation.py | 108 +++++++++++++++++-
 2 files changed, 125 insertions(+), 6 deletions(-)

diff --git a/litellm/proxy/hooks/batch_rate_limiter.py b/litellm/proxy/hooks/batch_rate_limiter.py
index b4abd86102a5..435b6eea45b9 100644
--- a/litellm/proxy/hooks/batch_rate_limiter.py
+++ b/litellm/proxy/hooks/batch_rate_limiter.py
@@ -103,9 +103,11 @@ def __init__(
     def _get_file_bound_batch_model(self, data: Dict) -> Optional[str]:
         """Resolve the model bound to the batch input file ID.
 
-        Used only as a fallback routing model when the request omits a
-        top-level ``model``; the provider is then read from that deployment's
-        trusted credentials for the provider-level skip decision.
+        ``create_batch`` routes a file-bound id (model-embedded ``file-...`` or
+        unified managed file) on that bound model and ignores the top-level
+        ``model``, so this is the authoritative routing model whenever the file
+        binds one. The provider is then read from that deployment's trusted
+        credentials for the provider-level skip decision.
         """
         input_file_id = data.get("input_file_id")
         if not isinstance(input_file_id, str) or not input_file_id:
@@ -130,12 +132,23 @@ def _get_file_bound_batch_model(self, data: Dict) -> Optional[str]:
         return None
 
     def _get_batch_routing_model(self, data: Dict) -> Optional[str]:
-        """Resolve the deployment/model used for this batch from request data."""
+        """Resolve the deployment/model used for this batch from request data.
+
+        Mirrors ``create_batch`` routing precedence: a model bound to the input
+        file id wins over the top-level ``model``, because the batch endpoint
+        ignores the top-level model for file-bound ids. Resolving the provider
+        skip from the top-level model first would let a caller point ``model``
+        at a skip-listed provider while the file routes a rate-limited one.
+        """
+        file_bound_model = self._get_file_bound_batch_model(data)
+        if file_bound_model:
+            return file_bound_model
+
         model = data.get("model")
         if isinstance(model, str) and model:
             return model
 
-        return self._get_file_bound_batch_model(data)
+        return None
 
     def _resolve_batch_provider(self, batch_model: Optional[str]) -> Optional[str]:
         """Resolve the provider from the deployment that serves ``batch_model``.
diff --git a/tests/test_litellm/proxy/hooks/test_batch_file_validation.py b/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
index e9bfaf7205e1..af5a5cde8ba0 100644
--- a/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
+++ b/tests/test_litellm/proxy/hooks/test_batch_file_validation.py
@@ -521,13 +521,34 @@ def _make_rate_limiter():
     )
 
 
-def test_get_batch_routing_model_prefers_request_model():
+def test_get_batch_routing_model_uses_request_model_for_plain_file():
     rate_limiter = _make_rate_limiter()
     assert (
         rate_limiter._get_batch_routing_model({"model": "gpt-4o-mini"}) == "gpt-4o-mini"
     )
 
 
+def test_get_batch_routing_model_prefers_file_bound_over_request_model():
+    """``create_batch`` routes a model-embedded file id on its bound model and
+    ignores the top-level ``model``. The skip decision must use the same
+    precedence, otherwise a caller could point ``model`` at a skip-listed
+    provider while the file routes a rate-limited one."""
+    import base64
+
+    rate_limiter = _make_rate_limiter()
+    encoded = (
+        base64.urlsafe_b64encode(b"litellm:file-xyz;model,vllm-batch")
+        .decode()
+        .rstrip("=")
+    )
+    assert (
+        rate_limiter._get_batch_routing_model(
+            {"input_file_id": f"file-{encoded}", "model": "gpt-4o-mini"}
+        )
+        == "vllm-batch"
+    )
+
+
 def test_get_batch_routing_model_returns_none_without_model_or_file():
     rate_limiter = _make_rate_limiter()
     assert rate_limiter._get_batch_routing_model({}) is None
@@ -708,6 +729,91 @@ def test_should_not_skip_for_skip_listed_top_level_model():
     assert should_skip is False
 
 
+def test_should_not_skip_when_file_bound_provider_is_rate_limited():
+    """A caller must not bypass batch rate limits by pointing the top-level
+    ``model`` at a skip-listed provider while the model-embedded ``input_file_id``
+    routes to a rate-limited provider. ``create_batch`` runs the batch on the
+    file-bound model, so the skip decision must resolve the provider from that
+    model and still process the file when its provider is not skip-listed."""
+    import base64
+
+    rate_limiter = _make_rate_limiter()
+    rate_limiter.parallel_request_limiter._create_rate_limit_descriptors.return_value = [
+        {"rate_limit": {"requests_per_unit": 5}}
+    ]
+    user = UserAPIKeyAuth(api_key="sk", models=["*"])
+    encoded = (
+        base64.urlsafe_b64encode(b"litellm:file-orig;model,vllm-batch")
+        .decode()
+        .rstrip("=")
+    )
+
+    def _creds(model_id, **kwargs):
+        provider = "hosted_vllm" if model_id == "vllm-batch" else "openai"
+        return {"custom_llm_provider": provider}
+
+    with (
+        patch(
+            "litellm.proxy.proxy_server.general_settings",
+            {"skip_batch_input_file_rate_limiting_for_providers": ["openai"]},
+        ),
+        patch("litellm.proxy.proxy_server.llm_router", MagicMock()),
+        patch(
+            "litellm.proxy.openai_files_endpoints.common_utils.get_credentials_for_model",
+            side_effect=_creds,
+        ),
+    ):
+        should_skip, descriptors = (
+            rate_limiter._should_skip_batch_input_file_processing(
+                data={"input_file_id": f"file-{encoded}", "model": "gpt-skip"},
+                user_api_key_dict=user,
+            )
+        )
+    assert should_skip is False
+    assert descriptors is not None
+
+
+def test_should_skip_when_file_bound_provider_is_skip_listed():
+    """The provider skip must still fire when the model the batch actually runs
+    on (the file-bound model) resolves to a skip-listed provider, even if the
+    top-level ``model`` resolves to a different, non-skipped provider."""
+    import base64
+
+    rate_limiter = _make_rate_limiter()
+    rate_limiter.parallel_request_limiter._create_rate_limit_descriptors.return_value = [
+        {"rate_limit": {"requests_per_unit": 5}}
+    ]
+    user = UserAPIKeyAuth(api_key="sk", models=["*"])
+    encoded = (
+        base64.urlsafe_b64encode(b"litellm:file-orig;model,vllm-batch")
+        .decode()
+        .rstrip("=")
+    )
+
+    def _creds(model_id, **kwargs):
+        provider = "hosted_vllm" if model_id == "vllm-batch" else "openai"
+        return {"custom_llm_provider": provider}
+
+    with (
+        patch(
+            "litellm.proxy.proxy_server.general_settings",
+            {"skip_batch_input_file_rate_limiting_for_providers": ["hosted_vllm"]},
+        ),
+        patch("litellm.proxy.proxy_server.llm_router", MagicMock()),
+        patch(
+            "litellm.proxy.openai_files_endpoints.common_utils.get_credentials_for_model",
+            side_effect=_creds,
+        ),
+    ):
+        should_skip, descriptors = (
+            rate_limiter._should_skip_batch_input_file_processing(
+                data={"input_file_id": f"file-{encoded}", "model": "gpt-skip"},
+                user_api_key_dict=user,
+            )
+        )
+    assert should_skip is True
+
+
 def test_warns_once_for_unsupported_model_skip_setting():
     """Operators who set the no-op per-model skip key get a single warning so a
     misconfigured deployment does not silently leave batch limits unenforced."""