From 74a49b527c6db53bb8f89d83cdebec576589b62b Mon Sep 17 00:00:00 2001
From: user <70670632+stuxf@users.noreply.github.com>
Date: Thu, 16 Apr 2026 02:24:10 +0000
Subject: [PATCH 01/15] fix(proxy): read guardrail config from admin metadata,
 fix tag routing consistency

Read guardrail control flags (disable_global_guardrails, opted_out_global_guardrails)
from admin-configured key metadata instead of the request body. This ensures
callers cannot override admin security policies.

Fix tag-based routing to enforce strict tag checks regardless of whether the
request includes tags. Fix budget limiter to use the same dynamic metadata
key resolution as the tag router for consistent tag extraction.
---
 litellm/integrations/custom_guardrail.py     | 29 ++++++++++++--------
 litellm/proxy/litellm_pre_call_utils.py      |  6 ++++
 litellm/router_strategy/budget_limiter.py    | 24 ++++++++++++----
 litellm/router_strategy/tag_based_routing.py |  4 +--
 4 files changed, 42 insertions(+), 21 deletions(-)

diff --git a/litellm/integrations/custom_guardrail.py b/litellm/integrations/custom_guardrail.py
index 6046f1bb581..1e6a46044e0 100644
--- a/litellm/integrations/custom_guardrail.py
+++ b/litellm/integrations/custom_guardrail.py
@@ -257,24 +257,24 @@ class CustomGuardrail(CustomLogger):
 
     def get_disable_global_guardrail(self, data: dict) -> Optional[bool]:
         """
-        Returns True if the global guardrail should be disabled
+        Returns True if the global guardrail should be disabled.
+
+        Reads from admin-configured key/team metadata only, not from
+        the request body, to prevent callers from disabling guardrails.
         """
-        if "disable_global_guardrails" in data:
-            return data["disable_global_guardrails"]
         metadata = data.get("litellm_metadata") or data.get("metadata", {})
-        if "disable_global_guardrails" in metadata:
-            return metadata["disable_global_guardrails"]
-        return False
+        admin_metadata = metadata.get("user_api_key_metadata") or {}
+        return admin_metadata.get("disable_global_guardrails", False)
 
     def get_opted_out_global_guardrails_from_metadata(self, data: dict) -> List[str]:
         """
         Returns the list of global guardrail names the team/key has opted out of.
+
+        Reads from admin-configured key/team metadata only.
         """
-        if "opted_out_global_guardrails" in data:
-            value = data["opted_out_global_guardrails"]
-            return value if isinstance(value, list) else []
         metadata = data.get("litellm_metadata") or data.get("metadata", {})
-        value = metadata.get("opted_out_global_guardrails")
+        admin_metadata = metadata.get("user_api_key_metadata") or {}
+        value = admin_metadata.get("opted_out_global_guardrails")
         return value if isinstance(value, list) else []
 
     def _is_valid_response_type(self, result: Any) -> bool:
@@ -417,7 +417,9 @@ class CustomGuardrail(CustomLogger):
         """
         requested_guardrails = self.get_guardrail_from_metadata(data)
         disable_global_guardrail = self.get_disable_global_guardrail(data)
-        opted_out_global_guardrails = self.get_opted_out_global_guardrails_from_metadata(data)
+        opted_out_global_guardrails = (
+            self.get_opted_out_global_guardrails_from_metadata(data)
+        )
         verbose_logger.debug(
             "inside should_run_guardrail for guardrail=%s event_type= %s guardrail_supported_event_hooks= %s requested_guardrails= %s self.default_on= %s",
             self.guardrail_name,
@@ -426,7 +428,10 @@ class CustomGuardrail(CustomLogger):
             requested_guardrails,
             self.default_on,
         )
-        if self.default_on is True and self.guardrail_name in opted_out_global_guardrails:
+        if (
+            self.default_on is True
+            and self.guardrail_name in opted_out_global_guardrails
+        ):
             return False
 
         if self.default_on is True and disable_global_guardrail is not True:
diff --git a/litellm/proxy/litellm_pre_call_utils.py b/litellm/proxy/litellm_pre_call_utils.py
index b0adf7aa6ee..95e1cbed44e 100644
--- a/litellm/proxy/litellm_pre_call_utils.py
+++ b/litellm/proxy/litellm_pre_call_utils.py
@@ -977,6 +977,12 @@ async def add_litellm_data_to_request(  # noqa: PLR0915
             "Setting client-provided x-api-key as api_key parameter (will override deployment key)"
         )
 
+    # Strip internal pipeline state from user input
+    for _meta_key in ("metadata", "litellm_metadata"):
+        _user_meta = data.get(_meta_key)
+        if isinstance(_user_meta, dict):
+            _user_meta.pop("_pipeline_managed_guardrails", None)
+
     ##########################################################
     # Init - Proxy Server Request
     # we do this as soon as entering so we track the original request
diff --git a/litellm/router_strategy/budget_limiter.py b/litellm/router_strategy/budget_limiter.py
index 64dc5fe4741..261e659644a 100644
--- a/litellm/router_strategy/budget_limiter.py
+++ b/litellm/router_strategy/budget_limiter.py
@@ -29,6 +29,9 @@ from litellm.caching.redis_cache import RedisPipelineIncrementOperation
 from litellm.integrations.custom_logger import CustomLogger, Span
 from litellm.litellm_core_utils.duration_parser import duration_in_seconds
 from litellm.router_strategy.tag_based_routing import _get_tags_from_request_kwargs
+from litellm.litellm_core_utils.core_helpers import (
+    get_metadata_variable_name_from_kwargs,
+)
 from litellm.router_utils.cooldown_callbacks import (
     _get_prometheus_logger_from_callbacks,
 )
@@ -100,9 +103,9 @@ class RouterBudgetLimiting(CustomLogger):
         self.dual_cache = dual_cache
         self.redis_increment_operation_queue: List[RedisPipelineIncrementOperation] = []
         asyncio.create_task(self.periodic_sync_in_memory_spend_with_redis())
-        self.provider_budget_config: Optional[
-            GenericBudgetConfigType
-        ] = provider_budget_config
+        self.provider_budget_config: Optional[GenericBudgetConfigType] = (
+            provider_budget_config
+        )
         self.deployment_budget_config: Optional[GenericBudgetConfigType] = None
         self.tag_budget_config: Optional[GenericBudgetConfigType] = None
         self._init_provider_budgets()
@@ -175,7 +178,10 @@ class RouterBudgetLimiting(CustomLogger):
                 spend_map=spend_map,
                 potential_deployments=potential_deployments,
                 request_tags=_get_tags_from_request_kwargs(
-                    request_kwargs=request_kwargs
+                    request_kwargs=request_kwargs,
+                    metadata_variable_name=get_metadata_variable_name_from_kwargs(
+                        request_kwargs or {}
+                    ),
                 ),
             )
 
@@ -333,7 +339,10 @@ class RouterBudgetLimiting(CustomLogger):
             # Check tag budgets
             if self.tag_budget_config:
                 request_tags = _get_tags_from_request_kwargs(
-                    request_kwargs=request_kwargs
+                    request_kwargs=request_kwargs,
+                    metadata_variable_name=get_metadata_variable_name_from_kwargs(
+                        request_kwargs or {}
+                    ),
                 )
                 for _tag in request_tags:
                     _tag_budget_config = self._get_budget_config_for_tag(_tag)
@@ -459,7 +468,10 @@ class RouterBudgetLimiting(CustomLogger):
                 response_cost=response_cost,
             )
 
-        request_tags = _get_tags_from_request_kwargs(kwargs)
+        request_tags = _get_tags_from_request_kwargs(
+            kwargs,
+            metadata_variable_name=get_metadata_variable_name_from_kwargs(kwargs or {}),
+        )
         if len(request_tags) > 0:
             for _tag in request_tags:
                 _tag_budget_config = self._get_budget_config_for_tag(_tag)
diff --git a/litellm/router_strategy/tag_based_routing.py b/litellm/router_strategy/tag_based_routing.py
index 1188ce9d592..b0b154fc778 100644
--- a/litellm/router_strategy/tag_based_routing.py
+++ b/litellm/router_strategy/tag_based_routing.py
@@ -106,9 +106,7 @@ def _match_deployment(
     # the strict tag check has already failed (step 1 returned None).  Allow
     # the regex to fire only when the deployment has NO plain tags, so we never
     # use regex as a backdoor around the operator's strict-tag policy.
-    strict_tag_check_failed = (
-        not match_any and bool(deployment_tags) and bool(request_tags)
-    )
+    strict_tag_check_failed = not match_any and bool(deployment_tags)
     if deployment_tag_regex and header_strings and not strict_tag_check_failed:
         regex_match = _is_valid_deployment_tag_regex(
             deployment_tag_regex, header_strings

From 3cd5796fc7947d0302d27d1ba214c68d0f663007 Mon Sep 17 00:00:00 2001
From: user <70670632+stuxf@users.noreply.github.com>
Date: Thu, 16 Apr 2026 02:28:23 +0000
Subject: [PATCH 02/15] refactor: extract admin metadata helper, hoist
 loop-invariant tag resolution

Extract _get_admin_metadata() in CustomGuardrail to deduplicate metadata
lookup. Hoist tag resolution above the deployment loop in budget limiter.
Update stale comment in tag routing.
---
 litellm/integrations/custom_guardrail.py     | 14 +++++----
 litellm/router_strategy/budget_limiter.py    | 30 +++++++++++---------
 litellm/router_strategy/tag_based_routing.py |  8 +++---
 3 files changed, 29 insertions(+), 23 deletions(-)

diff --git a/litellm/integrations/custom_guardrail.py b/litellm/integrations/custom_guardrail.py
index 1e6a46044e0..4c294f6272e 100644
--- a/litellm/integrations/custom_guardrail.py
+++ b/litellm/integrations/custom_guardrail.py
@@ -255,6 +255,12 @@ class CustomGuardrail(CustomLogger):
                     f"Event hook {event_hook} is not in the supported event hooks {supported_event_hooks}"
                 )
 
+    @staticmethod
+    def _get_admin_metadata(data: dict) -> dict:
+        """Return the admin-configured key/team metadata from the request data."""
+        metadata = data.get("litellm_metadata") or data.get("metadata", {})
+        return metadata.get("user_api_key_metadata") or {}
+
     def get_disable_global_guardrail(self, data: dict) -> Optional[bool]:
         """
         Returns True if the global guardrail should be disabled.
@@ -262,9 +268,7 @@ class CustomGuardrail(CustomLogger):
         Reads from admin-configured key/team metadata only, not from
         the request body, to prevent callers from disabling guardrails.
         """
-        metadata = data.get("litellm_metadata") or data.get("metadata", {})
-        admin_metadata = metadata.get("user_api_key_metadata") or {}
-        return admin_metadata.get("disable_global_guardrails", False)
+        return self._get_admin_metadata(data).get("disable_global_guardrails", False)
 
     def get_opted_out_global_guardrails_from_metadata(self, data: dict) -> List[str]:
         """
@@ -272,9 +276,7 @@ class CustomGuardrail(CustomLogger):
 
         Reads from admin-configured key/team metadata only.
         """
-        metadata = data.get("litellm_metadata") or data.get("metadata", {})
-        admin_metadata = metadata.get("user_api_key_metadata") or {}
-        value = admin_metadata.get("opted_out_global_guardrails")
+        value = self._get_admin_metadata(data).get("opted_out_global_guardrails")
         return value if isinstance(value, list) else []
 
     def _is_valid_response_type(self, result: Any) -> bool:
diff --git a/litellm/router_strategy/budget_limiter.py b/litellm/router_strategy/budget_limiter.py
index 261e659644a..be27b852478 100644
--- a/litellm/router_strategy/budget_limiter.py
+++ b/litellm/router_strategy/budget_limiter.py
@@ -310,6 +310,16 @@ class RouterBudgetLimiting(CustomLogger):
         deployment_configs: Dict[str, GenericBudgetInfo] = {}
         deployment_providers: List[Optional[str]] = []
 
+        # Resolve tags once before the loop (loop-invariant)
+        _request_tags: List[str] = []
+        if self.tag_budget_config:
+            _request_tags = _get_tags_from_request_kwargs(
+                request_kwargs=request_kwargs,
+                metadata_variable_name=get_metadata_variable_name_from_kwargs(
+                    request_kwargs or {}
+                ),
+            )
+
         for deployment in healthy_deployments:
             # Check provider budgets
             if self.provider_budget_config:
@@ -336,20 +346,14 @@ class RouterBudgetLimiting(CustomLogger):
                         cache_keys.append(
                             f"deployment_spend:{model_id}:{budget_config.budget_duration}"
                         )
-            # Check tag budgets
-            if self.tag_budget_config:
-                request_tags = _get_tags_from_request_kwargs(
-                    request_kwargs=request_kwargs,
-                    metadata_variable_name=get_metadata_variable_name_from_kwargs(
-                        request_kwargs or {}
-                    ),
+
+        # Check tag budgets (outside loop — tags are per-request, not per-deployment)
+        for _tag in _request_tags:
+            _tag_budget_config = self._get_budget_config_for_tag(_tag)
+            if _tag_budget_config:
+                cache_keys.append(
+                    f"tag_spend:{_tag}:{_tag_budget_config.budget_duration}"
                 )
-                for _tag in request_tags:
-                    _tag_budget_config = self._get_budget_config_for_tag(_tag)
-                    if _tag_budget_config:
-                        cache_keys.append(
-                            f"tag_spend:{_tag}:{_tag_budget_config.budget_duration}"
-                        )
         return (
             cache_keys,
             provider_configs,
diff --git a/litellm/router_strategy/tag_based_routing.py b/litellm/router_strategy/tag_based_routing.py
index b0b154fc778..0163f3bbd4f 100644
--- a/litellm/router_strategy/tag_based_routing.py
+++ b/litellm/router_strategy/tag_based_routing.py
@@ -102,10 +102,10 @@ def _match_deployment(
             return {"matched_via": "tags", "matched_value": matched_value}
 
     # 2. Regex match against request headers.
-    # When match_any=False and the deployment has both plain tags and tag_regex,
-    # the strict tag check has already failed (step 1 returned None).  Allow
-    # the regex to fire only when the deployment has NO plain tags, so we never
-    # use regex as a backdoor around the operator's strict-tag policy.
+    # When match_any=False and the deployment has plain tags, the strict tag
+    # check either didn't run (no request tags) or failed (step 1 returned
+    # None).  Block the regex path so it cannot circumvent the operator's
+    # strict-tag policy.
     strict_tag_check_failed = not match_any and bool(deployment_tags)
     if deployment_tag_regex and header_strings and not strict_tag_check_failed:
         regex_match = _is_valid_deployment_tag_regex(

From 34e9be1ba70f8f498a6a5c7c34b3e3663aeb763d Mon Sep 17 00:00:00 2001
From: user <70670632+stuxf@users.noreply.github.com>
Date: Thu, 16 Apr 2026 02:41:41 +0000
Subject: [PATCH 03/15] fix: merge team metadata in admin helper, remove
 turn_off_message_logging from dynamic params

Include user_api_key_team_metadata alongside user_api_key_metadata in
_get_admin_metadata() so team-level guardrail settings are respected.
Key-level settings take precedence over team-level.

Remove turn_off_message_logging from _supported_callback_params so it
cannot be set via request metadata. Admin controls logging globally
or via key/team configuration.

Update tests to verify user-injected guardrail flags are ignored while
admin-configured flags are respected.
---
 litellm/integrations/custom_guardrail.py      |   7 +-
 .../initialize_dynamic_callback_params.py     |   1 -
 .../integrations/test_custom_guardrail.py     | 118 +++++++-----------
 3 files changed, 53 insertions(+), 73 deletions(-)

diff --git a/litellm/integrations/custom_guardrail.py b/litellm/integrations/custom_guardrail.py
index 4c294f6272e..89431847e90 100644
--- a/litellm/integrations/custom_guardrail.py
+++ b/litellm/integrations/custom_guardrail.py
@@ -257,9 +257,12 @@ class CustomGuardrail(CustomLogger):
 
     @staticmethod
     def _get_admin_metadata(data: dict) -> dict:
-        """Return the admin-configured key/team metadata from the request data."""
+        """Return merged admin-configured key and team metadata from the request data."""
         metadata = data.get("litellm_metadata") or data.get("metadata", {})
-        return metadata.get("user_api_key_metadata") or {}
+        team_meta = metadata.get("user_api_key_team_metadata") or {}
+        key_meta = metadata.get("user_api_key_metadata") or {}
+        # Key-level settings override team-level
+        return {**team_meta, **key_meta}
 
     def get_disable_global_guardrail(self, data: dict) -> Optional[bool]:
         """
diff --git a/litellm/litellm_core_utils/initialize_dynamic_callback_params.py b/litellm/litellm_core_utils/initialize_dynamic_callback_params.py
index 92c97a59924..563609af1e4 100644
--- a/litellm/litellm_core_utils/initialize_dynamic_callback_params.py
+++ b/litellm/litellm_core_utils/initialize_dynamic_callback_params.py
@@ -48,7 +48,6 @@ _supported_callback_params = [
     "braintrust_host",
     "slack_webhook_url",
     "lunary_public_key",
-    "turn_off_message_logging",
 ]
 
 
diff --git a/tests/test_litellm/integrations/test_custom_guardrail.py b/tests/test_litellm/integrations/test_custom_guardrail.py
index 3a959d599b5..4c1ef853ab4 100644
--- a/tests/test_litellm/integrations/test_custom_guardrail.py
+++ b/tests/test_litellm/integrations/test_custom_guardrail.py
@@ -173,17 +173,16 @@ class TestCustomGuardrailShouldRunGuardrail:
         assert result is False
 
     def test_should_run_guardrail_with_disable_global_guardrail(self):
