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

litellm/litellm_core_utils/cloud_storage_security.py

@@ -20,15 +20,21 @@ MANAGED_CLOUD_STORAGE_SCHEMES: Final = ("s3://", "gs://")
 _MAPPING_PROXY_TYPE: Final[type] = type(MappingProxyType({}))
 
 
+def _fully_unquoted(value: str) -> str:
+    decoded: Final = unquote(value)
+    return value if decoded == value else _fully_unquoted(decoded)
+
+
 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``),
+    however many times it was percent-encoded on the way in.
 
     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)
+    return isinstance(file_id, str) and _fully_unquoted(file_id).startswith(MANAGED_CLOUD_STORAGE_SCHEMES)
 
 
 _SAFE_OBJECT_COMPONENT_PATTERN: Final = re.compile(r"[^A-Za-z0-9._-]+")

litellm/proxy/openai_files_endpoints/files_endpoints.py

@@ -892,14 +892,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/test_litellm/litellm_core_utils/test_cloud_storage_security.py

@@ -1,3 +1,5 @@
+import pytest
+
 from litellm.litellm_core_utils.cloud_storage_security import (
     is_managed_cloud_storage_uri,
 )
@@ -13,3 +15,16 @@ 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)

tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py

@@ -235,23 +235,96 @@ 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.
-    """
+
+def _raw_cloud_file_ids_in_every_encoding() -> tuple[tuple[str, str, str], ...]:
     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"},
+    return tuple(
+        (encoded_id, raw_id, provider)
+        for raw_id, provider in ((RAW_GCS_OUTPUT_FILE_ID, "vertex_ai"), (RAW_S3_OUTPUT_FILE_ID, "bedrock"))
+        for encoded_id in (quote(raw_id, safe=""), quote(quote(raw_id, safe=""), safe=""))
     )
 
-    assert response.status_code == 400
-    assert "managed file id" in response.json()["error"]["message"].lower()
+
+def _override_auth(monkeypatch, mocker: MockerFixture, llm_router: Router, user_role) -> ProxyLogging:
+    import litellm.proxy.proxy_server as ps
+
+    proxy_logging_obj = 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
+
+
+@pytest.mark.parametrize(("encoded_id", "raw_id", "provider"), _raw_cloud_file_ids_in_every_encoding())
+def test_get_file_content_answers_403_for_a_raw_cloud_id_from_a_non_admin_key(
+    mocker: MockerFixture, monkeypatch, llm_router: Router, encoded_id: str, raw_id: str, provider: str
+):
+    import litellm.proxy.proxy_server as ps
+    from litellm.proxy._types import LitellmUserRoles
+
+    _override_auth(monkeypatch, mocker, llm_router, LitellmUserRoles.INTERNAL_USER)
+    afile_content = mocker.AsyncMock()
+    monkeypatch.setattr(litellm, "afile_content", afile_content)
+
+    try:
+        response = client.get(
+            f"/v1/files/{encoded_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(("encoded_id", "raw_id", "provider"), _raw_cloud_file_ids_in_every_encoding())
+def test_get_file_content_forwards_a_raw_cloud_id_from_a_proxy_admin_key(
+    mocker: MockerFixture, monkeypatch, llm_router: Router, encoded_id: str, raw_id: str, provider: str
+):
+    import litellm.proxy.proxy_server as ps
+    from litellm.proxy._types import LitellmUserRoles
+
+    _override_auth(monkeypatch, mocker, llm_router, LitellmUserRoles.PROXY_ADMIN)
+    captured_kwargs: dict = {}
+
+    async def _mock_afile_content(**kwargs):
+        captured_kwargs.update(kwargs)
+        return HttpxBinaryResponseContent(
+            response=httpx.Response(
+                status_code=200,
+                content=b'{"custom_id": "request-1"}\n',
+                headers={"content-type": "application/octet-stream"},
+            )
+        )
+
+    monkeypatch.setattr(litellm, "afile_content", _mock_afile_content)
+
+    try:
+        response = client.get(
+            f"/v1/files/{encoded_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 == b'{"custom_id": "request-1"}\n'
+    assert captured_kwargs["custom_llm_provider"] == provider
+    assert captured_kwargs["file_id"] == raw_id
 
 
 def test_mock_create_audio_file(mocker: MockerFixture, monkeypatch, llm_router: Router):