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 55746e5e527..4ae3f6b35fe 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 @@ -329,10 +329,10 @@ class ContentFilterGuardrail(CustomGuardrail): action if action else category_config_obj.default_action ) - # Handle conditional categories (with identifier_words + inherit_from) - if ( - category_config_obj.identifier_words - and category_config_obj.inherit_from + # Handle conditional categories (with identifier_words + block words) + if category_config_obj.identifier_words and ( + category_config_obj.inherit_from + or category_config_obj.additional_block_words ): self._load_conditional_category( category_name, @@ -387,51 +387,55 @@ class ContentFilterGuardrail(CustomGuardrail): categories_dir: str, ) -> None: """ - Load a conditional category that uses identifier_words + inherited block_words. + Load a conditional category that uses identifier_words + block_words. + Block words can come from inherited category or additional_block_words. Args: category_name: Name of the category - category_config_obj: CategoryConfig object with identifier_words and inherit_from + category_config_obj: CategoryConfig object with identifier_words category_action: Action to take when match is found severity_threshold: Minimum severity threshold categories_dir: Directory containing category files """ - # Load the inherited category to get block words - inherit_from = category_config_obj.inherit_from - if not inherit_from: - return - - # Remove .json or .yaml extension if included - inherit_base = inherit_from.replace(".json", "").replace(".yaml", "") - - # Find the inherited category file - inherit_yaml_path = os.path.join(categories_dir, f"{inherit_base}.yaml") - inherit_json_path = os.path.join(categories_dir, f"{inherit_base}.json") - - if os.path.exists(inherit_yaml_path): - inherit_file_path = inherit_yaml_path - elif os.path.exists(inherit_json_path): - inherit_file_path = inherit_json_path - else: - verbose_proxy_logger.warning( - f"Category {category_name}: inherit_from '{inherit_from}' file not found at {categories_dir}" - ) - verbose_proxy_logger.debug( - f"Tried paths: {inherit_yaml_path}, {inherit_json_path}" - ) - return - try: - # Load the inherited category - inherited_category = self._load_category_file(inherit_file_path) - - # Extract block words from inherited category that meet severity threshold block_words = [] - for keyword_data in inherited_category.keywords: - keyword = keyword_data["keyword"].lower() - severity = keyword_data["severity"] - if self._should_apply_severity(severity, severity_threshold): - block_words.append(keyword) + inherit_from = category_config_obj.inherit_from + + # Load inherited block words if specified + if inherit_from: + # Remove .json or .yaml extension if included + inherit_base = inherit_from.replace(".json", "").replace(".yaml", "") + + # Find the inherited category file + inherit_yaml_path = os.path.join(categories_dir, f"{inherit_base}.yaml") + inherit_json_path = os.path.join(categories_dir, f"{inherit_base}.json") + + inherit_file_path = None + if os.path.exists(inherit_yaml_path): + inherit_file_path = inherit_yaml_path + elif os.path.exists(inherit_json_path): + inherit_file_path = inherit_json_path + else: + verbose_proxy_logger.warning( + f"Category {category_name}: inherit_from '{inherit_from}' file not found at {categories_dir}" + ) + verbose_proxy_logger.debug( + f"Tried paths: {inherit_yaml_path}, {inherit_json_path}" + ) + + if inherit_file_path: + # Load the inherited category + inherited_category = self._load_category_file(inherit_file_path) + + # Extract block words from inherited category that meet severity threshold + for keyword_data in inherited_category.keywords: + keyword = keyword_data["keyword"].lower() + severity = keyword_data["severity"] + if self._should_apply_severity(severity, severity_threshold): + block_words.append(keyword) + else: + # If inherit file not found, set inherit_from to None for logging + inherit_from = None # Add additional block words specific to this category if category_config_obj.additional_block_words: @@ -445,16 +449,29 @@ class ContentFilterGuardrail(CustomGuardrail): "severity": "high", # Combinations are always high severity } - verbose_proxy_logger.info( + # Build log message + log_msg = ( f"Loaded conditional category {category_name}: " f"{len(category_config_obj.identifier_words)} identifiers + " - f"{len(block_words)} block words " - f"({len(category_config_obj.additional_block_words)} additional + " - f"{len(block_words) - len(category_config_obj.additional_block_words)} from {inherit_from})" + f"{len(block_words)} block words" ) + if inherit_from and category_config_obj.additional_block_words: + inherited_count = len(block_words) - len( + category_config_obj.additional_block_words + ) + log_msg += ( + f" ({len(category_config_obj.additional_block_words)} additional + " + f"{inherited_count} from {inherit_from})" + ) + elif inherit_from: + log_msg += f" (from {inherit_from})" + elif category_config_obj.additional_block_words: + log_msg += f" ({len(block_words)} from additional_block_words)" + + verbose_proxy_logger.info(log_msg) except Exception as e: verbose_proxy_logger.error( - f"Error loading inherited category for {category_name}: {e}" + f"Error loading conditional category for {category_name}: {e}" ) def _load_category_file(self, file_path: str) -> CategoryConfig: diff --git a/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/policy_templates/eu_ai_act_article5_fr.yaml b/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/policy_templates/eu_ai_act_article5_fr.yaml new file mode 100644 index 00000000000..7c7c8aab4bf --- /dev/null +++ b/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/policy_templates/eu_ai_act_article5_fr.yaml @@ -0,0 +1,242 @@ +# Loi sur l'IA (UE) Article 5 - Détection des pratiques interdites +# Utilise une logique conditionnelle : BLOQUER si un mot identificateur + un mot de blocage apparaissent ensemble +# Référence : https://artificialintelligenceact.eu/article/5/ +category_name: "eu_ai_act_article5_prohibited_practices_fr" +description: "Détecte les pratiques interdites de l'Article 5 de la loi sur l'IA de l'UE (français)" +default_action: "BLOCK" + +# MOTS IDENTIFICATEURS - Actions qui pourraient créer des systèmes interdits +identifier_words: + # Actions de développement + - "construire" + - "créer" + - "développer" + - "concevoir" + - "implémenter" + - "mettre en place" + - "établir" + - "bâtir" + - "élaborer" + + # Actions de notation/évaluation + - "noter" + - "classer" + - "évaluer" + - "estimer" + - "classifier" + - "profiler" + - "coter" + - "juger" + - "attribuer une note" + - "attribuer un score" + - "donner une note" + - "donner un score" + + # Actions de détection + - "reconnaître" + - "détecter" + - "identifier" + - "inférer" + - "déduire" + - "prédire" + - "déterminer" + - "analyser" + - "repérer" + - "déceler" + - "mesurer" + - "surveiller" + - "monitorer" + +# MOTS DE BLOCAGE SUPPLÉMENTAIRES - Contextes interdits +additional_block_words: + # Notation sociale (Article 5.1.c) + - "comportement social" + - "crédit social" + - "score social" + - "note sociale" + - "notation sociale" + - "fiabilité" + - "score de citoyen" + - "note de citoyen" + - "score de réputation" + - "note de réputation" + - "score comportemental" + - "note comportementale" + - "comportement sur les réseaux sociaux" + - "comportement médias sociaux" + - "réputation sociale" + + # Reconnaissance d'émotions au travail/éducation (Article 5.1.f) + - "émotion des employés" + - "émotion des salariés" + - "émotion du personnel" + - "émotion au travail" + - "émotion des travailleurs" + - "émotion des étudiants" + - "émotion des élèves" + - "émotion en classe" + - "sentiment des employés" + - "sentiment des salariés" + - "sentiment des travailleurs" + - "sentiment des étudiants" + - "sentiment des élèves" + - "humeur des employés" + - "humeur des salariés" + - "humeur des travailleurs" + - "état émotionnel employés" + - "état émotionnel salariés" + - "ressenti des employés" + - "ressenti des salariés" + + # Catégorisation biométrique (Article 5.1.h) + - "race à partir du visage" + - "ethnie à partir du visage" + - "race à partir de biométrie" + - "race à partir du biométrique" + - "ethnie à partir de biométrie" + - "ethnie à partir du biométrique" + - "opinions politiques à partir" + - "vues politiques à partir" + - "orientation sexuelle à partir" + - "religion à partir de biométrie" + - "religion à partir du biométrique" + - "croyances religieuses à partir" + + # Profilage prédictif + - "prédiction de crime" + - "prédiction criminelle" + - "comportement criminel" + - "récidive" + - "prédiction de la récidive" + - "prédiction de performance des employés" + - "prédiction de performance des salariés" + - "solvabilité à partir des réseaux sociaux" + - "solvabilité à partir du social" + + # Manipulation (Article 5.1.a) + - "subliminal" + - "subliminale" + - "subconscient" + - "inconscient" + - "manipulation de comportement" + - "manipulation comportementale" + + # Exploitation de vulnérabilités (Article 5.1.b) + - "vulnérabilité des enfants" + - "vulnérabilité des personnes âgées" + - "vulnérabilité des handicapés" + - "vulnérabilité des personnes handicapées" + +# TOUJOURS BLOQUER - Pratiques interdites explicites +always_block_keywords: + # Systèmes de notation sociale + - keyword: "système de crédit social" + severity: "high" + - keyword: "système de notation sociale" + severity: "high" + - keyword: "système de score social" + severity: "high" + - keyword: "notation des citoyens" + severity: "high" + - keyword: "système de notation des citoyens" + severity: "high" + - keyword: "système pour noter le comportement social" + severity: "high" + - keyword: "système pour évaluer le comportement social" + severity: "high" + - keyword: "système pour classer le comportement social" + severity: "high" + + # Reconnaissance d'émotions au travail/éducation + - keyword: "reconnaissance d'émotions au travail" + severity: "high" + - keyword: "détection d'émotions des employés" + severity: "high" + - keyword: "détection d'émotions des salariés" + severity: "high" + - keyword: "reconnaissance d'émotions en classe" + severity: "high" + - keyword: "détection d'émotions des étudiants" + severity: "high" + - keyword: "détection d'émotions des élèves" + severity: "high" + + # Catégorisation biométrique + - keyword: "inférer la race à partir du visage" + severity: "high" + - keyword: "prédire la race à partir du visage" + severity: "high" + - keyword: "déduire la race à partir du visage" + severity: "high" + - keyword: "inférer l'ethnie à partir de biométrie" + severity: "high" + - keyword: "prédire l'ethnie à partir de biométrie" + severity: "high" + - keyword: "prédire les opinions politiques à partir" + severity: "high" + - keyword: "inférer les opinions politiques à partir" + severity: "high" + - keyword: "système de catégorisation biométrique" + severity: "high" + + # Profilage prédictif + - keyword: "police prédictive" + severity: "high" + - keyword: "algorithme de prédiction de crime" + severity: "high" + - keyword: "algorithme de prédiction criminelle" + severity: "high" + - keyword: "prédiction de récidive" + severity: "high" + +# EXCEPTIONS - Cas d'utilisation légitimes +exceptions: + # Recherche et éducation + - "recherche sur" + - "étude sur" + - "académique" + - "thèse sur" + - "mémoire sur" + - "recherche scientifique" + - "travail de recherche" + + # Surveillance de conformité + - "audit de biais" + - "audit des biais" + - "détecter la discrimination" + - "détection de discrimination" + - "surveillance de conformité" + - "contrôle de conformité" + - "examen éthique" + - "révision éthique" + - "test d'équité" + - "test de justice" + + # Contextes de divertissement/produits + - "dans un film" + - "dans le film" + - "pour un film" + - "cinéma" + - "dans un jeu" + - "dans le jeu" + - "jeu vidéo" + - "avis sur le produit" + - "avis produit" + - "retour client" + - "commentaires clients" + - "feedback client" + + # Méta-discussion + - "expliquer" + - "qu'est-ce que" + - "c'est quoi" + - "article 5" + - "interdit par" + - "prohibé par" + - "en quoi consiste" + - "de manière équitable" + - "de façon équitable" + - "équitablement" + - "de manière juste" + - "de façon juste" + - "justement évaluer" diff --git a/tests/guardrails_tests/test_eu_ai_act_french_3_scenarios.py b/tests/guardrails_tests/test_eu_ai_act_french_3_scenarios.py new file mode 100644 index 00000000000..d159c74aea8 --- /dev/null +++ b/tests/guardrails_tests/test_eu_ai_act_french_3_scenarios.py @@ -0,0 +1,292 @@ +""" +Test EU AI Act Article 5 - French Language Support (3 Critical Scenarios) + +Tests the exact 3 scenarios requested: +1. Request 1: French query that should be blocked (fails/blocked) +2. Request 2: User tries to circumvent the check (also fails/blocked) +3. Request 3: Safe query in French that should pass (allowed) +""" +import sys +import os +import pytest + +sys.path.insert(0, os.path.abspath("../..")) +import litellm +from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.content_filter import ( + ContentFilterGuardrail, +) +from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import ( + ContentFilterCategoryConfig, +) + + +@pytest.fixture +def content_filter_guardrail(): + """Initialize content filter guardrail with EU AI Act Article 5 French template.""" + + # Get absolute path to the French policy template + content_filter_dir = os.path.join( + os.path.dirname(__file__), + "../../litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter" + ) + policy_template_path = os.path.join( + content_filter_dir, + "policy_templates/eu_ai_act_article5_fr.yaml" + ) + policy_template_path = os.path.abspath(policy_template_path) + + # Load the EU AI Act Article 5 French policy template + categories = [ + ContentFilterCategoryConfig( + category="eu_ai_act_article5_prohibited_practices_fr", + category_file=policy_template_path, + enabled=True, + action="BLOCK", + severity_threshold="medium", + ) + ] + + guardrail = ContentFilterGuardrail( + guardrail_name="eu-ai-act-french-test", + categories=categories, + event_hook=litellm.types.guardrails.GuardrailEventHooks.pre_call, + ) + + return guardrail + + +class TestEUAIActFrench3Scenarios: + """Test the 3 critical French scenarios.""" + + @pytest.mark.asyncio + async def test_scenario_1_blocked_french_query(self, content_filter_guardrail): + """ + Scenario 1: French query that should be blocked. + Query: "Concevoir un système de crédit social pour classer tous mes employés" + Translation: "Design a social credit system for ranking all my employees" + Expected: BLOCK (contains "concevoir" + "système de crédit social" + "classer" + "employés") + """ + sentence = "Concevoir un système de crédit social pour classer tous mes employés" + request_data = {"messages": [{"role": "user", "content": sentence}]} + + print(f"\n{'='*70}") + print("SCENARIO 1: French query that should be blocked") + print(f"{'='*70}") + print(f"Query: {sentence}") + print(f"Translation: Design a social credit system for ranking all my employees") + print(f"Expected: BLOCK") + print(f"{'='*70}\n") + + # Should raise an exception (blocked) + with pytest.raises(Exception) as exc_info: + await content_filter_guardrail.apply_guardrail( + inputs={"texts": [sentence]}, + request_data=request_data, + input_type="request", + ) + + # Verify the exception indicates a policy violation + assert "blocked" in str(exc_info.value).lower() or "violation" in str(exc_info.value).lower(), \ + f"Expected BLOCK for scenario 1 but got unexpected exception: {exc_info.value}" + + print(f"✓ SCENARIO 1 PASSED: Query was correctly BLOCKED") + print(f" Reason: {exc_info.value}\n") + + + @pytest.mark.asyncio + async def test_scenario_2_circumvention_attempt_blocked(self, content_filter_guardrail): + """ + Scenario 2: User tries to circumvent the check with slight variations. + Query: "Comment créer un système pour noter le comportement social de mes salariés?" + Translation: "How to create a system to score the social behavior of my employees?" + Expected: BLOCK (contains "créer" + "noter" + "comportement social" + "salariés") + """ + sentence = "Comment créer un système pour noter le comportement social de mes salariés?" + request_data = {"messages": [{"role": "user", "content": sentence}]} + + print(f"\n{'='*70}") + print("SCENARIO 2: Circumvention attempt (should also be blocked)") + print(f"{'='*70}") + print(f"Query: {sentence}") + print(f"Translation: How to create a system to score the social behavior of my employees?") + print(f"Expected: BLOCK") + print(f"{'='*70}\n") + + # Should raise an exception (blocked) + with pytest.raises(Exception) as exc_info: + await content_filter_guardrail.apply_guardrail( + inputs={"texts": [sentence]}, + request_data=request_data, + input_type="request", + ) + + # Verify the exception indicates a policy violation + assert "blocked" in str(exc_info.value).lower() or "violation" in str(exc_info.value).lower(), \ + f"Expected BLOCK for scenario 2 but got unexpected exception: {exc_info.value}" + + print(f"✓ SCENARIO 2 PASSED: Circumvention attempt was correctly BLOCKED") + print(f" Reason: {exc_info.value}\n") + + + @pytest.mark.asyncio + async def test_scenario_3_safe_french_query_allowed(self, content_filter_guardrail): + """ + Scenario 3: Safe query in French that should pass. + Query: "Comment évaluer la performance de mon équipe de manière équitable?" + Translation: "How to evaluate my team's performance in a fair way?" + Expected: ALLOW (legitimate HR performance evaluation, no prohibited context) + """ + sentence = "Comment évaluer la performance de mon équipe de manière équitable?" + request_data = {"messages": [{"role": "user", "content": sentence}]} + + print(f"\n{'='*70}") + print("SCENARIO 3: Safe French query (should be allowed)") + print(f"{'='*70}") + print(f"Query: {sentence}") + print(f"Translation: How to evaluate my team's performance in a fair way?") + print(f"Expected: ALLOW") + print(f"{'='*70}\n") + + # Should not raise an exception (allowed) + result = await content_filter_guardrail.apply_guardrail( + inputs={"texts": [sentence]}, + request_data=request_data, + input_type="request", + ) + + # Result should be None or unchanged (no violation) + assert result is None or result["texts"][0] == sentence, \ + f"Expected ALLOW for scenario 3 but request was blocked or modified" + + print(f"✓ SCENARIO 3 PASSED: Safe query was correctly ALLOWED\n") + + + @pytest.mark.asyncio + async def test_pure_conditional_matching(self, content_filter_guardrail): + """ + Test pure conditional matching (identifier + block word, not in always_block_keywords). + This ensures conditional matching works independently of always_block_keywords. + """ + # This sentence contains "analyser" (identifier) + "sentiment des salariés" (block word) + # but doesn't match any always_block_keywords + sentence = "Développer un outil pour analyser le sentiment des salariés en continu" + request_data = {"messages": [{"role": "user", "content": sentence}]} + + print(f"\n{'='*70}") + print("PURE CONDITIONAL MATCHING TEST") + print(f"{'='*70}") + print(f"Query: {sentence}") + print(f"Translation: Develop a tool to analyze employee sentiment continuously") + print(f"Expected: BLOCK (conditional match: analyser + sentiment des salariés)") + print(f"{'='*70}\n") + + # Should raise an exception (blocked by conditional matching) + with pytest.raises(Exception) as exc_info: + await content_filter_guardrail.apply_guardrail( + inputs={"texts": [sentence]}, + request_data=request_data, + input_type="request", + ) + + # Verify it's a conditional match, not an always_block match + error_detail = getattr(exc_info.value, 'detail', {}) + error_msg = error_detail.get("error", str(exc_info.value)) if isinstance(error_detail, dict) else str(exc_info.value) + assert "conditional match" in error_msg.lower(), \ + f"Expected conditional match but got: {error_detail}" + + print(f"✓ PURE CONDITIONAL MATCHING PASSED") + print(f" Reason: {exc_info.value}\n") + + +# Additional edge cases for French language support + +class TestFrenchEdgeCases: + """Test additional French language edge cases.""" + + @pytest.mark.asyncio + async def test_mixed_french_english(self, content_filter_guardrail): + """Test mixed French and English query.""" + sentence = "Build a système de crédit social for employees" + request_data = {"messages": [{"role": "user", "content": sentence}]} + + # Should block (contains "build" and "système de crédit social") + with pytest.raises(Exception): + await content_filter_guardrail.apply_guardrail( + inputs={"texts": [sentence]}, + request_data=request_data, + input_type="request", + ) + + + @pytest.mark.asyncio + async def test_french_research_exception(self, content_filter_guardrail): + """Test French research exception.""" + sentence = "Je fais une recherche sur les systèmes de crédit social en Chine" + request_data = {"messages": [{"role": "user", "content": sentence}]} + + # Should allow (contains "recherche sur" exception) + result = await content_filter_guardrail.apply_guardrail( + inputs={"texts": [sentence]}, + request_data=request_data, + input_type="request", + ) + + assert result is None or result["texts"][0] == sentence + + + @pytest.mark.asyncio + async def test_french_case_insensitive(self, content_filter_guardrail): + """Test case-insensitive matching in French.""" + sentence = "CONCEVOIR UN SYSTÈME DE CRÉDIT SOCIAL" + request_data = {"messages": [{"role": "user", "content": sentence}]} + + # Should block (case-insensitive) + with pytest.raises(Exception): + await content_filter_guardrail.apply_guardrail( + inputs={"texts": [sentence]}, + request_data=request_data, + input_type="request", + ) + + + @pytest.mark.asyncio + async def test_exception_bypass_prevention(self, content_filter_guardrail): + """ + Test that short exception words don't create bypasses. + Words like "enjeu" (stake) should not match "jeu" (game) exception. + """ + # "enjeu" contains "jeu" but should NOT trigger exception + sentence = "Créer un système de crédit social pour l'enjeu principal de l'entreprise" + request_data = {"messages": [{"role": "user", "content": sentence}]} + + # Should still block (no exception bypass) + with pytest.raises(Exception) as exc_info: + await content_filter_guardrail.apply_guardrail( + inputs={"texts": [sentence]}, + request_data=request_data, + input_type="request", + ) + + # Verify it was blocked + assert "blocked" in str(exc_info.value).lower() + + + @pytest.mark.asyncio + async def test_legitimate_game_context_allowed(self, content_filter_guardrail): + """Test that legitimate game context with proper phrasing is allowed.""" + sentence = "Détecter les émotions des joueurs dans un jeu vidéo" + request_data = {"messages": [{"role": "user", "content": sentence}]} + + # Should allow (contains "dans un jeu" exception with proper context) + result = await content_filter_guardrail.apply_guardrail( + inputs={"texts": [sentence]}, + request_data=request_data, + input_type="request", + ) + + assert result is None or result["texts"][0] == sentence + + +if __name__ == "__main__": + # Run tests with: pytest test_eu_ai_act_french_3_scenarios.py -v -s + pytest.main([__file__, "-v", "-s"])