-        """Test that disable_global_guardrail disables a global guardrail when set to True"""
+        """Test that disable_global_guardrails only works from admin metadata"""
         from litellm.types.guardrails import GuardrailEventHooks
 
-        # Create a guardrail with default_on=True (global guardrail)
         custom_guardrail = CustomGuardrail(
             guardrail_name="global_guardrail",
             default_on=True,
             event_hook=GuardrailEventHooks.pre_call,
         )
 
-        # Test 1: Global guardrail runs by default when default_on=True
+        # Test 1: Global guardrail runs by default
         data = {
             "model": "gpt-3.5-turbo",
             "messages": [{"role": "user", "content": "test"}],
@@ -193,7 +192,7 @@ class TestCustomGuardrailShouldRunGuardrail:
         )
         assert result is True, "Global guardrail should run when default_on=True"
 
-        # Test 2: Global guardrail is disabled when disable_global_guardrail=True at root level
+        # Test 2: User-injected disable at root level is IGNORED
         data_with_disable_root = {
             "model": "gpt-3.5-turbo",
             "messages": [{"role": "user", "content": "test"}],
@@ -203,23 +202,10 @@ class TestCustomGuardrailShouldRunGuardrail:
             data=data_with_disable_root, event_type=GuardrailEventHooks.pre_call
         )
         assert (
-            result is False
-        ), "Global guardrail should be disabled when disable_global_guardrail=True"
+            result is True
+        ), "User-injected disable_global_guardrails should be ignored"
 
-        # Test 3: Global guardrail is disabled when disable_global_guardrail=True in litellm_metadata
-        data_with_disable_litellm = {
-            "model": "gpt-3.5-turbo",
-            "messages": [{"role": "user", "content": "test"}],
-            "litellm_metadata": {"disable_global_guardrails": True},
-        }
-        result = custom_guardrail.should_run_guardrail(
-            data=data_with_disable_litellm, event_type=GuardrailEventHooks.pre_call
-        )
-        assert (
-            result is False
-        ), "Global guardrail should be disabled when disable_global_guardrail=True in litellm_metadata"
-
-        # Test 4: Global guardrail is disabled when disable_global_guardrail=True in metadata
+        # Test 3: User-injected disable in metadata is IGNORED
         data_with_disable_metadata = {
             "model": "gpt-3.5-turbo",
             "messages": [{"role": "user", "content": "test"}],
@@ -228,25 +214,21 @@ class TestCustomGuardrailShouldRunGuardrail:
         result = custom_guardrail.should_run_guardrail(
             data=data_with_disable_metadata, event_type=GuardrailEventHooks.pre_call
         )
-        assert (
-            result is False
-        ), "Global guardrail should be disabled when disable_global_guardrail=True in metadata"
+        assert result is True, "User-injected metadata disable should be ignored"
 
-        # Test 5: Global guardrail runs when disable_global_guardrail=False
-        data_with_disable_false = {
+        # Test 4: Admin-configured disable via user_api_key_metadata IS respected
+        data_with_admin_disable = {
             "model": "gpt-3.5-turbo",
             "messages": [{"role": "user", "content": "test"}],
-            "disable_global_guardrails": False,
+            "metadata": {"user_api_key_metadata": {"disable_global_guardrails": True}},
         }
         result = custom_guardrail.should_run_guardrail(
-            data=data_with_disable_false, event_type=GuardrailEventHooks.pre_call
+            data=data_with_admin_disable, event_type=GuardrailEventHooks.pre_call
         )
-        assert (
-            result is True
-        ), "Global guardrail should still run when disable_global_guardrail=False"
+        assert result is False, "Admin-configured disable should be respected"
 
     def test_should_run_guardrail_with_opted_out_global_guardrails(self):
-        """Test the per-guardrail opt-out list for global (default_on=True) guardrails"""
+        """Test that per-guardrail opt-out only works from admin metadata"""
         from litellm.types.guardrails import GuardrailEventHooks
 
         custom_guardrail = CustomGuardrail(
@@ -255,7 +237,7 @@ class TestCustomGuardrailShouldRunGuardrail:
             event_hook=GuardrailEventHooks.pre_call,
         )
 
-        # Test 1: guardrail in the opt-out list at root level → skipped
+        # Test 1: User-injected opt-out at root level is IGNORED
         data_root = {
             "model": "gpt-3.5-turbo",
             "messages": [{"role": "user", "content": "test"}],
@@ -265,23 +247,10 @@ class TestCustomGuardrailShouldRunGuardrail:
             custom_guardrail.should_run_guardrail(
                 data=data_root, event_type=GuardrailEventHooks.pre_call
             )
-            is False
+            is True
         )
 
-        # Test 2: guardrail in the opt-out list inside litellm_metadata → skipped
-        data_litellm = {
-            "model": "gpt-3.5-turbo",
-            "messages": [{"role": "user", "content": "test"}],
-            "litellm_metadata": {"opted_out_global_guardrails": ["global_guardrail"]},
-        }
-        assert (
-            custom_guardrail.should_run_guardrail(
-                data=data_litellm, event_type=GuardrailEventHooks.pre_call
-            )
-            is False
-        )
-
-        # Test 3: guardrail in the opt-out list inside metadata → skipped
+        # Test 2: User-injected opt-out in metadata is IGNORED
         data_metadata = {
             "model": "gpt-3.5-turbo",
             "messages": [{"role": "user", "content": "test"}],
@@ -291,7 +260,7 @@ class TestCustomGuardrailShouldRunGuardrail:
             custom_guardrail.should_run_guardrail(
                 data=data_metadata, event_type=GuardrailEventHooks.pre_call
             )
-            is False
+            is True
         )
 
         # Test 4: a different guardrail in the opt-out list → still runs
@@ -588,7 +557,9 @@ class TestGuardrailSensitiveFieldStripping:
             duration=1.0,
         )
 
-        logged_response = request_data["metadata"]["standard_logging_guardrail_information"][0]["guardrail_response"]
+        logged_response = request_data["metadata"][
+            "standard_logging_guardrail_information"
+        ][0]["guardrail_response"]
         assert "secret_fields" not in logged_response
         assert "sk-live-SHOULD-NOT-APPEAR" not in json.dumps(logged_response)
 
@@ -599,7 +570,12 @@ class TestGuardrailSensitiveFieldStripping:
 
         guardrail.add_standard_logging_guardrail_information_to_request_data(
             guardrail_json_response=[
-                {"result": "ok", "secret_fields": {"raw_headers": {"authorization": "Bearer sk-secret"}}},
+                {
+                    "result": "ok",
+                    "secret_fields": {
+                        "raw_headers": {"authorization": "Bearer sk-secret"}
+                    },
+                },
                 {"result": "also_ok"},
             ],
             request_data=request_data,
@@ -608,6 +584,7 @@ class TestGuardrailSensitiveFieldStripping:
         )
 
         import json
+
         serialized = json.dumps(request_data)
         assert "secret_fields" not in serialized
         assert "sk-secret" not in serialized
@@ -621,21 +598,21 @@ class TestCustomGuardrailPassthroughSupport:
         """
         Test that async_post_call_success_deployment_hook handles raw httpx.Response objects
         from passthrough endpoints without crashing with TypeError.
-        
+
         This tests Fix #3: TypeError: TypedDict does not support instance and class checks
         """
         import httpx
 
         custom_guardrail = CustomGuardrail()
-        
+
         # Mock the async_post_call_success_hook to return None (guardrail didn't modify response)
         custom_guardrail.async_post_call_success_hook = AsyncMock(return_value=None)
-        
+
         # Create a mock httpx.Response object (typical passthrough response)
         mock_response = AsyncMock(spec=httpx.Response)
         mock_response.status_code = 200
         mock_response.text = "Mock response"
-        
+
         request_data = {
             "guardrails": ["test_guardrail"],
             "user_api_key_user_id": "test_user",
@@ -644,14 +621,14 @@ class TestCustomGuardrailPassthroughSupport:
             "user_api_key_hash": "test_hash",
             "user_api_key_request_route": "passthrough_route",
         }
-        
+
         # This should not raise TypeError: TypedDict does not support instance and class checks
         result = await custom_guardrail.async_post_call_success_deployment_hook(
             request_data=request_data,
             response=mock_response,
             call_type=CallTypes.allm_passthrough_route,
         )
-        
+
         # When result is None, should return the original response
         assert result == mock_response
 
@@ -659,53 +636,53 @@ class TestCustomGuardrailPassthroughSupport:
     async def test_async_post_call_success_deployment_hook_with_none_call_type(self):
         """
         Test that async_post_call_success_deployment_hook handles None call_type gracefully.
-        
+
         This ensures that even if call_type is None (before fix #1), the guardrail doesn't crash.
         """
         custom_guardrail = CustomGuardrail()
-        
+
         # Mock the async_post_call_success_hook to return None
         custom_guardrail.async_post_call_success_hook = AsyncMock(return_value=None)
-        
+
         mock_response = AsyncMock()
-        
+
         request_data = {
             "guardrails": ["test_guardrail"],
             "user_api_key_user_id": "test_user",
         }
-        
+
         # Call with None call_type - should not crash
         result = await custom_guardrail.async_post_call_success_deployment_hook(
             request_data=request_data,
             response=mock_response,
             call_type=None,
         )
-        
+
         # Should return the original response when result is None
         assert result == mock_response
 
     def test_is_valid_response_type_with_none(self):
         """
         Test _is_valid_response_type helper method correctly identifies None as invalid.
-        
+
         This is part of Fix #3: Safely handling TypedDict types that don't support isinstance checks.
         """
         custom_guardrail = CustomGuardrail()
-        
+
         # None should be invalid
         assert custom_guardrail._is_valid_response_type(None) is False
 
     def test_is_valid_response_type_with_typeddict_error(self):
         """
         Test _is_valid_response_type gracefully handles TypeError from TypedDict.
-        
+
         This tests Fix #3: When isinstance() is called with TypedDict types, it raises TypeError.
         The method should catch this and allow the response through.
         """
         from litellm.types.utils import ModelResponse
-        
+
         custom_guardrail = CustomGuardrail()
-        
+
         # Create a valid LiteLLM response object
         response = ModelResponse(
             id="test-id",
@@ -714,13 +691,12 @@ class TestCustomGuardrailPassthroughSupport:
             model="test-model",
             object="chat.completion",
         )
-        
+
         # This should return True (it's a valid response type or TypeError is caught)
         result = custom_guardrail._is_valid_response_type(response)
         assert result is True
 
 
-
 class TestEventTypeLogging:
     """Tests for event_type logging in guardrail information."""
 
@@ -1014,7 +990,9 @@ class TestTracingFieldsPopulation:
             guardrail_json_response="blocked",
             request_data=request_data,
             guardrail_status="guardrail_intervened",
-            tracing_detail=GuardrailTracingDetail(policy_template="EU AI Act Article 5"),
+            tracing_detail=GuardrailTracingDetail(
+                policy_template="EU AI Act Article 5"
+            ),
         )
 
         slg_list = request_data["metadata"]["standard_logging_guardrail_information"]

From 413f89892b54c44d626f7c25be0a2546dc1557b3 Mon Sep 17 00:00:00 2001
From: user <70670632+stuxf@users.noreply.github.com>
Date: Thu, 16 Apr 2026 02:47:03 +0000
Subject: [PATCH 04/15] test: update dynamic callback params test for
 turn_off_message_logging removal

Verify turn_off_message_logging is no longer extracted from request
kwargs since it is now admin-only.
---
 .../test_initialize_dynamic_callback_params.py           | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/tests/test_litellm/litellm_core_utils/test_initialize_dynamic_callback_params.py b/tests/test_litellm/litellm_core_utils/test_initialize_dynamic_callback_params.py
index 91969a2b8e2..55f3c2ba3aa 100644
--- a/tests/test_litellm/litellm_core_utils/test_initialize_dynamic_callback_params.py
+++ b/tests/test_litellm/litellm_core_utils/test_initialize_dynamic_callback_params.py
@@ -82,13 +82,18 @@ def test_env_reference_in_litellm_params_metadata_raises():
 def test_non_string_values_are_not_flagged():
     kwargs = {
         "langsmith_sampling_rate": 0.5,
-        "turn_off_message_logging": True,
     }
 
     params = initialize_standard_callback_dynamic_params(kwargs)
 
     assert params.get("langsmith_sampling_rate") == 0.5
-    assert params.get("turn_off_message_logging") is True
+
+
+def test_turn_off_message_logging_not_extracted_from_request():
+    """turn_off_message_logging is admin-only — must not be settable via request."""
+    kwargs = {"turn_off_message_logging": True}
+    params = initialize_standard_callback_dynamic_params(kwargs)
+    assert params.get("turn_off_message_logging") is None
 
 
 def test_empty_kwargs_returns_empty_params():

From 22572eafaf3d59caec1da10b8cf9fc073abb5a29 Mon Sep 17 00:00:00 2001
From: user <70670632+stuxf@users.noreply.github.com>
Date: Thu, 16 Apr 2026 21:29:13 +0000
Subject: [PATCH 05/15] fix: merge admin metadata from both metadata and
 litellm_metadata

Greptile P2: _get_admin_metadata used 'litellm_metadata or metadata',
meaning a caller sending a non-empty litellm_metadata would shadow
admin config the proxy had injected into data['metadata']. Admin
exemptions would be silently ignored.

Check both keys and prefer whichever contains admin fields. Add
regression test covering the shadowing scenario.
---
 litellm/integrations/custom_guardrail.py      | 19 ++++++++++++++-----
 .../integrations/test_custom_guardrail.py     | 14 ++++++++++++++
 2 files changed, 28 insertions(+), 5 deletions(-)

diff --git a/litellm/integrations/custom_guardrail.py b/litellm/integrations/custom_guardrail.py
index 89431847e90..b0931964cc6 100644
--- a/litellm/integrations/custom_guardrail.py
+++ b/litellm/integrations/custom_guardrail.py
@@ -257,11 +257,20 @@ class CustomGuardrail(CustomLogger):
 
     @staticmethod
     def _get_admin_metadata(data: dict) -> dict:
-        """Return merged admin-configured key and team metadata from the request data."""
-        metadata = data.get("litellm_metadata") or data.get("metadata", {})
-        team_meta = metadata.get("user_api_key_team_metadata") or {}
-        key_meta = metadata.get("user_api_key_metadata") or {}
-        # Key-level settings override team-level
+        """Return merged admin-configured key and team metadata from the request data.
+
+        The proxy may inject admin metadata (user_api_key_metadata,
+        user_api_key_team_metadata) into either ``metadata`` or
+        ``litellm_metadata`` depending on endpoint. Check both so a caller
+        cannot shadow admin config by pre-populating the other key.
+        Key-level settings override team-level.
+        """
+        team_meta: dict = {}
+        key_meta: dict = {}
+        for key in ("metadata", "litellm_metadata"):
+            meta = data.get(key) or {}
+            team_meta = meta.get("user_api_key_team_metadata") or team_meta
+            key_meta = meta.get("user_api_key_metadata") or key_meta
         return {**team_meta, **key_meta}
 
     def get_disable_global_guardrail(self, data: dict) -> Optional[bool]:
diff --git a/tests/test_litellm/integrations/test_custom_guardrail.py b/tests/test_litellm/integrations/test_custom_guardrail.py
index 4c1ef853ab4..b8eb0857364 100644
--- a/tests/test_litellm/integrations/test_custom_guardrail.py
+++ b/tests/test_litellm/integrations/test_custom_guardrail.py
@@ -227,6 +227,20 @@ class TestCustomGuardrailShouldRunGuardrail:
         )
         assert result is False, "Admin-configured disable should be respected"
 
+        # Test 5: Admin config in metadata isn't shadowed by user-supplied litellm_metadata
+        data_cross_key = {
+            "model": "gpt-3.5-turbo",
+            "messages": [{"role": "user", "content": "test"}],
+            "metadata": {"user_api_key_metadata": {"disable_global_guardrails": True}},
+            "litellm_metadata": {"request_tags": ["user-supplied"]},
+        }
+        result = custom_guardrail.should_run_guardrail(
+            data=data_cross_key, event_type=GuardrailEventHooks.pre_call
+        )
+        assert (
+            result is False
+        ), "Admin config in metadata must not be shadowed by user-supplied litellm_metadata"
+
     def test_should_run_guardrail_with_opted_out_global_guardrails(self):
         """Test that per-guardrail opt-out only works from admin metadata"""
         from litellm.types.guardrails import GuardrailEventHooks

From d0601692b8dc5fff8f31714372f65c584a2abf20 Mon Sep 17 00:00:00 2001
From: user <70670632+stuxf@users.noreply.github.com>
Date: Thu, 16 Apr 2026 21:48:36 +0000
Subject: [PATCH 06/15] fix(proxy): strip user_api_key_metadata injection slots
 from user input

Expand the pre-call metadata strip to also remove user_api_key_metadata
and user_api_key_team_metadata. The proxy writes these fields into
data[_metadata_variable_name] with admin-authoritative values, but only
into that one metadata key; the caller's value in the OTHER metadata
key (metadata vs litellm_metadata) would otherwise persist and be
picked up by _get_admin_metadata, letting a caller supply their own
'admin' config to disable guardrails, opt out of global policies, etc.

VERIA-28 (High): Security Policy and Guardrail Bypass via Unsanitized
Request Metadata.

Add regression test at the proxy boundary verifying the strip, and
extend the guardrail test to cover the post-strip admin-config path.
---
 litellm/proxy/litellm_pre_call_utils.py       |   8 +-
 .../integrations/test_custom_guardrail.py     |  16 ++
 .../proxy/test_litellm_pre_call_utils.py      | 225 +++++++++++++-----
 3 files changed, 188 insertions(+), 61 deletions(-)

