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

@@ -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

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