fix(responses_id_security): fail closed for raw and ownerless response ids

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

enterprise/litellm_enterprise/proxy/common_utils/check_responses_cost.py

@@ -122,7 +122,8 @@ class CheckResponsesCost:
                 model_name = stored_response.get("model", None)
                 
                 # Decrypt the response ID
-                responses_id_security, _, _ = ResponsesIDSecurity()._decrypt_response_id(unified_object_id)
+                decrypted_id = ResponsesIDSecurity()._decrypt_response_id(unified_object_id)
+                responses_id_security = decrypted_id.response_id if decrypted_id else unified_object_id
                 
                 # Prepare metadata with model information for cost tracking
                 litellm_metadata = {

litellm/proxy/hooks/responses_id_security.py

@@ -5,7 +5,8 @@ This hook uses the DBSpendUpdateWriter to batch-write response IDs to the databa
 instead of writing immediately on each request.
 """
 
-from typing import TYPE_CHECKING, Any, AsyncGenerator, Optional, Tuple, Union, cast
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any, AsyncGenerator, Optional, Union, cast
 
 from fastapi import HTTPException
 
@@ -27,6 +28,14 @@ if TYPE_CHECKING:
     from litellm.proxy._types import UserAPIKeyAuth
 
 
+@dataclass(frozen=True, slots=True)
+class DecryptedResponseID:
+    response_id: str
+    user_id: Optional[str]
+    team_id: Optional[str]
+    key_hash: Optional[str]
+
+
 class ResponsesIDSecurity(CustomLogger):
     def __init__(self):
         pass
@@ -51,112 +60,137 @@ class ResponsesIDSecurity(CustomLogger):
         if call_type == "aresponses":
             # check 'previous_response_id' if present in the data
             previous_response_id = data.get("previous_response_id")
-            if previous_response_id and self._is_encrypted_response_id(previous_response_id):
-                original_response_id, user_id, team_id = self._decrypt_response_id(previous_response_id)
-                self.check_user_access_to_response_id(user_id, team_id, user_api_key_dict)
-                data["previous_response_id"] = original_response_id
+            if previous_response_id:
+                data["previous_response_id"] = self._authorize_response_id(previous_response_id, user_api_key_dict)
         elif call_type in {"aget_responses", "adelete_responses", "acancel_responses", "alist_input_items"}:
             response_id = data.get("response_id")
 
-            if response_id and self._is_encrypted_response_id(response_id):
-                original_response_id, user_id, team_id = self._decrypt_response_id(response_id)
-
-                self.check_user_access_to_response_id(user_id, team_id, user_api_key_dict)
-                data["response_id"] = original_response_id
+            if response_id:
+                data["response_id"] = self._authorize_response_id(response_id, user_api_key_dict)
         return data
 
+    def _authorize_response_id(self, response_id: str, user_api_key_dict: "UserAPIKeyAuth") -> str:
+        """
+        Returns the provider-side response id the caller is allowed to act on.
+
+        Any id that is not an ownership-bound id issued by this proxy is rejected, unless the caller is a proxy
+        admin, the feature is disabled, or no signing key is configured (in which case this proxy never issued
+        bound ids in the first place).
+        """
+        decrypted = self._decrypt_response_id(response_id)
+        if decrypted is None:
+            self._reject_unverifiable_response_id(user_api_key_dict)
+            return response_id
+
+        self.check_user_access_to_response_id(
+            decrypted.user_id,
+            decrypted.team_id,
+            user_api_key_dict,
+            response_id_key_hash=decrypted.key_hash,
+        )
+        return decrypted.response_id
+
+    def _reject_unverifiable_response_id(self, user_api_key_dict: "UserAPIKeyAuth") -> None:
+        from litellm.proxy.proxy_server import general_settings
+
+        if self._is_proxy_admin(user_api_key_dict) or general_settings.get("disable_responses_id_security", False):
+            return
+
+        if self._get_signing_key() is None:
+            verbose_proxy_logger.warning(
+                "Responses ID security is enabled but no signing key is configured, so response ids are neither "
+                "encrypted nor ownership-checked. Set LITELLM_SALT_KEY or a master_key. "
+                "See: https://docs.litellm.ai/docs/proxy/prod#5-set-litellm-salt-key"
+            )
+            return
+
+        raise HTTPException(
+            status_code=403,
+            detail="Forbidden. This response id was not issued by this proxy, so its owner cannot be verified. To "
+            "disable this security feature, set general_settings::disable_responses_id_security to True in the "
+            "config.yaml file.",
+        )
+
+    @staticmethod
+    def _is_proxy_admin(user_api_key_dict: "UserAPIKeyAuth") -> bool:
+        return user_api_key_dict.user_role in (
+            LitellmUserRoles.PROXY_ADMIN.value,
+            LitellmUserRoles.PROXY_ADMIN,
+        )
+
     def check_user_access_to_response_id(
         self,
         response_id_user_id: Optional[str],
         response_id_team_id: Optional[str],
         user_api_key_dict: "UserAPIKeyAuth",
+        response_id_key_hash: Optional[str] = None,
     ) -> bool:
         from litellm.proxy.proxy_server import general_settings
 
-        if (
-            user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN.value
-            or user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN
-        ):
+        if self._is_proxy_admin(user_api_key_dict):
+            return True
+
+        if general_settings.get("disable_responses_id_security", False):
+            verbose_proxy_logger.debug(
+                f"Responses ID Security is disabled. User {user_api_key_dict.user_id} is accessing response id owned "
+                f"by user {response_id_user_id} / team {response_id_team_id}."
+            )
+            return True
+
+        if response_id_key_hash and response_id_key_hash == user_api_key_dict.api_key:
             return True
 
         if response_id_user_id and response_id_user_id != user_api_key_dict.user_id:
-            if general_settings.get("disable_responses_id_security", False):
-                verbose_proxy_logger.debug(
-                    f"Responses ID Security is disabled. User {user_api_key_dict.user_id} is accessing response id {response_id_user_id} which is not associated with them."
-                )
-                return True
             raise HTTPException(
                 status_code=403,
                 detail="Forbidden. The response id is not associated with the user, who this key belongs to. To disable this security feature, set general_settings::disable_responses_id_security to True in the config.yaml file.",
             )
 
         if response_id_team_id and response_id_team_id != user_api_key_dict.team_id:
-            if general_settings.get("disable_responses_id_security", False):
-                verbose_proxy_logger.debug(
-                    f"Responses ID Security is disabled. Response belongs to team {response_id_team_id} but user {user_api_key_dict.user_id} is accessing it with team id {user_api_key_dict.team_id}."
-                )
-                return True
             raise HTTPException(
                 status_code=403,
                 detail="Forbidden. The response id is not associated with the team, who this key belongs to. To disable this security feature, set general_settings::disable_responses_id_security to True in the config.yaml file.",
             )
 
+        if not response_id_user_id and not response_id_team_id:
+            raise HTTPException(
+                status_code=403,
+                detail="Forbidden. The response id carries no owner, so it cannot be verified as belonging to this "
+                "key. To disable this security feature, set general_settings::disable_responses_id_security to True "
+                "in the config.yaml file.",
+            )
+
         return True
 
     def _is_encrypted_response_id(self, response_id: str) -> bool:
-        split_result = response_id.split("resp_")
-        if len(split_result) < 2:
-            return False
+        return self._decrypt_response_id(response_id) is not None
 
-        remaining_string = split_result[1]
-        decrypted_value = decrypt_value_helper(value=remaining_string, key="response_id", return_original_value=True)
-
-        if decrypted_value is None:
-            return False
-
-        if decrypted_value.startswith(SpecialEnums.LITELM_MANAGED_FILE_ID_PREFIX.value):
-            return True
-        return False
-
-    def _decrypt_response_id(self, response_id: str) -> Tuple[str, Optional[str], Optional[str]]:
+    def _decrypt_response_id(self, response_id: str) -> Optional[DecryptedResponseID]:
         """
-        Returns:
-         - original_response_id: the original response id
-         - user_id: the user id
-         - team_id: the team id
+        Decrypt an id issued by this proxy, or return None when the id is not one (raw provider id, litellm base64
+        managed id, forged id, or an id encrypted under a different signing key).
         """
         split_result = response_id.split("resp_")
         if len(split_result) < 2:
-            return response_id, None, None
+            return None
 
-        remaining_string = split_result[1]
-        decrypted_value = decrypt_value_helper(value=remaining_string, key="response_id", return_original_value=True)
+        decrypted_value = decrypt_value_helper(value=split_result[1], key="response_id", return_original_value=True)
 
-        if decrypted_value is None:
-            return response_id, None, None
+        if decrypted_value is None or not decrypted_value.startswith(SpecialEnums.LITELM_MANAGED_FILE_ID_PREFIX.value):
+            return None
 
-        if decrypted_value.startswith(SpecialEnums.LITELM_MANAGED_FILE_ID_PREFIX.value):
-            # Expected format: "litellm_proxy:responses_api:response_id:{response_id};user_id:{user_id}"
-            parts = decrypted_value.split(";")
+        # Format: "litellm_proxy:responses_api:response_id:{};user_id:{};team_id:{};key_hash:{}"
+        # key_hash is absent on ids issued before it was added.
+        parts = decrypted_value.split(";")
+        if len(parts) < 3:
+            return None
 
-            if len(parts) >= 2:
-                # Extract response_id from "litellm_proxy:responses_api:response_id:{response_id}"
-                response_id_part = parts[0]
-                original_response_id = response_id_part.split("response_id:")[-1]
-
-                # Extract user_id from "user_id:{user_id}"
-                user_id_part = parts[1]
-                user_id = user_id_part.split("user_id:")[-1]
-
-                # Extract team_id from "team_id:{team_id}"
-                team_id_part = parts[2]
-                team_id = team_id_part.split("team_id:")[-1]
-
-                return original_response_id, user_id, team_id
-            else:
-                # Fallback if format is unexpected
-                return response_id, None, None
-        return response_id, None, None
+        return DecryptedResponseID(
+            response_id=parts[0].split("response_id:")[-1],
+            user_id=parts[1].split("user_id:")[-1] or None,
+            team_id=parts[2].split("team_id:")[-1] or None,
+            key_hash=(parts[3].split("key_hash:")[-1] or None) if len(parts) > 3 else None,
+        )
 
     def _get_signing_key(self) -> Optional[str]:
         """Get the signing key for encryption/decryption."""
@@ -181,7 +215,7 @@ class ResponsesIDSecurity(CustomLogger):
         # Check if signing key is available
         signing_key = self._get_signing_key()
         if signing_key is None:
-            verbose_proxy_logger.debug(
+            verbose_proxy_logger.warning(
                 "Response ID encryption is enabled but no signing key is configured. "
                 "Please set LITELLM_SALT_KEY environment variable or configure a master_key. "
                 "Skipping response ID encryption. "
@@ -201,6 +235,7 @@ class ResponsesIDSecurity(CustomLogger):
                     response_id,
                     user_api_key_dict.user_id or "",
                     user_api_key_dict.team_id or "",
+                    user_api_key_dict.api_key or "",
                 )
 
                 encoded_user_id_and_response_id = encrypt_value_helper(value=encrypted_response_id)
@@ -218,6 +253,7 @@ class ResponsesIDSecurity(CustomLogger):
                     response_obj.id,
                     user_api_key_dict.user_id or "",
                     user_api_key_dict.team_id or "",
+                    user_api_key_dict.api_key or "",
                 )
                 encoded_user_id_and_response_id = encrypt_value_helper(value=encrypted_response_id)
                 encrypted_id = f"resp_{encoded_user_id_and_response_id}"

litellm/types/utils.py

@@ -3831,7 +3831,7 @@ class SpecialEnums(Enum):
     LITELLM_MANAGED_BATCH_COMPLETE_STR = "litellm_proxy;model_id:{};llm_batch_id:{}"
 
     LITELLM_MANAGED_RESPONSE_API_RESPONSE_ID_COMPLETE_STR = (
-        "litellm_proxy:responses_api:response_id:{};user_id:{};team_id:{}"
+        "litellm_proxy:responses_api:response_id:{};user_id:{};team_id:{};key_hash:{}"
     )
 
     LITELLM_MANAGED_GENERIC_RESPONSE_COMPLETE_STR = "litellm_proxy;model_id:{};generic_response_id:{}"  # generic implementation of 'managed batches' - used for finetuning and any future work.

tests/test_litellm/test_responses_id_security.py

@@ -9,7 +9,10 @@ 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 (
+    DecryptedResponseID,
+    ResponsesIDSecurity,
+)
 from litellm.types.llms.openai import ResponsesAPIResponse
 from litellm.types.utils import SpecialEnums
 
@@ -46,7 +49,7 @@ class TestIsEncryptedResponseId:
         import litellm.proxy.hooks.responses_id_security as responses_module
 
         with patch.object(responses_module, "decrypt_value_helper") as mock_decrypt:
-            mock_decrypt.return_value = f"{SpecialEnums.LITELM_MANAGED_FILE_ID_PREFIX.value}response_id:resp_123;user_id:user-456"
+            mock_decrypt.return_value = f"{SpecialEnums.LITELM_MANAGED_FILE_ID_PREFIX.value}response_id:resp_123;user_id:user-456;team_id:team-789"
 
             result = responses_id_security._is_encrypted_response_id(
                 "resp_encrypted_value"
@@ -77,15 +80,36 @@ class TestDecryptResponseId:
         import litellm.proxy.hooks.responses_id_security as responses_module
 
         with patch.object(responses_module, "decrypt_value_helper") as mock_decrypt:
-            mock_decrypt.return_value = f"{SpecialEnums.LITELM_MANAGED_FILE_ID_PREFIX.value}response_id:resp_original_123;user_id:user-456;team_id:team-789"
+            mock_decrypt.return_value = f"{SpecialEnums.LITELM_MANAGED_FILE_ID_PREFIX.value}response_id:resp_original_123;user_id:user-456;team_id:team-789;key_hash:hashed-key-1"
 
-            original_id, user_id, team_id = responses_id_security._decrypt_response_id(
+            decrypted = responses_id_security._decrypt_response_id(
                 "resp_encrypted_value"
             )
 
-            assert original_id == "resp_original_123"
-            assert user_id == "user-456"
-            assert team_id == "team-789"
+            assert decrypted == DecryptedResponseID(
+                response_id="resp_original_123",
+                user_id="user-456",
+                team_id="team-789",
+                key_hash="hashed-key-1",
+            )
+
+    def test_decrypt_response_id_without_key_hash(self, responses_id_security):
+        """Ids issued before key_hash was added must still decrypt"""
+        import litellm.proxy.hooks.responses_id_security as responses_module
+
+        with patch.object(responses_module, "decrypt_value_helper") as mock_decrypt:
+            mock_decrypt.return_value = f"{SpecialEnums.LITELM_MANAGED_FILE_ID_PREFIX.value}response_id:resp_original_123;user_id:user-456;team_id:team-789"
+
+            decrypted = responses_id_security._decrypt_response_id(
+                "resp_encrypted_value"
+            )
+
+            assert decrypted == DecryptedResponseID(
+                response_id="resp_original_123",
+                user_id="user-456",
+                team_id="team-789",
+                key_hash=None,
+            )
 
     def test_decrypt_response_id_no_encryption(self, responses_id_security):
         """Test decrypting a non-encrypted response ID"""
@@ -95,14 +119,10 @@ class TestDecryptResponseId:
         with patch.object(responses_module, "decrypt_value_helper") as mock_decrypt:
             mock_decrypt.return_value = None
 
-            original_id, user_id, team_id = responses_id_security._decrypt_response_id(
-                "resp_plain_value"
+            assert (
+                responses_id_security._decrypt_response_id("resp_plain_value") is None
             )
 
-            assert original_id == "resp_plain_value"
-            assert user_id is None
-            assert team_id is None
-
 
 class TestEncryptResponseId:
     """Test _encrypt_response_id function"""
@@ -320,7 +340,7 @@ class TestAsyncPreCallHook:
             with patch.object(
                 responses_id_security,
                 "_decrypt_response_id",
-                return_value=("resp_original_123", "test-user-123", "test-team-123"),
+                return_value=DecryptedResponseID("resp_original_123", "test-user-123", "test-team-123", None),
             ):
                 result = await responses_id_security.async_pre_call_hook(
                     user_api_key_dict=mock_user_api_key_dict,
@@ -344,7 +364,7 @@ class TestAsyncPreCallHook:
             with patch.object(
                 responses_id_security,
                 "_decrypt_response_id",
-                return_value=("resp_original_456", "test-user-123", "test-team-123"),
+                return_value=DecryptedResponseID("resp_original_456", "test-user-123", "test-team-123", None),
             ):
                 result = await responses_id_security.async_pre_call_hook(
                     user_api_key_dict=mock_user_api_key_dict,
@@ -374,7 +394,7 @@ class TestAsyncPreCallHook:
             with patch.object(
                 responses_id_security,
                 "_decrypt_response_id",
-                return_value=("resp_original_team_b", None, "team-b"),
+                return_value=DecryptedResponseID("resp_original_team_b", None, "team-b", None),
             ):
                 with patch("litellm.proxy.proxy_server.general_settings", {}):
                     with pytest.raises(HTTPException) as exc_info:
@@ -407,7 +427,7 @@ class TestAsyncPreCallHook:
             with patch.object(
                 responses_id_security,
                 "_decrypt_response_id",
-                return_value=("resp_original_team_b", "user-from-team-b", "team-b"),
+                return_value=DecryptedResponseID("resp_original_team_b", "user-from-team-b", "team-b", None),
             ):
                 with patch("litellm.proxy.proxy_server.general_settings", {}):
                     with pytest.raises(HTTPException) as exc_info:
@@ -441,7 +461,7 @@ class TestAsyncPreCallHook:
             with patch.object(
                 responses_id_security,
                 "_decrypt_response_id",
-                return_value=("resp_original_team_a", None, "team-a"),
+                return_value=DecryptedResponseID("resp_original_team_a", None, "team-a", None),
             ):
                 result = await responses_id_security.async_pre_call_hook(
                     user_api_key_dict=mock_auth_team_a,
@@ -471,7 +491,7 @@ class TestAsyncPreCallHook:
             with patch.object(
                 responses_id_security,
                 "_decrypt_response_id",
-                return_value=("resp_original_team_b", None, "team-b"),
+                return_value=DecryptedResponseID("resp_original_team_b", None, "team-b", None),
             ):
                 with patch("litellm.proxy.proxy_server.general_settings", {}):
                     with pytest.raises(HTTPException) as exc_info:
@@ -504,7 +524,7 @@ class TestAsyncPreCallHook:
             with patch.object(
                 responses_id_security,
                 "_decrypt_response_id",
-                return_value=("resp_original_team_b", None, "team-b"),
+                return_value=DecryptedResponseID("resp_original_team_b", None, "team-b", None),
             ):
                 with patch("litellm.proxy.proxy_server.general_settings", {}):
                     with pytest.raises(HTTPException) as exc_info:
@@ -531,7 +551,7 @@ class TestAsyncPreCallHook:
             with patch.object(
                 responses_id_security,
                 "_decrypt_response_id",
-                return_value=("resp_original_789", "test-user-123", "test-team-123"),
+                return_value=DecryptedResponseID("resp_original_789", "test-user-123", "test-team-123", None),
             ):
                 result = await responses_id_security.async_pre_call_hook(
                     user_api_key_dict=mock_user_api_key_dict,
@@ -560,7 +580,7 @@ class TestAsyncPreCallHook:
             with patch.object(
                 responses_id_security,
                 "_decrypt_response_id",
-                return_value=("resp_original_team_b", None, "team-b"),
+                return_value=DecryptedResponseID("resp_original_team_b", None, "team-b", None),
             ):
                 with patch("litellm.proxy.proxy_server.general_settings", {}):
                     with pytest.raises(HTTPException) as exc_info:
@@ -617,3 +637,162 @@ class TestAsyncPostCallSuccessHook:
         )
 
         assert result == mock_response
+
+
+class TestUnverifiableResponseIds:
+    """Ids that were not issued by this proxy carry no ownership binding, so they must be rejected"""
+
+    @pytest.fixture
+    def tenant_key(self):
+        from litellm.proxy._types import UserAPIKeyAuth
+
+        return UserAPIKeyAuth(api_key="hashed-key-a", user_id="user-a", team_id="team-a")
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "call_type, id_field",
+        [
+            ("aresponses", "previous_response_id"),
+            ("aget_responses", "response_id"),
+            ("adelete_responses", "response_id"),
+            ("acancel_responses", "response_id"),
+            ("alist_input_items", "response_id"),
+        ],
+    )
+    async def test_raw_response_id_is_rejected(
+        self, responses_id_security, mock_cache, tenant_key, call_type, id_field, monkeypatch
+    ):
+        monkeypatch.setenv("LITELLM_SALT_KEY", "test-salt-key")
+
+        with patch("litellm.proxy.proxy_server.general_settings", {}):
+            with pytest.raises(HTTPException) as exc_info:
+                await responses_id_security.async_pre_call_hook(
+                    user_api_key_dict=tenant_key,
+                    cache=mock_cache,
+                    data={id_field: "resp_raw_provider_id"},
+                    call_type=call_type,
+                )
+
+        assert exc_info.value.status_code == 403
+        assert "not issued by this proxy" in exc_info.value.detail
+
+    @pytest.mark.asyncio
+    async def test_raw_response_id_allowed_for_proxy_admin(
+        self, responses_id_security, mock_cache, monkeypatch
+    ):
+        from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
+
+        monkeypatch.setenv("LITELLM_SALT_KEY", "test-salt-key")
+        admin_key = UserAPIKeyAuth(api_key="hashed-admin", user_role=LitellmUserRoles.PROXY_ADMIN.value)
+
+        with patch("litellm.proxy.proxy_server.general_settings", {}):
+            result = await responses_id_security.async_pre_call_hook(
+                user_api_key_dict=admin_key,
+                cache=mock_cache,
+                data={"response_id": "resp_raw_provider_id"},
+                call_type="aget_responses",
+            )
+
+        assert result["response_id"] == "resp_raw_provider_id"
+
+    @pytest.mark.asyncio
+    async def test_raw_response_id_allowed_when_security_disabled(
+        self, responses_id_security, mock_cache, tenant_key, monkeypatch
+    ):
+        monkeypatch.setenv("LITELLM_SALT_KEY", "test-salt-key")
+
+        with patch(
+            "litellm.proxy.proxy_server.general_settings",
+            {"disable_responses_id_security": True},
+        ):
+            result = await responses_id_security.async_pre_call_hook(
+                user_api_key_dict=tenant_key,
+                cache=mock_cache,
+                data={"response_id": "resp_raw_provider_id"},
+                call_type="aget_responses",
+            )
+
+        assert result["response_id"] == "resp_raw_provider_id"
+
+    @pytest.mark.asyncio
+    async def test_raw_response_id_allowed_when_no_signing_key(
+        self, responses_id_security, mock_cache, tenant_key, monkeypatch
+    ):
+        monkeypatch.delenv("LITELLM_SALT_KEY", raising=False)
+
+        with patch("litellm.proxy.proxy_server.master_key", None):
+            with patch("litellm.proxy.proxy_server.general_settings", {}):
+                result = await responses_id_security.async_pre_call_hook(
+                    user_api_key_dict=tenant_key,
+                    cache=mock_cache,
+                    data={"response_id": "resp_raw_provider_id"},
+                    call_type="aget_responses",
+                )
+
+        assert result["response_id"] == "resp_raw_provider_id"
+
+
+class TestOwnerlessResponseIds:
+    """A key with no user_id and no team_id must not hand every other key access to its responses"""
+
+    @staticmethod
+    def _issue_id(responses_id_security, issuing_key) -> str:
+        response = ResponsesAPIResponse(
+            id="resp_provider_original", created_at=1234567890, output=[], status="completed"
+        )
+        return responses_id_security._encrypt_response_id(response, issuing_key).id
+
+    @pytest.mark.asyncio
+    async def test_issuing_key_can_reuse_its_own_ownerless_id(
+        self, responses_id_security, mock_cache, monkeypatch
+    ):
+        from litellm.proxy._types import UserAPIKeyAuth
+
+        monkeypatch.setenv("LITELLM_SALT_KEY", "test-salt-key")
+        issuing_key = UserAPIKeyAuth(api_key="hashed-key-1")
+        encrypted_id = self._issue_id(responses_id_security, issuing_key)
+
+        with patch("litellm.proxy.proxy_server.general_settings", {}):
+            result = await responses_id_security.async_pre_call_hook(
+                user_api_key_dict=issuing_key,
+                cache=mock_cache,
+                data={"response_id": encrypted_id},
+                call_type="aget_responses",
+            )
+
+        assert result["response_id"] == "resp_provider_original"
+
+    @pytest.mark.asyncio
+    async def test_other_ownerless_key_is_rejected(
+        self, responses_id_security, mock_cache, monkeypatch
+    ):
+        from litellm.proxy._types import UserAPIKeyAuth
+
+        monkeypatch.setenv("LITELLM_SALT_KEY", "test-salt-key")
+        encrypted_id = self._issue_id(responses_id_security, UserAPIKeyAuth(api_key="hashed-key-1"))
+
+        with patch("litellm.proxy.proxy_server.general_settings", {}):
+            with pytest.raises(HTTPException) as exc_info:
+                await responses_id_security.async_pre_call_hook(
+                    user_api_key_dict=UserAPIKeyAuth(api_key="hashed-key-2"),
+                    cache=mock_cache,
+                    data={"response_id": encrypted_id},
+                    call_type="aget_responses",
+                )
+
+        assert exc_info.value.status_code == 403
+        assert "no owner" in exc_info.value.detail
+
+    def test_legacy_ownerless_id_without_key_hash_is_rejected(self, responses_id_security):
+        from litellm.proxy._types import UserAPIKeyAuth
+
+        with patch("litellm.proxy.proxy_server.general_settings", {}):
+            with pytest.raises(HTTPException) as exc_info:
+                responses_id_security.check_user_access_to_response_id(
+                    response_id_user_id=None,
+                    response_id_team_id=None,
+                    user_api_key_dict=UserAPIKeyAuth(api_key="hashed-key-1"),
+                )
+
+        assert exc_info.value.status_code == 403
+        assert "no owner" in exc_info.value.detail