diff --git a/litellm/proxy/guardrails/guardrail_hooks/tool_permission.py b/litellm/proxy/guardrails/guardrail_hooks/tool_permission.py index 4d3fe39b91d..5f99b1a4611 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/tool_permission.py +++ b/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, ...]] diff --git a/tests/unit/proxy/guardrails/guardrail_hooks/test_tool_permission.py b/tests/unit/proxy/guardrails/guardrail_hooks/test_tool_permission.py index 7d84c1ddb32..46e766f753c 100644 --- a/tests/unit/proxy/guardrails/guardrail_hooks/test_tool_permission.py +++ b/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"),