fix(guardrails): bound strip_patterns input and warn when shaping outlives enforcement

Two review findings on the payload-shaping options.

Python's re has no match timeout, and the substitution cap bounds how many matches
are replaced rather than how long the engine spends finding one, so a configured
pattern that backtracks catastrophically would burn the worker for as long as the
caller's text allows. Text past 100k characters is therefore never matched: the
block is replaced with a placeholder, which is the fail-closed choice, since
sending it unstripped would leak exactly what strip_patterns exists to remove. The
placeholder is marked lossy like any other shaped text, so it cannot be written
back either.

Whatever the shaping leaves out is never scanned, so a guardrail that can still
block is only enforcing on the subset it receives. That is the intended trade for
an observer and a surprise for an enforcer, so it is now stated at init whenever a
lossy option is set on a guardrail that has not also been put in fire_and_forget
mode.

litellm/proxy/guardrails/guardrail_hooks/generic_guardrail_api/generic_guardrail_api.py

@@ -375,6 +375,17 @@ class GenericGuardrailAPI(CustomGuardrail):
             )
             self.streaming_end_of_stream_only = True
 
+        if self._payload_policy.is_lossy and not self.fire_and_forget:
+            # Whatever the shaping leaves out is never scanned, so an enforcing
+            # guardrail cannot act on it. Fine for an observer, worth saying out
+            # loud for a guardrail that can still block.
+            verbose_proxy_logger.warning(
+                "Generic Guardrail API (%s): max_messages / max_text_chars / strip_patterns / excluded text "
+                "keep part of the request from ever reaching the guardrail, so it can only enforce on what it "
+                "is sent. Content outside the configured window cannot be blocked or masked.",
+                configured_name,
+            )
+
         if self._skip_policy.filters_requests:
             verbose_proxy_logger.warning(
                 "Generic Guardrail API (%s): skip_if_system_prompt_matches / skip_if_first_role_in match on the "

litellm/proxy/guardrails/guardrail_hooks/generic_guardrail_api/payload_policy.py

@@ -36,6 +36,17 @@ PROTECTED_PAYLOAD_FIELDS: Final = frozenset({"input_type", "litellm_call_id"})
 # huge transcript cannot stall the request path.
 MAX_STRIP_SUBSTITUTIONS: Final = 64
 
+# Python's re has no match timeout, and the substitution cap above bounds how many
+# matches are replaced, not how long the engine spends finding one. A configured
+# pattern that backtracks catastrophically would therefore burn the worker for as
+# long as the caller's text lets it, so text past this size is not matched at all.
+MAX_STRIP_INPUT_CHARS: Final = 100_000
+
+# What a too-large block is replaced with. Dropping it is the fail-closed choice:
+# sending it unstripped would leak exactly the content strip_patterns exists to
+# remove, and it cannot be written back either, since it is marked lossy.
+OVERSIZED_TEXT_PLACEHOLDER: Final = "[omitted: exceeds strip size limit]"
+
 # Stands in for image data when send_images=False, mirroring the "[present]"
 # convention used for headers whose value is not forwarded.
 IMAGE_OMITTED_PLACEHOLDER: Final = "[omitted]"
@@ -57,6 +68,11 @@ class PayloadPolicy:
     def shapes_text(self) -> bool:
         return self.max_text_chars is not None or bool(self.strip_patterns)
 
+    @property
+    def is_lossy(self) -> bool:
+        """Whether any option keeps part of the request from reaching the guardrail."""
+        return self.shapes_text or self.max_messages is not None or not self.send_images or bool(self.exclude_fields)
+
 
 @dataclass(frozen=True, slots=True)
 class PayloadLoss:
@@ -118,11 +134,19 @@ def _strip_text(text: str, patterns: tuple[re.Pattern[str], ...]) -> str:
 
 
 def _shape_text(text: str, policy: PayloadPolicy) -> str:
+    if policy.strip_patterns and len(text) > MAX_STRIP_INPUT_CHARS:
+        verbose_proxy_logger.warning(
+            "Generic Guardrail API: a %d character text block exceeds the %d character strip_patterns "
+            "limit and was replaced with a placeholder instead of being matched.",
+            len(text),
+            MAX_STRIP_INPUT_CHARS,
+        )
+        return OVERSIZED_TEXT_PLACEHOLDER
     stripped: Final = _strip_text(text, policy.strip_patterns)
     return stripped[: policy.max_text_chars] if policy.max_text_chars is not None else stripped
 
 
-def _retained_messages(messages: JsonValue, policy: PayloadPolicy) -> list[JsonValue] | None:
+def _retained_messages(messages: JsonValue, policy: PayloadPolicy) -> JsonValue:
     """The message window ``max_messages`` keeps, or None when there is nothing to window."""
     if not isinstance(messages, list) or not messages:
         return None
@@ -131,7 +155,7 @@ def _retained_messages(messages: JsonValue, policy: PayloadPolicy) -> list[JsonV
     return messages[-policy.max_messages :]
 
 
-def _text_window_size(retained: list[JsonValue] | None, policy: PayloadPolicy, total_texts: int) -> int:
+def _text_window_size(retained: JsonValue, policy: PayloadPolicy, total_texts: int) -> int:
     """How many trailing ``texts`` entries belong to the retained messages.
 
     ``texts`` is fragment-based (a multimodal turn contributes several entries, a
@@ -144,7 +168,7 @@ def _text_window_size(retained: list[JsonValue] | None, policy: PayloadPolicy, t
     """
     if policy.max_messages is None:
         return total_texts
-    if retained is None:
+    if not isinstance(retained, list):
         return min(policy.max_messages, total_texts)
     fragments: Final = sum(1 for _ in iter_messages_text(retained))
     return min(fragments, total_texts)
@@ -171,8 +195,8 @@ def _shape_texts(
     return shaped, frozenset(index for index, text in enumerate(texts) if text != shaped[index])
 
 
-def _shape_messages(retained: list[JsonValue] | None, policy: PayloadPolicy) -> JsonValue:
-    if retained is None:
+def _shape_messages(retained: JsonValue, policy: PayloadPolicy) -> JsonValue:
+    if not isinstance(retained, list):
         return None
     texted: Final = (
         map_messages_text(retained, lambda text: _shape_text(text, policy)) if policy.shapes_text else retained

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

@@ -3303,3 +3303,65 @@ class TestSessionScopeIsolation:
             await guardrail.apply_guardrail(inputs={"texts": ["a"]}, request_data=data, input_type="request")
 
         assert len(_recorded_entries(data)) == 1
+
+
+class TestLossyShapingSafeguards:
+    """Bounds and warnings for what payload shaping keeps from the guardrail."""
+
+    @pytest.mark.asyncio
+    async def test_oversized_text_is_dropped_rather_than_matched(self):
+        """re has no match timeout, so a huge block is replaced instead of scanned."""
+        handler = _RecordingHandler()
+        guardrail = _make_guardrail(handler, strip_patterns=[r"SECRET-\d+"])
+        huge = "SECRET-1 " + ("a" * 200_000)
+
+        await guardrail.apply_guardrail(
+            inputs={"texts": [huge, "SECRET-2 small"]},
+            request_data={},
+            input_type="request",
+        )
+
+        sent = handler.payloads[0]["texts"]
+        assert sent[0] == "[omitted: exceeds strip size limit]"
+        # A normal-sized block is still stripped as usual.
+        assert sent[1] == " small"
+
+    @pytest.mark.asyncio
+    async def test_oversized_placeholder_is_not_written_back(self):
+        handler = _RecordingHandler(action="GUARDRAIL_INTERVENED", texts=["MASKED"])
+        guardrail = _make_guardrail(handler, strip_patterns=[r"SECRET-\d+"])
+        huge = "SECRET-1 " + ("a" * 200_000)
+
+        result = await guardrail.apply_guardrail(
+            inputs={"texts": [huge]},
+            request_data={},
+            input_type="request",
+        )
+
+        assert result["texts"] == [huge]
+
+    @pytest.mark.asyncio
+    async def test_size_limit_only_applies_with_strip_patterns(self):
+        """Without patterns there is nothing to match, so a big block is sent as-is."""
+        handler = _RecordingHandler()
+        guardrail = _make_guardrail(handler)
+        huge = "a" * 200_000
+
+        await guardrail.apply_guardrail(inputs={"texts": [huge]}, request_data={}, input_type="request")
+
+        assert handler.payloads[0]["texts"] == [huge]
+
+    def test_enforcing_guardrail_warns_that_shaped_content_cannot_be_blocked(self, caplog):
+        with caplog.at_level("WARNING", logger="LiteLLM Proxy"):
+            _make_guardrail(_RecordingHandler(), max_messages=4)
+        assert "cannot be blocked or masked" in caplog.text
+
+    def test_observer_does_not_get_the_enforcement_warning(self, caplog):
+        with caplog.at_level("WARNING", logger="LiteLLM Proxy"):
+            _make_guardrail(_RecordingHandler(), max_messages=4, fire_and_forget=True)
+        assert "cannot be blocked or masked" not in caplog.text
+
+    def test_unshaped_guardrail_does_not_warn(self, caplog):
+        with caplog.at_level("WARNING", logger="LiteLLM Proxy"):
+            _make_guardrail(_RecordingHandler())
+        assert "cannot be blocked or masked" not in caplog.text