fix(guardrails): judge a namespace member by a rule on its namespace's name

litellm/proxy/guardrails/guardrail_hooks/tool_permission.py

@@ -179,6 +179,13 @@ def _judged_call_names(
     return judged_names.get(returned_name, ())
 
 
+def _pair_namespace(qualified: str, bare: str) -> str | None:
+    suffix: Final = qualified_member_name("", bare)
+    if qualified == bare or not qualified.endswith(suffix):
+        return None
+    return _non_empty_string(qualified.removesuffix(suffix))
+
+
 def _qualified_target(namespace: str | None, target: _ToolTarget) -> _ToolTarget:
     name, tool_type = target
     return (qualified_member_name(namespace, name), tool_type) if namespace is not None and name is not None else target
@@ -737,6 +744,23 @@ class ToolPermissionGuardrail(CustomGuardrail):
             None,
         )
 
+    def _narrower_name(
+        self,
+        name: str,
+        other: str,
+        tool_type: str | None,
+        tool_call: ChatCompletionMessageToolCall | None,
+    ) -> str:
+        rule: Final = self._first_name_rule(name, tool_type, tool_call)
+        other_rule: Final = self._first_name_rule(other, tool_type, tool_call)
+        if other_rule is None or other_rule is rule:
+            return name
+        if rule is None or self._is_name_rule_for(rule, other, tool_type):
+            return other
+        if self._is_name_rule_for(other_rule, name, tool_type):
+            return name
+        return other if other_rule.decision == "deny" else name
+
     def _deciding_member_name(
         self,
         qualified: str,
@@ -744,15 +768,9 @@ class ToolPermissionGuardrail(CustomGuardrail):
         tool_type: str | None,
         tool_call: ChatCompletionMessageToolCall | None = None,
     ) -> str:
