fix(proxy): re-authorize the retained responses id instead of trusting it

litellm/proxy/hooks/responses_id_security.py

@@ -31,7 +31,7 @@ if TYPE_CHECKING:
 _RESPONSES_API_PROVIDER_PREFIX: Final = "/openai"
 _RESPONSES_API_CREATE_ROUTES: Final = frozenset({"/v1/responses", "/responses"})
 
-_AUTHORIZED_RESPONSE_ID_KEY: Final = "_litellm_authorized_response_id"
+_ADDRESSED_RESPONSE_ID_KEY: Final = "_litellm_addressed_response_id"
 _UNMANAGED_RESPONSE_ID_DETAIL: Final = (
     "Forbidden. This response id was not issued by this proxy, so the proxy cannot tell who owns it. "
     "To let keys address responses this proxy did not issue, set "
@@ -93,14 +93,15 @@ class ResponsesIDSecurity(CustomLogger):
         if call_type not in responses_api_call_types:
             return None
         addressed_id_field: Final = "previous_response_id" if call_type == "aresponses" else "response_id"
-        addressed_id: Final = data.get(addressed_id_field)
+        retained_id: Final = data.get(_ADDRESSED_RESPONSE_ID_KEY)
+        addressed_id: Final = (
+            retained_id if isinstance(retained_id, str) and retained_id else data.get(addressed_id_field)
+        )
         if not isinstance(addressed_id, str) or not addressed_id:
             return data
-        if data.get(_AUTHORIZED_RESPONSE_ID_KEY) == addressed_id:
-            return data
         authorized_id: Final = self._authorize_response_id(addressed_id, user_api_key_dict)
         data[addressed_id_field] = authorized_id
-        data[_AUTHORIZED_RESPONSE_ID_KEY] = authorized_id
+        data[_ADDRESSED_RESPONSE_ID_KEY] = addressed_id
         return data
 
     def _authorize_response_id(

tests/test_litellm/test_responses_id_security.py

@@ -913,3 +913,62 @@ class TestUnmanagedResponseIdEscapeHatches:
         )
 
         assert result["response_id"] == _FABRICATED_UNMANAGED_ID
+
+
+class TestClientSuppliedRetainedIdCannotBypassAuthorization:
+    """The retained-id key travels in the request body, so it is re-authorized, never trusted."""
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("call_type", sorted(_ADDRESSED_ID_FIELD_BY_CALL_TYPE))
+    async def test_forged_retained_id_is_still_authorized(self, mock_cache, salt_key_env, call_type):
+        field = _ADDRESSED_ID_FIELD_BY_CALL_TYPE[call_type]
+        data = {
+            field: _FABRICATED_UNMANAGED_ID,
+            "_litellm_addressed_response_id": _FABRICATED_UNMANAGED_ID,
+        }
+
+        with pytest.raises(HTTPException) as exc_info:
+            await _hook().async_pre_call_hook(
+                user_api_key_dict=_auth(),
+                cache=mock_cache,
+                data=data,
+                call_type=call_type,
+            )
+
+        assert exc_info.value.status_code == 403
+        assert data[field] == _FABRICATED_UNMANAGED_ID
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("forged", [{"nested": "value"}, ["list"], 42, "", None])
+    async def test_non_string_retained_id_falls_back_to_the_addressed_field(self, mock_cache, salt_key_env, forged):
+        data = {"response_id": _FABRICATED_UNMANAGED_ID, "_litellm_addressed_response_id": forged}
+
+        with pytest.raises(HTTPException) as exc_info:
+            await _hook().async_pre_call_hook(
+                user_api_key_dict=_auth(),
+                cache=mock_cache,
+                data=data,
+                call_type="aget_responses",
+            )
+
+        assert exc_info.value.status_code == 403
+
+    @pytest.mark.asyncio
+    async def test_stranger_forging_their_own_id_never_reaches_someone_elses_response(
+        self, mock_cache, salt_key_env
+    ):
+        hook = _hook()
+        stranger = _auth(user_id="stranger-user", team_id="stranger-team")
+        stranger_id = _issue_managed_id(hook, stranger, provider_response_id="resp_strangerownprovideridcccccccc")
+        victim_provider_id = "resp_victimprovideriddddddddddddddddddddd"
+        data = {"response_id": victim_provider_id, "_litellm_addressed_response_id": stranger_id}
+
+        result = await hook.async_pre_call_hook(
+            user_api_key_dict=stranger,
+            cache=mock_cache,
+            data=data,
+            call_type="aget_responses",
+        )
+
+        assert result["response_id"] == "resp_strangerownprovideridcccccccc"
+        assert result["response_id"] != victim_provider_id