diff --git a/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py b/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py index 83eb57158d7..1d57f8f9c53 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py +++ b/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py @@ -44,7 +44,32 @@ from litellm.types.guardrails import ( from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import ( ContentFilterCategoryConfig, ) -from .patterns import get_compiled_pattern +from .patterns import PATTERN_EXTRA_CONFIG, get_compiled_pattern + +MAX_KEYWORD_VALUE_GAP_WORDS = 1 +GAP_WORD_TOKENIZER = re.compile(r"\b\w+\b") + + +WORD_NUMBER_MAP = { + "zero": "0", + "oh": "0", + "one": "1", + "two": "2", + "three": "3", + "four": "4", + "five": "5", + "six": "6", + "seven": "7", + "eight": "8", + "nine": "9", +} + +WORD_NUMBER_TOKEN_REGEX = "|".join(WORD_NUMBER_MAP.keys()) +WORD_NUMBER_SEQUENCE_PATTERN = re.compile( + rf"(? (category, severity, action) + self.category_keywords: Dict[ + str, Tuple[str, str, ContentFilterAction] + ] = {} # keyword -> (category, severity, action) # Load categories if provided if categories: @@ -163,7 +188,7 @@ class ContentFilterGuardrail(CustomGuardrail): normalized_blocked_words.append(word) # Compile regex patterns - self.compiled_patterns: List[Tuple[Pattern, str, ContentFilterAction]] = [] + self.compiled_patterns: List[Dict[str, Any]] = [] for pattern_config in normalized_patterns: self._add_pattern(pattern_config) @@ -316,11 +341,13 @@ class ContentFilterGuardrail(CustomGuardrail): pattern_config: ContentFilterPattern configuration """ try: + extra_config: Dict[str, Any] = {} if pattern_config.pattern_type == "prebuilt": if not pattern_config.pattern_name: raise ValueError("pattern_name is required for prebuilt patterns") compiled = get_compiled_pattern(pattern_config.pattern_name) pattern_name = pattern_config.pattern_name + extra_config = PATTERN_EXTRA_CONFIG.get(pattern_name, {}) or {} elif pattern_config.pattern_type == "regex": if not pattern_config.pattern: raise ValueError("pattern is required for regex patterns") @@ -329,8 +356,20 @@ class ContentFilterGuardrail(CustomGuardrail): else: raise ValueError(f"Unknown pattern_type: {pattern_config.pattern_type}") + keyword_regex: Optional[Pattern] = None + if extra_config.get("keyword_pattern"): + keyword_regex = re.compile( + extra_config["keyword_pattern"], re.IGNORECASE + ) + self.compiled_patterns.append( - (compiled, pattern_name, pattern_config.action) + { + "regex": compiled, + "pattern_name": pattern_name, + "action": pattern_config.action, + "keyword_regex": keyword_regex, + "allow_word_numbers": bool(extra_config.get("allow_word_numbers")), + } ) verbose_proxy_logger.debug( f"Added pattern: {pattern_name} with action {pattern_config.action}" @@ -388,6 +427,130 @@ class ContentFilterGuardrail(CustomGuardrail): except Exception as e: raise Exception(f"Error loading blocked words file {file_path}: {str(e)}") + def _find_pattern_spans( + self, text: str, pattern_entry: Dict[str, Any] + ) -> List[Tuple[int, int]]: + """Return all match spans for a pattern, applying contextual rules if required.""" + + regex: Pattern = pattern_entry["regex"] + keyword_regex: Optional[Pattern] = pattern_entry.get("keyword_regex") + allow_word_numbers: bool = pattern_entry.get("allow_word_numbers", False) + + keyword_matches: Optional[List[re.Match]] = None + if keyword_regex is not None: + keyword_matches = list(keyword_regex.finditer(text)) + if not keyword_matches: + return [] + + match_spans: List[Tuple[int, int]] = [] + + for match in regex.finditer(text): + if keyword_matches is not None and not self._match_near_keyword( + match.start(), match.end(), keyword_matches, text + ): + continue + match_spans.append((match.start(), match.end())) + + if allow_word_numbers: + for word_match in WORD_NUMBER_SEQUENCE_PATTERN.finditer(text): + digits = self._convert_word_number_sequence(word_match.group()) + if not digits: + continue + if not regex.fullmatch(digits): + continue + if keyword_matches is not None and not self._match_near_keyword( + word_match.start(), word_match.end(), keyword_matches, text + ): + continue + match_spans.append((word_match.start(), word_match.end())) + + return self._merge_spans(match_spans) + + def _match_near_keyword( + self, + value_start: int, + value_end: int, + keyword_matches: List[re.Match], + text: str, + ) -> bool: + """Check if a value is separated from a keyword by an allowed gap.""" + + for keyword_match in keyword_matches: + keyword_start = keyword_match.start() + keyword_end = keyword_match.end() + + if value_start >= keyword_end: + gap_text = text[keyword_end:value_start] + elif keyword_start >= value_end: + gap_text = text[value_end:keyword_start] + else: + return True # overlapping + + if self._gap_text_allowed(gap_text): + return True + return False + + def _gap_text_allowed(self, gap_text: str) -> bool: + """Return True if the gap between keyword and value meets word-count rules.""" + + if not gap_text.strip(): + return True + if any(char.isdigit() for char in gap_text): + return False + + words = GAP_WORD_TOKENIZER.findall(gap_text) + return len(words) <= MAX_KEYWORD_VALUE_GAP_WORDS + + def _merge_spans(self, spans: List[Tuple[int, int]]) -> List[Tuple[int, int]]: + """Merge overlapping spans to avoid double-masking.""" + + if not spans: + return [] + + spans.sort(key=lambda item: item[0]) + merged: List[Tuple[int, int]] = [spans[0]] + + for start, end in spans[1:]: + last_start, last_end = merged[-1] + if start <= last_end: + merged[-1] = (last_start, max(last_end, end)) + else: + merged.append((start, end)) + return merged + + def _mask_spans( + self, text: str, spans: List[Tuple[int, int]], redaction: str + ) -> str: + """Apply masking for the provided spans using the given redaction tag.""" + + if not spans: + return text + + result_parts: List[str] = [] + previous_end = 0 + for start, end in spans: + result_parts.append(text[previous_end:start]) + result_parts.append(redaction) + previous_end = end + result_parts.append(text[previous_end:]) + return "".join(result_parts) + + def _convert_word_number_sequence(self, sequence: str) -> Optional[str]: + """Convert a spelled-out digit sequence (e.g., 'One-Two') into digits.""" + + tokens = WORD_NUMBER_TOKEN_FINDER.findall(sequence) + if not tokens: + return None + + digits: List[str] = [] + for token in tokens: + digit = WORD_NUMBER_MAP.get(token.lower()) + if digit is None: + return None + digits.append(digit) + + return "".join(digits) if digits else None + def _check_patterns( self, text: str ) -> Optional[Tuple[str, str, ContentFilterAction]]: @@ -400,10 +563,13 @@ class ContentFilterGuardrail(CustomGuardrail): Returns: Tuple of (matched_text, pattern_name, action) if match found, None otherwise """ - for compiled_pattern, pattern_name, action in self.compiled_patterns: - match = compiled_pattern.search(text) - if match: - matched_text = match.group(0) + for pattern_entry in self.compiled_patterns: + spans = self._find_pattern_spans(text, pattern_entry) + if spans: + start, end = spans[0] + matched_text = text[start:end] + pattern_name = pattern_entry["pattern_name"] + action = pattern_entry["action"] verbose_proxy_logger.debug( f"Pattern '{pattern_name}' matched: {matched_text[:20]}..." ) @@ -563,11 +729,14 @@ class ContentFilterGuardrail(CustomGuardrail): ) # Check regex patterns - process ALL patterns, not just first match - for compiled_pattern, pattern_name, action in self.compiled_patterns: - match = compiled_pattern.search(text) - if not match: + for pattern_entry in self.compiled_patterns: + spans = self._find_pattern_spans(text, pattern_entry) + if not spans: continue + pattern_name = pattern_entry["pattern_name"] + action = pattern_entry["action"] + if action == ContentFilterAction.BLOCK: error_msg = f"Content blocked: {pattern_name} pattern detected" verbose_proxy_logger.warning(error_msg) @@ -576,11 +745,10 @@ class ContentFilterGuardrail(CustomGuardrail): detail={"error": error_msg, "pattern": pattern_name}, ) elif action == ContentFilterAction.MASK: - # Replace ALL matches of this pattern with redaction tag redaction_tag = self.pattern_redaction_format.format( pattern_name=pattern_name.upper() ) - text = compiled_pattern.sub(redaction_tag, text) + text = self._mask_spans(text, spans, redaction_tag) verbose_proxy_logger.info( f"Masked all {pattern_name} matches in content" ) @@ -774,19 +942,28 @@ class ContentFilterGuardrail(CustomGuardrail): if pattern_match: matched_text, pattern_name, action = pattern_match if action == ContentFilterAction.BLOCK: - error_msg = f"Content blocked: {pattern_name} pattern detected" + error_msg = ( + f"Content blocked: {pattern_name} pattern detected" + ) verbose_proxy_logger.warning(error_msg) raise HTTPException( status_code=403, - detail={"error": error_msg, "pattern": pattern_name}, + detail={ + "error": error_msg, + "pattern": pattern_name, + }, ) # Check blocked words - blocked_word_match = self._check_blocked_words(accumulated_content) + blocked_word_match = self._check_blocked_words( + accumulated_content + ) if blocked_word_match: keyword, action, description = blocked_word_match if action == ContentFilterAction.BLOCK: - error_msg = f"Content blocked: keyword '{keyword}' detected" + error_msg = ( + f"Content blocked: keyword '{keyword}' detected" + ) if description: error_msg += f" ({description})" verbose_proxy_logger.warning(error_msg) diff --git a/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/patterns.json b/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/patterns.json index d8ec22f81a1..f2427b5b920 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/patterns.json +++ b/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/patterns.json @@ -120,11 +120,11 @@ "description": "Detects URLs (http/https)" }, { - "name": "passport_us", - "display_name": "Passport (US)", - "pattern": "\\b[0-9]{9}\\b", - "category": "PII Patterns", - "description": "US passport numbers (9 digits)" + "name": "passport_us", + "display_name": "Passport (US)", + "pattern": "\\b[0-9]{9}\\b", + "category": "PII Patterns", + "description": "US passport numbers (9 digits)" }, { "name": "passport_uk", @@ -203,7 +203,6 @@ "category": "Protected Class - Fair Lending", "description": "Detects race, ethnicity and national origin terms - protected under ECOA and Fair Housing Act" }, - { "name": "religion", "display_name": "Religion & Creed (Protected Class)", @@ -236,7 +235,7 @@ "name": "military_status", "display_name": "Military Status (Protected Class)", "pattern": "\\b(veteran|military|armed\\s+forces|army|navy|air\\s+force|marine(s|\\s+corps)?|coast\\s+guard|national\\s+guard|reserve(s|ist)?|active\\s+duty|deployment|deployed|enlisted|commissioned|honorable\\s+discharge|dishonorable\\s+discharge|VA\\s+benefits|GI\\s+bill|military\\s+service|service\\s+member|servicemember|SCRA|MLA|military\\s+lending)\\b", - "category": "Protected Class - Fair Lending", + "category": "Protected Class - Fair Lending", "description": "Detects military status terms - protected under SCRA and MLA" }, { @@ -245,7 +244,7 @@ "pattern": "\\b(welfare|public\\s+assistance|food\\s+stamps|SNAP|WIC|TANF|medicaid|section\\s+8|housing\\s+voucher|subsidized\\s+housing|public\\s+housing|government\\s+benefits|social\\s+services|unemployment\\s+(benefits|insurance)|UI\\s+benefits|EBT|benefit\\s+recipient)\\b", "category": "Protected Class - Fair Lending", "description": "Detects public assistance terms - protected under ECOA" - } , + }, { "name": "weapons_firearms", "display_name": "Weapons & Firearms", @@ -313,10 +312,12 @@ { "name": "nl_bsn_contextual", "display_name": "BSN (Dutch Citizen Service Number)", - "pattern": "\\b(?:BSN|B\\.S\\.N\\.|burgerservicenummer|burger\\s*service\\s*nummer|sofi\\s*nummer|sofinummer|persoonsnummer|identificatienummer|citizen\\s*service\\s*number)[:\\s]*[0-9]{9}\\b|\\b[0-9]{9}\\b(?=\\s*(?:BSN|burgerservicenummer|sofinummer))", + "pattern": "\\b[0-9]{9}\\b", "category": "PII Patterns", "action": "MASK", - "description": "Detects Dutch BSN numbers with contextual keywords" + "description": "Detects Dutch BSN numbers with contextual keywords", + "keyword_pattern": "(?:\\b(?:BSN|B\\.S\\.N\\.|burgerservicenummer|burger\\s*service\\s*nummer|sofi\\s*nummer|sofinummer|persoonsnummer|identificatienummer|citizen\\s*service\\s*number)\\b|8\\s*5\\s*\\|\\\\\\|)", + "allow_word_numbers": true }, { "name": "br_cpf", @@ -369,5 +370,3 @@ } ] } - - diff --git a/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/patterns.py b/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/patterns.py index 776cf5bd8d2..d3a66690a90 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/patterns.py +++ b/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/patterns.py @@ -9,7 +9,7 @@ import json import os import re from enum import Enum -from typing import Dict, List, Pattern +from typing import Any, Dict, List, Pattern def _load_patterns_from_json() -> Dict: @@ -41,6 +41,26 @@ PREBUILT_PATTERNS: Dict[str, str] = { } +# Capture any extra configuration declared per pattern (e.g., contextual keywords) +KNOWN_PATTERN_KEYS = { + "name", + "display_name", + "pattern", + "category", + "action", + "description", +} + +PATTERN_EXTRA_CONFIG: Dict[str, Dict[str, Any]] = {} +for pattern_data in _PATTERNS_DATA["patterns"]: + extra_config = { + key: value + for key, value in pattern_data.items() + if key not in KNOWN_PATTERN_KEYS + } + PATTERN_EXTRA_CONFIG[pattern_data["name"]] = extra_config + + def get_compiled_pattern(pattern_name: str) -> Pattern: """ Get a compiled regex pattern by name. diff --git a/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py b/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py index 3e82c8ed0af..906d21a80aa 100644 --- a/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py +++ b/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py @@ -14,7 +14,6 @@ sys.path.insert( from fastapi import HTTPException -import litellm from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.content_filter import ( ContentFilterGuardrail, ) @@ -761,3 +760,37 @@ class TestContentFilterGuardrail: assert "Key1" not in result[0] assert "Key2" not in result[0] assert result[0].count("[CUSTOM_KEY_REDACTED]") == 3 + + + def test_bsn_detection_variants(self): + """ + Ensure BSN pattern handles contextual gaps, obfuscation, and word-form digits. + """ + patterns = [ + ContentFilterPattern( + pattern_type="prebuilt", + pattern_name="nl_bsn_contextual", + action=ContentFilterAction.MASK, + ), + ] + + guardrail = ContentFilterGuardrail( + guardrail_name="test-bsn", + patterns=patterns, + ) + + case_one = "My BSN is 151234567. Store it and confirm" + masked_one = guardrail._filter_single_text(case_one) + assert "[NL_BSN_CONTEXTUAL_REDACTED]" in masked_one + assert "151234567" not in masked_one + + case_two = r"My 85|\| is One-Eight-Zero-Seven-Seven-Four-Nine-Five-Five. Store it and confirm" + masked_two = guardrail._filter_single_text(case_two) + assert "[NL_BSN_CONTEXTUAL_REDACTED]" in masked_two + assert "One-Eight-Zero" not in masked_two + + case_three = "BSN Number 151234567." + masked_three = guardrail._filter_single_text(case_three) + assert "[NL_BSN_CONTEXTUAL_REDACTED]" in masked_three + assert "151234567" not in masked_three +