fix(proxy)!: let proxy admin keys read raw cloud storage file ids on GET /files/{id}/content (#40621)

* fix(proxy): let proxy admin keys read raw cloud storage file ids on GET /files/{id}/content

Since #30595 the endpoint answered 400 for every raw gs:// or s3:// id, which broke the
documented unmanaged Vertex batch flow at its last step, while a double-encoded id slipped
past the guard entirely. Raw ids are now readable by proxy admin keys only, every other key
gets 403, and the guard decodes the id fully before deciding, so encoding depth no longer matters

* fix(proxy): decode nested percent-encoded file ids iteratively

An id percent-encoded a thousand or more times sent the recursive decoder past the
interpreter's recursion limit, so a proxy admin key got a 500 instead of a verdict on
the id. The decoder now walks the encodings in a loop, so any depth resolves

* fix(proxy): cap the decode passes the raw cloud storage file id guard makes

Any authenticated key could send a file id wrapped in tens of thousands of percent-encoding layers and the guard would unwrap every one of them, each pass walking the whole id, so a 120 KB request line held a worker for about five seconds and a 2 MB one for minutes. The guard now unwraps at most eight layers, and an id still changing after that is treated as a raw cloud storage id, so it stays admin-only instead of being decoded any further

* refactor(proxy): type the decode step the raw cloud storage file id guard folds over

* test(proxy): return a streaming result from the admin raw id test mock now that vertex_ai file content streams

* test(proxy): send raw cloud ids to the content route the way uvicorn delivers them

Starlette's TestClient decodes the path twice, so the twice-encoded cases reached the route fully decoded and never hit the decode loop. The cases now name the id the route receives, and the helper tests take a typed quote helper and Final constants.

* test(proxy): mark the raw cloud id test locals Final

litellm/litellm_core_utils/cloud_storage_security.py

@@ -1,6 +1,7 @@
 import posixpath
 import re
 from collections.abc import Mapping, Sequence
+from itertools import accumulate, repeat
 from types import MappingProxyType
 from typing import Any, Final, cast
 from urllib.parse import quote, unquote
@@ -20,15 +21,33 @@ MANAGED_CLOUD_STORAGE_SCHEMES: Final = ("s3://", "gs://")
 _MAPPING_PROXY_TYPE: Final[type] = type(MappingProxyType({}))
 
 
+MAX_FILE_ID_DECODE_PASSES: Final = 8
+
+
+def _unquote_once(current: str, _: str) -> str:
+    return unquote(current)
+
+
+def _decodings(value: str) -> tuple[str, ...]:
+    return tuple(accumulate(repeat(value, MAX_FILE_ID_DECODE_PASSES + 2), _unquote_once))
+
+
 def is_managed_cloud_storage_uri(file_id: str) -> bool:
     """
-    True if file_id is a raw cloud-storage object URI (e.g. ``s3://bucket/key``).
+    True if file_id is a raw cloud-storage object URI (e.g. ``s3://bucket/key``),
+    up to ``MAX_FILE_ID_DECODE_PASSES`` layers of percent-encoding deep, or an id
+    encoded deeper than that, which no client produces and which is treated as raw
+    rather than decoded any further.
 
     These are internal provider artifacts. On the multi-tenant proxy they must be
     retrieved through their managed unified file id so owner/team access is enforced;
     a raw URI supplied by a caller bypasses that check.
     """
-    return isinstance(file_id, str) and file_id.startswith(MANAGED_CLOUD_STORAGE_SCHEMES)
+    if not isinstance(file_id, str):
+        return False
+    decodings: Final = _decodings(file_id)
+    settled: Final = decodings[-1] == decodings[-2]
+    return decodings[-1].startswith(MANAGED_CLOUD_STORAGE_SCHEMES) or not settled
 
 
 _SAFE_OBJECT_COMPONENT_PATTERN: Final = re.compile(r"[^A-Za-z0-9._-]+")

litellm/proxy/openai_files_endpoints/files_endpoints.py

@@ -1102,14 +1102,10 @@ async def get_file_content(
                     }
                 )
         else:
