From 94193b34c8a39d648565a57c5fe801c7d197a5d1 Mon Sep 17 00:00:00 2001
From: Dominic White <singe-github@singe.za.net>
Date: Mon, 7 Sep 2026 11:14:27 +0200
Subject: [PATCH] fix: remove unchecked safety identifier cast

---
 litellm/litellm_core_utils/safety_identifier.py | 15 ++++++++++++---
 litellm/responses/streaming_iterator.py         |  8 +++-----
 2 files changed, 15 insertions(+), 8 deletions(-)

diff --git a/litellm/litellm_core_utils/safety_identifier.py b/litellm/litellm_core_utils/safety_identifier.py
index b231e096fc5..f7e7954a8fb 100644
--- a/litellm/litellm_core_utils/safety_identifier.py
+++ b/litellm/litellm_core_utils/safety_identifier.py
@@ -1,11 +1,20 @@
 import hashlib
-from collections.abc import MutableMapping
-from typing import Final
+from typing import Final, Protocol
+
+
+class _SafetyIdentifierPayload(Protocol):
+    def get(self, key: str, /) -> object | None: ...
+
+    def __setitem__(self, key: str, value: object, /) -> None: ...
+
+    def __contains__(self, key: object, /) -> bool: ...
+
+    def pop(self, key: str, default: object | None = None, /) -> object | None: ...
 
 
 def enforce_safety_identifier(
     *,
-    data: MutableMapping[str, object],  # mutable-ok: trusted enforcement rewrites the request payload in place
+    data: _SafetyIdentifierPayload,
     user_id: str | None,
     enabled: bool,
 ) -> bool:
diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py
index 49968451e28..14f1af19d23 100644
--- a/litellm/responses/streaming_iterator.py
+++ b/litellm/responses/streaming_iterator.py
@@ -6,11 +6,11 @@ import os
 import time
 import traceback
 import uuid
-from collections.abc import Awaitable, Callable, Iterable, Mapping, MutableMapping, Sequence
+from collections.abc import Awaitable, Callable, Iterable, Mapping, Sequence
 from datetime import datetime
 from functools import lru_cache
 from types import MappingProxyType
-from typing import TYPE_CHECKING, Any, Final, Literal, Protocol, cast, overload, runtime_checkable
+from typing import TYPE_CHECKING, Any, Final, Literal, Protocol, overload, runtime_checkable
 
 import httpx
 from openai._streaming import SSEDecoder
@@ -100,9 +100,7 @@ def _enforce_responses_ws_safety_identifier(
     user_api_key_dict: UserAPIKeyAuth | None,
 ) -> bool:
     return enforce_safety_identifier(
-        data=cast(  # cast-ok: JSON protocol is backed by a mutable response.create dictionary
-            MutableMapping[str, object], msg_obj
-        ),
+        data=msg_obj,
         user_id=user_api_key_dict.user_id if user_api_key_dict is not None else None,
         enabled=str_to_bool(os.getenv("LITELLM_ENFORCE_SAFETY_IDENTIFIER")) is True,
     )