diff --git a/litellm/proxy/litellm_pre_call_utils.py b/litellm/proxy/litellm_pre_call_utils.py
index 95e1cbed44e..0b2c20b25a5 100644
--- a/litellm/proxy/litellm_pre_call_utils.py
+++ b/litellm/proxy/litellm_pre_call_utils.py
@@ -977,11 +977,17 @@ async def add_litellm_data_to_request(  # noqa: PLR0915
             "Setting client-provided x-api-key as api_key parameter (will override deployment key)"
         )
 
-    # Strip internal pipeline state from user input
+    # Strip internal pipeline state and admin-injection slots from user input.
+    # The proxy writes user_api_key_metadata / user_api_key_team_metadata
+    # into data[_metadata_variable_name] below; if a caller pre-populates
+    # either key on the OTHER metadata field, _get_admin_metadata lookups
+    # would treat the caller's payload as admin-configured.
     for _meta_key in ("metadata", "litellm_metadata"):
         _user_meta = data.get(_meta_key)
         if isinstance(_user_meta, dict):
             _user_meta.pop("_pipeline_managed_guardrails", None)
+            _user_meta.pop("user_api_key_metadata", None)
+            _user_meta.pop("user_api_key_team_metadata", None)
 
     ##########################################################
     # Init - Proxy Server Request
diff --git a/tests/test_litellm/integrations/test_custom_guardrail.py b/tests/test_litellm/integrations/test_custom_guardrail.py
index b8eb0857364..0c904e9df50 100644
--- a/tests/test_litellm/integrations/test_custom_guardrail.py
+++ b/tests/test_litellm/integrations/test_custom_guardrail.py
@@ -241,6 +241,22 @@ class TestCustomGuardrailShouldRunGuardrail:
             result is False
         ), "Admin config in metadata must not be shadowed by user-supplied litellm_metadata"
 
+        # Test 6: After the pre-call strip runs, user-injected
+        # user_api_key_metadata in the non-authoritative metadata key is gone.
+        # _get_admin_metadata must then surface admin config unchanged.
+        data_post_strip = {
+            "model": "gpt-3.5-turbo",
+            "messages": [{"role": "user", "content": "test"}],
+            "metadata": {"user_api_key_metadata": {"disable_global_guardrails": True}},
+            "litellm_metadata": {},  # post-strip: attacker payload removed
+        }
+        result = custom_guardrail.should_run_guardrail(
+            data=data_post_strip, event_type=GuardrailEventHooks.pre_call
+        )
+        assert (
+            result is False
+        ), "Admin config in metadata must be respected when other metadata key is empty"
+
     def test_should_run_guardrail_with_opted_out_global_guardrails(self):
         """Test that per-guardrail opt-out only works from admin metadata"""
         from litellm.types.guardrails import GuardrailEventHooks
diff --git a/tests/test_litellm/proxy/test_litellm_pre_call_utils.py b/tests/test_litellm/proxy/test_litellm_pre_call_utils.py
index cf7e71b14d4..e5adc673990 100644
--- a/tests/test_litellm/proxy/test_litellm_pre_call_utils.py
+++ b/tests/test_litellm/proxy/test_litellm_pre_call_utils.py
@@ -207,6 +207,79 @@ async def test_add_litellm_data_to_request_parses_string_metadata():
     assert updated_data["metadata"]["generation_name"] == "gen123"
 
 
+@pytest.mark.asyncio
+async def test_add_litellm_data_to_request_strips_admin_injection_slots():
+    """User-supplied user_api_key_metadata / user_api_key_team_metadata /
+    _pipeline_managed_guardrails must be stripped from both metadata keys
+    before the proxy writes its own admin-populated values. Otherwise a
+    caller can shadow admin config via the non-`_metadata_variable_name`
+    metadata key (e.g. litellm_metadata while the proxy writes to metadata).
+    """
+    from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request
+
+    request_mock = MagicMock(spec=Request)
+    request_mock.url.path = "/v1/chat/completions"
+    request_mock.url = MagicMock()
+    request_mock.url.__str__.return_value = "http://localhost/v1/chat/completions"
+    request_mock.method = "POST"
+    request_mock.query_params = {}
+    request_mock.headers = {"Content-Type": "application/json"}
+    request_mock.client = MagicMock()
+    request_mock.client.host = "127.0.0.1"
+
+    # Caller tries to inject admin config into BOTH metadata keys
+    attacker_admin_payload = {"disable_global_guardrails": True}
+    data = {
+        "model": "gpt-3.5-turbo",
+        "metadata": {
+            "user_api_key_metadata": attacker_admin_payload,
+            "user_api_key_team_metadata": attacker_admin_payload,
+            "_pipeline_managed_guardrails": ["evaded"],
+        },
+        "litellm_metadata": {
+            "user_api_key_metadata": attacker_admin_payload,
+            "user_api_key_team_metadata": attacker_admin_payload,
+            "_pipeline_managed_guardrails": ["evaded"],
+        },
+    }
+
+    real_admin_metadata = {"admin_flag": "from_proxy"}
+    user_api_key_dict = UserAPIKeyAuth(
+        api_key="hashed-key",
+        metadata=real_admin_metadata,
+        team_metadata=real_admin_metadata,
+        spend=0.0,
+        max_budget=100.0,
+        model_max_budget={},
+        team_spend=0.0,
+        team_max_budget=200.0,
+    )
+
+    updated = await add_litellm_data_to_request(
+        data=data,
+        request=request_mock,
+        user_api_key_dict=user_api_key_dict,
+        proxy_config=MagicMock(),
+        general_settings={},
+        version="test-version",
+    )
+
+    # The key that matches `_metadata_variable_name` gets proxy-populated
+    # with the real admin payload; the OTHER key must not retain the
+    # attacker's injection.
+    populated = updated["metadata"]
+    assert populated["user_api_key_metadata"] == real_admin_metadata
+    assert populated["user_api_key_team_metadata"] == real_admin_metadata
+    assert "_pipeline_managed_guardrails" not in populated or populated[
+        "_pipeline_managed_guardrails"
+    ] != ["evaded"]
+
+    other = updated.get("litellm_metadata") or {}
+    assert other.get("user_api_key_metadata") in (None, {}, real_admin_metadata)
+    assert other.get("user_api_key_team_metadata") in (None, {}, real_admin_metadata)
+    assert "_pipeline_managed_guardrails" not in other
+
+
 @pytest.mark.asyncio
 async def test_add_litellm_data_to_request_user_spend_and_budget():
     from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request
@@ -221,7 +294,10 @@ async def test_add_litellm_data_to_request_user_spend_and_budget():
     request_mock.client = MagicMock()
     request_mock.client.host = "127.0.0.1"
 