-            # A raw cloud-storage URI (s3://, gs://) supplied here would skip the
-            # managed-file owner/team check that only runs for unified ids, letting
-            # a caller read another tenant's object by its key. Such objects are only
-            # reachable through their managed unified id.
-            if is_managed_cloud_storage_uri(file_id):
+            if is_managed_cloud_storage_uri(file_id) and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN:
                 raise HTTPException(
-                    status_code=400,
-                    detail="Raw cloud storage file ids cannot be retrieved directly. Use the LiteLLM managed file id returned when the file was created.",
+                    status_code=403,
+                    detail="Raw cloud storage file ids can only be retrieved by a proxy admin key. Use the LiteLLM managed file id returned when the file was created.",
                 )
             # Check for model-based credential routing
             (

tests/unit/litellm_core_utils/test_cloud_storage_security.py

@@ -1,4 +1,11 @@
+from functools import reduce
+from typing import Final
+from urllib.parse import quote
+
+import pytest
+
 from litellm.litellm_core_utils.cloud_storage_security import (
+    MAX_FILE_ID_DECODE_PASSES,
     is_managed_cloud_storage_uri,
 )
 
@@ -13,3 +20,38 @@ def test_is_managed_cloud_storage_uri_ignores_provider_and_unified_ids():
     assert not is_managed_cloud_storage_uri("file-abc123")
     assert not is_managed_cloud_storage_uri("bGl0ZWxsbV9wcm94eQ==")
     assert not is_managed_cloud_storage_uri("")
+
+
+@pytest.mark.parametrize(
+    "file_id",
+    (
+        "gs%3A%2F%2Fbucket%2Flitellm-vertex-files%2Fprediction-model%2Fpredictions.jsonl",
+        "gs%253A%252F%252Fbucket%252Flitellm-vertex-files%252Fprediction-model%252Fpredictions.jsonl",
+        "s3%3A%2F%2Fbucket%2Flitellm-batch-outputs%2Fx.jsonl.out",
+        "s3%253A%252F%252Fbucket%252Flitellm-batch-outputs%252Fx.jsonl.out",
+    ),
+)
+def test_is_managed_cloud_storage_uri_sees_through_percent_encoding(file_id: str):
+    assert is_managed_cloud_storage_uri(file_id)
+
+
+def _quote_once(current: str, _: int) -> str:
+    return quote(current, safe="")
+
+
+def _quoted_times(value: str, times: int) -> str:
+    return reduce(_quote_once, range(times), value)
+
+
+def test_is_managed_cloud_storage_uri_decodes_up_to_the_pass_cap():
+    assert is_managed_cloud_storage_uri(_quoted_times("gs://bucket/x", MAX_FILE_ID_DECODE_PASSES))
+    assert not is_managed_cloud_storage_uri(_quoted_times("file-abc/x", MAX_FILE_ID_DECODE_PASSES))
+
+
+def test_is_managed_cloud_storage_uri_fails_closed_on_an_id_encoded_past_the_pass_cap():
+    nested_gs_id: Final = "gs" + "%" + "25" * 5000 + "3A//bucket/x"
+    nested_plain_id: Final = "%" + "25" * 5000
+
+    assert is_managed_cloud_storage_uri(_quoted_times("file-abc/x", MAX_FILE_ID_DECODE_PASSES + 1))
+    assert is_managed_cloud_storage_uri(nested_gs_id)
+    assert is_managed_cloud_storage_uri(nested_plain_id)

tests/unit/proxy/openai_files_endpoint/test_files_endpoint.py

@@ -1,5 +1,6 @@
 import json
 from typing import Final, List, Optional
+from urllib.parse import quote
 from unittest.mock import ANY, AsyncMock, MagicMock
 
 import pytest
@@ -12,7 +13,7 @@ from pytest_mock import MockerFixture
 import litellm
 from litellm import Router
 from litellm.files.types import FileContentStreamingResult
-from litellm.proxy._types import LiteLLM_UserTableFiltered, UserAPIKeyAuth
+from litellm.proxy._types import LiteLLM_UserTableFiltered, LitellmUserRoles, UserAPIKeyAuth
 from litellm.proxy.hooks import get_proxy_hook
 from litellm.proxy.management_endpoints.internal_user_endpoints import ui_view_users
 from litellm.proxy.openai_files_endpoints.file_content_streaming_handler import (
@@ -235,23 +236,112 @@ def test_invalid_purpose(mocker: MockerFixture, monkeypatch, llm_router: Router)
     assert error["param"] == "purpose"
 
 
-def test_get_file_content_rejects_raw_cloud_storage_uri(llm_router: Router):
-    """A raw s3:// file id must be rejected on the proxy content endpoint.
+RAW_GCS_OUTPUT_FILE_ID: Final = (
+    "gs://my-gcs-bucket/litellm-vertex-files/publishers/google/models/gemini-3.8-flash/"
+    "prediction-model-2026-09-10T20:27:18.178864Z/predictions.jsonl"
+)
+RAW_S3_OUTPUT_FILE_ID: Final = "s3://my-bucket/litellm-batch-outputs/job-123/input.jsonl.out"
 
-    Such an id is not a managed unified id, so it would otherwise skip the
-    owner/team access check and let a caller read another tenant's batch output
-    object by its key. Callers must use the managed unified file id.
-    """
-    from urllib.parse import quote
 
-    s3_file_id = "s3://my-bucket/litellm-batch-outputs/job-123/input.jsonl.out"
-    response = client.get(
-        f"/v1/files/{quote(s3_file_id, safe='')}/content?provider=bedrock",
-        headers={"Authorization": "Bearer test-key"},
+def _ids_a_server_hands_the_route_for(raw_id: str, provider: str) -> tuple[tuple[str, str], ...]:
+    encoded_once: Final = quote(raw_id, safe="")
+    return ((raw_id, provider), (encoded_once, provider), (quote(encoded_once, safe=""), provider))
+
+
+RAW_CLOUD_FILE_IDS_AS_THE_ROUTE_RECEIVES_THEM: Final = _ids_a_server_hands_the_route_for(
+    RAW_GCS_OUTPUT_FILE_ID, "vertex_ai"
+) + _ids_a_server_hands_the_route_for(RAW_S3_OUTPUT_FILE_ID, "bedrock")
+
+
+def _testclient_path_segment_delivering(route_file_id: str) -> str:
+    return quote(quote(route_file_id, safe=""), safe="")
+
+
+def _override_auth(
+    monkeypatch: pytest.MonkeyPatch, mocker: MockerFixture, llm_router: Router, user_role: LitellmUserRoles
+) -> ProxyLogging:
+    import litellm.proxy.proxy_server as ps
+
+    proxy_logging_obj: Final = setup_proxy_logging_object(monkeypatch, llm_router)
+    monkeypatch.setattr("litellm.proxy.proxy_server.master_key", None)
+    monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None)
+    monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", llm_router)
+    proxy_logging_obj.update_request_status = mocker.AsyncMock()
+    proxy_logging_obj.post_call_failure_hook = mocker.AsyncMock()
+    app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
+        api_key="test-key", user_role=user_role, user_id="test-user"
     )
+    return proxy_logging_obj
 
-    assert response.status_code == 400
-    assert "managed file id" in response.json()["error"]["message"].lower()
+
+@pytest.mark.parametrize(("route_file_id", "provider"), RAW_CLOUD_FILE_IDS_AS_THE_ROUTE_RECEIVES_THEM)
+def test_get_file_content_answers_403_for_a_raw_cloud_id_from_a_non_admin_key(
+    mocker: MockerFixture,
+    monkeypatch: pytest.MonkeyPatch,
+    llm_router: Router,
+    route_file_id: str,
+    provider: str,
+):
+    import litellm.proxy.proxy_server as ps
+
+    _override_auth(monkeypatch, mocker, llm_router, LitellmUserRoles.INTERNAL_USER)
+    afile_content: Final = mocker.AsyncMock()
+    monkeypatch.setattr(litellm, "afile_content", afile_content)
+
+    try:
+        response: Final = client.get(
+            f"/v1/files/{_testclient_path_segment_delivering(route_file_id)}/content?provider={provider}",
+            headers={"Authorization": "Bearer test-key"},
+        )
+    finally:
+        app.dependency_overrides.pop(ps.user_api_key_auth, None)
+
+    assert response.status_code == 403, response.text
+    assert "proxy admin" in response.json()["error"]["message"]
+    afile_content.assert_not_called()
+
+
+@pytest.mark.parametrize(("route_file_id", "provider"), RAW_CLOUD_FILE_IDS_AS_THE_ROUTE_RECEIVES_THEM)
+def test_get_file_content_forwards_a_raw_cloud_id_from_a_proxy_admin_key(
+    mocker: MockerFixture,
+    monkeypatch: pytest.MonkeyPatch,
+    llm_router: Router,
+    route_file_id: str,
+    provider: str,
+):
+    import litellm.proxy.proxy_server as ps
+
+    _override_auth(monkeypatch, mocker, llm_router, LitellmUserRoles.PROXY_ADMIN)
+    content: Final = b'{"custom_id": "request-1"}\n'
+
+    async def _stream():
+        yield content
+
+    async def _provider_file_content(**kwargs: object) -> FileContentStreamingResult | HttpxBinaryResponseContent:
+        if kwargs.get("stream"):
+            return FileContentStreamingResult(stream_iterator=_stream(), headers={"content-length": str(len(content))})
+        return HttpxBinaryResponseContent(
+            response=httpx.Response(
+                status_code=200, content=content, headers={"content-type": "application/octet-stream"}
+            )
+        )
+
+    afile_content: Final = mocker.AsyncMock(side_effect=_provider_file_content)
+    monkeypatch.setattr(litellm, "afile_content", afile_content)
+
+    try:
+        response: Final = client.get(
+            f"/v1/files/{_testclient_path_segment_delivering(route_file_id)}/content?provider={provider}",
+            headers={"Authorization": "Bearer test-key"},
+        )
+    finally:
+        app.dependency_overrides.pop(ps.user_api_key_auth, None)
+
+    assert response.status_code == 200, response.text
+    assert response.content == content
+    afile_content.assert_called_once()
+    assert afile_content.call_args.kwargs["custom_llm_provider"] == provider
+    assert afile_content.call_args.kwargs["file_id"] == route_file_id
 
 
 def test_mock_create_audio_file(mocker: MockerFixture, monkeypatch, llm_router: Router):