Fix block_code_execution: response-side bypass and phrase tightening

litellm/proxy/guardrails/guardrail_hooks/block_code_execution/block_code_execution.py

@@ -8,24 +8,40 @@
 
 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 @@
     "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).",
     "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 @@
     "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 @@
     "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 @@
     "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 @@ def __init__(
 
     @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 @@ def _scan_text(
         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 @@ def _scan_text(
         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 @@ async def apply_guardrail(
             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

tests/test_litellm/proxy/guardrails/guardrail_hooks/test_block_code_execution.py

@@ -5,9 +5,13 @@
 
 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 @@ def test_normalize_escaped_newlines_mixed_content_detects_block(self):
         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