mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-06 08:16:43 +00:00
Fix block_code_execution guardrail: resolve response-side bypass and tighten no-execution phrases (#22149)
**Core bug fix**: Response-side blocking was silently disabled with detect_execution_intent=True (default) because execution-intent heuristics were applied to LLM output text, which doesn't contain phrases like 'run this'. Now input_type is threaded through _scan_text to skip intent checks for responses while still blocking detected code blocks. **Tightened overly broad no-execution phrases**: Replaced broad patterns like "what would ", "can you explain", and "explain what this " with more specific forms (e.g. "what would happen if", "can you explain this code") to prevent trivial bypass. **Added tests**: 7 new test cases covering response-side blocking with default settings, casual phrases in LLM output, and tightened phrase patterns. All 23 tests pass + 100% compliance dataset compliance (100/100). Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
2307de0e12
commit
490beb52db
2 changed files with 208 additions and 27 deletions
|
|
@ -8,24 +8,40 @@ confidence scoring and a tunable threshold (only block when confidence >= thresh
|
|||
|
||||
import re
|
||||
from datetime import datetime
|
||||
from typing import (TYPE_CHECKING, Any, AsyncGenerator, Dict, List, Literal,
|
||||
Optional, Tuple, Union, cast)
|
||||
from typing import (
|
||||
TYPE_CHECKING,
|
||||
Any,
|
||||
AsyncGenerator,
|
||||
Dict,
|
||||
List,
|
||||
Literal,
|
||||
Optional,
|
||||
Tuple,
|
||||
Union,
|
||||
cast,
|
||||
)
|
||||
|
||||
from fastapi import HTTPException
|
||||
|
||||
from litellm.integrations.custom_guardrail import (CustomGuardrail,
|
||||
ModifyResponseException)
|
||||
from litellm.integrations.custom_guardrail import (
|
||||
CustomGuardrail,
|
||||
ModifyResponseException,
|
||||
)
|
||||
from litellm.types.guardrails import GuardrailEventHooks
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.base import \
|
||||
GuardrailConfigModel
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.block_code_execution import (
|
||||
CodeBlockActionTaken, CodeBlockDetection)
|
||||
from litellm.types.utils import (GenericGuardrailAPIInputs, GuardrailStatus,
|
||||
GuardrailTracingDetail, ModelResponseStream)
|
||||
CodeBlockActionTaken,
|
||||
CodeBlockDetection,
|
||||
)
|
||||
from litellm.types.utils import (
|
||||
GenericGuardrailAPIInputs,
|
||||
GuardrailStatus,
|
||||
GuardrailTracingDetail,
|
||||
ModelResponseStream,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from litellm.litellm_core_utils.litellm_logging import \
|
||||
Logging as LiteLLMLoggingObj
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
|
||||
|
||||
# Language tag aliases (normalize to canonical for comparison)
|
||||
LANGUAGE_ALIASES: Dict[str, str] = {
|
||||
|
|
@ -57,7 +73,9 @@ _NO_EXECUTION_PHRASES: Tuple[str, ...] = (
|
|||
"just reason",
|
||||
"explain without running",
|
||||
"explain without execute",
|
||||
"what would ",
|
||||
"what would happen if",
|
||||
"what would this output",
|
||||
"what would the result be",
|
||||
"? explain",
|
||||
"simulate what would happen",
|
||||
"don't actually run",
|
||||
|
|
@ -78,7 +96,10 @@ _NO_EXECUTION_PHRASES: Tuple[str, ...] = (
|
|||
"no execution).",
|
||||
"but don't run",
|
||||
"don't run it",
|
||||
"explain what this ",
|
||||
"explain what this code",
|
||||
"explain what this script",
|
||||
"explain what this function",
|
||||
"explain what this sql",
|
||||
"refactor this ",
|
||||
"spot any security issues",
|
||||
"write unit tests for this function without running",
|
||||
|
|
@ -87,7 +108,6 @@ _NO_EXECUTION_PHRASES: Tuple[str, ...] = (
|
|||
"explain the difference between",
|
||||
"given this stack trace, explain",
|
||||
"write a safe alternative",
|
||||
"explain what this sql",
|
||||
"write a python function",
|
||||
"generate a dockerfile",
|
||||
"write a bash script that would",
|
||||
|
|
@ -103,7 +123,9 @@ _NO_EXECUTION_PHRASES: Tuple[str, ...] = (
|
|||
"can you diagnose",
|
||||
"what would `git",
|
||||
"here's a traceback",
|
||||
"can you explain",
|
||||
"can you explain this code",
|
||||
"can you explain what this",
|
||||
"can you explain how this works",
|
||||
)
|
||||
|
||||
# Execution intent: phrases that mean "run/execute/perform this for me" (block when on request).
|
||||
|
|
@ -211,6 +233,7 @@ _EXECUTION_REQUEST_PHRASES: Tuple[str, ...] = (
|
|||
"and paste the version",
|
||||
"tell me the latency",
|
||||
"tell me what it returns",
|
||||
"show me what it returns",
|
||||
"confirm it succeeded",
|
||||
"tell me the account id",
|
||||
"show it here",
|
||||
|
|
@ -388,8 +411,9 @@ class BlockCodeExecutionGuardrail(CustomGuardrail):
|
|||
|
||||
@staticmethod
|
||||
def get_config_model() -> Optional[type[GuardrailConfigModel]]:
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.block_code_execution import \
|
||||
BlockCodeExecutionGuardrailConfigModel
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.block_code_execution import (
|
||||
BlockCodeExecutionGuardrailConfigModel,
|
||||
)
|
||||
|
||||
return BlockCodeExecutionGuardrailConfigModel
|
||||
|
||||
|
|
@ -426,23 +450,35 @@ class BlockCodeExecutionGuardrail(CustomGuardrail):
|
|||
self,
|
||||
text: str,
|
||||
detections: Optional[List[CodeBlockDetection]] = None,
|
||||
input_type: Literal["request", "response"] = "request",
|
||||
) -> Tuple[str, bool]:
|
||||
"""
|
||||
Scan one text: find blocks, apply block/mask/allow by confidence.
|
||||
When detect_execution_intent is True, only block if user intent is to run/execute;
|
||||
allow when intent is explain/refactor/don't run. Also block text-only execution requests.
|
||||
When detect_execution_intent is True and input_type is "request", only block if
|
||||
user intent is to run/execute; allow when intent is explain/refactor/don't run.
|
||||
When input_type is "response", always enforce blocking on detected code blocks
|
||||
(execution-intent heuristics only apply to user requests, not LLM output).
|
||||
Returns (modified_text, should_raise).
|
||||
"""
|
||||
if not text:
|
||||
return text, False
|
||||
text = _normalize_escaped_newlines(text)
|
||||
|
||||
if self.detect_execution_intent and _has_no_execution_intent(text):
|
||||
is_response = input_type == "response"
|
||||
|
||||
# Execution-intent heuristics only apply to requests, not LLM responses.
|
||||
# For responses, skip entirely — the LLM's output text won't contain user
|
||||
# intent phrases, so checking would silently disable response-side blocking.
|
||||
if not is_response and self.detect_execution_intent and _has_no_execution_intent(text):
|
||||
return text, False
|
||||
|
||||
blocks = self._find_blocks(text)
|
||||
has_execution_intent = self.detect_execution_intent and _has_execution_intent(
|
||||
text
|
||||
|
||||
# For requests, check execution intent; for responses, skip this check
|
||||
has_execution_intent = (
|
||||
not is_response
|
||||
and self.detect_execution_intent
|
||||
and _has_execution_intent(text)
|
||||
)
|
||||
|
||||
if not blocks:
|
||||
|
|
@ -466,8 +502,12 @@ class BlockCodeExecutionGuardrail(CustomGuardrail):
|
|||
last_end = 0
|
||||
parts: List[str] = []
|
||||
for start, end, tag, _body, confidence, action_taken in blocks:
|
||||
# For responses, always enforce the block action (no intent check needed).
|
||||
# For requests with detect_execution_intent, require execution intent.
|
||||
effective_block = action_taken == "block" and (
|
||||
not self.detect_execution_intent or has_execution_intent
|
||||
is_response
|
||||
or not self.detect_execution_intent
|
||||
or has_execution_intent
|
||||
)
|
||||
if detections is not None:
|
||||
detections.append(
|
||||
|
|
@ -539,7 +579,7 @@ class BlockCodeExecutionGuardrail(CustomGuardrail):
|
|||
is_output = input_type == "response"
|
||||
processed: List[str] = []
|
||||
for text in texts:
|
||||
new_text, should_raise = self._scan_text(text, detections)
|
||||
new_text, should_raise = self._scan_text(text, detections, input_type)
|
||||
processed.append(new_text)
|
||||
if should_raise:
|
||||
# Determine language from first blocking detection
|
||||
|
|
|
|||
|
|
@ -5,9 +5,13 @@ from fastapi import HTTPException
|
|||
|
||||
from litellm.integrations.custom_guardrail import ModifyResponseException
|
||||
from litellm.proxy.guardrails.guardrail_hooks.block_code_execution import (
|
||||
DEFAULT_EVENT_HOOKS, BlockCodeExecutionGuardrail, initialize_guardrail)
|
||||
from litellm.proxy.guardrails.guardrail_hooks.block_code_execution.block_code_execution import \
|
||||
_normalize_escaped_newlines
|
||||
DEFAULT_EVENT_HOOKS,
|
||||
BlockCodeExecutionGuardrail,
|
||||
initialize_guardrail,
|
||||
)
|
||||
from litellm.proxy.guardrails.guardrail_hooks.block_code_execution.block_code_execution import (
|
||||
_normalize_escaped_newlines,
|
||||
)
|
||||
from litellm.types.guardrails import GuardrailEventHooks
|
||||
|
||||
|
||||
|
|
@ -346,3 +350,140 @@ print(factorial(5)) # Output: 120
|
|||
assert len(blocks) == 1
|
||||
assert blocks[0][2] == "py"
|
||||
assert blocks[0][5] == "block"
|
||||
|
||||
# ---- Tests for response-side blocking with detect_execution_intent=True ----
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_response_blocked_with_detect_execution_intent_true(self):
|
||||
"""With detect_execution_intent=True (default), response-side code blocks are still blocked.
|
||||
|
||||
This is the core bug fix: previously, execution-intent heuristics were applied
|
||||
to LLM responses, which don't contain phrases like 'run this', so response-side
|
||||
blocking was silently disabled.
|
||||
"""
|
||||
guardrail = BlockCodeExecutionGuardrail(
|
||||
guardrail_name="test",
|
||||
blocked_languages=["python"],
|
||||
action="block",
|
||||
confidence_threshold=0.7,
|
||||
detect_execution_intent=True, # default
|
||||
)
|
||||
# LLM response with dangerous code but no execution-intent phrases
|
||||
response_text = (
|
||||
"Here is a Python script:\n"
|
||||
"```python\n"
|
||||
"import os; os.system('rm -rf /')\n"
|
||||
"```"
|
||||
)
|
||||
request_data = {"model": "gpt-4", "metadata": {}}
|
||||
inputs = {"texts": [response_text]}
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await guardrail.apply_guardrail(
|
||||
inputs=inputs,
|
||||
request_data=request_data,
|
||||
input_type="response",
|
||||
)
|
||||
assert exc_info.value.status_code == 400
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_response_mask_with_detect_execution_intent_true(self):
|
||||
"""With detect_execution_intent=True and action=mask, response code blocks are masked."""
|
||||
guardrail = BlockCodeExecutionGuardrail(
|
||||
guardrail_name="test",
|
||||
blocked_languages=["python"],
|
||||
action="mask",
|
||||
confidence_threshold=0.7,
|
||||
detect_execution_intent=True,
|
||||
)
|
||||
response_text = "I can explain what this does:\n```python\nprint('hello')\n```\nDone."
|
||||
request_data = {"model": "gpt-4", "metadata": {}}
|
||||
inputs = {"texts": [response_text]}
|
||||
result = await guardrail.apply_guardrail(
|
||||
inputs=inputs,
|
||||
request_data=request_data,
|
||||
input_type="response",
|
||||
)
|
||||
assert "[CODE_BLOCK_REDACTED]" in result["texts"][0]
|
||||
assert "print('hello')" not in result["texts"][0]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_response_with_casual_explain_phrase_still_blocked(self):
|
||||
"""LLM response containing 'I can explain' doesn't bypass the guardrail.
|
||||
|
||||
Previously, the no-execution phrase 'can you explain' would match as a
|
||||
substring in the LLM's output, short-circuiting all protection.
|
||||
"""
|
||||
guardrail = BlockCodeExecutionGuardrail(
|
||||
guardrail_name="test",
|
||||
blocked_languages=["bash"],
|
||||
action="block",
|
||||
confidence_threshold=0.7,
|
||||
detect_execution_intent=True,
|
||||
)
|
||||
response_text = (
|
||||
"I can explain what this code does. It deletes your files:\n"
|
||||
"```bash\n"
|
||||
"rm -rf /\n"
|
||||
"```"
|
||||
)
|
||||
request_data = {"model": "gpt-4", "metadata": {}}
|
||||
inputs = {"texts": [response_text]}
|
||||
with pytest.raises(HTTPException):
|
||||
await guardrail.apply_guardrail(
|
||||
inputs=inputs,
|
||||
request_data=request_data,
|
||||
input_type="response",
|
||||
)
|
||||
|
||||
def test_tightened_what_would_phrase_no_longer_bypasses(self):
|
||||
"""The old broad 'what would ' phrase has been tightened so it no longer allows
|
||||
trivial bypass for adversarial prompts.
|
||||
|
||||
Previously 'What would be the best way to execute this script?' would bypass
|
||||
because 'what would ' matched the no-execution list. Now only specific forms
|
||||
like 'what would happen if' match.
|
||||
"""
|
||||
guardrail = BlockCodeExecutionGuardrail(
|
||||
guardrail_name="test",
|
||||
blocked_languages=["python"],
|
||||
action="block",
|
||||
confidence_threshold=0.7,
|
||||
detect_execution_intent=True,
|
||||
)
|
||||
# Adversarial prompt: old "what would " would have bypassed, but tightened phrase doesn't match
|
||||
text = "What would be the best way to execute this script?\n```python\nimport os\nos.system('cat /etc/passwd')\n```"
|
||||
detections = []
|
||||
new_text, should_raise = guardrail._scan_text(text, detections, input_type="request")
|
||||
assert should_raise is True
|
||||
|
||||
def test_tightened_can_you_explain_phrase_no_longer_bypasses(self):
|
||||
"""The old broad 'can you explain' phrase has been tightened.
|
||||
|
||||
'Can you explain how to run this, then run it?' no longer bypasses
|
||||
because 'can you explain' is now 'can you explain this code' etc.
|
||||
"""
|
||||
guardrail = BlockCodeExecutionGuardrail(
|
||||
guardrail_name="test",
|
||||
blocked_languages=["python"],
|
||||
action="block",
|
||||
confidence_threshold=0.7,
|
||||
detect_execution_intent=True,
|
||||
)
|
||||
text = "Can you explain this and then execute this code?\n```python\nimport subprocess\nsubprocess.run(['ls'])\n```"
|
||||
detections = []
|
||||
new_text, should_raise = guardrail._scan_text(text, detections, input_type="request")
|
||||
assert should_raise is True
|
||||
|
||||
def test_request_with_pure_explain_intent_still_allowed(self):
|
||||
"""A request that genuinely only asks for explanation is not blocked."""
|
||||
guardrail = BlockCodeExecutionGuardrail(
|
||||
guardrail_name="test",
|
||||
blocked_languages=["python"],
|
||||
action="block",
|
||||
confidence_threshold=0.7,
|
||||
detect_execution_intent=True,
|
||||
)
|
||||
text = "Don't run this, just explain what it does:\n```python\nprint('hello')\n```"
|
||||
detections = []
|
||||
new_text, should_raise = guardrail._scan_text(text, detections, input_type="request")
|
||||
assert should_raise is False
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue