Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py: 11%
993 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""
2Content Filter Guardrail for LiteLLM.
4This guardrail provides regex pattern matching and keyword filtering
5to detect and block/mask sensitive content.
6"""
8import asyncio
9import json
10import os
11import re
12import time
13from collections.abc import AsyncGenerator, Coroutine, Mapping, Sequence
14from dataclasses import dataclass, replace
15from datetime import datetime
16from re import Pattern
17from typing import TYPE_CHECKING, Any, Final, Literal, Optional, TypedDict, cast
19import yaml
20from fastapi import HTTPException
22from litellm import Router
23from litellm._logging import verbose_proxy_logger
24from litellm.constants import (
25 CONTENT_FILTER_STREAMING_HOLDBACK_CHARS,
26 CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS,
27 DEFAULT_MAX_RECURSE_DEPTH,
28)
29from litellm.integrations.custom_guardrail import CustomGuardrail
30from litellm.proxy._types import UserAPIKeyAuth
31from litellm.types.utils import (
32 CallTypes,
33 Function,
34 GenericGuardrailAPIInputs,
35 GuardrailStatus,
36 GuardrailTracingDetail,
37 ModelResponse,
38 ModelResponseStream,
39)
41if TYPE_CHECKING: 41 ↛ 42line 41 didn't jump to line 42 because the condition on line 41 was never true
42 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
44from litellm.types.guardrails import (
45 BlockedWord,
46 ContentFilterAction,
47 ContentFilterPattern,
48 GuardrailEventHooks,
49 Mode,
50)
51from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import (
52 BlockedWordDetection,
53 CategoryKeywordDetection,
54 CompetitorIntentDetection,
55 CompetitorIntentResult,
56 ContentFilterCategoryConfig,
57 ContentFilterDetection,
58 PatternDetection,
59)
61from .competitor_intent import (
62 AirlineCompetitorIntentChecker,
63 BaseCompetitorIntentChecker,
64)
65from .patterns import PATTERN_EXTRA_CONFIG, get_compiled_pattern
67MAX_KEYWORD_VALUE_GAP_WORDS: Final = 1
68GAP_WORD_TOKENIZER: Final = re.compile(r"\b\w+\b")
69SENTENCE_TERMINATORS: Final = re.compile(r"[.!?]+")
72WORD_NUMBER_MAP: Final = {
73 "zero": "0",
74 "oh": "0",
75 "one": "1",
76 "two": "2",
77 "three": "3",
78 "four": "4",
79 "five": "5",
80 "six": "6",
81 "seven": "7",
82 "eight": "8",
83 "nine": "9",
84}
86WORD_NUMBER_TOKEN_REGEX: Final = "|".join(WORD_NUMBER_MAP.keys())
87WORD_NUMBER_SEQUENCE_PATTERN: Final = re.compile(
88 rf"(?<![A-Za-z])(?:{WORD_NUMBER_TOKEN_REGEX})(?:[\s\-]+(?:{WORD_NUMBER_TOKEN_REGEX}))+(?![A-Za-z])",
89 re.IGNORECASE,
90)
91WORD_NUMBER_TOKEN_FINDER: Final = re.compile(rf"(?:{WORD_NUMBER_TOKEN_REGEX})", re.IGNORECASE)
94class ConditionalCategoryConfig(TypedDict):
95 identifier_words: Sequence[str]
96 block_words: Sequence[str]
97 action: ContentFilterAction
98 severity: str
101class CompiledPatternEntry(TypedDict):
102 regex: Pattern[str]
103 pattern_name: str
104 action: ContentFilterAction
105 keyword_regex: Pattern[str] | None
106 allow_word_numbers: bool
109class _PatternExtraLookup(TypedDict):
110 keyword_pattern: str | None
111 allow_word_numbers: bool
114class _CategoryConfigView(TypedDict):
115 category: object
116 enabled: object
117 action: object
118 category_file: str | None
121@dataclass(frozen=True, slots=True)
122class _StreamedChoiceState:
123 buffered_text: str = ""
124 yielded_masked_text_len: int = 0
125 committed_detections: tuple[ContentFilterDetection, ...] = ()
126 latest_detections: tuple[ContentFilterDetection, ...] = ()
127 next_trim_len: int = 0
130@dataclass(frozen=True, slots=True)
131class _StreamedScanPlan:
132 context_chars: int
133 exception_phrases: tuple[str, ...]
134 conditional_words: tuple[str, ...]
137class CategoryFileData(TypedDict, total=False):
138 category_name: str
139 description: str
140 default_action: str
141 keywords: Sequence[Mapping[str, str]]
142 exceptions: Sequence[str]
143 identifier_words: Sequence[str]
144 always_block_keywords: Sequence[Mapping[str, str]]
145 inherit_from: str
146 additional_block_words: Sequence[str]
147 phrase_patterns: Sequence[str]
150# Helper data structure for category-based detection
151class CategoryConfig:
152 """Configuration for a content category."""
154 def __init__(
155 self,
156 category_name: str,
157 description: str,
158 default_action: ContentFilterAction,
159 keywords: Sequence[Mapping[str, str]],
160 exceptions: Sequence[str],
161 identifier_words: Sequence[str] | None = None,
162 always_block_keywords: Sequence[Mapping[str, str]] | None = None,
163 inherit_from: str | None = None,
164 additional_block_words: Sequence[str] | None = None,
165 phrase_patterns: Sequence[str] | None = None,
166 ):
167 self.category_name = category_name
168 self.description = description
169 self.default_action = default_action
170 self.keywords = keywords
171 self.exceptions = [e.lower() for e in exceptions]
172 # New fields for conditional child safety logic
173 self.identifier_words = [w.lower() for w in identifier_words] if identifier_words else []
174 self.always_block_keywords = always_block_keywords or []
175 self.inherit_from = inherit_from
176 self.additional_block_words = [w.lower() for w in additional_block_words] if additional_block_words else []
177 # Phrase patterns: regex patterns for catching paraphrases
178 self.phrase_patterns: list[tuple[str, Pattern]] = []
179 for p in phrase_patterns or []:
180 try:
181 self.phrase_patterns.append((p, re.compile(p, re.IGNORECASE)))
182 except re.error:
183 verbose_proxy_logger.warning("Invalid phrase pattern in %s: %s", category_name, p)
186class ContentFilterGuardrail(CustomGuardrail):
187 """
188 Content filter guardrail that detects sensitive information using:
189 - Prebuilt regex patterns (SSN, credit cards, API keys, etc.)
190 - Custom user-defined regex patterns
191 - Dictionary-based keyword matching
193 Actions:
194 - BLOCK: Reject the request with an error
195 - MASK: Replace the sensitive content with a redacted placeholder
196 """
198 # Redaction format constants
199 PATTERN_REDACTION_FORMAT = "[{pattern_name}_REDACTED]"
200 KEYWORD_REDACTION_STR = "[KEYWORD_REDACTED]"
202 def __init__(
203 self,
204 guardrail_name: str | None = None,
205 guardrail_id: str | None = None,
206 policy_template: str | None = None,
207 patterns: list[ContentFilterPattern] | None = None,
208 blocked_words: list[BlockedWord] | None = None,
209 blocked_words_file: str | None = None,
210 event_hook: GuardrailEventHooks | list[GuardrailEventHooks] | Mode | None = None,
211 default_on: bool = False,
212 pattern_redaction_format: str | None = None,
213 keyword_redaction_tag: str | None = None,
214 categories: list[ContentFilterCategoryConfig] | None = None,
215 severity_threshold: str = "medium",
216 llm_router: Router | None = None,
217 image_model: str | None = None,
218 competitor_intent_config: dict[str, object] | None = None,
219 **kwargs,
220 ):
221 """
222 Initialize the Content Filter Guardrail.
224 Args:
225 guardrail_name: Name of this guardrail instance
226 patterns: List of ContentFilterPattern objects to detect
227 blocked_words: List of BlockedWord objects with keywords and actions
228 blocked_words_file: Path to YAML file containing blocked_words list
229 event_hook: When to run this guardrail (pre_call, post_call, etc.)
230 default_on: If True, runs on all requests by default
231 pattern_redaction_format: Format string for pattern redaction (use {pattern_name} placeholder)
232 keyword_redaction_tag: Tag to use for keyword redaction
233 categories: List of category configurations with enabled/action/severity settings
234 severity_threshold: Minimum severity to block ("high", "medium", "low")
235 """
237 super().__init__(
238 guardrail_name=guardrail_name,
239 supported_event_hooks=list(self.get_supported_event_hooks()),
240 event_hook=event_hook or GuardrailEventHooks.pre_call,
241 default_on=default_on,
242 **kwargs,
243 )
245 self.guardrail_provider = "litellm_content_filter"
246 self.config_guardrail_id = guardrail_id
247 self.config_policy_template = policy_template
248 self.pattern_redaction_format = pattern_redaction_format or self.PATTERN_REDACTION_FORMAT
249 self.keyword_redaction_tag = keyword_redaction_tag or self.KEYWORD_REDACTION_STR
250 self.severity_threshold = severity_threshold
251 self.llm_router = llm_router
252 self.image_model = image_model
253 # Store loaded categories
254 self.loaded_categories: dict[str, CategoryConfig] = {}
255 self.category_keywords: dict[
256 str, tuple[str, str, ContentFilterAction]
257 ] = {} # keyword -> (category, severity, action)
258 # Always-block keywords are checked after exceptions (exceptions take precedence)
259 self.always_block_category_keywords: dict[str, tuple[str, str, ContentFilterAction]] = {}
260 # Store conditional categories (identifier_words + block_words)
261 self.conditional_categories: dict[str, ConditionalCategoryConfig] = {}
263 # Competitor intent checker (optional; airline uses major_airlines.json, generic requires competitors)
264 self._competitor_intent_checker: BaseCompetitorIntentChecker | None = None
265 if competitor_intent_config and isinstance(competitor_intent_config, dict):
266 self._init_competitor_intent_checker(competitor_intent_config)
268 # Load categories if provided
269 if categories:
270 self._load_categories(categories)
272 # Normalize inputs: convert dicts to Pydantic models for consistent handling
273 normalized_patterns: Final = self._normalize_patterns(patterns)
274 normalized_blocked_words: Final = self._normalize_blocked_words(blocked_words)
276 # Compile regex patterns
277 self.compiled_patterns: list[CompiledPatternEntry] = []
278 for pattern_config in normalized_patterns:
279 self._add_pattern(pattern_config)
281 # Warn if using during_call with MASK action (unstable)
282 if self.event_hook == GuardrailEventHooks.during_call and any(
283 p["action"] == ContentFilterAction.MASK for p in self.compiled_patterns
284 ):
285 verbose_proxy_logger.warning(
286 "ContentFilterGuardrail '%s': 'during_call' mode with 'MASK' action is unstable due to race conditions. Use 'pre_call' mode for reliable request masking.",
287 self.guardrail_name,
288 )
290 # Load blocked words - always initialize as dict
291 self.blocked_words: dict[str, tuple[ContentFilterAction, str | None]] = {}
292 for word in normalized_blocked_words:
293 self.blocked_words[word.keyword.lower()] = (word.action, word.description)
295 # Defensive check: ensure blocked_words is a dict (not a list)
296 if not isinstance(self.blocked_words, dict):
297 verbose_proxy_logger.error(
298 "blocked_words is not a dict, got %s. Resetting to empty dict.", type(self.blocked_words)
299 )
300 self.blocked_words = {}
302 # Load blocked words from file if provided
303 if blocked_words_file:
304 self._load_blocked_words_file(blocked_words_file)
306 verbose_proxy_logger.debug(
307 "ContentFilterGuardrail initialized with %s patterns and %s blocked words",
308 len(self.compiled_patterns),
309 len(self.blocked_words),
310 )
311 verbose_proxy_logger.debug(
312 "Loaded %s categories with %s keywords", len(self.loaded_categories), len(self.category_keywords)
313 )
315 def _init_competitor_intent_checker(self, competitor_intent_config: dict[str, object]) -> None:
316 try:
317 competitor_intent_type: Final = competitor_intent_config.get("competitor_intent_type", "airline")
318 if competitor_intent_type == "generic":
319 self._competitor_intent_checker = BaseCompetitorIntentChecker(competitor_intent_config)
320 else:
321 self._competitor_intent_checker = AirlineCompetitorIntentChecker(competitor_intent_config)
322 verbose_proxy_logger.debug(
323 "ContentFilterGuardrail: competitor intent checker enabled (%s)",
324 competitor_intent_type,
325 )
326 except Exception as e:
327 verbose_proxy_logger.warning(
328 "ContentFilterGuardrail: failed to init competitor intent checker: %s",
329 e,
330 )
332 @staticmethod
333 def _normalize_patterns(
334 patterns: list[ContentFilterPattern] | None,
335 ) -> list[ContentFilterPattern]:
336 result: Final[list[ContentFilterPattern]] = []
337 if patterns:
338 for pattern_config in patterns:
339 if isinstance(pattern_config, dict):
340 result.append(ContentFilterPattern(**pattern_config))
341 else:
342 result.append(pattern_config)
343 return result
345 @staticmethod
346 def _normalize_blocked_words(
347 blocked_words: list[BlockedWord] | None,
348 ) -> list[BlockedWord]:
349 result: Final[list[BlockedWord]] = []
350 if blocked_words:
351 for word in blocked_words:
352 if isinstance(word, dict):
353 result.append(BlockedWord(**word))
354 else:
355 result.append(word)
356 return result
358 @staticmethod
359 def _category_config_view(cat_config: ContentFilterCategoryConfig) -> _CategoryConfigView:
360 return {
361 "category": cat_config.get("category"),
362 "enabled": cat_config.get("enabled", True),
363 "action": cat_config.get("action"),
364 "category_file": cat_config.get("category_file"),
365 }
367 @staticmethod
368 def _assert_within_categories_dir(path: str, categories_dir: str) -> None:
369 """Raise ValueError if path escapes the categories directory."""
370 resolved: Final = os.path.realpath(path)
371 allowed: Final = os.path.realpath(categories_dir)
372 try:
373 common: Final = os.path.commonpath([resolved, allowed])
374 except ValueError:
375 # commonpath() raises ValueError on Windows when paths span different drives
376 raise ValueError(f"Category file path '{path}' is outside the allowed categories directory")
377 if common != allowed:
378 raise ValueError(
379 f"Category file path '{path}' is outside the allowed categories directory '{categories_dir}'"
380 )
382 def _resolve_category_file_path(self, file_path: str) -> str:
383 """
384 Resolve a category file path that may be relative.
386 Paths in policy templates (e.g. category_file) are often stored as
387 relative paths like "litellm/proxy/.../policy_templates/file.yaml".
388 These only work when the CWD is the project root. In production
389 (Docker, installed packages, etc.) the CWD is different, so the
390 file isn't found.
392 Resolution order:
393 1. Return as-is if absolute or already exists (jailed to module dir).
394 2. Try joining the full path relative to this module's directory (jailed).
395 3. Progressively strip leading path components and try each suffix
396 relative to this module's directory (jailed).
398 The directory jail can be disabled for deployments that legitimately
399 store category files outside the package (e.g. mounted volumes) by
400 setting the environment variable
401 ``LITELLM_CONTENT_FILTER_ALLOW_EXTERNAL_PATHS=true``. Use only in
402 trusted environments where the proxy configuration cannot be influenced
403 by untrusted input.
405 Args:
406 file_path: The file path to resolve (absolute or relative).
408 Returns:
409 The resolved absolute-ish path, or the original path if
410 resolution fails (caller should check existence).
412 Raises:
413 ValueError: If the resolved path escapes the module directory
414 and ``LITELLM_CONTENT_FILTER_ALLOW_EXTERNAL_PATHS`` is not set.
415 """
416 module_dir: Final = os.path.dirname(__file__)
417 allow_external: Final = os.environ.get("LITELLM_CONTENT_FILTER_ALLOW_EXTERNAL_PATHS", "").lower() == "true"
419 if os.path.isabs(file_path) or os.path.exists(file_path):
420 if not allow_external:
421 self._assert_within_categories_dir(file_path, module_dir)
422 else:
423 verbose_proxy_logger.warning(
424 "LITELLM_CONTENT_FILTER_ALLOW_EXTERNAL_PATHS is set — "
425 "skipping directory jail for category_file '%s'",
426 file_path,
427 )
428 return file_path
430 # Try the full relative path joined to the module directory
431 candidate = os.path.join(module_dir, file_path)
432 if os.path.exists(candidate):
433 if not allow_external:
434 self._assert_within_categories_dir(candidate, module_dir)
435 return candidate
437 # Progressively strip leading components to find a matching suffix
438 parts: Final = file_path.split("/")
439 for i in range(1, len(parts)):
440 suffix = os.path.join(*parts[i:])
441 candidate = os.path.join(module_dir, suffix)
442 if os.path.exists(candidate):
443 if not allow_external:
444 self._assert_within_categories_dir(candidate, module_dir)
445 return candidate
447 # File not found via any resolution strategy — jail the module-relative
448 # path anyway to reject traversal attempts (e.g. "../../../../etc/passwd")
449 # regardless of CWD or whether the target file exists.
450 if not allow_external:
451 self._assert_within_categories_dir(os.path.join(module_dir, file_path), module_dir)
452 return file_path
454 def _load_categories(self, categories: list[ContentFilterCategoryConfig]) -> None:
455 """
456 Load content categories from configuration.
458 Args:
459 categories: List of category configurations with format:
460 - category: "harmful_self_harm"
461 enabled: true
462 action: "BLOCK"
463 severity_threshold: "medium"
464 category_file: "/path/to/custom_file.yaml" # optional override
465 """
466 categories_dir: Final = os.path.join(os.path.dirname(__file__), "categories")
468 for cat_config in categories:
469 view = self._category_config_view(cat_config)
470 category_name = view["category"]
471 if not category_name or not isinstance(category_name, str):
472 verbose_proxy_logger.warning("Category name missing or invalid in config, skipping")
473 continue
475 # Prevent path traversal via category_name (e.g. "../../etc/passwd")
476 if not re.match(r"^[a-zA-Z0-9_\-]+$", category_name):
477 verbose_proxy_logger.warning("Category name '%s' contains invalid characters, skipping", category_name)
478 continue
480 enabled = view["enabled"]
481 action = view["action"]
482 severity_threshold = (
483 cat_config.get("severity_threshold", self.severity_threshold) or self.severity_threshold
484 )
485 custom_file = view["category_file"]
487 if not enabled:
488 verbose_proxy_logger.debug("Category %s is disabled, skipping", category_name)
489 continue
491 # Load category file (custom or default)
492 if custom_file:
493 try:
494 category_file_path = self._resolve_category_file_path(custom_file)
495 except ValueError as e:
496 verbose_proxy_logger.warning(
497 "Category %s: invalid category_file path, skipping. %s", category_name, e
498 )
499 continue
500 else:
501 # Try .yaml first, then .json (e.g. harm_toxic_abuse.json)
502 yaml_path = os.path.join(categories_dir, f"{category_name}.yaml")
503 json_path = os.path.join(categories_dir, f"{category_name}.json")
504 if os.path.exists(yaml_path):
505 category_file_path = yaml_path
506 elif os.path.exists(json_path):
507 category_file_path = json_path
508 else:
509 category_file_path = yaml_path # will trigger "not found" below
511 if not os.path.exists(category_file_path):
512 verbose_proxy_logger.warning("Category file not found: %s, skipping", category_file_path)
513 continue
515 try:
516 category_config_obj = self._load_category_file(category_file_path)
517 self.loaded_categories[category_name] = category_config_obj
519 # Use action from config, or default from category file
520 category_action = ContentFilterAction(action if action else category_config_obj.default_action)
522 # Handle conditional categories (with identifier_words + block words)
523 if category_config_obj.identifier_words and (
524 category_config_obj.inherit_from or category_config_obj.additional_block_words
525 ):
526 self._load_conditional_category(
527 category_name,
528 category_config_obj,
529 category_action,
530 severity_threshold,
531 categories_dir,
532 )
534 # Add always_block_keywords if present
535 if category_config_obj.always_block_keywords:
536 for keyword_data in category_config_obj.always_block_keywords:
537 keyword = keyword_data["keyword"].lower()
538 severity = keyword_data.get("severity", "high")
539 if self._should_apply_severity(severity, severity_threshold):
540 self.always_block_category_keywords[keyword] = (
541 category_name,
542 severity,
543 category_action,
544 )
546 # Add regular keywords from this category
547 for keyword_data in category_config_obj.keywords:
548 keyword = keyword_data["keyword"].lower()
549 severity = keyword_data["severity"]
551 # Check if keyword meets severity threshold
552 if self._should_apply_severity(severity, severity_threshold):
553 self.category_keywords[keyword] = (
554 category_name,
555 severity,
556 category_action,
557 )
559 verbose_proxy_logger.info(
560 "Loaded category %s: %s keywords, %s always-block keywords, conditional: %s",
561 category_name,
562 len(category_config_obj.keywords),
563 len(category_config_obj.always_block_keywords),
564 bool(category_config_obj.identifier_words),
565 )
566 except Exception as e:
567 verbose_proxy_logger.error("Error loading category %s: %s", category_name, e)
569 def _load_conditional_category(
570 self,
571 category_name: str,
572 category_config_obj: CategoryConfig,
573 category_action: ContentFilterAction,
574 severity_threshold: str,
575 categories_dir: str,
576 ) -> None:
577 """
578 Load a conditional category that uses identifier_words + block_words.
579 Block words can come from inherited category or additional_block_words.
581 Args:
582 category_name: Name of the category
583 category_config_obj: CategoryConfig object with identifier_words
584 category_action: Action to take when match is found
585 severity_threshold: Minimum severity threshold
586 categories_dir: Directory containing category files
587 """
588 try:
589 block_words: Final[list[str]] = []
590 inherit_from = category_config_obj.inherit_from
592 # Load inherited block words if specified
593 if inherit_from:
594 # Remove .json or .yaml extension if included
595 inherit_base: Final = inherit_from.replace(".json", "").replace(".yaml", "")
597 # Find the inherited category file
598 inherit_yaml_path: Final = os.path.join(categories_dir, f"{inherit_base}.yaml")
599 inherit_json_path: Final = os.path.join(categories_dir, f"{inherit_base}.json")
601 inherit_file_path = None
602 if os.path.exists(inherit_yaml_path):
603 inherit_file_path = inherit_yaml_path
604 elif os.path.exists(inherit_json_path):
605 inherit_file_path = inherit_json_path
606 else:
607 verbose_proxy_logger.warning(
608 "Category %s: inherit_from '%s' file not found at %s",
609 category_name,
610 inherit_from,
611 categories_dir,
612 )
613 verbose_proxy_logger.debug("Tried paths: %s, %s", inherit_yaml_path, inherit_json_path)
615 if inherit_file_path:
616 # Load the inherited category
617 inherited_category: Final = self._load_category_file(inherit_file_path)
619 # Extract block words from inherited category that meet severity threshold
620 for keyword_data in inherited_category.keywords:
621 keyword = keyword_data["keyword"].lower()
622 severity = keyword_data["severity"]
623 if self._should_apply_severity(severity, severity_threshold):
624 block_words.append(keyword)
625 else:
626 # If inherit file not found, set inherit_from to None for logging
627 inherit_from = None
629 # Add additional block words specific to this category
630 if category_config_obj.additional_block_words:
631 block_words.extend(category_config_obj.additional_block_words)
633 # Store the conditional category configuration
634 self.conditional_categories[category_name] = {
635 "identifier_words": category_config_obj.identifier_words,
636 "block_words": block_words,
637 "action": category_action,
638 "severity": "high", # Combinations are always high severity
639 }
641 # Build log message
642 log_msg = (
643 f"Loaded conditional category {category_name}: "
644 f"{len(category_config_obj.identifier_words)} identifiers + "
645 f"{len(block_words)} block words"
646 )
647 if inherit_from and category_config_obj.additional_block_words:
648 inherited_count: Final = len(block_words) - len(category_config_obj.additional_block_words)
649 log_msg += (
650 f" ({len(category_config_obj.additional_block_words)} additional + "
651 f"{inherited_count} from {inherit_from})"
652 )
653 elif inherit_from:
654 log_msg += f" (from {inherit_from})"
655 elif category_config_obj.additional_block_words:
656 log_msg += f" ({len(block_words)} from additional_block_words)"
658 verbose_proxy_logger.info(log_msg)
659 except Exception as e:
660 verbose_proxy_logger.error("Error loading conditional category for %s: %s", category_name, e)
662 def _load_category_file(self, file_path: str) -> CategoryConfig:
663 """
664 Load a category definition from a YAML or JSON file.
666 YAML format: category_name, description, default_action, keywords (list of
667 {keyword, severity}), exceptions.
668 Optional: identifier_words, always_block_keywords, inherit_from.
669 JSON format: list of {id, match, tags, severity}; match is pipe-separated
670 phrases; severity 1-4 mapped to low/medium/high. Used for harm_toxic_abuse.
672 Args:
673 file_path: Path to category YAML or JSON file
675 Returns:
676 CategoryConfig object
677 """
678 if file_path.lower().endswith(".json"):
679 return self._load_category_file_json(file_path)
680 data: Final = self._read_category_yaml(file_path)
682 return CategoryConfig(
683 category_name=data.get("category_name", "unknown"),
684 description=data.get("description", ""),
685 default_action=ContentFilterAction(data.get("default_action", "BLOCK")),
686 keywords=data.get("keywords", []),
687 exceptions=data.get("exceptions", []),
688 identifier_words=data.get("identifier_words"),
689 always_block_keywords=data.get("always_block_keywords", []),
690 inherit_from=data.get("inherit_from"),
691 additional_block_words=data.get("additional_block_words"),
692 phrase_patterns=data.get("phrase_patterns"),
693 )
695 @staticmethod
696 def _read_category_yaml(file_path: str) -> CategoryFileData:
697 with open(file_path, "r") as f:
698 return yaml.safe_load(f)
700 def _load_category_file_json(self, file_path: str) -> CategoryConfig:
701 """
702 Load a category from the harm_toxic_abuse-style JSON format.
704 Each entry has: id, match (pipe-separated phrases), tags, severity (1-4).
705 Severity mapping: 4,3 -> high; 2 -> medium; 1 -> low.
706 """
707 with open(file_path, "r") as f:
708 entries = json.load(f)
709 if not isinstance(entries, list):
710 entries = [entries]
711 # Derive category name from filename (e.g. harm_toxic_abuse.json -> harm_toxic_abuse)
712 category_name: Final = os.path.splitext(os.path.basename(file_path))[0]
713 severity_map: Final = {4: "high", 3: "high", 2: "medium", 1: "low"}
714 keywords: Final[list[dict[str, str]]] = []
715 seen: Final = set()
716 for item in entries:
717 if not isinstance(item, dict):
718 continue
719 match_str = item.get("match") or ""
720 raw_severity = item.get("severity", 2)
721 severity = severity_map.get(raw_severity if isinstance(raw_severity, int) else 2, "medium")
722 for phrase in match_str.split("|"):
723 phrase = phrase.strip().lower()
724 if not phrase or phrase in seen:
725 continue
726 seen.add(phrase)
727 keywords.append({"keyword": phrase, "severity": severity})
728 return CategoryConfig(
729 category_name=category_name,
730 description="Detects harmful, toxic, or abusive language and content",
731 default_action=ContentFilterAction("BLOCK"),
732 keywords=keywords,
733 exceptions=[],
734 )
736 def _should_apply_severity(self, severity: str, threshold: str) -> bool:
737 """
738 Check if a given severity meets the threshold.
740 Args:
741 severity: The severity level of the item ("high", "medium", "low")
742 threshold: The minimum severity threshold
744 Returns:
745 True if severity meets or exceeds threshold
746 """
747 severity_order: Final = {"low": 0, "medium": 1, "high": 2}
748 return severity_order.get(severity, 0) >= severity_order.get(threshold, 1)
750 def _add_pattern(self, pattern_config: ContentFilterPattern) -> None:
751 """
752 Add a pattern to the compiled patterns list.
754 Args:
755 pattern_config: ContentFilterPattern configuration
756 """
757 try:
758 extra_config: _PatternExtraLookup = {"keyword_pattern": None, "allow_word_numbers": False}
759 if pattern_config.pattern_type == "prebuilt":
760 if not pattern_config.pattern_name:
761 raise ValueError("pattern_name is required for prebuilt patterns")
762 compiled = get_compiled_pattern(pattern_config.pattern_name)
763 pattern_name = pattern_config.pattern_name
764 extra_config = self._lookup_pattern_extra(pattern_name)
765 elif pattern_config.pattern_type == "regex":
766 if not pattern_config.pattern:
767 raise ValueError("pattern is required for regex patterns")
768 compiled = re.compile(pattern_config.pattern, re.IGNORECASE)
769 pattern_name = pattern_config.name or "custom_regex"
770 else:
771 raise ValueError(f"Unknown pattern_type: {pattern_config.pattern_type}")
773 keyword_pattern: Final = extra_config["keyword_pattern"]
774 keyword_regex: Final = re.compile(keyword_pattern, re.IGNORECASE) if keyword_pattern else None
776 self.compiled_patterns.append(
777 {
778 "regex": compiled,
779 "pattern_name": pattern_name,
780 "action": pattern_config.action,
781 "keyword_regex": keyword_regex,
782 "allow_word_numbers": extra_config["allow_word_numbers"],
783 }
784 )
785 verbose_proxy_logger.debug("Added pattern: %s with action %s", pattern_name, pattern_config.action)
786 except Exception as e:
787 verbose_proxy_logger.error("Error adding pattern %s: %s", pattern_config, e)
788 raise
790 @staticmethod
791 def _lookup_pattern_extra(pattern_name: str) -> _PatternExtraLookup:
792 extra: Final = PATTERN_EXTRA_CONFIG.get(pattern_name)
793 return {
794 "keyword_pattern": extra.get("keyword_pattern") if extra is not None else None,
795 "allow_word_numbers": bool(extra.get("allow_word_numbers")) if extra is not None else False,
796 }
798 def _load_blocked_words_file(self, file_path: str) -> None:
799 """
800 Load blocked words from a YAML file.
802 Args:
803 file_path: Path to YAML file containing blocked_words list
805 Expected format:
806 ```yaml
807 blocked_words:
808 - keyword: "sensitive_term"
809 action: "BLOCK"
810 description: "Optional description"
811 ```
812 """
813 try:
814 with open(file_path, "r") as f:
815 data: Final = yaml.safe_load(f)
817 if not isinstance(data, dict) or "blocked_words" not in data:
818 raise ValueError("Invalid format: file must contain 'blocked_words' key with list of words")
820 for word_data in data["blocked_words"]:
821 if not isinstance(word_data, dict) or "keyword" not in word_data or "action" not in word_data:
822 verbose_proxy_logger.warning("Skipping invalid word entry: %s", word_data)
823 continue
825 keyword = word_data["keyword"].lower()
826 action = ContentFilterAction(word_data["action"])
827 description = word_data.get("description")
829 self.blocked_words[keyword] = (action, description)
831 verbose_proxy_logger.info("Loaded %s blocked words from %s", len(data["blocked_words"]), file_path)
832 except FileNotFoundError:
833 raise FileNotFoundError(f"Blocked words file not found: {file_path}")
834 except Exception as e:
835 raise Exception(f"Error loading blocked words file {file_path}: {e}")
837 def _find_pattern_spans(self, text: str, pattern_entry: CompiledPatternEntry) -> list[tuple[int, int]]:
838 """Return all match spans for a pattern, applying contextual rules if required."""
840 regex: Final[Pattern[str]] = pattern_entry["regex"]
841 keyword_regex: Final[Pattern[str] | None] = pattern_entry.get("keyword_regex")
842 allow_word_numbers: Final[bool] = pattern_entry.get("allow_word_numbers", False)
844 keyword_matches: Final = list(keyword_regex.finditer(text)) if keyword_regex is not None else None
845 if keyword_matches is not None and not keyword_matches:
846 return []
848 match_spans: Final[list[tuple[int, int]]] = []
850 for match in regex.finditer(text):
851 if keyword_matches is not None and not self._match_near_keyword(
852 match.start(), match.end(), keyword_matches, text
853 ):
854 continue
855 match_spans.append((match.start(), match.end()))
857 if allow_word_numbers:
858 for word_match in WORD_NUMBER_SEQUENCE_PATTERN.finditer(text):
859 digits = self._convert_word_number_sequence(word_match.group())
860 if not digits:
861 continue
862 if not regex.fullmatch(digits):
863 continue
864 if keyword_matches is not None and not self._match_near_keyword(
865 word_match.start(), word_match.end(), keyword_matches, text
866 ):
867 continue
868 match_spans.append((word_match.start(), word_match.end()))
870 return self._merge_spans(match_spans)
872 def _match_near_keyword(
873 self,
874 value_start: int,
875 value_end: int,
876 keyword_matches: Sequence[re.Match[str]],
877 text: str,
878 ) -> bool:
879 """Check if a value is separated from a keyword by an allowed gap."""
881 for keyword_match in keyword_matches:
882 keyword_start = keyword_match.start()
883 keyword_end = keyword_match.end()
885 if value_start >= keyword_end:
886 gap_text = text[keyword_end:value_start]
887 elif keyword_start >= value_end:
888 gap_text = text[value_end:keyword_start]
889 else:
890 return True # overlapping
892 if self._gap_text_allowed(gap_text):
893 return True
894 return False
896 def _gap_text_allowed(self, gap_text: str) -> bool:
897 """Return True if the gap between keyword and value meets word-count rules."""
899 if not gap_text.strip():
900 return True
901 if any(char.isdigit() for char in gap_text):
902 return False
904 words: Final = GAP_WORD_TOKENIZER.findall(gap_text)
905 return len(words) <= MAX_KEYWORD_VALUE_GAP_WORDS
907 def _merge_spans(self, spans: list[tuple[int, int]]) -> list[tuple[int, int]]:
908 """Merge overlapping spans to avoid double-masking."""
910 if not spans:
911 return []
913 spans.sort(key=lambda item: item[0])
914 merged: Final[list[tuple[int, int]]] = [spans[0]]
916 for start, end in spans[1:]:
917 last_start, last_end = merged[-1]
918 if start <= last_end:
919 merged[-1] = (last_start, max(last_end, end))
920 else:
921 merged.append((start, end))
922 return merged
924 def _mask_spans(self, text: str, spans: list[tuple[int, int]], redaction: str) -> str:
925 """Apply masking for the provided spans using the given redaction tag."""
927 if not spans:
928 return text
930 result_parts: Final[list[str]] = []
931 previous_end = 0
932 for start, end in spans:
933 result_parts.append(text[previous_end:start])
934 result_parts.append(redaction)
935 previous_end = end
936 result_parts.append(text[previous_end:])
937 return "".join(result_parts)
939 def _convert_word_number_sequence(self, sequence: str) -> str | None:
940 """Convert a spelled-out digit sequence (e.g., 'One-Two') into digits."""
942 tokens: Final[list[str]] = WORD_NUMBER_TOKEN_FINDER.findall(sequence)
943 if not tokens:
944 return None
946 digits: Final[list[str]] = []
947 for token in tokens:
948 digit = WORD_NUMBER_MAP.get(token.lower())
949 if digit is None:
950 return None
951 digits.append(digit)
953 return "".join(digits) if digits else None
955 def _check_patterns(self, text: str) -> tuple[str, str, ContentFilterAction] | None:
956 """
957 Check text against all compiled regex patterns.
959 Args:
960 text: Text to check
962 Returns:
963 Tuple of (matched_text, pattern_name, action) if match found, None otherwise
964 """
965 for pattern_entry in self.compiled_patterns:
966 spans = self._find_pattern_spans(text, pattern_entry)
967 if spans:
968 start, end = spans[0]
969 matched_text = text[start:end]
970 pattern_name = pattern_entry["pattern_name"]
971 action = pattern_entry["action"]
972 verbose_proxy_logger.debug("Pattern '%s' matched: %s...", pattern_name, matched_text[:20])
973 return (matched_text, pattern_name, action)
974 return None
976 def _check_conditional_categories(
977 self, text: str, exceptions: list[str]
978 ) -> tuple[str, str, str, ContentFilterAction] | None:
979 """
980 Check text for conditional category matches (identifier + block word in same sentence).
982 This implements logic like: if text contains both an identifier word (e.g., "minor")
983 AND a block word (e.g., "romantic"), then block it.
985 Args:
986 text: Text to check
987 exceptions: List of exception phrases to ignore
989 Returns:
990 Tuple of (matched_phrase, category, severity, action) if match found, None otherwise
991 """
992 text_lower: Final = text.lower()
994 # First check if any exception applies
995 for exception in exceptions:
996 if exception in text_lower:
997 return None
999 # Split text into sentences for more precise matching
1000 # Simple sentence splitting on common terminators
1001 sentences: Final = SENTENCE_TERMINATORS.split(text)
1003 for category_name, config in self.conditional_categories.items():
1004 identifier_words = config["identifier_words"]
1005 block_words = config["block_words"]
1006 action = config["action"]
1007 severity = config["severity"]
1009 # Check category-specific exceptions
1010 category_obj = self.loaded_categories.get(category_name)
1011 if category_obj:
1012 exception_found = False
1013 for exception in category_obj.exceptions:
1014 if exception in text_lower:
1015 verbose_proxy_logger.debug(
1016 "Category exception '%s' found for %s, skipping", exception, category_name
1017 )
1018 exception_found = True
1019 break
1020 if exception_found:
1021 continue
1023 # Check each sentence for identifier + block word combination
1024 for sentence in sentences:
1025 sentence_lower = sentence.lower().strip()
1026 if not sentence_lower:
1027 continue
1029 # Check if sentence contains ANY identifier word
1030 identifier_found = None
1031 for identifier in identifier_words:
1032 if identifier in sentence_lower:
1033 identifier_found = identifier
1034 break
1036 if not identifier_found:
1037 continue
1039 # Check if sentence also contains ANY block word
1040 block_word_found = None
1041 for block_word in block_words:
1042 # Use word boundary for single words to avoid false positives
1043 if " " in block_word:
1044 # Multi-word phrase
1045 if block_word in sentence_lower:
1046 block_word_found = block_word
1047 break
1048 else:
1049 # Single word - use word boundary
1050 pattern = r"\b" + re.escape(block_word) + r"\b"
1051 if re.search(pattern, sentence_lower):
1052 block_word_found = block_word
1053 break
1055 if block_word_found:
1056 matched_phrase = f"{identifier_found} + {block_word_found}"
1057 verbose_proxy_logger.warning(
1058 "Conditional match in %s: '%s' in sentence", category_name, matched_phrase
1059 )
1060 return (matched_phrase, category_name, severity, action)
1062 return None
1064 def _check_phrase_patterns(
1065 self, text: str, exceptions: list[str]
1066 ) -> tuple[str, str, str, ContentFilterAction] | None:
1067 """
1068 Check text against phrase patterns from loaded categories.
1070 Phrase patterns are regex patterns that catch paraphrased requests
1071 (e.g., "put my money to make it grow" for financial advice).
1073 Args:
1074 text: Text to check
1075 exceptions: List of exception phrases to ignore
1077 Returns:
1078 Tuple of (matched_pattern, category, severity, action) if match found, None otherwise
1079 """
1080 text_lower: Final = text.lower()
1082 for exception in exceptions:
1083 if exception in text_lower:
1084 return None
1086 for category_name, config in self.loaded_categories.items():
1087 if not config.phrase_patterns:
1088 continue
1090 # Check category-specific exceptions
1091 for exception in config.exceptions:
1092 if exception in text_lower:
1093 break
1094 else:
1095 # Determine action for this category
1096 action = ContentFilterAction(config.default_action)
1097 # Check if we have a configured action in conditional_categories
1098 if category_name in self.conditional_categories:
1099 action = self.conditional_categories[category_name]["action"]
1101 for pattern_str, pattern in config.phrase_patterns:
1102 if pattern.search(text):
1103 verbose_proxy_logger.warning("Phrase pattern match in %s: '%s'", category_name, pattern_str)
1104 return (
1105 f"phrase: {pattern_str}",
1106 category_name,
1107 "high",
1108 action,
1109 )
1111 return None
1113 def _check_category_keywords(
1114 self, text: str, exceptions: list[str]
1115 ) -> tuple[str, str, str, ContentFilterAction] | None:
1116 """
1117 Check text for category keywords.
1119 Args:
1120 text: Text to check
1121 exceptions: List of exception phrases to ignore
1123 Returns:
1124 Tuple of (keyword, category, severity, action) if match found, None otherwise
1125 """
1126 text_lower: Final = text.lower()
1128 # Check exceptions first — they take precedence over always-block keywords too.
1129 for exception in exceptions:
1130 if exception in text_lower:
1131 verbose_proxy_logger.debug("Exception phrase '%s' found, skipping category keyword check", exception)
1132 return None
1134 # Always-block keywords are checked after exceptions.
1135 for keyword, (
1136 category,
1137 severity,
1138 action,
1139 ) in self.always_block_category_keywords.items():
1140 keyword_pattern_str = self._keyword_to_regex_pattern(keyword)
1141 if " " in keyword:
1142 keyword_found = bool(re.search(keyword_pattern_str, text_lower))
1143 else:
1144 keyword_pattern = r"\b" + keyword_pattern_str + r"\b"
1145 keyword_found = bool(re.search(keyword_pattern, text_lower))
1146 if keyword_found:
1147 verbose_proxy_logger.debug("Always-block keyword '%s' found in category '%s'", keyword, category)
1148 return (keyword, category, severity, action)
1150 # Check category keywords
1151 for keyword, (category, severity, action) in self.category_keywords.items():
1152 # Convert asterisks (*) in keywords to regex wildcards
1153 # Asterisks are used in the source data to obfuscate profanity (e.g., "fu*c*k" -> "fuck")
1154 # We treat * as a wildcard matching zero or one character
1155 keyword_pattern_str = self._keyword_to_regex_pattern(keyword)
1157 # Use word boundary matching for single words to avoid false positives
1158 # (e.g., "men" should not match "recommend")
1159 # For multi-word phrases, use substring matching
1160 if " " in keyword:
1161 # Multi-word phrase - use substring matching with wildcards
1162 keyword_pattern = keyword_pattern_str
1163 keyword_found = bool(re.search(keyword_pattern, text_lower))
1164 else:
1165 # Single word - use word boundary matching to match whole words only
1166 keyword_pattern = r"\b" + keyword_pattern_str + r"\b"
1167 keyword_found = bool(re.search(keyword_pattern, text_lower))
1169 if keyword_found:
1170 # Check if this keyword has exceptions
1171 category_obj = self.loaded_categories.get(category)
1172 if category_obj:
1173 # Check category-specific exceptions
1174 exception_found = False
1175 for exception in category_obj.exceptions:
1176 if exception in text_lower:
1177 verbose_proxy_logger.debug(
1178 "Category exception '%s' found for keyword '%s', skipping", exception, keyword
1179 )
1180 exception_found = True
1181 break
1182 if exception_found:
1183 continue
1185 verbose_proxy_logger.debug(
1186 "Category keyword '%s' found in category '%s' with severity %s", keyword, category, severity
1187 )
1188 return (keyword, category, severity, action)
1189 return None
1191 def _check_blocked_words(self, text: str) -> tuple[str, ContentFilterAction, str | None] | None:
1192 """
1193 Check text for blocked keywords.
1195 Args:
1196 text: Text to check
1198 Returns:
1199 Tuple of (keyword, action, description) if match found, None otherwise
1200 """
1201 # Handle case where blocked_words might still be a list (old instances)
1202 if isinstance(self.blocked_words, list):
1203 verbose_proxy_logger.warning(
1204 "blocked_words is a list instead of dict. Re-initializing. "
1205 "This suggests an old guardrail instance is still in use. Please restart the server."
1206 )
1207 # Convert list to dict on-the-fly
1208 temp_dict: Final[dict[str, tuple[ContentFilterAction, str | None]]] = {}
1209 for word in self.blocked_words:
1210 if isinstance(word, dict):
1211 temp_dict[word.get("keyword", "").lower()] = (
1212 word.get("action", ContentFilterAction.BLOCK),
1213 word.get("description"),
1214 )
1215 self.blocked_words = temp_dict
1217 if not self.blocked_words:
1218 return None
1220 text_lower: Final = text.lower()
1221 for keyword, (action, description) in self.blocked_words.items():
1222 if keyword in text_lower:
1223 verbose_proxy_logger.debug("Blocked word '%s' found with action %s", keyword, action)
1224 return (keyword, action, description)
1225 return None
1227 def _handle_conditional_match(
1228 self,
1229 matched_phrase: str,
1230 category_name: str,
1231 severity: str,
1232 action: ContentFilterAction,
1233 detections: list[ContentFilterDetection] | None,
1234 ) -> None:
1235 """Handle conditional category match detection and action."""
1236 if detections is not None:
1237 category_detection: Final[CategoryKeywordDetection] = {
1238 "type": "category_keyword",
1239 "category": category_name,
1240 "keyword": matched_phrase,
1241 "severity": severity,
1242 "action": action.value,
1243 }
1244 detections.append(category_detection)
1246 if action == ContentFilterAction.BLOCK:
1247 error_msg: Final = (
1248 f"Content blocked: {category_name} conditional match '{matched_phrase}' detected (severity: {severity})"
1249 )
1250 verbose_proxy_logger.warning(error_msg)
1251 raise HTTPException(
1252 status_code=400,
1253 detail={
1254 "error": error_msg,
1255 "category": category_name,
1256 "matched_phrase": matched_phrase,
1257 "severity": severity,
1258 },
1259 )
1260 elif action == ContentFilterAction.MASK:
1261 verbose_proxy_logger.warning(
1262 "Conditional match '%s' from %s detected but MASK action not supported for conditional categories",
1263 matched_phrase,
1264 category_name,
1265 )
1267 def _handle_category_keyword_match(
1268 self,
1269 keyword: str,
1270 category_name: str,
1271 severity: str,
1272 action: ContentFilterAction,
1273 text: str,
1274 detections: list[ContentFilterDetection] | None,
1275 ) -> str:
1276 """Handle category keyword match detection and action."""
1277 if detections is not None:
1278 category_detection: Final[CategoryKeywordDetection] = {
1279 "type": "category_keyword",
1280 "category": category_name,
1281 "keyword": keyword,
1282 "severity": severity,
1283 "action": action.value,
1284 }
1285 detections.append(category_detection)
1287 if action == ContentFilterAction.BLOCK:
1288 error_msg = f"Content blocked: {category_name} category keyword '{keyword}' detected (severity: {severity})"
1289 verbose_proxy_logger.warning(error_msg)
1290 raise HTTPException(
1291 status_code=400,
1292 detail={
1293 "error": error_msg,
1294 "category": category_name,
1295 "keyword": keyword,
1296 "severity": severity,
1297 },
1298 )
1299 elif action == ContentFilterAction.MASK:
1300 keyword_pattern_for_masking: Final = self._keyword_to_regex_pattern(keyword)
1301 text = re.sub(
1302 keyword_pattern_for_masking,
1303 self.keyword_redaction_tag,
1304 text,
1305 flags=re.IGNORECASE,
1306 )
1307 verbose_proxy_logger.info(
1308 "Masked category keyword '%s' from %s (severity: %s)", keyword, category_name, severity
1309 )
1311 return text
1313 def _handle_pattern_match(
1314 self,
1315 pattern_name: str,
1316 action: ContentFilterAction,
1317 text: str,
1318 spans: list[tuple[int, int]],
1319 detections: list[ContentFilterDetection] | None,
1320 ) -> str:
1321 """Handle regex pattern match detection and action."""
1322 if detections is not None:
1323 pattern_detection: Final[PatternDetection] = {
1324 "type": "pattern",
1325 "pattern_name": pattern_name,
1326 "action": action.value,
1327 }
1328 detections.append(pattern_detection)
1330 if action == ContentFilterAction.BLOCK:
1331 error_msg: Final = f"Content blocked: {pattern_name} pattern detected"
1332 verbose_proxy_logger.warning(error_msg)
1333 raise HTTPException(
1334 status_code=400,
1335 detail={"error": error_msg, "pattern": pattern_name},
1336 )
1337 elif action == ContentFilterAction.MASK:
1338 redaction_tag: Final = self.pattern_redaction_format.format(pattern_name=pattern_name.upper())
1339 text = self._mask_spans(text, spans, redaction_tag)
1340 verbose_proxy_logger.info("Masked all %s matches in content", pattern_name)
1342 return text
1344 def _handle_blocked_word_match(
1345 self,
1346 keyword: str,
1347 action: ContentFilterAction,
1348 description: str | None,
1349 text: str,
1350 detections: list[ContentFilterDetection] | None,
1351 ) -> str:
1352 """Handle blocked word match detection and action."""
1353 verbose_proxy_logger.debug("Blocked word '%s' found with action %s", keyword, action)
1355 if detections is not None:
1356 blocked_word_detection: Final[BlockedWordDetection] = {
1357 "type": "blocked_word",
1358 "keyword": keyword,
1359 "action": action.value,
1360 "description": description,
1361 }
1362 detections.append(blocked_word_detection)
1364 if action == ContentFilterAction.BLOCK:
1365 error_msg = f"Content blocked: keyword '{keyword}' detected"
1366 if description:
1367 error_msg += f" ({description})"
1368 verbose_proxy_logger.warning(error_msg)
1369 raise HTTPException(
1370 status_code=400,
1371 detail={
1372 "error": error_msg,
1373 "keyword": keyword,
1374 "description": description,
1375 },
1376 )
1377 elif action == ContentFilterAction.MASK:
1378 keyword_pattern_for_masking: Final = self._keyword_to_regex_pattern(keyword)
1379 text = re.sub(
1380 keyword_pattern_for_masking,
1381 self.keyword_redaction_tag,
1382 text,
1383 flags=re.IGNORECASE,
1384 )
1385 verbose_proxy_logger.info("Masked keyword '%s' in content", keyword)
1387 return text
1389 def _filter_single_text(self, text: str, detections: list[ContentFilterDetection] | None = None) -> str:
1390 """
1391 Apply all content filtering checks to a single text.
1393 This method performs:
1394 1. Category keyword checks
1395 2. Regex pattern checks
1396 3. Blocked word checks
1398 Args:
1399 text: Text to filter
1400 detections: Optional list to append detection information
1402 Returns:
1403 Filtered text (with masking applied if action is MASK)
1405 Raises:
1406 HTTPException: If sensitive content is detected and action is BLOCK
1407 """
1408 # Collect all exceptions from loaded categories
1409 all_exceptions: Final[list[str]] = []
1410 for category in self.loaded_categories.values():
1411 all_exceptions.extend(category.exceptions)
1413 # Check conditional categories first
1414 conditional_match: Final = self._check_conditional_categories(text, all_exceptions)
1415 if conditional_match:
1416 matched_phrase, category_name, severity, action = conditional_match
1417 self._handle_conditional_match(matched_phrase, category_name, severity, action, detections)
1419 # Check phrase patterns (regex-based paraphrase detection)
1420 phrase_match: Final = self._check_phrase_patterns(text, all_exceptions)
1421 if phrase_match:
1422 matched_phrase, category_name, severity, action = phrase_match
1423 self._handle_conditional_match(matched_phrase, category_name, severity, action, detections)
1425 # Check category keywords
1426 category_keyword_match: Final = self._check_category_keywords(text, all_exceptions)
1427 if category_keyword_match:
1428 keyword, category_name, severity, action = category_keyword_match
1429 text = self._handle_category_keyword_match(keyword, category_name, severity, action, text, detections)
1431 # Check regex patterns - process ALL patterns, not just first match
1432 for pattern_entry in self.compiled_patterns:
1433 spans = self._find_pattern_spans(text, pattern_entry)
1434 if spans:
1435 pattern_name = pattern_entry["pattern_name"]
1436 action = pattern_entry["action"]
1437 text = self._handle_pattern_match(pattern_name, action, text, spans, detections)
1439 # Check blocked words - iterate through ALL blocked words
1440 text_lower = text.lower()
1441 for keyword, (action, description) in self.blocked_words.items():
1442 keyword_pattern_str = self._keyword_to_regex_pattern(keyword)
1443 if re.search(keyword_pattern_str, text_lower):
1444 text = self._handle_blocked_word_match(keyword, action, description, text, detections)
1445 text_lower = text.lower() # Update after masking
1447 return text
1449 @staticmethod
1450 def _keyword_to_regex_pattern(keyword: str) -> str:
1451 """
1452 Convert a keyword into a safe regex pattern.
1454 Escape all regex metacharacters to prevent malformed patterns from
1455 user/template keywords. Preserve existing '*' wildcard semantics by
1456 translating escaped '*' into '.?'.
1457 """
1458 return re.escape(keyword).replace(r"\*", ".?")
1460 def _mask_content(self, text: str, pattern_name: str) -> str:
1461 """
1462 Mask sensitive content in text.
1464 Args:
1465 text: Text containing sensitive content
1466 pattern_name: Name of the pattern that matched
1468 Returns:
1469 Text with sensitive content masked
1470 """
1471 redaction_tag: Final = self.pattern_redaction_format.format(pattern_name=pattern_name.upper())
1472 return redaction_tag
1474 async def _process_images(self, images: list[str], detections: list[ContentFilterDetection]) -> None:
1475 """
1476 Process images by describing them and applying content filtering.
1478 Args:
1479 images: List of image URLs
1480 detections: List to append detection information
1481 """
1482 if not (images and self.image_model and self.llm_router):
1483 return
1485 tasks: Final[list[Coroutine[object, object, ModelResponse]]] = []
1486 for image in images:
1487 task = self.llm_router.acompletion(
1488 model=self.image_model,
1489 messages=[
1490 {
1491 "role": "system",
1492 "content": "Describe the image in detail.",
1493 },
1494 {
1495 "role": "user",
1496 "content": [
1497 {"type": "image_url", "image_url": {"url": image}},
1498 ],
1499 },
1500 ],
1501 stream=False,
1502 )
1503 tasks.append(task)
1505 responses: Final = await asyncio.gather(*tasks)
1506 descriptions: Final[list[str]] = []
1507 for response in responses:
1508 image_description = self._describe_image_response_content(response)
1509 if image_description:
1510 verbose_proxy_logger.debug("Image description: %s", image_description)
1511 descriptions.append(image_description)
1512 else:
1513 verbose_proxy_logger.warning("No image description found")
1515 # Apply content filtering to image descriptions
1516 verbose_proxy_logger.debug(
1517 "ContentFilterGuardrail: Applying guardrail to %s image description(s)", len(descriptions)
1518 )
1519 for description in descriptions:
1520 # This will raise HTTPException if BLOCK action is triggered
1521 try:
1522 self._filter_single_text(description, detections=detections)
1523 except HTTPException as e:
1524 # e.detail can be a string or dict
1525 if isinstance(e.detail, dict) and "error" in e.detail:
1526 detail_dict = cast(dict[str, str], e.detail)
1527 detail_dict["error"] = detail_dict["error"] + " (Image description): " + description
1528 elif isinstance(e.detail, str):
1529 e.detail = e.detail + " (Image description): " + description
1530 else:
1531 e.detail = "Content blocked: Image description detected" + description
1532 raise e
1534 @staticmethod
1535 def _describe_image_response_content(response: ModelResponse) -> str | None:
1536 choice = response.choices[0]
1537 message = getattr(choice, "message", None)
1538 if message and getattr(message, "content", None):
1539 return message.content
1540 return None
1542 def _count_masked_entities(
1543 self,
1544 detections: list[ContentFilterDetection],
1545 masked_entity_count: dict[str, int],
1546 ) -> None:
1547 """
1548 Count masked entities by type from detections.
1550 Args:
1551 detections: List of detection dictionaries
1552 masked_entity_count: Dictionary to update with counts
1553 """
1554 for detection in detections:
1555 if detection.get("type") == "competitor_intent":
1556 continue
1557 if detection.get("action") == ContentFilterAction.MASK.value:
1558 detection_type = detection["type"]
1559 if detection_type == "pattern":
1560 pattern_detection = cast(PatternDetection, detection)
1561 pattern_name = pattern_detection["pattern_name"]
1562 masked_entity_count[pattern_name] = masked_entity_count.get(pattern_name, 0) + 1
1563 elif detection_type == "blocked_word":
1564 entity_type = "blocked_word"
1565 masked_entity_count[entity_type] = masked_entity_count.get(entity_type, 0) + 1
1566 elif detection_type == "category_keyword":
1567 category_detection = cast(CategoryKeywordDetection, detection)
1568 category = category_detection["category"]
1569 masked_entity_count[category] = masked_entity_count.get(category, 0) + 1
1571 def _build_match_details(self, detections: list[ContentFilterDetection]) -> list[dict[str, object]]:
1572 """Build match_details list from content filter detections."""
1573 match_details: Final[list[dict[str, object]]] = []
1574 for detection in detections:
1575 action_taken = detection.get("action", detection.get("action_hint", ""))
1576 detail: dict[str, object] = {"type": detection["type"], "action_taken": action_taken}
1577 if detection["type"] == "pattern":
1578 detail["detection_method"] = "regex"
1579 detail["snippet"] = cast(PatternDetection, detection).get("pattern_name", "")
1580 elif detection["type"] == "blocked_word":
1581 detail["detection_method"] = "keyword"
1582 detail["snippet"] = cast(BlockedWordDetection, detection).get("keyword", "")
1583 elif detection["type"] == "category_keyword":
1584 detail["detection_method"] = "keyword"
1585 cat_det = cast(CategoryKeywordDetection, detection)
1586 detail["snippet"] = cat_det.get("keyword", "")
1587 detail["category"] = cat_det.get("category", "")
1588 elif detection["type"] == "competitor_intent":
1589 detail["detection_method"] = "intent"
1590 detail["snippet"] = detection.get("intent", "")
1591 detail["confidence"] = detection.get("confidence")
1592 match_details.append(detail)
1593 return match_details
1595 def _get_detection_methods(self, detections: list[ContentFilterDetection]) -> str:
1596 """Get comma-separated detection methods used."""
1597 methods: Final[set[str]] = set()
1598 for detection in detections:
1599 if detection["type"] == "pattern":
1600 methods.add("regex")
1601 elif detection["type"] == "competitor_intent":
1602 methods.add("intent")
1603 else:
1604 methods.add("keyword")
1605 return ",".join(sorted(methods)) if methods else ""
1607 def _get_patterns_checked_count(self) -> int:
1608 """Get total number of patterns and keywords that were evaluated."""
1609 return (
1610 len(self.compiled_patterns)
1611 + len(self.blocked_words)
1612 + len(self.category_keywords)
1613 + len(self.always_block_category_keywords)
1614 )
1616 def _get_policy_templates(self) -> str | None:
1617 """Get comma-separated policy template names from loaded categories."""
1618 if not self.loaded_categories:
1619 return None
1620 names: Final = [cat.description or cat.category_name for cat in self.loaded_categories.values()]
1621 return ", ".join(names) if names else None
1623 def _compute_risk_score(
1624 self,
1625 detections: list[ContentFilterDetection],
1626 masked_entity_count: dict[str, int],
1627 status: "GuardrailStatus",
1628 ) -> float:
1629 """
1630 Compute a risk score from 0-10 for this guardrail evaluation.
1632 Factors:
1633 - Match ratio: how many patterns matched vs total checked
1634 - Number of entities masked
1635 - Whether the guardrail blocked the request (max risk)
1636 """
1637 if status == "guardrail_intervened":
1638 return 10.0
1640 total_masked: Final = sum(masked_entity_count.values()) if masked_entity_count else 0
1641 patterns_checked: Final = self._get_patterns_checked_count()
1643 # Match ratio contribution (0-7 points)
1644 match_ratio: Final = total_masked / patterns_checked if patterns_checked > 0 else 0.0
1645 ratio_score: Final = match_ratio * 7.0
1647 # Detection count contribution (0-3 points, capped)
1648 detection_score: Final = min(len(detections), 5) * 0.6
1650 score = ratio_score + detection_score
1652 # Floor: if anything matched, minimum risk is 2
1653 if total_masked > 0 and score < 2.0:
1654 score = 2.0
1656 return round(min(10.0, score), 1)
1658 def _apply_competitor_intent_policy(
1659 self,
1660 intent_result: CompetitorIntentResult,
1661 request_data: dict,
1662 detections: list[ContentFilterDetection],
1663 ) -> None:
1664 """
1665 Apply policy for competitor intent result: refuse (raise), reframe (passthrough), or log_only/allow (return).
1666 Appends competitor_intent detection to detections. Never returns for refuse/reframe.
1667 """
1668 intent_val: Final = intent_result.get("intent", "other")
1669 confidence_val: Final = intent_result.get("confidence", 0.0)
1670 action_hint_val: Final = intent_result.get("action_hint", "allow")
1671 evidence_list: Final = intent_result.get("evidence", [])
1672 detection: Final[CompetitorIntentDetection] = {
1673 "type": "competitor_intent",
1674 "intent": intent_val,
1675 "confidence": confidence_val,
1676 "action_hint": action_hint_val,
1677 "entities": intent_result.get("entities", {}),
1678 "signals": intent_result.get("signals", []),
1679 "evidence": [dict(e) for e in evidence_list],
1680 }
1681 detections.append(detection)
1683 if action_hint_val == "refuse":
1684 msg = "Content blocked: competitor comparison or ranking intent detected."
1685 if self._competitor_intent_checker and getattr(
1686 self._competitor_intent_checker, "refuse_message_template", None
1687 ):
1688 msg = self._competitor_intent_checker.refuse_message_template or msg
1689 verbose_proxy_logger.warning("ContentFilterGuardrail: competitor intent refuse - %s", intent_val)
1690 raise HTTPException(
1691 status_code=400,
1692 detail={
1693 "error": msg,
1694 "intent": intent_val,
1695 "confidence": confidence_val,
1696 },
1697 )
1698 if action_hint_val == "reframe":
1699 msg = (
1700 "I can help with questions about our products and services. "
1701 "Would you like to compare specific features or get more information?"
1702 )
1703 if self._competitor_intent_checker and getattr(
1704 self._competitor_intent_checker, "reframe_message_template", None
1705 ):
1706 msg = self._competitor_intent_checker.reframe_message_template or msg
1707 verbose_proxy_logger.info("ContentFilterGuardrail: competitor intent reframe - %s", intent_val)
1708 self.raise_passthrough_exception(
1709 violation_message=msg,
1710 request_data=request_data,
1711 detection_info=dict(intent_result),
1712 )
1713 # log_only or allow: just log (detection already appended)
1714 verbose_proxy_logger.debug(
1715 "ContentFilterGuardrail: competitor intent %s (action_hint=%s)",
1716 intent_val,
1717 action_hint_val,
1718 )
1720 def _log_guardrail_information(
1721 self,
1722 request_data: dict,
1723 detections: list[ContentFilterDetection],
1724 status: "GuardrailStatus",
1725 start_time: datetime,
1726 masked_entity_count: dict[str, int],
1727 exception_str: str,
1728 duration: float | None = None,
1729 ) -> None:
1730 """
1731 Log guardrail information to request_data metadata.
1733 Args:
1734 request_data: Request data dictionary
1735 detections: List of detection dictionaries
1736 status: Guardrail status
1737 start_time: Start time of guardrail execution
1738 masked_entity_count: Count of masked entities by type
1739 exception_str: Exception string if guardrail failed
1740 duration: Seconds spent inside the guardrail; defaults to the wall clock since start_time
1741 """
1742 # Convert TypedDict detections to regular dicts for JSON serialization
1743 guardrail_json_response: Exception | str | dict | list[dict] = [dict(detection) for detection in detections]
1744 if status != "success":
1745 guardrail_json_response = exception_str if exception_str else [dict(detection) for detection in detections]
1747 # Competitor intent: add confidence and classification to tracing if present
1748 tracing_kw: Final[GuardrailTracingDetail] = {
1749 "guardrail_id": self.config_guardrail_id or self.guardrail_name,
1750 "policy_template": self.config_policy_template or self._get_policy_templates(),
1751 "detection_method": (self._get_detection_methods(detections) if detections else None),
1752 "match_details": (self._build_match_details(detections) if detections else None),
1753 "patterns_checked": self._get_patterns_checked_count(),
1754 "risk_score": self._compute_risk_score(detections, masked_entity_count, status),
1755 }
1756 for d in detections:
1757 if isinstance(d, dict) and d.get("type") == "competitor_intent":
1758 tracing_kw["confidence_score"] = d.get("confidence")
1759 tracing_kw["classification"] = dict(d)
1760 break
1762 self.add_standard_logging_guardrail_information_to_request_data(
1763 guardrail_provider=self.guardrail_provider,
1764 guardrail_json_response=guardrail_json_response,
1765 request_data=request_data,
1766 guardrail_status=status,
1767 start_time=start_time.timestamp(),
1768 end_time=datetime.now().timestamp(),
1769 duration=(datetime.now() - start_time).total_seconds() if duration is None else duration,
1770 masked_entity_count=masked_entity_count,
1771 tracing_detail=GuardrailTracingDetail(**tracing_kw),
1772 )
1774 @staticmethod
1775 def _get_mcp_tool_name(request_data: dict) -> str | None:
1776 raw_name: Final[object] = request_data.get("mcp_tool_name")
1777 if isinstance(raw_name, str) and raw_name:
1778 return raw_name
1779 return None
1781 def _assert_argument_label_clean(
1782 self, text: str, detections: list[ContentFilterDetection], context_label: str
1783 ) -> None:
1784 if self._filter_single_text(text, detections=detections) != text:
1785 raise HTTPException(
1786 status_code=400,
1787 detail={
1788 "error": (
1789 f"Content blocked: {context_label} argument matched a masking rule on a non-rewritable field"
1790 )
1791 },
1792 )
1794 def _filter_argument_value(
1795 self,
1796 value: object,
1797 detections: list[ContentFilterDetection],
1798 context_label: str,
1799 depth: int = 0,
1800 ) -> object:
1801 if depth > DEFAULT_MAX_RECURSE_DEPTH:
1802 raise HTTPException(
1803 status_code=400,
1804 detail={"error": f"Content blocked: {context_label} arguments exceed the maximum nesting depth"},
1805 )
1806 if isinstance(value, str):
1807 return self._filter_single_text(value, detections=detections)
1808 if isinstance(value, (int, float)) and not isinstance(value, bool):
1809 self._assert_argument_label_clean(str(value), detections, context_label)
1810 return value
1811 if isinstance(value, dict):
1812 for key in value:
1813 if isinstance(key, str):
1814 self._assert_argument_label_clean(key, detections, context_label)
1815 return {
1816 key: self._filter_argument_value(item, detections, context_label, depth + 1)
1817 for key, item in value.items()
1818 }
1819 if isinstance(value, list):
1820 return [self._filter_argument_value(item, detections, context_label, depth + 1) for item in value]
1821 return value
1823 def _scan_mcp_tool_call_arguments(
1824 self,
1825 request_data: dict,
1826 detections: list[ContentFilterDetection],
1827 logging_obj: Optional["LiteLLMLoggingObj"] = None,
1828 ) -> None:
1829 if not self._event_hook_is_event_type(GuardrailEventHooks.pre_mcp_call):
1830 return
1831 call_type: Final[object] = getattr(logging_obj, "call_type", None)
1832 if logging_obj is not None and call_type != CallTypes.call_mcp_tool.value:
1833 return
1834 if self._get_mcp_tool_name(request_data) is None:
1835 return
1836 raw_arguments: Final[object] = request_data.get("mcp_arguments")
1837 if not isinstance(raw_arguments, dict) or not raw_arguments:
1838 return
1839 filtered_arguments: Final = self._filter_argument_value(raw_arguments, detections, "MCP tool call")
1840 if filtered_arguments == raw_arguments:
1841 return
1842 request_data["mcp_arguments"] = filtered_arguments
1843 request_data["modified_arguments"] = filtered_arguments
1845 @staticmethod
1846 def _get_tool_call_arguments(tool_call: object) -> str | None:
1847 function: Final[object] = (
1848 tool_call.get("function") if isinstance(tool_call, dict) else getattr(tool_call, "function", None)
1849 )
1850 arguments: Final[object] = (
1851 function.get("arguments") if isinstance(function, dict) else getattr(function, "arguments", None)
1852 )
1853 return arguments if isinstance(arguments, str) and arguments.strip() else None
1855 @staticmethod
1856 def _set_tool_call_arguments(tool_call: object, arguments: str) -> None:
1857 function: Final[object] = (
1858 tool_call.get("function") if isinstance(tool_call, dict) else getattr(tool_call, "function", None)
1859 )
1860 if isinstance(function, dict):
1861 function["arguments"] = arguments
1862 elif isinstance(function, Function):
1863 function.arguments = arguments
1865 def _filter_tool_call_arguments(
1866 self,
1867 arguments: str,
1868 detections: list[ContentFilterDetection], # mutable-ok: _filter_single_text appends into a caller-owned list
1869 ) -> str:
1870 try:
1871 parsed: Final[object] = json.loads(arguments)
1872 except (json.JSONDecodeError, TypeError, ValueError):
1873 return self._filter_single_text(arguments, detections=detections)
1874 if not isinstance(parsed, (dict, list)):
1875 return self._filter_single_text(arguments, detections=detections)
1876 filtered: Final = self._filter_argument_value(parsed, detections, "tool call")
1877 return arguments if filtered == parsed else json.dumps(filtered)
1879 def _scan_tool_call_arguments(
1880 self,
1881 inputs: "GenericGuardrailAPIInputs",
1882 detections: list[ContentFilterDetection], # mutable-ok: _filter_single_text appends into a caller-owned list
1883 ) -> None:
1884 for tool_call in inputs.get("tool_calls") or ():
1885 arguments = self._get_tool_call_arguments(tool_call)
1886 if arguments is None:
1887 continue
1888 filtered_arguments = self._filter_tool_call_arguments(arguments, detections)
1889 if filtered_arguments != arguments:
1890 self._set_tool_call_arguments(tool_call, filtered_arguments)
1892 async def apply_guardrail(
1893 self,
1894 inputs: "GenericGuardrailAPIInputs",
1895 request_data: dict,
1896 input_type: Literal["request", "response"],
1897 logging_obj: Optional["LiteLLMLoggingObj"] = None,
1898 ) -> "GenericGuardrailAPIInputs":
1899 """
1900 Apply content filtering guardrail to a batch of texts.
1902 This method checks for sensitive patterns and blocked keywords,
1903 either blocking the request or masking the sensitive content.
1905 Args:
1906 inputs: Dictionary containing texts and optional images
1907 request_data: Request data dictionary for logging metadata
1908 input_type: Whether this is a "request" or "response"
1909 logging_obj: Optional logging object
1911 Returns:
1912 GenericGuardrailAPIInputs - processed_texts may be masked, images unchanged
1914 Raises:
1915 HTTPException: If sensitive content is detected and action is BLOCK
1916 """
1917 start_time: Final = datetime.now()
1918 detections: Final[list[ContentFilterDetection]] = []
1919 masked_entity_count: Final[dict[str, int]] = {}
1920 status: GuardrailStatus = "success"
1921 exception_str: str = ""
1923 try:
1924 texts: Final = inputs.get("texts", [])
1925 images: Final = inputs.get("images", [])
1927 # Process images if present
1928 await self._process_images(images, detections)
1930 # Process texts
1931 verbose_proxy_logger.debug("ContentFilterGuardrail: Applying guardrail to %s text(s)", len(texts))
1933 processed_texts: Final = []
1934 for text in texts:
1935 # Competitor intent check first (optional; may refuse/reframe)
1936 if self._competitor_intent_checker and text:
1937 intent_result = self._competitor_intent_checker.run(text)
1938 if intent_result.get("intent", "other") != "other":
1939 self._apply_competitor_intent_policy(intent_result, request_data, detections)
1940 filtered_text = self._filter_single_text(text, detections=detections)
1941 processed_texts.append(filtered_text)
1943 verbose_proxy_logger.debug("ContentFilterGuardrail: Guardrail applied successfully")
1944 inputs["texts"] = processed_texts
1946 self._scan_tool_call_arguments(inputs=inputs, detections=detections)
1948 if input_type == "request":
1949 self._scan_mcp_tool_call_arguments(
1950 request_data=request_data, detections=detections, logging_obj=logging_obj
1951 )
1953 # Count masked entities by type
1954 self._count_masked_entities(detections, masked_entity_count)
1956 return inputs
1957 except HTTPException:
1958 status = "guardrail_intervened"
1959 raise
1960 except Exception as e:
1961 status = "guardrail_failed_to_respond"
1962 exception_str = str(e)
1963 raise e
1964 finally:
1965 # Log guardrail information
1966 self._log_guardrail_information(
1967 request_data=request_data,
1968 detections=detections,
1969 status=status,
1970 start_time=start_time,
1971 masked_entity_count=masked_entity_count,
1972 exception_str=exception_str,
1973 )
1975 def _streamed_scan_plan(self) -> _StreamedScanPlan:
1976 """
1977 Per-stream inputs for buffer trimming: the retained tail length (the default
1978 context, widened to the longest configured keyword), the category exception
1979 phrases, which suppress matches anywhere in the scanned text, and the conditional
1980 category words, which only match when paired inside one sentence.
1981 """
1982 longest_keyword: Final = max(
1983 map(len, (*self.blocked_words, *self.category_keywords, *self.always_block_category_keywords)),
1984 default=0,
1985 )
1986 return _StreamedScanPlan(
1987 context_chars=max(CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS, longest_keyword),
1988 exception_phrases=tuple(
1989 phrase for category in self.loaded_categories.values() for phrase in category.exceptions
1990 ),
1991 conditional_words=tuple(
1992 word
1993 for config in self.conditional_categories.values()
1994 for word in (*config["identifier_words"], *config["block_words"])
1995 ),
1996 )
1998 @staticmethod
1999 def _cut_breaks_wider_context(buffered_text: str, head: str, tail: str, plan: _StreamedScanPlan) -> bool:
2000 buffered_lower: Final = buffered_text.lower()
2001 tail_lower: Final = tail.lower()
2002 if any(phrase in buffered_lower and phrase not in tail_lower for phrase in plan.exception_phrases):
2003 return True
2004 cut_sentence: Final = (
2005 SENTENCE_TERMINATORS.split(head.lower())[-1] + SENTENCE_TERMINATORS.split(tail_lower, maxsplit=1)[0]
2006 )
2007 return any(word in cut_sentence for word in plan.conditional_words)
2009 def _trim_streamed_choice_buffer(
2010 self, state: _StreamedChoiceState, masked_text: str, plan: _StreamedScanPlan
2011 ) -> _StreamedChoiceState:
2012 """
2013 Bound the per-choice buffer rescanned on every streamed chunk.
2015 Once the buffer exceeds twice the scan context, drop everything but the last
2016 context-sized tail, provided no exception phrase or unfinished conditional sentence
2017 would leave the buffer, the two halves mask to the same output as the whole (so no
2018 match or phrase straddles the cut), and the dropped prefix has already been yielded.
2019 Otherwise keep the buffer and retry once it has grown by another context length.
2021 Detections found in the dropped prefix move to the state's committed detections.
2022 """
2023 if len(state.buffered_text) <= max(2 * plan.context_chars, state.next_trim_len):
2024 return state
2025 deferred: Final = replace(state, next_trim_len=len(state.buffered_text) + plan.context_chars)
2026 head: Final = state.buffered_text[: -plan.context_chars]
2027 tail: Final = state.buffered_text[-plan.context_chars :]
2028 if self._cut_breaks_wider_context(state.buffered_text, head, tail, plan):
2029 return deferred
2030 head_detections: Final[list[ContentFilterDetection]] = [] # mutable-ok: filled by _filter_single_text
2031 try:
2032 masked_head: Final = self._filter_single_text(head, detections=head_detections)
2033 masked_tail: Final = self._filter_single_text(tail)
2034 except Exception:
2035 return deferred
2036 if masked_head + masked_tail != masked_text or len(masked_head) > state.yielded_masked_text_len:
2037 return deferred
2038 return replace(
2039 state,
2040 buffered_text=tail,
2041 yielded_masked_text_len=state.yielded_masked_text_len - len(masked_head),
2042 committed_detections=state.committed_detections + tuple(head_detections),
2043 next_trim_len=0,
2044 )
2046 @staticmethod
2047 def _merge_detections(detections: Sequence[ContentFilterDetection]) -> tuple[ContentFilterDetection, ...]:
2048 return tuple(detection for index, detection in enumerate(detections) if detection not in detections[:index])
2050 async def async_post_call_streaming_iterator_hook(
2051 self,
2052 user_api_key_dict: UserAPIKeyAuth,
2053 response: Any,
2054 request_data: dict,
2055 ) -> AsyncGenerator[ModelResponseStream, None]:
2056 """
2057 Process streaming response chunks and check for blocked content.
2059 For BLOCK action: Raises HTTPException immediately when blocked content is detected.
2060 For MASK action: Content is buffered to handle patterns split across chunks.
2062 At stream end (including when BLOCK raises HTTPException), write a
2063 standard_logging_guardrail_information entry into request_data["metadata"]
2064 so the post-call log row reaches standard_logging_object.guardrail_information
2065 and the UI Request Lifecycle panel. Mirrors apply_guardrail's finally-block
2066 contract.
2067 """
2068 state_by_choice: Final[dict[int, _StreamedChoiceState]] = {}
2069 plan: Final = self._streamed_scan_plan()
2071 start_time: Final = datetime.now()
2072 scan_seconds: float = 0.0 # rebind-ok: accumulates per-chunk scan time across the stream
2073 detections: list[ContentFilterDetection] = []
2074 masked_entity_count: Final[dict[str, int]] = {}
2075 status: GuardrailStatus = "success"
2076 exception_str: str = ""
2078 verbose_proxy_logger.info(
2079 "ContentFilterGuardrail: Starting robust streaming masking for model %s", request_data.get("model")
2080 )
2082 try:
2083 async for item in response:
2084 if isinstance(item, ModelResponseStream) and item.choices:
2085 for choice in item.choices:
2086 if not (hasattr(choice, "delta") and choice.delta):
2087 continue
2089 choice_index = getattr(choice, "index", 0)
2090 if not isinstance(choice_index, int):
2091 choice_index = 0
2093 content = getattr(choice.delta, "content", None)
2094 is_final = bool(getattr(choice, "finish_reason", None))
2095 new_content = content if isinstance(content, str) else ""
2096 if not new_content and not is_final:
2097 continue
2099 previous_state = state_by_choice.get(choice_index, _StreamedChoiceState())
2100 buffered_text = previous_state.buffered_text + new_content
2101 if not buffered_text:
2102 continue
2104 # Add a space at the end if it's the final chunk to trigger word boundaries (\b)
2105 text_to_scan = buffered_text + (" " if is_final else "")
2106 choice_detections: list[ContentFilterDetection] = []
2107 scan_started = time.perf_counter()
2109 try:
2110 masked_text = self._filter_single_text(text_to_scan, detections=choice_detections)
2111 if is_final and masked_text.endswith(" "):
2112 masked_text = masked_text[:-1]
2113 latest_detections = tuple(choice_detections)
2114 except HTTPException:
2115 state_by_choice[choice_index] = replace(
2116 previous_state, latest_detections=tuple(choice_detections)
2117 )
2118 raise
2119 except Exception as e:
2120 verbose_proxy_logger.error("ContentFilterGuardrail: Error in masking: %s", e)
2121 masked_text = text_to_scan # Fallback to current text
2122 latest_detections = previous_state.latest_detections
2123 finally:
2124 scan_seconds += time.perf_counter() - scan_started
2126 safe_to_yield_len = max(
2127 previous_state.yielded_masked_text_len,
2128 len(masked_text) - (0 if is_final else CONTENT_FILTER_STREAMING_HOLDBACK_CHARS),
2129 )
2130 choice.delta.content = masked_text[previous_state.yielded_masked_text_len : safe_to_yield_len]
2131 next_state = replace(
2132 previous_state,
2133 buffered_text=buffered_text,
2134 yielded_masked_text_len=safe_to_yield_len,
2135 latest_detections=latest_detections,
2136 )
2137 if is_final:
2138 state_by_choice[choice_index] = next_state
2139 continue
2141 trim_started = time.perf_counter()
2142 state_by_choice[choice_index] = self._trim_streamed_choice_buffer(next_state, masked_text, plan)
2143 scan_seconds += time.perf_counter() - trim_started
2145 yield item
2146 else:
2147 # Not a ModelResponseStream or no choices - yield as is
2148 yield item
2149 except HTTPException:
2150 status = "guardrail_intervened"
2151 raise
2152 except Exception as e:
2153 status = "guardrail_failed_to_respond"
2154 exception_str = str(e)
2155 raise e
2156 finally:
2157 detections = [
2158 detection
2159 for state in state_by_choice.values()
2160 for detection in self._merge_detections((*state.committed_detections, *state.latest_detections))
2161 ]
2162 self._count_masked_entities(detections, masked_entity_count)
2163 self._log_guardrail_information(
2164 request_data=request_data,
2165 detections=detections,
2166 status=status,
2167 start_time=start_time,
2168 masked_entity_count=masked_entity_count,
2169 exception_str=exception_str,
2170 duration=scan_seconds,
2171 )
2173 @staticmethod
2174 def get_config_model():
2175 from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import (
2176 LitellmContentFilterGuardrailConfigModel,
2177 )
2179 return LitellmContentFilterGuardrailConfigModel
2181 @classmethod
2182 def get_supported_event_hooks(cls) -> list[GuardrailEventHooks]:
2183 return [
2184 GuardrailEventHooks.pre_call,
2185 GuardrailEventHooks.post_call,
2186 GuardrailEventHooks.during_call,
2187 GuardrailEventHooks.realtime_input_transcription,
2188 GuardrailEventHooks.pre_mcp_call,
2189 GuardrailEventHooks.post_mcp_call,
2190 ]