From d8faaef3d2880228c86f1d4c05c4f5e6a5880145 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sun, 2 Aug 2026 01:20:20 +0000 Subject: [PATCH] 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> --- .../common_utils/check_responses_cost.py | 3 +- litellm/proxy/hooks/responses_id_security.py | 174 ++++++++------ litellm/types/utils.py | 2 +- .../test_responses_id_security.py | 223 ++++++++++++++++-- 4 files changed, 309 insertions(+), 93 deletions(-) diff --git a/enterprise/litellm_enterprise/proxy/common_utils/check_responses_cost.py b/enterprise/litellm_enterprise/proxy/common_utils/check_responses_cost.py index dc0168683c8..f73adf5fc8b 100644 --- a/enterprise/litellm_enterprise/proxy/common_utils/check_responses_cost.py +++ b/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 = { diff --git a/litellm/proxy/hooks/responses_id_security.py b/litellm/proxy/hooks/responses_id_security.py index cba58506a00..153a0077c49 100644 --- a/litellm/proxy/hooks/responses_id_security.py +++ b/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}" diff --git a/litellm/types/utils.py b/litellm/types/utils.py index 18991f53e6f..b80cf282925 100644 --- a/litellm/types/utils.py +++ b/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. diff --git a/tests/test_litellm/test_responses_id_security.py b/tests/test_litellm/test_responses_id_security.py index 17487030cc1..75368ace70c 100644 --- a/tests/test_litellm/test_responses_id_security.py +++ b/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