fix(guardrails): take caller identity on the apply routes from auth, not the body

/guardrails/apply_guardrail built the request data a guardrail sees straight from
the caller's metadata, so a caller could claim any identity it liked: several
user_api_key keys are aliases for the key hash, the bare user_api_key is the field
the in-line path fills from auth and that spend tracking, logging and the rate
limiter read, and agent_id feeds spend and session budgets. A guardrail keyed on any
of it, for an exemption or per-tenant behavior, could be steered by the request body.

Metadata now comes from the dict common_processing_pre_call_logic already assembled,
layered over the caller's, so the identity fields filled from auth win and
user_api_key_auth survives for tool_policy's per-key overrides. Client metadata that
is neither identity nor policy is still forwarded, which parameterized guardrails
need.

profile_id and profile_name are dropped outright: panw_prisma_airs resolves the
profile from per-request metadata as id first then name, so a caller supplying either
would choose the policy that judges it. presidio's guardrail_config is the only other
per-request knob and stays, since it narrows language and entity list inside the
scope its config already allows.

A request that sent neither messages nor metadata still gets an empty dict, which is
exactly when the route was empty before. Guardrails read that emptiness as a signal:
singulr treats it as a playground call, and that is the only branch that puts the
text on the wire, so manufacturing metadata for a bare request made it send
playground_text: null and allow everything.

litellm/proxy/guardrails/guardrail_endpoints.py