-    data = {"model": "gpt-3.5-turbo", "messages": [{"role": "user", "content": "Hello"}]}
+    data = {
+        "model": "gpt-3.5-turbo",
+        "messages": [{"role": "user", "content": "Hello"}],
+    }
 
     user_api_key_dict = UserAPIKeyAuth(
         api_key="hashed-key",
@@ -1023,6 +1099,7 @@ def test_add_headers_to_llm_call_by_model_group_existing_headers_in_data():
         # Restore original model_group_settings
         litellm.model_group_settings = original_model_group_settings
 
+
 import json
 import time
 from typing import Optional
@@ -1040,15 +1117,16 @@ class TestCustomLogger(CustomLogger):
     def __init__(self):
         self.standard_logging_object: Optional[StandardLoggingPayload] = None
         super().__init__()
-        
+
     async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
         print(f"SUCCESS CALLBACK CALLED! kwargs keys: {list(kwargs.keys())}")
         self.standard_logging_object = kwargs.get("standard_logging_object")
         print(f"Captured standard_logging_object: {self.standard_logging_object}")
-        
+
     async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
         print(f"FAILURE CALLBACK CALLED! kwargs keys: {list(kwargs.keys())}")
 
+
 @pytest.mark.asyncio
 async def test_add_litellm_metadata_from_request_headers():
     """
@@ -1065,8 +1143,16 @@ async def test_add_litellm_metadata_from_request_headers():
 
     try:
         # Prepare test data (ensure no streaming, add mock_response and api_key to route to litellm.acompletion)
-        headers = {"x-litellm-spend-logs-metadata": '{"user_id": "12345", "project_id": "proj_abc", "request_type": "chat_completion", "timestamp": "2025-09-02T10:30:00Z"}'}
-        data = {"model": "gpt-4", "messages": [{"role": "user", "content": "Hello"}], "stream": False, "mock_response": "Hi", "api_key": "fake-key"}
+        headers = {
+            "x-litellm-spend-logs-metadata": '{"user_id": "12345", "project_id": "proj_abc", "request_type": "chat_completion", "timestamp": "2025-09-02T10:30:00Z"}'
+        }
+        data = {
+            "model": "gpt-4",
+            "messages": [{"role": "user", "content": "Hello"}],
+            "stream": False,
+            "mock_response": "Hi",
+            "api_key": "fake-key",
+        }
 
         # Create mock request with headers
         mock_request = MagicMock(spec=Request)
@@ -1078,9 +1164,7 @@ async def test_add_litellm_metadata_from_request_headers():
 
         # Create mock user API key dict
         mock_user_api_key_dict = UserAPIKeyAuth(
-            api_key="test-key",
-            user_id="test-user",
-            org_id="test-org"
+            api_key="test-key", user_id="test-user", org_id="test-org"
         )
 
         # Create mock proxy logging object
@@ -1095,7 +1179,7 @@ async def test_add_litellm_metadata_from_request_headers():
 
         async def mock_post_call_success_hook(*args, **kwargs):
             # Return the response unchanged
-            return kwargs.get('response', args[2] if len(args) > 2 else None)
+            return kwargs.get("response", args[2] if len(args) > 2 else None)
 
         mock_proxy_logging_obj.during_call_hook = mock_during_call_hook
         mock_proxy_logging_obj.pre_call_hook = mock_pre_call_hook
@@ -1108,10 +1192,15 @@ async def test_add_litellm_metadata_from_request_headers():
         general_settings = {}
 
         # Create mock select_data_generator with correct signature
-        def mock_select_data_generator(response=None, user_api_key_dict=None, request_data=None):
+        def mock_select_data_generator(
+            response=None, user_api_key_dict=None, request_data=None
+        ):
             async def mock_generator():
-                yield "data: " + json.dumps({"choices": [{"delta": {"content": "Hello"}}]}) + "\n\n"
+                yield "data: " + json.dumps(
+                    {"choices": [{"delta": {"content": "Hello"}}]}
+                ) + "\n\n"
                 yield "data: [DONE]\n\n"
+
             return mock_generator()
 
         # Create the processor
@@ -1129,22 +1218,28 @@ async def test_add_litellm_metadata_from_request_headers():
             select_data_generator=mock_select_data_generator,
             llm_router=None,
             model="gpt-4",
-            is_streaming_request=False
+            is_streaming_request=False,
         )
 
         # Sleep for 3 seconds to allow logging to complete
         await asyncio.sleep(3)
 
         # Check if standard_logging_object was set
-        assert test_logger.standard_logging_object is not None, "standard_logging_object should be populated after LLM request"
+        assert (
+            test_logger.standard_logging_object is not None
+        ), "standard_logging_object should be populated after LLM request"
 
         # Verify the logging object contains expected metadata
         standard_logging_obj = test_logger.standard_logging_object
 
-        print(f"Standard logging object captured: {json.dumps(standard_logging_obj, indent=4, default=str)}")
+        print(
+            f"Standard logging object captured: {json.dumps(standard_logging_obj, indent=4, default=str)}"
+        )
 
         SPEND_LOGS_METADATA = standard_logging_obj["metadata"]["spend_logs_metadata"]
-        assert SPEND_LOGS_METADATA == dict(json.loads(headers["x-litellm-spend-logs-metadata"])), "spend_logs_metadata should be the same as the headers"
+        assert SPEND_LOGS_METADATA == dict(
+            json.loads(headers["x-litellm-spend-logs-metadata"])
+        ), "spend_logs_metadata should be the same as the headers"
     finally:
         litellm.callbacks = original_callbacks
 
@@ -1197,7 +1292,9 @@ def test_get_internal_user_header_from_mapping_returns_expected_header():
         {"header_name": "X-OpenWebUI-User-Email", "litellm_user_role": "customer"},
     ]
 
-    header_name = LiteLLMProxyRequestSetup.get_internal_user_header_from_mapping(mappings)
+    header_name = LiteLLMProxyRequestSetup.get_internal_user_header_from_mapping(
+        mappings
+    )
     assert header_name == "X-OpenWebUI-User-Id"
 
 
@@ -1205,7 +1302,9 @@ def test_get_internal_user_header_from_mapping_none_when_absent():
     mappings = [
         {"header_name": "X-OpenWebUI-User-Email", "litellm_user_role": "customer"}
     ]
-    header_name = LiteLLMProxyRequestSetup.get_internal_user_header_from_mapping(mappings)
+    header_name = LiteLLMProxyRequestSetup.get_internal_user_header_from_mapping(
+        mappings
+    )
     assert header_name is None
 
     single = {"header_name": "X-Only-Customer", "litellm_user_role": "customer"}
@@ -1218,7 +1317,10 @@ def test_add_internal_user_from_user_mapping_sets_user_id_when_header_present():
     headers = {"X-OpenWebUI-User-Id": "internal-user-123"}
     general_settings = {
         "user_header_mappings": [
-            {"header_name": "X-OpenWebUI-User-Id", "litellm_user_role": "internal_user"},
+            {
+                "header_name": "X-OpenWebUI-User-Id",
+                "litellm_user_role": "internal_user",
+            },
             {"header_name": "X-OpenWebUI-User-Email", "litellm_user_role": "customer"},
         ]
     }
@@ -1312,7 +1414,7 @@ async def test_team_guardrails_append_to_key_guardrails():
 
     metadata = updated_data.get("metadata", {})
     guardrails = metadata.get("guardrails", [])
-    
+
     assert "key-guardrail-1" in guardrails
     assert "key-guardrail-2" in guardrails
     assert "team-guardrail-1" in guardrails
@@ -1341,7 +1443,7 @@ async def test_request_guardrails_do_not_override_key_guardrails():
         metadata={"guardrails": ["key-guardrail-1"]},
         team_metadata={},
     )
-    
+
     # Test case: Request with empty guardrails should not result in empty guardrails
     data_with_empty = {
         "model": "gpt-3.5-turbo",
@@ -1361,7 +1463,7 @@ async def test_request_guardrails_do_not_override_key_guardrails():
 
     _metadata = updated_data_empty.get("metadata", {})
     requested_guardrails = _metadata.get("guardrails", [])
-    
+
     assert "guardrails" not in updated_data_empty
     assert "key-guardrail-1" in requested_guardrails
     assert len(requested_guardrails) == 1
@@ -1476,7 +1578,10 @@ def test_update_model_if_key_alias_exists():
     assert data["model"] == "xai/grok-4-fast-non-reasoning"
 
     # Test case 2: Key alias doesn't exist
-    data = {"model": "unknown-model", "messages": [{"role": "user", "content": "Hello"}]}
+    data = {
+        "model": "unknown-model",
+        "messages": [{"role": "user", "content": "Hello"}],
+    }
     user_api_key_dict = UserAPIKeyAuth(
         api_key="test-key",
         aliases={"modelAlias": "xai/grok-4-fast-non-reasoning"},
@@ -1594,16 +1699,22 @@ async def test_embedding_header_forwarding_with_model_group():
 
         # Verify that only x- prefixed headers (except x-stainless) were forwarded
         forwarded_headers = updated_data["headers"]
-        assert "X-Custom-Header" in forwarded_headers, "X-Custom-Header should be forwarded"
+        assert (
+            "X-Custom-Header" in forwarded_headers
+        ), "X-Custom-Header should be forwarded"
         assert forwarded_headers["X-Custom-Header"] == "custom-value"
         assert "X-Request-ID" in forwarded_headers, "X-Request-ID should be forwarded"
         assert forwarded_headers["X-Request-ID"] == "test-request-123"
 
         # Verify that authorization header was NOT forwarded (sensitive header)
-        assert "Authorization" not in forwarded_headers, "Authorization header should not be forwarded"
+        assert (
+            "Authorization" not in forwarded_headers
+        ), "Authorization header should not be forwarded"
 
         # Verify that Content-Type was NOT forwarded (doesn't start with x-)
-        assert "Content-Type" not in forwarded_headers, "Content-Type should not be forwarded"
+        assert (
+            "Content-Type" not in forwarded_headers
+        ), "Content-Type should not be forwarded"
 
         # Verify original data fields are preserved
         assert updated_data["model"] == "local-openai/text-embedding-3-small"
@@ -1659,8 +1770,9 @@ async def test_embedding_header_forwarding_without_model_group_config():
         )
 
         # Verify that headers were NOT added since model is not in forward list
-        assert "headers" not in updated_data or updated_data.get("headers") is None, \
-            "Headers should not be forwarded for models not in forward_client_headers_to_llm_api list"
+        assert (
+            "headers" not in updated_data or updated_data.get("headers") is None
+        ), "Headers should not be forwarded for models not in forward_client_headers_to_llm_api list"
 
         # Verify original data fields are preserved
         assert updated_data["model"] == "text-embedding-ada-002"
@@ -1714,7 +1826,9 @@ async def test_add_guardrails_from_policy_engine():
     attachment_registry = get_attachment_registry()
     attachment_registry._attachments = [
         PolicyAttachment(policy="global-baseline", scope="*"),  # applies to all
-        PolicyAttachment(policy="healthcare", teams=["healthcare-team"]),  # applies to healthcare team
+        PolicyAttachment(
+            policy="healthcare", teams=["healthcare-team"]
+        ),  # applies to healthcare team
     ]
     attachment_registry._initialized = True
 
@@ -1757,7 +1871,10 @@ async def test_add_guardrails_from_policy_engine_accepts_dynamic_policies_and_po
     data = {
         "model": "gpt-4",
         "messages": [{"role": "user", "content": "Hello"}],
-        "policies": ["PII-POLICY-GLOBAL", "HIPAA-POLICY"],  # Dynamic policies - should be accepted and removed
+        "policies": [
+            "PII-POLICY-GLOBAL",
+            "HIPAA-POLICY",
+        ],  # Dynamic policies - should be accepted and removed
         "metadata": {},
     }
 
@@ -1780,7 +1897,9 @@ async def test_add_guardrails_from_policy_engine_accepts_dynamic_policies_and_po
     )
 
     # Verify that 'policies' was removed from the request body
-    assert "policies" not in data, "'policies' should be removed from request body to prevent forwarding to LLM provider"
+    assert (
+        "policies" not in data
+    ), "'policies' should be removed from request body to prevent forwarding to LLM provider"
 
     # Verify that other fields are preserved
     assert "model" in data
@@ -1869,7 +1988,9 @@ async def test_bearer_token_not_in_debug_logs():
     from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request
     from litellm.proxy.proxy_server import ProxyConfig
 
-    secret_token = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.fakesignature"
+    secret_token = (
+        "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.fakesignature"
+    )
 
     mock_request = MagicMock(spec=Request)
     mock_request.headers = {
@@ -1898,8 +2019,10 @@ async def test_bearer_token_not_in_debug_logs():
     logger.setLevel(logging.DEBUG)
 
     try:
-        with patch("litellm.proxy.proxy_server.llm_router", None), \
-             patch("litellm.proxy.proxy_server.premium_user", True):
+        with (
+            patch("litellm.proxy.proxy_server.llm_router", None),
+            patch("litellm.proxy.proxy_server.premium_user", True),
+        ):
             await add_litellm_data_to_request(
                 data=data,
                 request=mock_request,
@@ -2020,9 +2143,7 @@ def test_resolve_project_model_specific_wins():
         "gpt-4": {"azure": {"litellm_credentials": "team-gpt4"}},
         "defaultconfig": {"azure": {"litellm_credentials": "team-default"}},
     }
-    result = _resolve_credential_from_model_config(
-        "gpt-4", project_config, team_config
-    )
+    result = _resolve_credential_from_model_config("gpt-4", project_config, team_config)
     assert result == "proj-gpt4"
 
 
@@ -2034,9 +2155,7 @@ def test_resolve_project_default_wins_over_team():
         "gpt-4": {"azure": {"litellm_credentials": "team-gpt4"}},
         "defaultconfig": {"azure": {"litellm_credentials": "team-default"}},
     }
-    result = _resolve_credential_from_model_config(
-        "gpt-4", project_config, team_config
-    )
+    result = _resolve_credential_from_model_config("gpt-4", project_config, team_config)
     assert result == "proj-default"
 
 
@@ -2091,12 +2210,8 @@ def test_apply_overrides_project_model_specific(setup_test_credentials):
         },
         project_metadata={
             "model_config": {
-                "defaultconfig": {
-                    "azure": {"litellm_credentials": "hotel-rec-azure"}
-                },
-                "gpt-4-vision": {
-                    "azure": {"litellm_credentials": "hotel-rec-vision"}
-                },
+                "defaultconfig": {"azure": {"litellm_credentials": "hotel-rec-azure"}},
+                "gpt-4-vision": {"azure": {"litellm_credentials": "hotel-rec-vision"}},
             }
         },
     )
@@ -2123,12 +2238,8 @@ def test_apply_overrides_project_default(setup_test_credentials):
         },
         project_metadata={
             "model_config": {
-                "defaultconfig": {
-                    "azure": {"litellm_credentials": "hotel-rec-azure"}
-                },
-                "gpt-4-vision": {
-                    "azure": {"litellm_credentials": "hotel-rec-vision"}
-                },
+                "defaultconfig": {"azure": {"litellm_credentials": "hotel-rec-azure"}},
+                "gpt-4-vision": {"azure": {"litellm_credentials": "hotel-rec-vision"}},
             }
         },
     )
@@ -2231,9 +2342,7 @@ def test_apply_overrides_missing_credential_name(setup_test_credentials):
         api_key="test-key",
         team_metadata={
             "model_config": {
-                "gpt-4": {
-                    "azure": {"litellm_credentials": "nonexistent-credential"}
-                }
+                "gpt-4": {"azure": {"litellm_credentials": "nonexistent-credential"}}
             }
         },
     )
@@ -2272,9 +2381,7 @@ def test_apply_overrides_no_model_in_data(setup_test_credentials):
         api_key="test-key",
         team_metadata={
             "model_config": {
-                "defaultconfig": {
-                    "azure": {"litellm_credentials": "some-cred"}
-                }
+                "defaultconfig": {"azure": {"litellm_credentials": "some-cred"}}
             }
         },
     )
@@ -2305,9 +2412,7 @@ def test_apply_overrides_clientside_api_version_preserved(setup_test_credentials
         api_key="test-key",
         team_metadata={
             "model_config": {
-                "gpt-4-vision": {
-                    "azure": {"litellm_credentials": "hotel-rec-vision"}
-                }
+                "gpt-4-vision": {"azure": {"litellm_credentials": "hotel-rec-vision"}}
             }
         },
     )

From 0e62addd947087a9b75307af39d23ee8173a11a2 Mon Sep 17 00:00:00 2001
From: user <70670632+stuxf@users.noreply.github.com>
Date: Thu, 16 Apr 2026 22:31:00 +0000
Subject: [PATCH 07/15] fix(proxy): gate caller-supplied routing/budget tags
 behind allow_client_tags

VERIA-28 (High) follow-up: tag-based routing and tag budget enforcement
read metadata.tags directly from the request, letting an attacker reach
restricted tag-routed deployments or misattribute spend to a victim
team's tag.

Strip metadata.tags (and litellm_metadata.tags) at the pre-call boundary
unless the caller's key or team metadata opts in with
allow_client_tags=True. Default-deny: existing clients that need to pass
routing tags must have the flag set explicitly on their key or team.

Preserves the tag-routing feature for admins who trust their callers;
closes the injection path for everyone else.
---
 litellm/proxy/litellm_pre_call_utils.py       |  22 +++
 .../proxy/test_litellm_pre_call_utils.py      | 135 ++++++++++++++++++
 2 files changed, 157 insertions(+)

diff --git a/litellm/proxy/litellm_pre_call_utils.py b/litellm/proxy/litellm_pre_call_utils.py
index 0b2c20b25a5..f7b66a55b8e 100644
--- a/litellm/proxy/litellm_pre_call_utils.py
+++ b/litellm/proxy/litellm_pre_call_utils.py
@@ -989,6 +989,28 @@ async def add_litellm_data_to_request(  # noqa: PLR0915
             _user_meta.pop("user_api_key_metadata", None)
             _user_meta.pop("user_api_key_team_metadata", None)
 
+    # Strip caller-supplied routing/budget tags unless the admin has opted
+    # this key or team in via metadata.allow_client_tags=True. Tags drive
+    # tag-based routing and tag budget attribution — accepting them from
+    # untrusted callers lets an attacker reach restricted deployments or
+    # misattribute spend to a victim team's tag.
+    _admin_allow_client_tags = False
+    for _admin_meta in (
+        user_api_key_dict.metadata,
+        user_api_key_dict.team_metadata,
+    ):
+        if (
+            isinstance(_admin_meta, dict)
+            and _admin_meta.get("allow_client_tags") is True
+        ):
+            _admin_allow_client_tags = True
+            break
+    if not _admin_allow_client_tags:
+        for _meta_key in ("metadata", "litellm_metadata"):
+            _user_meta = data.get(_meta_key)
+            if isinstance(_user_meta, dict):
+                _user_meta.pop("tags", None)
+
     ##########################################################
     # Init - Proxy Server Request
     # we do this as soon as entering so we track the original request
diff --git a/tests/test_litellm/proxy/test_litellm_pre_call_utils.py b/tests/test_litellm/proxy/test_litellm_pre_call_utils.py
index e5adc673990..e433bebb3ac 100644
--- a/tests/test_litellm/proxy/test_litellm_pre_call_utils.py
+++ b/tests/test_litellm/proxy/test_litellm_pre_call_utils.py
@@ -280,6 +280,141 @@ async def test_add_litellm_data_to_request_strips_admin_injection_slots():
     assert "_pipeline_managed_guardrails" not in other
 
 
+@pytest.mark.asyncio
+async def test_add_litellm_data_to_request_strips_user_tags_without_permission():
+    """Caller-supplied metadata.tags must be stripped when the key/team
+    metadata does not opt in via allow_client_tags=True. Otherwise an
+    attacker can reach restricted tag-routed deployments or attribute
+    spend to a victim team's tag."""
+    from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request
+
+    request_mock = MagicMock(spec=Request)
+    request_mock.url.path = "/v1/chat/completions"
+    request_mock.url = MagicMock()
+    request_mock.url.__str__.return_value = "http://localhost/v1/chat/completions"
+    request_mock.method = "POST"
+    request_mock.query_params = {}
+    request_mock.headers = {"Content-Type": "application/json"}
+    request_mock.client = MagicMock()
+    request_mock.client.host = "127.0.0.1"
+
+    data = {
+        "model": "gpt-3.5-turbo",
+        "metadata": {"tags": ["restricted-tier", "victim-team"]},
+        "litellm_metadata": {"tags": ["also-stripped"]},
+    }
+
+    user_api_key_dict = UserAPIKeyAuth(
+        api_key="hashed-key",
+        metadata={},
+        team_metadata={},
+        spend=0.0,
+        max_budget=100.0,
+        model_max_budget={},
+        team_spend=0.0,
+        team_max_budget=200.0,
+    )
+
+    updated = await add_litellm_data_to_request(
+        data=data,
+        request=request_mock,
+        user_api_key_dict=user_api_key_dict,
+        proxy_config=MagicMock(),
+        general_settings={},
+        version="test-version",
+    )
+
+    assert "tags" not in (updated.get("metadata") or {})
+    assert "tags" not in (updated.get("litellm_metadata") or {})
+
+
+@pytest.mark.asyncio
+async def test_add_litellm_data_to_request_preserves_user_tags_when_key_opts_in():
+    """When key.metadata.allow_client_tags=True, caller-supplied tags are
+    preserved and reach the router."""
+    from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request
+
+    request_mock = MagicMock(spec=Request)
+    request_mock.url.path = "/v1/chat/completions"
+    request_mock.url = MagicMock()
+    request_mock.url.__str__.return_value = "http://localhost/v1/chat/completions"
+    request_mock.method = "POST"
+    request_mock.query_params = {}
+    request_mock.headers = {"Content-Type": "application/json"}
+    request_mock.client = MagicMock()
+    request_mock.client.host = "127.0.0.1"
+
+    data = {
+        "model": "gpt-3.5-turbo",
+        "metadata": {"tags": ["opted-in-tag"]},
+    }
+
+    user_api_key_dict = UserAPIKeyAuth(
+        api_key="hashed-key",
+        metadata={"allow_client_tags": True},
+        team_metadata={},
+        spend=0.0,
+        max_budget=100.0,
+        model_max_budget={},
+        team_spend=0.0,
+        team_max_budget=200.0,
+    )
+
+    updated = await add_litellm_data_to_request(
+        data=data,
+        request=request_mock,
+        user_api_key_dict=user_api_key_dict,
+        proxy_config=MagicMock(),
+        general_settings={},
+        version="test-version",
+    )
+
+    assert updated["metadata"].get("tags") == ["opted-in-tag"]
+
+
+@pytest.mark.asyncio
+async def test_add_litellm_data_to_request_preserves_user_tags_when_team_opts_in():
+    """Team-level allow_client_tags is also honored (not just key-level)."""
+    from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request
+
+    request_mock = MagicMock(spec=Request)
+    request_mock.url.path = "/v1/chat/completions"
+    request_mock.url = MagicMock()
+    request_mock.url.__str__.return_value = "http://localhost/v1/chat/completions"
+    request_mock.method = "POST"
+    request_mock.query_params = {}
+    request_mock.headers = {"Content-Type": "application/json"}
+    request_mock.client = MagicMock()
+    request_mock.client.host = "127.0.0.1"
+
+    data = {
+        "model": "gpt-3.5-turbo",
+        "metadata": {"tags": ["team-allowed"]},
+    }
+
+    user_api_key_dict = UserAPIKeyAuth(
+        api_key="hashed-key",
+        metadata={},
+        team_metadata={"allow_client_tags": True},
+        spend=0.0,
+        max_budget=100.0,
+        model_max_budget={},
+        team_spend=0.0,
+        team_max_budget=200.0,
+    )
+
+    updated = await add_litellm_data_to_request(
+        data=data,
+        request=request_mock,
+        user_api_key_dict=user_api_key_dict,
+        proxy_config=MagicMock(),
+        general_settings={},
+        version="test-version",
+    )
+
+    assert updated["metadata"].get("tags") == ["team-allowed"]
+
+
 @pytest.mark.asyncio
 async def test_add_litellm_data_to_request_user_spend_and_budget():
     from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request

From 8526628a8fac6400b75b5bdb78958a656fa3abd4 Mon Sep 17 00:00:00 2001
From: user <70670632+stuxf@users.noreply.github.com>
Date: Thu, 16 Apr 2026 22:40:43 +0000
Subject: [PATCH 08/15] test: update tag-merge tests for default-deny client
 tag policy

Two pre-existing tests codified the pre-fix behavior where any caller-
supplied metadata.tags would flow through to spend logs and routing:

- test_add_key_or_team_level_spend_logs_metadata_to_request exercised
  the request/key/team tag merge. Set allow_client_tags=True on the key
  metadata so the merge path is still tested under the new regime.

- test_create_file_with_nested_litellm_metadata asserted that
  litellm_metadata[tags] form-data propagated to the handler. Drop the
  tag field; the test still proves nested form-parser correctness via
  spend_logs_metadata and environment.
---
 tests/proxy_unit_tests/test_proxy_utils.py                  | 4 ++++
 .../proxy/openai_files_endpoint/test_files_endpoint.py      | 6 +++---
 2 files changed, 7 insertions(+), 3 deletions(-)

diff --git a/tests/proxy_unit_tests/test_proxy_utils.py b/tests/proxy_unit_tests/test_proxy_utils.py
index 9f5f14457e8..a84bf7a4e7c 100644
--- a/tests/proxy_unit_tests/test_proxy_utils.py
+++ b/tests/proxy_unit_tests/test_proxy_utils.py
@@ -162,9 +162,13 @@ async def test_add_key_or_team_level_spend_logs_metadata_to_request(
 
     print(f"team_sl_metadata: {team_sl_metadata}")
     mock_request.url.path = "/chat/completions"
+    # Opt the key into client-supplied tags so request_tags are preserved
+    # and merged with admin-configured key/team tags. Without this flag,
+    # request_tags would be stripped by add_litellm_data_to_request.
     key_metadata = {
         "tags": key_tags,
         "spend_logs_metadata": key_sl_metadata,
+        "allow_client_tags": True,
     }
     team_metadata = {
         "tags": team_tags,
diff --git a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py
index 09d11388d84..49fff6de0ff 100644
--- a/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py
+++ b/tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py
@@ -1295,7 +1295,6 @@ def test_create_file_with_nested_litellm_metadata(
             "target_model_names": "gpt-3.5-turbo",
             "litellm_metadata[spend_logs_metadata][owner]": "john_doe",
             "litellm_metadata[spend_logs_metadata][team]": "engineering",
-            "litellm_metadata[tags]": "production",
             "litellm_metadata[environment]": "prod",
         },
         headers={"Authorization": "Bearer test-key"},
@@ -1306,11 +1305,12 @@ def test_create_file_with_nested_litellm_metadata(
     result = response.json()
     assert result["id"] == "file-test-123"
 
-    # Verify nested metadata was correctly parsed
+    # Verify nested metadata was correctly parsed.
+    # Note: caller-supplied `tags` is stripped by default; test removed
+    # to keep the parsing test focused on parser correctness.
     assert "spend_logs_metadata" in captured_litellm_metadata
     assert captured_litellm_metadata["spend_logs_metadata"]["owner"] == "john_doe"
     assert captured_litellm_metadata["spend_logs_metadata"]["team"] == "engineering"
-    assert captured_litellm_metadata["tags"] == "production"
     assert captured_litellm_metadata["environment"] == "prod"
 
 

From af8d479482d8c41e8abc17e144c68b92e0d82d64 Mon Sep 17 00:00:00 2001
From: user <70670632+stuxf@users.noreply.github.com>
Date: Thu, 16 Apr 2026 22:42:14 +0000
Subject: [PATCH 09/15] chore(proxy): emit warning when caller-supplied tags
 are stripped

Silent strip is the worst debug UX: admin's client sends routing tags,
they disappear, admin can't figure out why. Emit a warning naming the
metadata key the tags came from and telling the admin exactly which
flag to set if this is intentional.
---
 litellm/proxy/litellm_pre_call_utils.py | 11 ++++++++++-
 1 file changed, 10 insertions(+), 1 deletion(-)

diff --git a/litellm/proxy/litellm_pre_call_utils.py b/litellm/proxy/litellm_pre_call_utils.py
index f7b66a55b8e..b1dae22c9fe 100644
--- a/litellm/proxy/litellm_pre_call_utils.py
+++ b/litellm/proxy/litellm_pre_call_utils.py
@@ -1006,10 +1006,19 @@ async def add_litellm_data_to_request(  # noqa: PLR0915
             _admin_allow_client_tags = True
             break
     if not _admin_allow_client_tags:
+        _stripped_from: List[str] = []
         for _meta_key in ("metadata", "litellm_metadata"):
             _user_meta = data.get(_meta_key)
-            if isinstance(_user_meta, dict):
+            if isinstance(_user_meta, dict) and "tags" in _user_meta:
                 _user_meta.pop("tags", None)
+                _stripped_from.append(_meta_key)
+        if _stripped_from:
+            verbose_proxy_logger.warning(
+                "Stripped caller-supplied tags from %s: this key/team does "
+                "not have `allow_client_tags: true` in its metadata. Set it "
+                "to opt into client-supplied routing/budget tags.",
+                ", ".join(_stripped_from),
+            )
 
     ##########################################################
     # Init - Proxy Server Request

From 9622864ef20048065c19cd0992e5acb33c97c67e Mon Sep 17 00:00:00 2001
From: user <70670632+stuxf@users.noreply.github.com>
Date: Thu, 16 Apr 2026 22:50:54 +0000
Subject: [PATCH 10/15] test: set allow_client_tags on duplicate-tag merge test
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

test_add_litellm_data_to_request_duplicate_tags tests the request/key
tag merge when tags overlap. The merge requires caller-supplied tags to
flow through — set allow_client_tags=True on the key so the merge path
stays testable under the new default-deny regime.
---
 tests/proxy_unit_tests/test_proxy_utils.py | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/tests/proxy_unit_tests/test_proxy_utils.py b/tests/proxy_unit_tests/test_proxy_utils.py
index a84bf7a4e7c..916e01b0f2c 100644
--- a/tests/proxy_unit_tests/test_proxy_utils.py
+++ b/tests/proxy_unit_tests/test_proxy_utils.py
@@ -842,12 +842,13 @@ async def test_add_litellm_data_to_request_duplicate_tags(
     mock_request.headers = {}
     mock_request.state = State()
 
-    # Setup key with tags in metadata
+    # Setup key with tags in metadata. Opt into client-supplied tags so the
+    # request_tags are preserved for the merge under test.
     user_api_key_dict = UserAPIKeyAuth(
         api_key="test_api_key",
         user_id="test_user_id",
         org_id="test_org_id",
-        metadata={"tags": key_tags},
+        metadata={"tags": key_tags, "allow_client_tags": True},
     )
 
     # Setup request data with tags

From db19f24d694ec82037f06361e4aa6ae10d194da5 Mon Sep 17 00:00:00 2001
From: user <70670632+stuxf@users.noreply.github.com>
Date: Thu, 16 Apr 2026 23:00:52 +0000
Subject: [PATCH 11/15] fix(proxy): move metadata strip after JSON-string parse
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Veria AI caught a bypass: metadata can arrive as a JSON string via
multipart/form-data or extra_body, and the existing strip block ran
before the string-to-dict parse. The isinstance(_user_meta, dict)
guard returned False on the string, the strip was skipped, and then
the parse turned the string into a dict — leaving user_api_key_metadata
/ user_api_key_team_metadata / _pipeline_managed_guardrails / tags
intact in the parsed dict.

Move the strip to run AFTER the parse and BEFORE the merge of
litellm_metadata into data[_metadata_variable_name], closing the bypass
for both raw-dict and string-encoded payloads.

Regression test: test_add_litellm_data_to_request_strips_string_encoded_admin_injection.
---
 litellm/proxy/litellm_pre_call_utils.py       | 103 ++++++++++--------
 .../proxy/test_litellm_pre_call_utils.py      |  68 +++++++++++-
 2 files changed, 122 insertions(+), 49 deletions(-)

diff --git a/litellm/proxy/litellm_pre_call_utils.py b/litellm/proxy/litellm_pre_call_utils.py
index b1dae22c9fe..33affa5351c 100644
--- a/litellm/proxy/litellm_pre_call_utils.py
+++ b/litellm/proxy/litellm_pre_call_utils.py
@@ -977,49 +977,6 @@ async def add_litellm_data_to_request(  # noqa: PLR0915
             "Setting client-provided x-api-key as api_key parameter (will override deployment key)"
         )
 
-    # Strip internal pipeline state and admin-injection slots from user input.
-    # The proxy writes user_api_key_metadata / user_api_key_team_metadata
-    # into data[_metadata_variable_name] below; if a caller pre-populates
-    # either key on the OTHER metadata field, _get_admin_metadata lookups
-    # would treat the caller's payload as admin-configured.
-    for _meta_key in ("metadata", "litellm_metadata"):
-        _user_meta = data.get(_meta_key)
-        if isinstance(_user_meta, dict):
-            _user_meta.pop("_pipeline_managed_guardrails", None)
-            _user_meta.pop("user_api_key_metadata", None)
-            _user_meta.pop("user_api_key_team_metadata", None)
-
-    # Strip caller-supplied routing/budget tags unless the admin has opted
-    # this key or team in via metadata.allow_client_tags=True. Tags drive
-    # tag-based routing and tag budget attribution — accepting them from
-    # untrusted callers lets an attacker reach restricted deployments or
-    # misattribute spend to a victim team's tag.
-    _admin_allow_client_tags = False
-    for _admin_meta in (
-        user_api_key_dict.metadata,
-        user_api_key_dict.team_metadata,
-    ):
-        if (
-            isinstance(_admin_meta, dict)
-            and _admin_meta.get("allow_client_tags") is True
-        ):
-            _admin_allow_client_tags = True
-            break
-    if not _admin_allow_client_tags:
-        _stripped_from: List[str] = []
-        for _meta_key in ("metadata", "litellm_metadata"):
-            _user_meta = data.get(_meta_key)
-            if isinstance(_user_meta, dict) and "tags" in _user_meta:
-                _user_meta.pop("tags", None)
-                _stripped_from.append(_meta_key)
-        if _stripped_from:
-            verbose_proxy_logger.warning(
-                "Stripped caller-supplied tags from %s: this key/team does "
-                "not have `allow_client_tags: true` in its metadata. Set it "
-                "to opt into client-supplied routing/budget tags.",
-                ", ".join(_stripped_from),
-            )
-
     ##########################################################
     # Init - Proxy Server Request
     # we do this as soon as entering so we track the original request
@@ -1126,11 +1083,61 @@ async def add_litellm_data_to_request(  # noqa: PLR0915
                 )
             else:
                 data["litellm_metadata"] = parsed_litellm_metadata
-        # Merge litellm_metadata into the metadata variable (preserving existing values)
-        if isinstance(data["litellm_metadata"], dict):
-            for key, value in data["litellm_metadata"].items():
-                if key not in data[_metadata_variable_name]:
-                    data[_metadata_variable_name][key] = value
+
+    # Strip internal pipeline state and admin-injection slots from user input.
+    # Runs AFTER the string-to-dict parse above so JSON-string metadata (sent
+    # via multipart/form-data or extra_body) cannot smuggle `user_api_key_metadata`
+    # past the isinstance(dict) guard.
+    #
+    # The proxy writes user_api_key_metadata / user_api_key_team_metadata into
+    # data[_metadata_variable_name] below; if a caller pre-populates either
+    # key on the OTHER metadata field, _get_admin_metadata lookups would treat
+    # the caller's payload as admin-configured.
+    for _meta_key in ("metadata", "litellm_metadata"):
+        _user_meta = data.get(_meta_key)
+        if isinstance(_user_meta, dict):
+            _user_meta.pop("_pipeline_managed_guardrails", None)
+            _user_meta.pop("user_api_key_metadata", None)
+            _user_meta.pop("user_api_key_team_metadata", None)
+
+    # Strip caller-supplied routing/budget tags unless the admin has opted
+    # this key or team in via metadata.allow_client_tags=True. Tags drive
+    # tag-based routing and tag budget attribution — accepting them from
+    # untrusted callers lets an attacker reach restricted deployments or
+    # misattribute spend to a victim team's tag.
+    _admin_allow_client_tags = False
+    for _admin_meta in (
+        user_api_key_dict.metadata,
+        user_api_key_dict.team_metadata,
+    ):
+        if (
+            isinstance(_admin_meta, dict)
+            and _admin_meta.get("allow_client_tags") is True
+        ):
+            _admin_allow_client_tags = True
+            break
+    if not _admin_allow_client_tags:
+        _stripped_from: List[str] = []
+        for _meta_key in ("metadata", "litellm_metadata"):
+            _user_meta = data.get(_meta_key)
+            if isinstance(_user_meta, dict) and "tags" in _user_meta:
+                _user_meta.pop("tags", None)
+                _stripped_from.append(_meta_key)
+        if _stripped_from:
+            verbose_proxy_logger.warning(
+                "Stripped caller-supplied tags from %s: this key/team does "
+                "not have `allow_client_tags: true` in its metadata. Set it "
+                "to opt into client-supplied routing/budget tags.",
+                ", ".join(_stripped_from),
+            )
+
+    # Now merge litellm_metadata into the metadata variable (preserving existing
+    # values) — runs AFTER the strip so attacker injections in litellm_metadata
+    # cannot cross-contaminate the admin-authoritative metadata dict.
+    if "litellm_metadata" in data and isinstance(data["litellm_metadata"], dict):
+        for key, value in data["litellm_metadata"].items():
+            if key not in data[_metadata_variable_name]:
+                data[_metadata_variable_name][key] = value
 
     data = LiteLLMProxyRequestSetup.add_user_api_key_auth_to_request_metadata(
         data=data,
diff --git a/tests/test_litellm/proxy/test_litellm_pre_call_utils.py b/tests/test_litellm/proxy/test_litellm_pre_call_utils.py
index e433bebb3ac..351857b9b10 100644
--- a/tests/test_litellm/proxy/test_litellm_pre_call_utils.py
+++ b/tests/test_litellm/proxy/test_litellm_pre_call_utils.py
@@ -280,6 +280,70 @@ async def test_add_litellm_data_to_request_strips_admin_injection_slots():
     assert "_pipeline_managed_guardrails" not in other
 
 
+@pytest.mark.asyncio
+async def test_add_litellm_data_to_request_strips_string_encoded_admin_injection():
+    """Regression: metadata arriving as a JSON string (multipart/form-data or
+    extra_body) must not bypass the admin-injection strip. The parse happens
+    AFTER receipt, so the strip has to run after the parse, not before.
+    """
+    from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request
+
+    request_mock = MagicMock(spec=Request)
+    request_mock.url.path = "/v1/chat/completions"
+    request_mock.url = MagicMock()
+    request_mock.url.__str__.return_value = "http://localhost/v1/chat/completions"
+    request_mock.method = "POST"
+    request_mock.query_params = {}
+    request_mock.headers = {"Content-Type": "multipart/form-data"}
+    request_mock.client = MagicMock()
+    request_mock.client.host = "127.0.0.1"
+
+    # Attacker encodes an admin-injection payload inside a JSON string.
+    attacker_payload = {
+        "user_api_key_metadata": {"disable_global_guardrails": True},
+        "user_api_key_team_metadata": {"disable_global_guardrails": True},
+        "_pipeline_managed_guardrails": ["evaded"],
+    }
+    data = {
+        "model": "gpt-3.5-turbo",
+        "metadata": json.dumps(attacker_payload),
+        "litellm_metadata": json.dumps(attacker_payload),
+    }
+
+    real_admin_metadata = {"admin_flag": "from_proxy"}
+    user_api_key_dict = UserAPIKeyAuth(
+        api_key="hashed-key",
+        metadata=real_admin_metadata,
+        team_metadata=real_admin_metadata,
+        spend=0.0,
+        max_budget=100.0,
+        model_max_budget={},
+        team_spend=0.0,
+        team_max_budget=200.0,
+    )
+
+    updated = await add_litellm_data_to_request(
+        data=data,
+        request=request_mock,
+        user_api_key_dict=user_api_key_dict,
+        proxy_config=MagicMock(),
+        general_settings={},
+        version="test-version",
+    )
+
+    populated = updated["metadata"]
+    # The real admin payload from user_api_key_dict wins.
+    assert populated["user_api_key_metadata"] == real_admin_metadata
+    assert populated["user_api_key_team_metadata"] == real_admin_metadata
+    assert populated.get("_pipeline_managed_guardrails") != ["evaded"]
+
+    other = updated.get("litellm_metadata") or {}
+    # After the strip, litellm_metadata has no admin-injection slots.
+    assert "user_api_key_metadata" not in other
+    assert "user_api_key_team_metadata" not in other
+    assert "_pipeline_managed_guardrails" not in other
+
+
 @pytest.mark.asyncio
 async def test_add_litellm_data_to_request_strips_user_tags_without_permission():
     """Caller-supplied metadata.tags must be stripped when the key/team
@@ -486,9 +550,11 @@ async def test_add_litellm_data_to_request_audio_transcription_multipart():
         "file": b"Fake audio bytes",
     }
 
+    # Opt the key in to client-supplied tags so the parsed tags from the
+    # JSON-string multipart body aren't stripped by the admin-injection strip.
     user_api_key_dict = UserAPIKeyAuth(
         api_key="hashed-key",
-        metadata={},
+        metadata={"allow_client_tags": True},
         team_metadata={},
         spend=0.0,
         max_budget=100.0,

From dc8e03b91f8fc71a6322078ad958465c25be3b21 Mon Sep 17 00:00:00 2001
From: user <70670632+stuxf@users.noreply.github.com>
Date: Thu, 16 Apr 2026 23:14:05 +0000
Subject: [PATCH 12/15] fix(proxy): expand _guardrail_modification_check to
 cover all bypass keys
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Per VERIA-28's secondary recommendation. The existing check only gated
metadata.guardrails. User-supplied values for disable_global_guardrails
(plural and the original singular typo variant) and opted_out_global_guardrails
are already silently ignored by _get_admin_metadata at read time, but the
silent-ignore makes diagnosis confusing and relies on one specific read
site catching them.

Reject at auth time with a 403 when any of:
- guardrails list (existing)
- disable_global_guardrails (new)
- disable_global_guardrail (new — historical singular-key variant)
- opted_out_global_guardrails (new)

are present in metadata, litellm_metadata, or at the request root, and the
caller's team lacks can_modify_guardrails. Defense in depth: the strip at
the pre-call layer still runs; this check fails loudly one layer earlier
so operators see an explicit 403 rather than a silent-ignore.
---
 litellm/proxy/auth/auth_checks.py             |  43 +++-
 .../proxy/auth/test_auth_checks.py            | 217 ++++++++++++++----
 2 files changed, 204 insertions(+), 56 deletions(-)

diff --git a/litellm/proxy/auth/auth_checks.py b/litellm/proxy/auth/auth_checks.py
index 56958a88f6d..1f4b5311baf 100644
--- a/litellm/proxy/auth/auth_checks.py
+++ b/litellm/proxy/auth/auth_checks.py
@@ -327,15 +327,46 @@ def _global_proxy_budget_check(
             )
 
 
+_GUARDRAIL_MODIFICATION_KEYS: tuple = (
+    "guardrails",
+    "disable_global_guardrails",
+    "disable_global_guardrail",
+    "opted_out_global_guardrails",
+)
+
+
 def _guardrail_modification_check(
     request_body: dict, team_object: Optional[LiteLLM_TeamTable]
 ) -> None:
-    _request_metadata: dict = request_body.get("metadata", {}) or {}
-    if not _request_metadata.get("guardrails"):
-        return
+    """
+    Reject user-supplied metadata flags that would modify guardrail behavior
+    unless the team has explicit permission. Checked keys include the plural
+    ``guardrails`` list plus the per-request toggles that influence whether
+    default-on guardrails run (``disable_global_guardrails``,
+    ``disable_global_guardrail`` singular, and ``opted_out_global_guardrails``).
 
+    User-supplied values for the bypass toggles are also silently ignored by
+    ``_get_admin_metadata`` at read time; this check adds defense in depth by
+    failing loudly at the auth layer so operators see an explicit 403 instead
+    of a confusing silent-ignore.
+    """
     from litellm.proxy.guardrails.guardrail_helpers import can_modify_guardrails
 
+    def _user_requested_modification(container: Any) -> bool:
+        if not isinstance(container, dict):
+            return False
+        return any(container.get(key) for key in _GUARDRAIL_MODIFICATION_KEYS)
+
+    # Check both metadata keys — callers can populate either depending on the
+    # endpoint. Cover the top-level too so root-level injection is rejected.
+    modifies = (
+        _user_requested_modification(request_body.get("metadata"))
+        or _user_requested_modification(request_body.get("litellm_metadata"))
+        or _user_requested_modification(request_body)
+    )
+    if not modifies:
+        return
+
     if not can_modify_guardrails(team_object):
         raise HTTPException(
             status_code=403,
@@ -451,9 +482,9 @@ async def common_checks(  # noqa: PLR0915
                 model=_model,
                 team_object=team_object,
                 llm_router=llm_router,
-                team_model_aliases=valid_token.team_model_aliases
-                if valid_token
-                else None,
+                team_model_aliases=(
+                    valid_token.team_model_aliases if valid_token else None
+                ),
             ):
                 raise ProxyException(
                     message=f"Team not allowed to access model. Team={team_object.team_id}, Model={_model}. Allowed team models = {team_object.models}",
diff --git a/tests/test_litellm/proxy/auth/test_auth_checks.py b/tests/test_litellm/proxy/auth/test_auth_checks.py
index bd659ed518f..a3d24fc8bf6 100644
--- a/tests/test_litellm/proxy/auth/test_auth_checks.py
+++ b/tests/test_litellm/proxy/auth/test_auth_checks.py
@@ -62,9 +62,9 @@ def reset_constants_module():
     # Reload modules before test
     importlib.reload(constants)
     importlib.reload(auth_checks)
-    
+
     yield
-    
+
     # Reload modules after test to clean up
     importlib.reload(constants)
     importlib.reload(auth_checks)
@@ -157,9 +157,9 @@ def test_experimental_ui_token_ignores_litellm_ui_session_duration(
     expires = datetime.fromisoformat(token_data["expires"].replace("Z", "+00:00"))
     now = get_utc_datetime()
     # Must be ~10 min, NOT 24h. If LITELLM_UI_SESSION_DURATION were incorrectly used, this would fail.
-    assert expires <= now + timedelta(minutes=11), (
-        "Experimental UI must use 10-min expiry, not LITELLM_UI_SESSION_DURATION"
-    )
+    assert expires <= now + timedelta(
+        minutes=11
+    ), "Experimental UI must use 10-min expiry, not LITELLM_UI_SESSION_DURATION"
 
 
 def test_get_experimental_ui_login_jwt_auth_token_invalid(
@@ -293,13 +293,15 @@ def test_get_cli_jwt_auth_token_custom_expiration(
 
     # Set custom expiration to 48 hours
     monkeypatch.setenv("LITELLM_CLI_JWT_EXPIRATION_HOURS", "48")
-    
+
     # Reload the constants module to pick up the new env var
     importlib.reload(constants)
     # Also reload auth_checks to pick up the new constant value
     importlib.reload(auth_checks)
-    
-    token = auth_checks.ExperimentalUIJWTToken.get_cli_jwt_auth_token(valid_sso_user_defined_values)
+
+    token = auth_checks.ExperimentalUIJWTToken.get_cli_jwt_auth_token(
+        valid_sso_user_defined_values
+    )
 
     # Decrypt and verify token contents
     decrypted_token = decrypt_value_helper(
@@ -315,7 +317,6 @@ def test_get_cli_jwt_auth_token_custom_expiration(
     assert expires <= get_utc_datetime() + timedelta(hours=48, minutes=1)
 
 
-
 @pytest.mark.asyncio
 async def test_default_internal_user_params_with_get_user_object(monkeypatch):
     """Test that default_internal_user_params is used when creating a new user via get_user_object"""
@@ -436,7 +437,9 @@ async def test_get_user_object_upsert_includes_user_email():
     mock_prisma_client.db.litellm_usertable.create.assert_called_once()
     creation_args = mock_prisma_client.db.litellm_usertable.create.call_args[1]["data"]
 
-    assert "user_email" in creation_args, "user_email should be included when upserting a new user"
+    assert (
+        "user_email" in creation_args
+    ), "user_email should be included when upserting a new user"
     assert creation_args["user_email"] == "test@example.com"
     assert creation_args["user_id"] == "new_test_user"
 
@@ -463,7 +466,9 @@ def test_log_budget_lookup_failure_skips_user_not_found():
 
 
 @pytest.mark.asyncio
-@patch("litellm.proxy.management_endpoints.team_endpoints.new_team", new_callable=AsyncMock)
+@patch(
+    "litellm.proxy.management_endpoints.team_endpoints.new_team", new_callable=AsyncMock
+)
 async def test_get_team_db_check_calls_new_team_on_upsert(mock_new_team, monkeypatch):
     """
     Test that _get_team_db_check correctly calls the `new_team` function
@@ -497,8 +502,12 @@ async def test_get_team_db_check_calls_new_team_on_upsert(mock_new_team, monkeyp
 
 
 @pytest.mark.asyncio
-@patch("litellm.proxy.management_endpoints.team_endpoints.new_team", new_callable=AsyncMock)
-async def test_get_team_db_check_does_not_call_new_team_if_exists(mock_new_team, monkeypatch):
+@patch(
+    "litellm.proxy.management_endpoints.team_endpoints.new_team", new_callable=AsyncMock
+)
+async def test_get_team_db_check_does_not_call_new_team_if_exists(
+    mock_new_team, monkeypatch
+):
     """
     Test that _get_team_db_check does NOT call the `new_team` function
     if the team already exists in the database.
@@ -541,8 +550,9 @@ async def test_vector_store_access_check_early_returns(
     if vector_store_registry:
         vector_store_registry.get_vector_store_ids_to_run.return_value = None
 
-    with patch("litellm.proxy.proxy_server.prisma_client", prisma_client), patch(
-        "litellm.vector_store_registry", vector_store_registry
+    with (
+        patch("litellm.proxy.proxy_server.prisma_client", prisma_client),
+        patch("litellm.vector_store_registry", vector_store_registry),
     ):
         result = await vector_store_access_check(
             request_body=request_body,
@@ -639,8 +649,9 @@ async def test_vector_store_access_check_with_permissions():
     mock_vector_store_registry = MagicMock()
     mock_vector_store_registry.get_vector_store_ids_to_run.return_value = ["store-1"]
 
-    with patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client), patch(
-        "litellm.vector_store_registry", mock_vector_store_registry
+    with (
+        patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client),
+        patch("litellm.vector_store_registry", mock_vector_store_registry),
     ):
         result = await vector_store_access_check(
             request_body=request_body,
@@ -653,8 +664,9 @@ async def test_vector_store_access_check_with_permissions():
     # Test with denied access
     mock_vector_store_registry.get_vector_store_ids_to_run.return_value = ["store-3"]
 
-    with patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client), patch(
-        "litellm.vector_store_registry", mock_vector_store_registry
+    with (
+        patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client),
+        patch("litellm.vector_store_registry", mock_vector_store_registry),
     ):
         with pytest.raises(ProxyException) as exc_info:
             await vector_store_access_check(
@@ -687,8 +699,9 @@ async def test_vector_store_access_check_with_team_permissions():
         "team-store-allowed"
     ]
 
-    with patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client), patch(
-        "litellm.vector_store_registry", mock_vector_store_registry
+    with (
+        patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client),
+        patch("litellm.vector_store_registry", mock_vector_store_registry),
     ):
         result = await vector_store_access_check(
             request_body=request_body,
@@ -702,8 +715,9 @@ async def test_vector_store_access_check_with_team_permissions():
         "team-store-denied"
     ]
 
-    with patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client), patch(
-        "litellm.vector_store_registry", mock_vector_store_registry
+    with (
+        patch("litellm.proxy.proxy_server.prisma_client", mock_prisma_client),
+        patch("litellm.vector_store_registry", mock_vector_store_registry),
     ):
         with pytest.raises(ProxyException) as exc_info:
             await vector_store_access_check(
@@ -1598,12 +1612,15 @@ async def test_custom_auth_common_checks_opt_in():
     mock_request = MagicMock()
 
     # Default (no flag) — common_checks should NOT be called
-    with patch(
-        "litellm.proxy.auth.user_api_key_auth.common_checks",
-        new_callable=AsyncMock,
-    ) as mock_common, patch(
-        "litellm.proxy.proxy_server.general_settings",
-        {},
+    with (
+        patch(
+            "litellm.proxy.auth.user_api_key_auth.common_checks",
+            new_callable=AsyncMock,
+        ) as mock_common,
+        patch(
+            "litellm.proxy.proxy_server.general_settings",
+            {},
+        ),
     ):
         mock_common.return_value = True
         result = await _run_post_custom_auth_checks(
@@ -1616,12 +1633,15 @@ async def test_custom_auth_common_checks_opt_in():
         mock_common.assert_not_called()
 
     # With flag=True — common_checks SHOULD be called
-    with patch(
-        "litellm.proxy.auth.user_api_key_auth.common_checks",
-        new_callable=AsyncMock,
-    ) as mock_common, patch(
-        "litellm.proxy.proxy_server.general_settings",
-        {"custom_auth_run_common_checks": True},
+    with (
+        patch(
+            "litellm.proxy.auth.user_api_key_auth.common_checks",
+            new_callable=AsyncMock,
+        ) as mock_common,
+        patch(
+            "litellm.proxy.proxy_server.general_settings",
+            {"custom_auth_run_common_checks": True},
+        ),
     ):
         mock_common.return_value = True
         result = await _run_post_custom_auth_checks(
@@ -1660,9 +1680,7 @@ async def test_virtual_key_budget_check_reads_from_spend_counter():
             return 1.5
         return fallback_spend
 
-    with patch(
-        "litellm.proxy.proxy_server.get_current_spend", mock_get_current_spend
-    ):
+    with patch("litellm.proxy.proxy_server.get_current_spend", mock_get_current_spend):
         with pytest.raises(litellm.BudgetExceededError) as exc_info:
             await _virtual_key_max_budget_check(
                 valid_token=valid_token,
@@ -1692,9 +1710,7 @@ async def test_virtual_key_budget_check_fallback_no_counter():
     async def mock_get_current_spend(counter_key, fallback_spend):
         return fallback_spend
 
-    with patch(
-        "litellm.proxy.proxy_server.get_current_spend", mock_get_current_spend
-    ):
+    with patch("litellm.proxy.proxy_server.get_current_spend", mock_get_current_spend):
         with pytest.raises(litellm.BudgetExceededError) as exc_info:
             await _virtual_key_max_budget_check(
                 valid_token=valid_token,
@@ -1723,9 +1739,7 @@ async def test_team_budget_check_reads_from_spend_counter():
             return 1.5
         return fallback_spend
 
-    with patch(
-        "litellm.proxy.proxy_server.get_current_spend", mock_get_current_spend
-    ):
+    with patch("litellm.proxy.proxy_server.get_current_spend", mock_get_current_spend):
         with pytest.raises(litellm.BudgetExceededError) as exc_info:
             await _team_max_budget_check(
                 team_object=team_object,
@@ -1763,12 +1777,13 @@ async def test_team_member_budget_check_reads_from_spend_counter():
             return 1.5
         return fallback_spend
 
-    with patch(
-        "litellm.proxy.proxy_server.get_current_spend", mock_get_current_spend
-    ), patch(
-        "litellm.proxy.auth.auth_checks.get_team_membership",
-        new_callable=AsyncMock,
-        return_value=team_membership,
+    with (
+        patch("litellm.proxy.proxy_server.get_current_spend", mock_get_current_spend),
+        patch(
+            "litellm.proxy.auth.auth_checks.get_team_membership",
+            new_callable=AsyncMock,
+            return_value=team_membership,
+        ),
     ):
         with pytest.raises(litellm.BudgetExceededError) as exc_info:
             await _check_team_member_budget(
@@ -1780,3 +1795,105 @@ async def test_team_member_budget_check_reads_from_spend_counter():
                 proxy_logging_obj=proxy_logging_obj,
             )
         assert exc_info.value.current_cost == 1.5
+
+
+class TestGuardrailModificationCheck:
+    """Defense-in-depth: `_guardrail_modification_check` must 403 when the
+    caller's metadata attempts to modify any guardrail-related key and the
+    team lacks the `modify_guardrails` permission. Checks both the
+    historically-covered `guardrails` list and the bypass toggles that
+    `_get_admin_metadata` silently ignores at read time.
+    """
+
+    def _call(self, request_body):
+        from litellm.proxy.auth.auth_checks import _guardrail_modification_check
+
+        team_object = MagicMock()
+        team_object.metadata = {}  # no permission
+        return _guardrail_modification_check(
+            request_body=request_body, team_object=team_object
+        )
+
+    def test_noop_when_no_guardrail_keys_present(self):
+        # no-op — should return silently
+        self._call({"metadata": {"unrelated": "value"}})
+
+    def test_rejects_guardrails_list(self):
+        from fastapi import HTTPException
+
+        with patch(
+            "litellm.proxy.guardrails.guardrail_helpers.can_modify_guardrails",
+            return_value=False,
+        ):
+            with pytest.raises(HTTPException) as exc:
+                self._call({"metadata": {"guardrails": ["custom"]}})
+            assert exc.value.status_code == 403
+
+    def test_rejects_disable_global_guardrails_plural(self):
+        from fastapi import HTTPException
+
+        with patch(
+            "litellm.proxy.guardrails.guardrail_helpers.can_modify_guardrails",
+            return_value=False,
+        ):
+            with pytest.raises(HTTPException) as exc:
+                self._call({"metadata": {"disable_global_guardrails": True}})
+            assert exc.value.status_code == 403
+
+    def test_rejects_disable_global_guardrail_singular(self):
+        """VERIA-28's originally-reported singular-key typo variant."""
+        from fastapi import HTTPException
+
+        with patch(
+            "litellm.proxy.guardrails.guardrail_helpers.can_modify_guardrails",
+            return_value=False,
+        ):
+            with pytest.raises(HTTPException) as exc:
+                self._call({"metadata": {"disable_global_guardrail": True}})
+            assert exc.value.status_code == 403
+
+    def test_rejects_opted_out_global_guardrails(self):
+        from fastapi import HTTPException
+
+        with patch(
+            "litellm.proxy.guardrails.guardrail_helpers.can_modify_guardrails",
+            return_value=False,
+        ):
+            with pytest.raises(HTTPException) as exc:
+                self._call(
+                    {"metadata": {"opted_out_global_guardrails": ["some_guardrail"]}}
+                )
+            assert exc.value.status_code == 403
+
+    def test_rejects_injection_via_litellm_metadata_key(self):
+        """Caller can populate the OTHER metadata key; that must also 403."""
+        from fastapi import HTTPException
+
+        with patch(
+            "litellm.proxy.guardrails.guardrail_helpers.can_modify_guardrails",
+            return_value=False,
+        ):
+            with pytest.raises(HTTPException) as exc:
+                self._call({"litellm_metadata": {"disable_global_guardrails": True}})
+            assert exc.value.status_code == 403
+
+    def test_rejects_root_level_injection(self):
+        """Top-level injection (`request_body["disable_global_guardrails"]`)
+        was VERIA-28's easiest variant to hit — keep it rejected."""
+        from fastapi import HTTPException
+
+        with patch(
+            "litellm.proxy.guardrails.guardrail_helpers.can_modify_guardrails",
+            return_value=False,
+        ):
+            with pytest.raises(HTTPException) as exc:
+                self._call({"disable_global_guardrails": True})
+            assert exc.value.status_code == 403
+
+    def test_allows_when_team_has_permission(self):
+        with patch(
+            "litellm.proxy.guardrails.guardrail_helpers.can_modify_guardrails",
+            return_value=True,
+        ):
+            # no-op, should not raise
+            self._call({"metadata": {"disable_global_guardrails": True}})

From 76aa97f77b0781b77e4ea248e7245b0bdeefa587 Mon Sep 17 00:00:00 2001
From: user <70670632+stuxf@users.noreply.github.com>
Date: Thu, 16 Apr 2026 23:35:26 +0000
Subject: [PATCH 13/15] fix(proxy): close three variant metadata/tag injection
 paths
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Close three variant bypasses adjacent to VERIA-28 found during post-fix
variant audit:

1. _guardrail_modification_check had the same isinstance(dict) bypass
   Veria-AI just flagged on the pre-call strip. A caller sending
   `{"metadata": "{…}"}` as a JSON-encoded string (multipart/form-data
   or extra_body) skipped the guard, got parsed to dict downstream, and
   reached guardrail logic with bypass flags intact. Coerce strings via
   safe_json_loads before evaluating.

2. The allow_client_tags strip only covered body metadata.tags and
   litellm_metadata.tags — caller-supplied tags arriving via the
   x-litellm-tags header or root-level data["tags"] bypassed it. Gate
   add_request_tag_to_metadata's result on the same flag.

3. requester_metadata was deepcopied BEFORE the strip, so attacker
   injections (user_api_key_metadata shadows, disallowed tags,
   _pipeline_managed_guardrails) persisted in the snapshot. The PANW
   guardrail (and any future consumer) trusting requester_metadata
   would see forged values. Move the deepcopy to after the strip.

Regression tests added for each.
---
 litellm/proxy/auth/auth_checks.py             |  22 ++-
 litellm/proxy/litellm_pre_call_utils.py       |  30 +++-
 .../proxy/auth/test_auth_checks.py            |  37 +++++
 .../proxy/test_litellm_pre_call_utils.py      | 131 ++++++++++++++++++
 4 files changed, 213 insertions(+), 7 deletions(-)

diff --git a/litellm/proxy/auth/auth_checks.py b/litellm/proxy/auth/auth_checks.py
index 1f4b5311baf..846d61384b4 100644
--- a/litellm/proxy/auth/auth_checks.py
+++ b/litellm/proxy/auth/auth_checks.py
@@ -350,12 +350,30 @@ def _guardrail_modification_check(
     failing loudly at the auth layer so operators see an explicit 403 instead
     of a confusing silent-ignore.
     """
+    from litellm.litellm_core_utils.safe_json_loads import safe_json_loads
     from litellm.proxy.guardrails.guardrail_helpers import can_modify_guardrails
 
+    def _coerce_to_dict(container: Any) -> Optional[dict]:
+        """Accept dict or JSON-string (from multipart/form-data or extra_body).
+
+        Without this, an attacker can smuggle guardrail keys past the check by
+        sending ``{"metadata": "{\\"disable_global_guardrails\\": true}"}`` —
+        ``isinstance(dict)`` on the string returns False, the check returns
+        no-modification, and ``add_litellm_data_to_request`` parses the string
+        to a dict downstream.
+        """
+        if isinstance(container, dict):
+            return container
+        if isinstance(container, str):
+            parsed = safe_json_loads(container)
+            return parsed if isinstance(parsed, dict) else None
+        return None
+
     def _user_requested_modification(container: Any) -> bool:
-        if not isinstance(container, dict):
+        coerced = _coerce_to_dict(container)
+        if coerced is None:
             return False
-        return any(container.get(key) for key in _GUARDRAIL_MODIFICATION_KEYS)
+        return any(coerced.get(key) for key in _GUARDRAIL_MODIFICATION_KEYS)
 
     # Check both metadata keys — callers can populate either depending on the
     # endpoint. Cover the top-level too so root-level injection is rejected.
diff --git a/litellm/proxy/litellm_pre_call_utils.py b/litellm/proxy/litellm_pre_call_utils.py
index 33affa5351c..51f82b7c00d 100644
--- a/litellm/proxy/litellm_pre_call_utils.py
+++ b/litellm/proxy/litellm_pre_call_utils.py
@@ -1069,9 +1069,10 @@ async def add_litellm_data_to_request(  # noqa: PLR0915
                 verbose_proxy_logger.warning(
                     f"Failed to parse 'metadata' as JSON dict. Received value: {data['metadata']}"
                 )
-        data[_metadata_variable_name]["requester_metadata"] = copy.deepcopy(
-            data["metadata"]
-        )
+        # requester_metadata is snapshotted AFTER the strip below so
+        # downstream consumers (e.g. PANW guardrail reading user_ip /
+        # profile_id) don't see attacker-injected admin slots preserved in
+        # the deepcopy.
 
     # Parse litellm_metadata if it's a string (e.g., from multipart/form-data or extra_body)
     if "litellm_metadata" in data and data["litellm_metadata"] is not None:
@@ -1131,6 +1132,16 @@ async def add_litellm_data_to_request(  # noqa: PLR0915
                 ", ".join(_stripped_from),
             )
 
+    # Snapshot the (now-cleaned) requester-supplied metadata for downstream
+    # consumers. Taking the deepcopy AFTER the strip prevents attacker-
+    # injected admin slots (user_api_key_metadata, tags without opt-in,
+    # _pipeline_managed_guardrails) from surviving in requester_metadata
+    # where guardrails and audit paths may read from it.
+    if "metadata" in data and isinstance(data["metadata"], dict):
+        data[_metadata_variable_name]["requester_metadata"] = copy.deepcopy(
+            data["metadata"]
+        )
+
     # Now merge litellm_metadata into the metadata variable (preserving existing
     # values) — runs AFTER the strip so attacker injections in litellm_metadata
     # cannot cross-contaminate the admin-authoritative metadata dict.
@@ -1300,15 +1311,24 @@ async def add_litellm_data_to_request(  # noqa: PLR0915
         user_agent = request.headers["user-agent"]
     data[_metadata_variable_name]["user_agent"] = user_agent
 
-    # Check if using tag based routing
+    # Check if using tag based routing. The helper reads caller-controlled
+    # sources (x-litellm-tags header, data["tags"] root-level), so its result
+    # is still gated by the same allow_client_tags flag that gated the
+    # body-metadata tag strip above. Otherwise the strip is trivially
+    # bypassed by sending tags via header or at the root of the body.
     tags = LiteLLMProxyRequestSetup.add_request_tag_to_metadata(
         llm_router=llm_router,
         headers=_headers,
         data=data,
     )
 
-    if tags is not None:
+    if tags is not None and _admin_allow_client_tags:
         data[_metadata_variable_name]["tags"] = tags
+    elif tags is not None:
+        verbose_proxy_logger.warning(
+            "Ignored caller-supplied tags from header/root body: this "
+            "key/team does not have `allow_client_tags: true` in its metadata."
+        )
 
     # Team Callbacks controls
     callback_settings_obj = _get_dynamic_logging_metadata(
diff --git a/tests/test_litellm/proxy/auth/test_auth_checks.py b/tests/test_litellm/proxy/auth/test_auth_checks.py
index a3d24fc8bf6..0391e224084 100644
--- a/tests/test_litellm/proxy/auth/test_auth_checks.py
+++ b/tests/test_litellm/proxy/auth/test_auth_checks.py
@@ -1897,3 +1897,40 @@ class TestGuardrailModificationCheck:
         ):
             # no-op, should not raise
             self._call({"metadata": {"disable_global_guardrails": True}})
+
+    def test_rejects_string_encoded_metadata_bypass(self):
+        """Regression: attacker sends metadata as JSON string to bypass the
+        isinstance(dict) guard. The check must coerce the string to dict
+        and evaluate guardrail modification keys inside it."""
+        import json as _json
+
+        from fastapi import HTTPException
+
+        attacker_payload = {"disable_global_guardrails": True}
+        with patch(
+            "litellm.proxy.guardrails.guardrail_helpers.can_modify_guardrails",
+            return_value=False,
+        ):
+            with pytest.raises(HTTPException) as exc:
+                self._call({"metadata": _json.dumps(attacker_payload)})
+            assert exc.value.status_code == 403
+
+    def test_rejects_string_encoded_litellm_metadata_bypass(self):
+        """Same bypass via the litellm_metadata key."""
+        import json as _json
+
+        from fastapi import HTTPException
+
+        attacker_payload = {"guardrails": ["evaded"]}
+        with patch(
+            "litellm.proxy.guardrails.guardrail_helpers.can_modify_guardrails",
+            return_value=False,
+        ):
+            with pytest.raises(HTTPException) as exc:
+                self._call({"litellm_metadata": _json.dumps(attacker_payload)})
+            assert exc.value.status_code == 403
+
+    def test_noop_when_string_is_not_json_object(self):
+        """Unparseable strings should not trigger a 403 — they have no keys."""
+        self._call({"metadata": "not-json"})
+        self._call({"metadata": '"just a string"'})
diff --git a/tests/test_litellm/proxy/test_litellm_pre_call_utils.py b/tests/test_litellm/proxy/test_litellm_pre_call_utils.py
index 351857b9b10..5f5ae0c64d9 100644
--- a/tests/test_litellm/proxy/test_litellm_pre_call_utils.py
+++ b/tests/test_litellm/proxy/test_litellm_pre_call_utils.py
@@ -344,6 +344,137 @@ async def test_add_litellm_data_to_request_strips_string_encoded_admin_injection
     assert "_pipeline_managed_guardrails" not in other
 
 
+@pytest.mark.asyncio
+async def test_add_litellm_data_to_request_ignores_x_litellm_tags_header_without_permission():
+    """Regression: the `x-litellm-tags` header bypassed the body-metadata
+    tag strip. Header tags must also be gated by `allow_client_tags`."""
+    from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request
+
+    request_mock = MagicMock(spec=Request)
+    request_mock.url.path = "/v1/chat/completions"
+    request_mock.url = MagicMock()
+    request_mock.url.__str__.return_value = "http://localhost/v1/chat/completions"
+    request_mock.method = "POST"
+    request_mock.query_params = {}
+    request_mock.headers = {
+        "Content-Type": "application/json",
+        "x-litellm-tags": "restricted-tier,victim-team",
+    }
+    request_mock.client = MagicMock()
+    request_mock.client.host = "127.0.0.1"
+
+    data = {"model": "gpt-3.5-turbo"}
+
+    user_api_key_dict = UserAPIKeyAuth(
+        api_key="hashed-key",
+        metadata={},
+        team_metadata={},
+        spend=0.0,
+        max_budget=100.0,
+        model_max_budget={},
+        team_spend=0.0,
+        team_max_budget=200.0,
+    )
+
+    updated = await add_litellm_data_to_request(
+        data=data,
+        request=request_mock,
+        user_api_key_dict=user_api_key_dict,
+        proxy_config=MagicMock(),
+        general_settings={},
+        version="test-version",
+    )
+
+    assert "tags" not in (updated.get("metadata") or {})
+
+
+@pytest.mark.asyncio
+async def test_add_litellm_data_to_request_ignores_root_level_tags_without_permission():
+    """Regression: root-level `data["tags"]` bypassed the body-metadata
+    tag strip. Root-level tags must also be gated by `allow_client_tags`."""
+    from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request
+
+    request_mock = MagicMock(spec=Request)
+    request_mock.url.path = "/v1/chat/completions"
+    request_mock.url = MagicMock()
+    request_mock.url.__str__.return_value = "http://localhost/v1/chat/completions"
+    request_mock.method = "POST"
+    request_mock.query_params = {}
+    request_mock.headers = {"Content-Type": "application/json"}
+    request_mock.client = MagicMock()
+    request_mock.client.host = "127.0.0.1"
+
+    data = {
+        "model": "gpt-3.5-turbo",
+        "tags": ["restricted-tier", "victim-team"],
+    }
+
+    user_api_key_dict = UserAPIKeyAuth(
+        api_key="hashed-key",
+        metadata={},
+        team_metadata={},
+        spend=0.0,
+        max_budget=100.0,
+        model_max_budget={},
+        team_spend=0.0,
+        team_max_budget=200.0,
+    )
+
+    updated = await add_litellm_data_to_request(
+        data=data,
+        request=request_mock,
+        user_api_key_dict=user_api_key_dict,
+        proxy_config=MagicMock(),
+        general_settings={},
+        version="test-version",
+    )
+
+    assert "tags" not in (updated.get("metadata") or {})
+
+
+@pytest.mark.asyncio
+async def test_add_litellm_data_to_request_honors_header_tags_when_opted_in():
+    """When allow_client_tags=True, header-supplied tags flow through."""
+    from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request
+
+    request_mock = MagicMock(spec=Request)
+    request_mock.url.path = "/v1/chat/completions"
+    request_mock.url = MagicMock()
+    request_mock.url.__str__.return_value = "http://localhost/v1/chat/completions"
+    request_mock.method = "POST"
+    request_mock.query_params = {}
+    request_mock.headers = {
+        "Content-Type": "application/json",
+        "x-litellm-tags": "production,ab-test",
+    }
+    request_mock.client = MagicMock()
+    request_mock.client.host = "127.0.0.1"
+
+    data = {"model": "gpt-3.5-turbo"}
+
+    user_api_key_dict = UserAPIKeyAuth(
+        api_key="hashed-key",
+        metadata={"allow_client_tags": True},
+        team_metadata={},
+        spend=0.0,
+        max_budget=100.0,
+        model_max_budget={},
+        team_spend=0.0,
+        team_max_budget=200.0,
+    )
+
+    updated = await add_litellm_data_to_request(
+        data=data,
+        request=request_mock,
+        user_api_key_dict=user_api_key_dict,
+        proxy_config=MagicMock(),
+        general_settings={},
+        version="test-version",
+    )
+
+    assert updated["metadata"].get("tags") == ["production", "ab-test"]
+
+
 @pytest.mark.asyncio
 async def test_add_litellm_data_to_request_strips_user_tags_without_permission():
     """Caller-supplied metadata.tags must be stripped when the key/team

From b4e98d190a42c04c1e7cf3c44abf7524a92826d5 Mon Sep 17 00:00:00 2001
From: user <70670632+stuxf@users.noreply.github.com>
Date: Fri, 17 Apr 2026 00:08:40 +0000
Subject: [PATCH 14/15] fix(proxy): close 6 more metadata/tag variant bypasses
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Post-merge audit found 6 adjacent variants of the VERIA-28 class. All
fixed here with regression tests:

1. Strip widened from 3 named keys to the full user_api_key_* prefix.
   The proxy writes a dozen user_api_key_* fields (user_id, alias,
   spend, team_id, request_route, end_user_id, …) into
   data[_metadata_variable_name]; the 3-key strip left the rest
   exploitable for identity/spend forgery in audit logs and guardrails.

2. proxy_server_request['body'] snapshot moved to AFTER the strip.
   Was captured at line ~990 before the strip ran, so
   standard_logging_object, lago, and spend_tracking readers saw the
   attacker-forged payload even though the live data dict was clean.

3. get_tags_from_request_body (auth-time) now coerces JSON-string
   metadata via safe_json_loads. Previously crashed with
   AttributeError on string metadata (DoS; potential RBAC bypass if
   a caller swallowed the exception).

4. get_end_user_id_from_request_body coerces JSON-string
   metadata/litellm_metadata. Previously isinstance(dict) guard
   caused end-user budget attribution to be silently skipped when
   the caller sent metadata as a JSON string.

5. Four hand-rolled 'if data.get("metadata") is None: data["metadata"] = {}'
   blocks in proxy_server.py (7160, 7341, 7590, 11375) now guard on
   isinstance(dict). They crashed with TypeError when metadata was a
   JSON string (DoS).

6. _get_admin_metadata defensively guards with isinstance(dict);
   previously AttributeError'd on any leaked string metadata.

Also hoists the inline safe_json_loads import in _guardrail_modification_check
to module level per CLAUDE.md style.
---
 litellm/integrations/custom_guardrail.py      |   6 +-
 litellm/proxy/auth/auth_checks.py             |   2 +-
 litellm/proxy/auth/auth_utils.py              |  31 ++--
 .../proxy/common_utils/http_parsing_utils.py  |  19 +-
 litellm/proxy/litellm_pre_call_utils.py       |  36 ++--
 litellm/proxy/proxy_server.py                 |  15 +-
 .../common_utils/test_http_parsing_utils.py   |  38 ++++
 .../proxy/test_litellm_pre_call_utils.py      | 166 ++++++++++++++++++
 8 files changed, 281 insertions(+), 32 deletions(-)

diff --git a/litellm/integrations/custom_guardrail.py b/litellm/integrations/custom_guardrail.py
index b0931964cc6..abf010e0d65 100644
--- a/litellm/integrations/custom_guardrail.py
+++ b/litellm/integrations/custom_guardrail.py
@@ -268,7 +268,11 @@ class CustomGuardrail(CustomLogger):
         team_meta: dict = {}
         key_meta: dict = {}
         for key in ("metadata", "litellm_metadata"):
-            meta = data.get(key) or {}
+            # Defensive: an unparsed JSON-string metadata could leak past the
+            # proxy's normal parse path; don't AttributeError on .get().
+            meta = data.get(key)
+            if not isinstance(meta, dict):
+                continue
             team_meta = meta.get("user_api_key_team_metadata") or team_meta
             key_meta = meta.get("user_api_key_metadata") or key_meta
         return {**team_meta, **key_meta}
diff --git a/litellm/proxy/auth/auth_checks.py b/litellm/proxy/auth/auth_checks.py
index 846d61384b4..1621f04213f 100644
--- a/litellm/proxy/auth/auth_checks.py
+++ b/litellm/proxy/auth/auth_checks.py
@@ -31,6 +31,7 @@ from litellm.constants import (
 )
 from litellm.litellm_core_utils.dd_tracing import tracer
 from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider
+from litellm.litellm_core_utils.safe_json_loads import safe_json_loads
 from litellm.proxy._types import (
     RBAC_ROLES,
     CallInfo,
@@ -350,7 +351,6 @@ def _guardrail_modification_check(
     failing loudly at the auth layer so operators see an explicit 403 instead
     of a confusing silent-ignore.
     """
-    from litellm.litellm_core_utils.safe_json_loads import safe_json_loads
     from litellm.proxy.guardrails.guardrail_helpers import can_modify_guardrails
 
     def _coerce_to_dict(container: Any) -> Optional[dict]:
diff --git a/litellm/proxy/auth/auth_utils.py b/litellm/proxy/auth/auth_utils.py
index 64766bbaadd..12d9bff91e0 100644
--- a/litellm/proxy/auth/auth_utils.py
+++ b/litellm/proxy/auth/auth_utils.py
@@ -823,19 +823,30 @@ def get_end_user_id_from_request_body(
         user_from_body_user_field = request_body["user"]
         return str(user_from_body_user_field)
 
+    def _as_dict(value: Any) -> dict:
+        # metadata / litellm_metadata can arrive as JSON strings from
+        # multipart/form-data or extra_body; coerce so string-encoded
+        # payloads can't evade end-user attribution.
+        if isinstance(value, dict):
+            return value
+        if isinstance(value, str):
+            from litellm.litellm_core_utils.safe_json_loads import safe_json_loads
+
+            parsed = safe_json_loads(value)
+            return parsed if isinstance(parsed, dict) else {}
+        return {}
+
     # Check 4: 'litellm_metadata.user' in request_body (commonly Anthropic)
-    litellm_metadata = request_body.get("litellm_metadata")
-    if isinstance(litellm_metadata, dict):
-        user_from_litellm_metadata = litellm_metadata.get("user")
-        if user_from_litellm_metadata is not None:
-            return str(user_from_litellm_metadata)
+    litellm_metadata = _as_dict(request_body.get("litellm_metadata"))
+    user_from_litellm_metadata = litellm_metadata.get("user")
+    if user_from_litellm_metadata is not None:
+        return str(user_from_litellm_metadata)
 
     # Check 5: 'metadata.user_id' in request_body (another common pattern)
-    metadata_dict = request_body.get("metadata")
-    if isinstance(metadata_dict, dict):
-        user_id_from_metadata_field = metadata_dict.get("user_id")
-        if user_id_from_metadata_field is not None:
-            return str(user_id_from_metadata_field)
+    metadata_dict = _as_dict(request_body.get("metadata"))
+    user_id_from_metadata_field = metadata_dict.get("user_id")
+    if user_id_from_metadata_field is not None:
+        return str(user_id_from_metadata_field)
 
     # Check 6: 'safety_identifier' in request body (OpenAI Responses API parameter)
     # SECURITY NOTE: safety_identifier can be set by any caller in the request body.
diff --git a/litellm/proxy/common_utils/http_parsing_utils.py b/litellm/proxy/common_utils/http_parsing_utils.py
index 1dd25262127..71abdfa5e9e 100644
--- a/litellm/proxy/common_utils/http_parsing_utils.py
+++ b/litellm/proxy/common_utils/http_parsing_utils.py
@@ -197,10 +197,10 @@ def check_file_size_under_limit(
 
     if llm_router is not None and request_data["model"] in router_model_names:
         try:
-            deployment: Optional[
-                Deployment
-            ] = llm_router.get_deployment_by_model_group_name(
-                model_group_name=request_data["model"]
+            deployment: Optional[Deployment] = (
+                llm_router.get_deployment_by_model_group_name(
+                    model_group_name=request_data["model"]
+                )
             )
             if (
                 deployment
@@ -426,7 +426,16 @@ def get_tags_from_request_body(request_body: dict) -> List[str]:
         List of tag names (strings), empty list if no valid tags found
     """
     metadata_variable_name = get_metadata_variable_name_from_kwargs(request_body)
-    metadata = request_body.get(metadata_variable_name) or {}
+    metadata = request_body.get(metadata_variable_name)
+    # metadata can arrive as a JSON string from multipart/form-data or extra_body;
+    # coerce defensively so .get() below never raises AttributeError.
+    if isinstance(metadata, str):
+        from litellm.litellm_core_utils.safe_json_loads import safe_json_loads
+
+        parsed = safe_json_loads(metadata)
+        metadata = parsed if isinstance(parsed, dict) else {}
+    elif not isinstance(metadata, dict):
+        metadata = {}
     tags_in_metadata: Any = metadata.get("tags", [])
     tags_in_request_body: Any = request_body.get("tags", [])
     combined_tags: List[str] = []
diff --git a/litellm/proxy/litellm_pre_call_utils.py b/litellm/proxy/litellm_pre_call_utils.py
index 51f82b7c00d..aa3a29e4f2f 100644
--- a/litellm/proxy/litellm_pre_call_utils.py
+++ b/litellm/proxy/litellm_pre_call_utils.py
@@ -981,13 +981,16 @@ async def add_litellm_data_to_request(  # noqa: PLR0915
     # Init - Proxy Server Request
     # we do this as soon as entering so we track the original request
     ##########################################################
-    # Track arrival time for queue time metric
+    # Track arrival time for queue time metric. The body snapshot is filled
+    # in after the admin-injection strip below so the audit / spend-tracking
+    # consumers of proxy_server_request["body"] see the cleaned metadata
+    # rather than attacker-forged user_api_key_* fields.
     arrival_time = time.time()
     data["proxy_server_request"] = {
         "url": str(request.url),
         "method": request.method,
         "headers": _headers,
-        "body": copy.copy(data),  # use copy instead of deepcopy
+        "body": None,  # filled in post-strip; see below
         "arrival_time": arrival_time,  # Track when request arrived at proxy
     }
 
@@ -1087,19 +1090,24 @@ async def add_litellm_data_to_request(  # noqa: PLR0915
 
     # Strip internal pipeline state and admin-injection slots from user input.
     # Runs AFTER the string-to-dict parse above so JSON-string metadata (sent
-    # via multipart/form-data or extra_body) cannot smuggle `user_api_key_metadata`
-    # past the isinstance(dict) guard.
+    # via multipart/form-data or extra_body) cannot smuggle admin fields past
+    # the isinstance(dict) guard.
     #
-    # The proxy writes user_api_key_metadata / user_api_key_team_metadata into
-    # data[_metadata_variable_name] below; if a caller pre-populates either
-    # key on the OTHER metadata field, _get_admin_metadata lookups would treat
-    # the caller's payload as admin-configured.
+    # The proxy populates a family of ``user_api_key_*`` fields below
+    # (user_api_key_metadata, user_api_key_user_id, user_api_key_alias,
+    # user_api_key_spend, user_api_key_team_metadata, …) into
+    # data[_metadata_variable_name]. Because the proxy only writes to ONE of
+    # the two metadata dicts, a caller pre-populating any of these keys on
+    # the OTHER metadata dict would have their forged values surface in
+    # guardrails, spend tracking, audit logs, and identity resolution. Strip
+    # by prefix so new ``user_api_key_*`` fields added in the future are
+    # covered without per-key maintenance.
     for _meta_key in ("metadata", "litellm_metadata"):
         _user_meta = data.get(_meta_key)
         if isinstance(_user_meta, dict):
             _user_meta.pop("_pipeline_managed_guardrails", None)
-            _user_meta.pop("user_api_key_metadata", None)
-            _user_meta.pop("user_api_key_team_metadata", None)
+            for _k in [k for k in _user_meta if k.startswith("user_api_key_")]:
+                _user_meta.pop(_k, None)
 
     # Strip caller-supplied routing/budget tags unless the admin has opted
     # this key or team in via metadata.allow_client_tags=True. Tags drive
@@ -1132,9 +1140,15 @@ async def add_litellm_data_to_request(  # noqa: PLR0915
                 ", ".join(_stripped_from),
             )
 
+    # Fill in the proxy_server_request body snapshot now that metadata has
+    # been parsed and stripped. Consumers (standard_logging_payload, lago,
+    # spend_tracking_utils, streaming_iterator) read `body` to audit the
+    # request; taking the snapshot here ensures they see cleaned metadata.
+    data["proxy_server_request"]["body"] = copy.copy(data)
+
     # Snapshot the (now-cleaned) requester-supplied metadata for downstream
     # consumers. Taking the deepcopy AFTER the strip prevents attacker-
-    # injected admin slots (user_api_key_metadata, tags without opt-in,
+    # injected admin slots (user_api_key_*, tags without opt-in,
     # _pipeline_managed_guardrails) from surviving in requester_metadata
     # where guardrails and audit paths may read from it.
     if "metadata" in data and isinstance(data["metadata"], dict):
diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py
index 2d789b982da..1d61bebee44 100644
--- a/litellm/proxy/proxy_server.py
+++ b/litellm/proxy/proxy_server.py
@@ -2201,9 +2201,11 @@ def run_ollama_serve():
         with open(os.devnull, "w") as devnull:
             subprocess.Popen(command, stdout=devnull, stderr=devnull)
     except Exception as e:
-        verbose_proxy_logger.debug(f"""
+        verbose_proxy_logger.debug(
+            f"""
             LiteLLM Warning: proxy started with `ollama` model\n`ollama serve` failed with Exception{e}. \nEnsure you run `ollama serve`
-        """)
+        """
+        )
 
 
 def _get_process_rss_mb() -> Optional[float]:
@@ -7157,7 +7159,10 @@ async def chat_completion(  # noqa: PLR0915
     global user_temperature, user_request_timeout, user_max_tokens, user_api_base
     data = await _read_request_body(request=request)
     if user_api_key_dict is not None:
-        if data.get("metadata") is None:
+        if not isinstance(data.get("metadata"), dict):
+            # Covers both missing and JSON-string metadata (multipart /
+            # extra_body); otherwise `data["metadata"][k] = v` below raises
+            # TypeError on a string value and 500s the request.
             data["metadata"] = {}
         if (
             hasattr(user_api_key_dict, "user_id")
@@ -11372,7 +11377,9 @@ async def async_queue_request(
             # if users are using user_api_key_auth, set `user` in `data`
             data["user"] = user_api_key_dict.user_id
 
-        if "metadata" not in data:
+        if not isinstance(data.get("metadata"), dict):
+            # Covers both missing and JSON-string metadata (multipart /
+            # extra_body); see above for the same guard upstream.
             data["metadata"] = {}
         data["metadata"]["user_api_key"] = user_api_key_dict.api_key
         data["metadata"]["user_api_key_metadata"] = user_api_key_dict.metadata
diff --git a/tests/test_litellm/proxy/common_utils/test_http_parsing_utils.py b/tests/test_litellm/proxy/common_utils/test_http_parsing_utils.py
index a1484bc263b..c9f595626ed 100644
--- a/tests/test_litellm/proxy/common_utils/test_http_parsing_utils.py
+++ b/tests/test_litellm/proxy/common_utils/test_http_parsing_utils.py
@@ -835,3 +835,41 @@ def test_safe_get_request_headers_state_unavailable():
 
     result = _safe_get_request_headers(mock_request)
     assert result == {"content-type": "application/json"}
+
+
+class TestGetTagsFromRequestBodyStringCoerce:
+    """Regression: the auth-time tag helper used `metadata.get("tags", ...)`
+    directly, which raised AttributeError when metadata arrived as a JSON
+    string (multipart/form-data or extra_body). That turned into a DoS at
+    auth time and potentially bypassed tag-based RBAC if the caller caught
+    the exception and fell through with empty tags.
+    """
+
+    def test_json_string_metadata_is_coerced_to_dict(self):
+        from litellm.proxy.common_utils.http_parsing_utils import (
+            get_tags_from_request_body,
+        )
+
+        metadata_json = json.dumps({"tags": ["a", "b"]})
+        # Must not raise
+        tags = get_tags_from_request_body({"metadata": metadata_json})
+        assert tags == ["a", "b"]
+
+    def test_unparseable_string_metadata_is_ignored(self):
+        from litellm.proxy.common_utils.http_parsing_utils import (
+            get_tags_from_request_body,
+        )
+
+        # Must not raise; must yield no metadata tags but keep root tags
+        tags = get_tags_from_request_body(
+            {"metadata": "not-json", "tags": ["root-only"]}
+        )
+        assert tags == ["root-only"]
+
+    def test_dict_metadata_still_works(self):
+        from litellm.proxy.common_utils.http_parsing_utils import (
+            get_tags_from_request_body,
+        )
+
+        tags = get_tags_from_request_body({"metadata": {"tags": ["x"]}})
+        assert tags == ["x"]
diff --git a/tests/test_litellm/proxy/test_litellm_pre_call_utils.py b/tests/test_litellm/proxy/test_litellm_pre_call_utils.py
index 5f5ae0c64d9..664a936b540 100644
--- a/tests/test_litellm/proxy/test_litellm_pre_call_utils.py
+++ b/tests/test_litellm/proxy/test_litellm_pre_call_utils.py
@@ -280,6 +280,172 @@ async def test_add_litellm_data_to_request_strips_admin_injection_slots():
     assert "_pipeline_managed_guardrails" not in other
 
 
+@pytest.mark.asyncio
+async def test_add_litellm_data_to_request_strips_all_user_api_key_prefix_keys():
+    """Strip must cover the full user_api_key_* family, not a hand-maintained
+    list of 2-3 names. Proxy writes a dozen such fields (user_id, alias,
+    spend, team_id, request_route, …) and an attacker populating any of them
+    in the non-authoritative metadata key would otherwise forge identity /
+    spend in audit logs and guardrails."""
+    from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request
+
+    request_mock = MagicMock(spec=Request)
+    request_mock.url.path = "/v1/chat/completions"
+    request_mock.url = MagicMock()
+    request_mock.url.__str__.return_value = "http://localhost/v1/chat/completions"
+    request_mock.method = "POST"
+    request_mock.query_params = {}
+    request_mock.headers = {"Content-Type": "application/json"}
+    request_mock.client = MagicMock()
+    request_mock.client.host = "127.0.0.1"
+
+    attacker_injected = {
+        "user_api_key_user_id": "victim",
+        "user_api_key_alias": "admin-key",
+        "user_api_key_spend": 0.0,
+        "user_api_key_team_id": "victim-team",
+        "user_api_key_end_user_id": "victim-user",
+        "user_api_key_request_route": "/fake/route",
+        "user_api_key_hash": "fake-hash",
+    }
+    data = {
+        "model": "gpt-3.5-turbo",
+        "metadata": {**attacker_injected},
+        "litellm_metadata": {**attacker_injected},
+    }
+
+    user_api_key_dict = UserAPIKeyAuth(
+        api_key="hashed-key",
+        user_id="real-user",
+        metadata={},
+        team_metadata={},
+        spend=42.0,
+        max_budget=100.0,
+        model_max_budget={},
+        team_spend=0.0,
+        team_max_budget=200.0,
+    )
+
+    updated = await add_litellm_data_to_request(
+        data=data,
+        request=request_mock,
+        user_api_key_dict=user_api_key_dict,
+        proxy_config=MagicMock(),
+        general_settings={},
+        version="test-version",
+    )
+
+    # The non-authoritative metadata dict must not retain ANY attacker-injected
+    # user_api_key_* key.
+    other = updated.get("litellm_metadata") or {}
+    attacker_leaks = [k for k in other if k.startswith("user_api_key_")]
+    assert attacker_leaks == [], f"Unexpected leaked keys: {attacker_leaks}"
+
+
+@pytest.mark.asyncio
+async def test_add_litellm_data_to_request_string_metadata_does_not_crash():
+    """Regression: pre-strip code that pre-populated data['metadata'][k]=v
+    before the string-to-dict parse would crash on JSON-string metadata.
+    The snapshot / strip / admin-population pipeline must survive metadata
+    arriving as a string."""
+    import json as _json
+
+    from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request
+
+    request_mock = MagicMock(spec=Request)
+    request_mock.url.path = "/v1/chat/completions"
+    request_mock.url = MagicMock()
+    request_mock.url.__str__.return_value = "http://localhost/v1/chat/completions"
+    request_mock.method = "POST"
+    request_mock.query_params = {}
+    request_mock.headers = {"Content-Type": "multipart/form-data"}
+    request_mock.client = MagicMock()
+    request_mock.client.host = "127.0.0.1"
+
+    data = {
+        "model": "gpt-3.5-turbo",
+        "metadata": _json.dumps({"generation_name": "test"}),
+    }
+
+    user_api_key_dict = UserAPIKeyAuth(
+        api_key="hashed-key",
+        metadata={},
+        team_metadata={},
+        spend=0.0,
+        max_budget=100.0,
+        model_max_budget={},
+        team_spend=0.0,
+        team_max_budget=200.0,
+    )
+
+    # Must not raise TypeError / AttributeError.
+    updated = await add_litellm_data_to_request(
+        data=data,
+        request=request_mock,
+        user_api_key_dict=user_api_key_dict,
+        proxy_config=MagicMock(),
+        general_settings={},
+        version="test-version",
+    )
+
+    # The parsed metadata should be a dict and the proxy snapshot body
+    # should have been taken AFTER the strip (so no leaked user_api_key_*
+    # from a raw string snapshot).
+    assert isinstance(updated["metadata"], dict)
+    assert updated["metadata"].get("generation_name") == "test"
+
+
+@pytest.mark.asyncio
+async def test_add_litellm_data_to_request_proxy_server_request_body_is_post_strip():
+    """Regression: proxy_server_request['body'] used to be snapshotted before
+    the admin-slot strip, so standard_logging_object and spend-tracking
+    readers saw attacker-injected payload. Snapshot must now be post-strip."""
+    from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request
+
+    request_mock = MagicMock(spec=Request)
+    request_mock.url.path = "/v1/chat/completions"
+    request_mock.url = MagicMock()
+    request_mock.url.__str__.return_value = "http://localhost/v1/chat/completions"
+    request_mock.method = "POST"
+    request_mock.query_params = {}
+    request_mock.headers = {"Content-Type": "application/json"}
+    request_mock.client = MagicMock()
+    request_mock.client.host = "127.0.0.1"
+
+    data = {
+        "model": "gpt-3.5-turbo",
+        "metadata": {"user_api_key_user_id": "victim"},
+    }
+
+    user_api_key_dict = UserAPIKeyAuth(
+        api_key="hashed-key",
+        user_id="real-user",
+        metadata={},
+        team_metadata={},
+        spend=0.0,
+        max_budget=100.0,
+        model_max_budget={},
+        team_spend=0.0,
+        team_max_budget=200.0,
+    )
+
+    updated = await add_litellm_data_to_request(
+        data=data,
+        request=request_mock,
+        user_api_key_dict=user_api_key_dict,
+        proxy_config=MagicMock(),
+        general_settings={},
+        version="test-version",
+    )
+
+    snapshot_body = updated["proxy_server_request"]["body"]
+    assert snapshot_body is not None
+    snapshot_metadata = snapshot_body.get("metadata") or {}
+    assert "user_api_key_user_id" not in snapshot_metadata or (
+        snapshot_metadata["user_api_key_user_id"] != "victim"
+    )
+
+
 @pytest.mark.asyncio
 async def test_add_litellm_data_to_request_strips_string_encoded_admin_injection():
     """Regression: metadata arriving as a JSON string (multipart/form-data or

From 132063289fca416c18b55fa99fa44919a2b0b41b Mon Sep 17 00:00:00 2001
From: user <70670632+stuxf@users.noreply.github.com>
Date: Fri, 17 Apr 2026 00:20:52 +0000
Subject: [PATCH 15/15] fix(proxy): strip root-level data['tags'] alongside
 metadata tags

Greptile P2. The admin-inject gate only removed tags from data['metadata']
and data['litellm_metadata'];  and the
policy engine read  directly, so a caller without
allow_client_tags could still drive tag-based policy decisions by moving
tags to the body root. Also strip the root key in the same branch.
---
 litellm/proxy/litellm_pre_call_utils.py                 | 7 +++++++
 tests/test_litellm/proxy/test_litellm_pre_call_utils.py | 5 +++++
 2 files changed, 12 insertions(+)

diff --git a/litellm/proxy/litellm_pre_call_utils.py b/litellm/proxy/litellm_pre_call_utils.py
index aa3a29e4f2f..7467bbae232 100644
--- a/litellm/proxy/litellm_pre_call_utils.py
+++ b/litellm/proxy/litellm_pre_call_utils.py
@@ -1132,6 +1132,13 @@ async def add_litellm_data_to_request(  # noqa: PLR0915
             if isinstance(_user_meta, dict) and "tags" in _user_meta:
                 _user_meta.pop("tags", None)
                 _stripped_from.append(_meta_key)
+        # Also strip the root-level `tags` field. get_tags_from_request_body
+        # reads request_body["tags"] directly and feeds it to the policy
+        # engine, so leaving it in place here would let the strip-in-metadata
+        # above be trivially bypassed by moving the tags to the body root.
+        if "tags" in data:
+            data.pop("tags", None)
+            _stripped_from.append("tags (root)")
         if _stripped_from:
             verbose_proxy_logger.warning(
                 "Stripped caller-supplied tags from %s: this key/team does "
diff --git a/tests/test_litellm/proxy/test_litellm_pre_call_utils.py b/tests/test_litellm/proxy/test_litellm_pre_call_utils.py
index 664a936b540..360ddd426a4 100644
--- a/tests/test_litellm/proxy/test_litellm_pre_call_utils.py
+++ b/tests/test_litellm/proxy/test_litellm_pre_call_utils.py
@@ -596,6 +596,11 @@ async def test_add_litellm_data_to_request_ignores_root_level_tags_without_permi
     )
 
     assert "tags" not in (updated.get("metadata") or {})
+    # Also ensure the root-level tags are removed. get_tags_from_request_body
+    # reads request_body["tags"] directly, so leaving it in place would let
+    # the policy engine see caller-supplied tags even after the metadata
+    # strip.
+    assert "tags" not in updated
 
 
 @pytest.mark.asyncio