-        qualified_rule: Final = self._first_name_rule(qualified, tool_type, tool_call)
-        bare_rule: Final = self._first_name_rule(bare, tool_type, tool_call)
-        if bare_rule is None or bare_rule is qualified_rule:
-            return qualified
-        if qualified_rule is None or self._is_name_rule_for(qualified_rule, bare, tool_type):
-            return bare
-        if self._is_name_rule_for(bare_rule, qualified, tool_type):
-            return qualified
-        return bare if bare_rule.decision == "deny" else qualified
+        member: Final = self._narrower_name(qualified, bare, tool_type, tool_call)
+        namespace: Final = _pair_namespace(qualified, bare)
+        return member if namespace is None else self._narrower_name(member, namespace, tool_type, tool_call)
 
     def _check_target_permission(
         self, target: _ToolTarget, bare_names: Mapping[str, tuple[str, ...]]

tests/unit/proxy/guardrails/guardrail_hooks/test_tool_permission.py

@@ -565,6 +565,78 @@ class TestToolPermissionGuardrail:
         assert [call.function.name for call in message.tool_calls or ()] == []
         assert "Tool 'exec'" in (message.content or "")
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        ("rules", "command", "kept"),
+        [
+            (
+                [
+                    {
+                        "id": "allow_ls",
+                        "tool_name": r"^exec$",
+                        "decision": "allow",
+                        "allowed_param_patterns": {"command": r"^ls$"},
+                    },
+                    {"id": "deny_exec", "tool_name": r"^exec$", "decision": "deny"},
+                ],
+                "ls",
+                ["run"],
+            ),
+            (
+                [
+                    {
+                        "id": "allow_ls",
+                        "tool_name": r"^exec$",
+                        "decision": "allow",
+                        "allowed_param_patterns": {"command": r"^ls$"},
+                    },
+                    {"id": "deny_exec", "tool_name": r"^exec$", "decision": "deny"},
+                ],
+                "pwd",
+                [],
+            ),
+            (
+                [
+                    {"id": "allow_run", "tool_name": r"^run$", "decision": "allow"},
+                    {"id": "deny_exec", "tool_name": r"^exec$", "decision": "deny"},
+                ],
+                "ls",
+                [],
+            ),
+        ],
+        ids=["namespace_rule_allows", "namespace_rule_denies", "namespace_deny_beats_member_allow"],
+    )
+    async def test_post_call_judges_a_returned_namespace_member_by_a_rule_on_its_namespace_name(
+        self, rules: list[dict[str, object]], command: str, kept: list[str]
+    ) -> None:
+        guardrail = ToolPermissionGuardrail(
+            guardrail_name="post-exec",
+            rules=rules,
+            default_action="allow",
+            on_disallowed_action="rewrite",
+            event_hook=GuardrailEventHooks.post_call,
+        )
+        namespace = {"type": "namespace", "name": "exec", "tools": [{"type": "function", "name": "run"}]}
+        function = Function(name="run", arguments=json.dumps({"command": command}))
+        response = ModelResponse(
+            choices=[
+                Choices(
+                    finish_reason="tool_calls",
+                    message={"tool_calls": [{"id": "call_run", "type": "function", "function": function}]},
+                )
+            ]
+        )
+
+        await guardrail.async_post_call_success_hook(
+            data={"guardrails": ["post-exec"], "tools": [namespace]},
+            user_api_key_dict=UserAPIKeyAuth(),
+            response=response,
+        )
+
+        message = response.choices[0].message
+        assert [call.function.name for call in message.tool_calls or ()] == kept
+        assert ("Tool 'exec'" in (message.content or "")) == (not kept)
+
     @pytest.mark.asyncio
     @pytest.mark.parametrize("namespace_first", [True, False], ids=["namespace_first", "top_level_first"])
     async def test_post_call_judges_a_returned_name_a_top_level_function_carries_as_that_function(
@@ -2286,6 +2358,56 @@ class TestToolPermissionGuardrailAnthropicMessages:
         ]
         assert rewritten["tool_choice"] == "required"
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        ("rules", "default_action", "kept"),
+        [
+            ([{"id": "allow_exec", "tool_name": r"^exec$", "decision": "allow"}], "deny", ["run"]),
+            ([{"id": "deny_exec", "tool_name": r"^exec$", "decision": "deny"}], "allow", []),
+            (
+                [
+                    {"id": "allow_run", "tool_name": r"^run$", "decision": "allow"},
+                    {"id": "deny_exec", "tool_name": r"^exec$", "decision": "deny"},
+                ],
+                "allow",
+                [],
+            ),
+            (
+                [
+                    {"id": "deny_exec", "tool_name": r"^exec$", "decision": "deny"},
+                    {"id": "allow_all", "tool_name": r"^.*$", "decision": "allow"},
+                ],
+                "allow",
+                [],
+            ),
+            (
+                [
+                    {"id": "allow_exec", "tool_name": r"^exec$", "decision": "allow"},
+                    {"id": "deny_all", "tool_name": r"^.*$", "decision": "deny"},
+                ],
+                "deny",
+                ["run"],
+            ),
+        ],
+        ids=[
+            "namespace_allowed",
+            "namespace_denied",
+            "namespace_deny_beats_member_allow",
+            "namespace_deny_beats_a_catch_all_allow",
+            "namespace_allow_beats_a_catch_all_deny",
+        ],
+    )
+    async def test_pre_call_judges_a_namespace_member_by_a_rule_on_its_namespace_name(
+        self, rules: list[dict[str, str]], default_action: Literal["allow", "deny"], kept: list[str]
+    ) -> None:
+        namespace = {"type": "namespace", "name": "exec", "tools": [{"type": "function", "name": "run"}]}
+        request = {"model": "gpt-5", "input": "Run pwd", "tools": [namespace]}
+
+        rewritten = await self._run_pre_call(self._pre_call(rules, "rewrite", default_action), request, "responses")
+
+        tools = rewritten.get("tools") or []
+        assert [member["name"] for member in chain.from_iterable(tool["tools"] for tool in tools)] == kept
+
     @pytest.mark.asyncio
     @pytest.mark.parametrize(
         ("call_type", "bash", "read"),