@@ -2278,6 +2278,50 @@ async def _emit_guardrail_success_logs(
     return response
 
 
+# Body metadata keys a caller must not be able to choose. The user_api_key prefix
+# covers identity: several of those keys are aliases for the key hash, and the proxy
+# fills them from auth. profile_id and profile_name both select which PANW profile
+# evaluates the content (panw_prisma_airs.py, per-request id then per-request name),
+# so a caller supplying either would pick the policy it is judged by. Everything else
+# is forwarded, which is the point of client metadata: presidio's guardrail_config,
+# for instance, only narrows language and entity list within its own configured scope.
+_CALLER_METADATA_PREFIX_DENYLIST: Final = ("user_api_key",)
+_CALLER_METADATA_DENYLIST: Final = frozenset({"profile_id", "profile_name"})
+
+
+def _guardrail_request_data(
+    request: ApplyGuardrailRequest,
+    processed_metadata: dict | None,  # mutable-ok: the proxy builds request metadata as a dict
+) -> dict:  # mutable-ok: guardrails receive request_data as a plain dict
+    """The request data a guardrail is given on the apply routes.
+
+    The caller's metadata is forwarded so parameterized guardrails keep working, but
+    the metadata the pre-call logic assembled is layered on top and wins: it carries
+    the identity fields filled from auth, plus user_api_key_auth, which per-key
+    guardrail overrides read. Keys a caller must not choose are dropped outright.
+
+    A request that sent neither messages nor metadata still gets an empty dict.
+    Guardrails treat that emptiness as a signal (singulr reads it as a playground
+    call and only then puts the text on the wire), so manufacturing metadata for a
+    bare request would silently change what they do.
+    """
+    if request.messages is None and request.metadata is None:
+        return {}  # mutable-ok: the guardrail contract passes a plain dict
+    caller_metadata: Final = {  # mutable-ok: merged into the request_data dict
+        key: value
+        for key, value in (request.metadata or {}).items()
+        if key not in _CALLER_METADATA_DENYLIST and not key.startswith(_CALLER_METADATA_PREFIX_DENYLIST)
+    }
+    return {  # mutable-ok: same reason
+        **({"messages": request.messages} if request.messages is not None else {}),
+        **(
+            {"metadata": {**caller_metadata, **(processed_metadata or {})}}  # mutable-ok: same reason
+            if request.metadata is not None or processed_metadata is not None
+            else {}
+        ),
+    }
+
+
 @router.post("/guardrails/apply_guardrail", response_model=ApplyGuardrailResponse)
 @router.post("/apply_guardrail", response_model=ApplyGuardrailResponse)
 async def apply_guardrail(
@@ -2344,10 +2388,7 @@ async def apply_guardrail(
         if litellm_logging_obj is not None:
             _patch_logging_obj_for_guardrail(litellm_logging_obj, request)
 
-        request_data: Final[dict] = {
-            **({"messages": request.messages} if request.messages is not None else {}),
-            **({"metadata": request.metadata} if request.metadata is not None else {}),
-        }
+        request_data: Final[dict] = _guardrail_request_data(request, data.get("metadata"))
         _input_type: Final = _resolve_guardrail_input_type(active_guardrail, request.input_type)
         guardrailed_inputs: Final = await active_guardrail.apply_guardrail(
             inputs={"texts": [request.text]},

tests/test_litellm/proxy/guardrails/test_guardrail_endpoints.py

@@ -1527,7 +1527,19 @@ async def test_apply_guardrail_invokes_logging_pipeline(mocker):
     }
 
 
-def _patch_apply_guardrail_env(mocker, guardrail_result):
+# What add_litellm_data_to_request leaves in data["metadata"]: the caller's own
+# metadata with every identity field overwritten from the authenticated key.
+PROCESSED_METADATA = {
+    "user_api_key": "real-hash",
+    "user_api_key_hash": "real-hash",
+    "user_api_key_alias": "real-caller",
+    "user_api_key_team_id": "real-team",
+    "agent_id": "real-agent",
+    "user_api_key_auth": "sentinel-auth-object",
+}
+
+
+def _patch_apply_guardrail_env(mocker, guardrail_result, processed_metadata=None):
     mock_guardrail = mocker.Mock()
     mock_guardrail.apply_guardrail = AsyncMock(return_value=guardrail_result)
 
@@ -1541,8 +1553,11 @@ def _patch_apply_guardrail_env(mocker, guardrail_result):
     mock_logging_obj.async_success_handler = AsyncMock()
     mock_logging_obj.model_call_details = {}
     mock_processor = mocker.Mock()
+    processed = {"guardrail_name": "test-guardrail"}
+    if processed_metadata is not None:
+        processed["metadata"] = processed_metadata
     mock_processor.common_processing_pre_call_logic = AsyncMock(
-        return_value=({"guardrail_name": "test-guardrail"}, mock_logging_obj)
+        return_value=(processed, mock_logging_obj)
     )
     mocker.patch(
         "litellm.proxy.common_request_processing.ProxyBaseLLMRequestProcessing",
@@ -1564,7 +1579,9 @@ def _patch_apply_guardrail_env(mocker, guardrail_result):
 async def test_apply_guardrail_forwards_metadata_to_guardrail(mocker):
     """Client-supplied metadata must reach apply_guardrail via request_data so
     parameterized custom guardrails can read per-request configuration."""
-    mock_guardrail = _patch_apply_guardrail_env(mocker, {"texts": ["ok"]})
+    mock_guardrail = _patch_apply_guardrail_env(
+        mocker, {"texts": ["ok"]}, processed_metadata={"route": "/apply_guardrail"}
+    )
 
     request = ApplyGuardrailRequest(
         guardrail_name="test-guardrail",
@@ -1577,18 +1594,19 @@ async def test_apply_guardrail_forwards_metadata_to_guardrail(mocker):
         user_api_key_dict=UserAPIKeyAuth(),
     )
 
-    mock_guardrail.apply_guardrail.assert_awaited_once_with(
-        inputs={"texts": ["What are tax loopholes?"]},
-        request_data={"metadata": {"forbidden_topics": ["tax"]}},
-        input_type="request",
-    )
+    call = mock_guardrail.apply_guardrail.await_args.kwargs
+    assert call["inputs"] == {"texts": ["What are tax loopholes?"]}
+    assert call["input_type"] == "request"
+    assert call["request_data"]["metadata"]["forbidden_topics"] == ["tax"]
 
 
 @pytest.mark.asyncio
 async def test_apply_guardrail_forwards_metadata_and_messages_together(mocker):
     """metadata and messages must coexist in request_data; the dict merge must
     not clobber messages when both fields are sent."""
-    mock_guardrail = _patch_apply_guardrail_env(mocker, {"texts": ["ok"]})
+    mock_guardrail = _patch_apply_guardrail_env(
+        mocker, {"texts": ["ok"]}, processed_metadata={"route": "/apply_guardrail"}
+    )
 
     messages = [{"role": "user", "content": "What are tax loopholes?"}]
     request = ApplyGuardrailRequest(
@@ -1603,19 +1621,19 @@ async def test_apply_guardrail_forwards_metadata_and_messages_together(mocker):
         user_api_key_dict=UserAPIKeyAuth(),
     )
 
-    mock_guardrail.apply_guardrail.assert_awaited_once_with(
-        inputs={"texts": ["What are tax loopholes?"]},
-        request_data={
-            "messages": messages,
-            "metadata": {"forbidden_topics": ["tax"]},
-        },
-        input_type="request",
-    )
+    call = mock_guardrail.apply_guardrail.await_args.kwargs
+    assert call["request_data"]["messages"] == messages
+    assert call["request_data"]["metadata"]["forbidden_topics"] == ["tax"]
 
 
 @pytest.mark.asyncio
 async def test_apply_guardrail_omits_metadata_when_not_sent(mocker):
-    """Without metadata, request_data stays empty (backward-compatible)."""
+    """Without metadata, request_data stays empty (backward-compatible).
+
+    Guardrails read that emptiness as a signal: singulr treats it as a playground
+    call and only then puts the text on the wire, so manufacturing metadata here
+    would silently stop it scanning.
+    """
     mock_guardrail = _patch_apply_guardrail_env(mocker, {"texts": ["ok"]})
 
     request = ApplyGuardrailRequest(guardrail_name="test-guardrail", text="hello")
@@ -1636,7 +1654,9 @@ async def test_apply_guardrail_omits_metadata_when_not_sent(mocker):
 async def test_apply_guardrail_forwards_explicit_empty_messages_and_metadata(mocker):
     """Explicitly-sent empty messages/metadata must be forwarded, not dropped;
     only omitted fields stay out of request_data."""
-    mock_guardrail = _patch_apply_guardrail_env(mocker, {"texts": ["ok"]})
+    mock_guardrail = _patch_apply_guardrail_env(
+        mocker, {"texts": ["ok"]}, processed_metadata={}
+    )
 
     request = ApplyGuardrailRequest(
         guardrail_name="test-guardrail",
@@ -1650,11 +1670,9 @@ async def test_apply_guardrail_forwards_explicit_empty_messages_and_metadata(moc
         user_api_key_dict=UserAPIKeyAuth(),
     )
 
-    mock_guardrail.apply_guardrail.assert_awaited_once_with(
-        inputs={"texts": ["hello"]},
-        request_data={"messages": [], "metadata": {}},
-        input_type="request",
-    )
+    call = mock_guardrail.apply_guardrail.await_args.kwargs
+    assert call["request_data"]["messages"] == []
+    assert "metadata" in call["request_data"]
 
 
 @pytest.mark.asyncio
@@ -2624,3 +2642,50 @@ def test_field_type_inference_handles_pep604_unions():
     assert _get_field_type_from_annotation(list[str] | None) == "array"
     assert _get_field_type_from_annotation(bool | None) == "boolean"
     assert _unwrap_optional_type(str | None) is str
+
+
+@pytest.mark.asyncio
+async def test_apply_guardrail_caller_cannot_choose_identity_or_policy(mocker):
+    """The assembled metadata wins over the body, and policy keys are dropped.
+
+    The pre-call logic fills the identity fields from auth, so layering its dict on
+    top of the caller's stops a body claiming another key, team or agent. profile_id
+    and profile_name are dropped outright: panw_prisma_airs reads both to pick which
+    profile evaluates the content, so a caller choosing either would choose the
+    policy it is judged by.
+    """
+    mock_guardrail = _patch_apply_guardrail_env(
+        mocker, {"texts": ["ok"]}, processed_metadata=dict(PROCESSED_METADATA)
+    )
+
+    request = ApplyGuardrailRequest(
+        guardrail_name="test-guardrail",
+        text="hello",
+        metadata={
+            "user_api_key": "forged-bare-hash",
+            "user_api_key_token": "forged-hash",
+            "user_api_key_alias": "forged-alias",
+            "user_api_key_team_id": "forged-team",
+            "agent_id": "forged-agent",
+            "profile_id": "permissive-profile",
+            "profile_name": "permissive-by-name",
+            "forbidden_topics": ["tax"],
+        },
+    )
+    await apply_guardrail(
+        fastapi_request=mocker.Mock(),
+        request=request,
+        user_api_key_dict=UserAPIKeyAuth(api_key="real-hash", key_alias="real-caller"),
+    )
+
+    metadata = mock_guardrail.apply_guardrail.await_args.kwargs["request_data"]["metadata"]
+    assert "forged" not in str(metadata)
+    assert "permissive" not in str(metadata)
+    assert "profile_id" not in metadata
+    assert "profile_name" not in metadata
+    assert metadata["user_api_key"] == "real-hash"
+    assert metadata["agent_id"] == "real-agent"
+    # user_api_key_auth survives; per-key guardrail overrides read it.
+    assert metadata["user_api_key_auth"] == "sentinel-auth-object"
+    # Parameterization the caller is allowed to send still arrives.
+    assert metadata["forbidden_topics"] == ["tax"]