fix(proxy): encrypt streamed responses ids on /openai/v1/responses and /responses aliases

The streaming security hook only encrypted response ids when request_route
matched "/v1/responses" exactly, so streamed creates on the /openai/v1/responses
and /responses aliases leaked the plain managed id. A second virtual key could
GET, continue, and DELETE another key's response. Normalize the route (strip the
provider prefix, accept the /responses alias) before gating, mirroring the
non-streaming hook which has no route gate.

litellm/proxy/hooks/responses_id_security.py

@@ -28,6 +28,21 @@ if TYPE_CHECKING:
     from litellm.proxy._types import UserAPIKeyAuth
 
 
+_RESPONSES_API_PROVIDER_PREFIX: Final = "/openai"
+_RESPONSES_API_CREATE_ROUTES: Final = frozenset({"/v1/responses", "/responses"})
+
+
+def _is_responses_api_create_route(request_route: str | None) -> bool:
+    if request_route is None:
+        return False
+    canonical: Final = (
+        request_route[len(_RESPONSES_API_PROVIDER_PREFIX) :]
+        if request_route.startswith(_RESPONSES_API_PROVIDER_PREFIX + "/")
+        else request_route
+    )
+    return canonical in _RESPONSES_API_CREATE_ROUTES
+
+
 class ResponsesIDSecurity(CustomLogger):
     def __init__(self):
         pass
@@ -267,8 +282,7 @@ class ResponsesIDSecurity(CustomLogger):
         async for chunk in response:
             if (
                 isinstance(chunk, BaseLiteLLMOpenAIResponseObject)
-                and user_api_key_dict.request_route
-                == "/v1/responses"  # only encrypt the response id for the responses api
+                and _is_responses_api_create_route(user_api_key_dict.request_route)
                 and not general_settings.get("disable_responses_id_security", False)
             ):
                 chunk = self._encrypt_response_id(chunk, user_api_key_dict, request_encryption_cache)

tests/test_litellm/test_responses_id_security.py

@@ -9,7 +9,11 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 from fastapi import HTTPException
 
-from litellm.proxy.hooks.responses_id_security import ResponsesIDSecurity
+from litellm.proxy.hooks.responses_id_security import (
+    ResponsesIDSecurity,
+    _is_responses_api_create_route,
+)
+from litellm.types.llms.base import BaseLiteLLMOpenAIResponseObject
 from litellm.types.llms.openai import ResponsesAPIResponse
 from litellm.types.utils import SpecialEnums
 
@@ -575,6 +579,97 @@ class TestAsyncPreCallHook:
                     assert "team" in exc_info.value.detail.lower()
 
 
+class TestIsResponsesApiCreateRoute:
+    """Test the route gate that decides whether a streamed response id is encrypted."""
+
+    @pytest.mark.parametrize(
+        "route",
+        [
+            "/v1/responses",
+            "/responses",
+            "/openai/v1/responses",
+        ],
+    )
+    def test_create_routes_match(self, route):
+        assert _is_responses_api_create_route(route) is True
+
+    @pytest.mark.parametrize(
+        "route",
+        [
+            None,
+            "/chat/completions",
+            "/openai/v1/chat/completions",
+            "/v1/responses/{response_id}",
+            "/openai/v1/responses/{response_id}",
+            "/v1/responsesX",
+            "/responsesX",
+        ],
+    )
+    def test_non_create_routes_do_not_match(self, route):
+        assert _is_responses_api_create_route(route) is False
+
+
+class TestAsyncPostCallStreamingIteratorHook:
+    """Regression test for LIT-6167: streamed responses on /openai/v1/responses and
+    /responses must have their ids security-encrypted, not just on the exact
+    /v1/responses path. Uses real encryption so the id must round-trip back to the
+    raw provider id plus the caller's user/team, which is the access-control wrapper
+    the aliases were leaking without."""
+
+    @staticmethod
+    async def _agen(chunks):
+        for chunk in chunks:
+            yield chunk
+
+    async def _drain_streamed_id(self, responses_id_security, route, monkeypatch):
+        monkeypatch.setenv("LITELLM_SALT_KEY", "sk-test-salt-key-abcdefghij")
+        chunk = BaseLiteLLMOpenAIResponseObject(id="resp_rawprovider123")
+
+        mock_auth = MagicMock()
+        mock_auth.user_id = "user-a"
+        mock_auth.team_id = "team-a"
+        mock_auth.request_route = route
+
+        collected = [
+            out
+            async for out in responses_id_security.async_post_call_streaming_iterator_hook(
+                user_api_key_dict=mock_auth,
+                response=self._agen([chunk]),
+                request_data={},
+            )
+        ]
+        return collected[0].id
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "route",
+        ["/v1/responses", "/responses", "/openai/v1/responses"],
+    )
+    async def test_streamed_id_encrypted_on_all_responses_routes(
+        self, responses_id_security, route, monkeypatch
+    ):
+        streamed_id = await self._drain_streamed_id(responses_id_security, route, monkeypatch)
+
+        assert streamed_id != "resp_rawprovider123"
+        assert responses_id_security._is_encrypted_response_id(streamed_id)
+        assert responses_id_security._decrypt_response_id(streamed_id) == (
+            "resp_rawprovider123",
+            "user-a",
+            "team-a",
+        )
+
+    @pytest.mark.asyncio
+    async def test_streamed_id_untouched_on_non_responses_route(
+        self, responses_id_security, monkeypatch
+    ):
+        streamed_id = await self._drain_streamed_id(
+            responses_id_security, "/chat/completions", monkeypatch
+        )
+
+        assert streamed_id == "resp_rawprovider123"
+        assert not responses_id_security._is_encrypted_response_id(streamed_id)
+
+
 class TestAsyncPostCallSuccessHook:
     """Test async_post_call_success_hook function"""