Add a validation layer for YAML based configuration

presidio-analyzer/presidio_analyzer/analyzer_engine_provider.py

@@ -36,7 +36,7 @@ def __init__(
     def get_configuration(
         self, conf_file: Optional[Union[Path, str]]
     ) -> Union[Dict[str, Any]]:
-        """Retrieve the analyzer engine configuration from the provided file."""
+        """Retrieve analyzer engine configuration from the provided file."""
 
         if not conf_file:
             default_conf_file = self._get_full_conf_path()
@@ -59,10 +59,18 @@ def get_configuration(
                 with open(self._get_full_conf_path()) as file:
                     configuration = yaml.safe_load(file)
             except Exception:
-                print(f"Failed to parse file {conf_file}, resorting to default")
+                logger.warning(
+                    f"Failed to parse file {conf_file}, resorting to default"
+                )
                 with open(self._get_full_conf_path()) as file:
                     configuration = yaml.safe_load(file)
 
+        # Validate configuration using Pydantic-based ConfigurationValidator
+        from presidio_analyzer.input_validation import ConfigurationValidator
+
+        ConfigurationValidator.validate_analyzer_configuration(configuration)
+        logger.debug("Analyzer configuration validation passed")
+
         return configuration
 
     def create_engine(self) -> AnalyzerEngine:

presidio-analyzer/presidio_analyzer/analyzer_request.py

@@ -37,5 +37,6 @@ def __init__(self, req_data: Dict):
         self.context = req_data.get("context")
         self.allow_list = req_data.get("allow_list")
         self.allow_list_match = req_data.get("allow_list_match", "exact")
-        self.regex_flags = req_data.get("regex_flags",
-                                        re.DOTALL | re.MULTILINE | re.IGNORECASE)
+        self.regex_flags = req_data.get(
+            "regex_flags", re.DOTALL | re.MULTILINE | re.IGNORECASE
+        )

presidio-analyzer/presidio_analyzer/input_validation/__init__.py

@@ -0,0 +1,19 @@
+"""Configuration validation module for Presidio."""
+
+from .schemas import ConfigurationValidator
+from .yaml_recognizer_models import (
+    BaseRecognizerConfig,
+    CustomRecognizerConfig,
+    LanguageContextConfig,
+    PredefinedRecognizerConfig,
+    RecognizerRegistryConfig,
+)
+
+__all__ = [
+    "ConfigurationValidator",
+    "BaseRecognizerConfig",
+    "CustomRecognizerConfig",
+    "LanguageContextConfig",
+    "PredefinedRecognizerConfig",
+    "RecognizerRegistryConfig",
+]

presidio-analyzer/presidio_analyzer/input_validation/schemas.py

@@ -0,0 +1,143 @@
+import re
+from pathlib import Path
+from typing import Any, Dict, List, Union
+
+from pydantic import ValidationError
+
+from .yaml_recognizer_models import RecognizerRegistryConfig
+
+
+class ConfigurationValidator:
+    """Class for validating configurations using Pydantic-enabled classes."""
+
+    @staticmethod
+    def validate_language_codes(languages: List[str]) -> List[str]:
+        """Validate language codes format.
+
+        :param languages: List of languages to validate.
+        """
+        for lang in languages:
+            if not re.match(r"^[a-z]{2}(-[A-Z]{2})?$", lang):
+                raise ValueError(
+                    f"Invalid language code format: {lang}. "
+                    f"Expected format: 'en' or 'en-US'"
+                )
+        return languages
+
+    @staticmethod
+    def validate_file_path(file_path: Union[str, Path]) -> Path:
+        """Validate file path exists and is readable.
+
+        :param file_path: Path to validate.
+        """
+        path = Path(file_path)
+        if not path.exists():
+            raise ValueError(f"Configuration file does not exist: {path}")
+        if not path.is_file():
+            raise ValueError(f"Path is not a file: {path}")
+        return path
+
+    @staticmethod
+    def validate_score_threshold(threshold: float) -> float:
+        """Validate score threshold is within valid range.
+
+        :param threshold: score threshold to validate.
+        """
+        if not 0.0 <= threshold <= 1.0:
+            raise ValueError(
+                f"Score threshold must be between 0.0 and 1.0, got: {threshold}"
+            )
+        return threshold
+
+    @staticmethod
+    def validate_nlp_configuration(config: Dict[str, Any]) -> Dict[str, Any]:
+        """Validate NLP configuration structure.
+
+        :param config: NLP Configuration to validate.
+        """
+        if not isinstance(config, dict):
+            raise ValueError("NLP configuration must be a dictionary")
+
+        required_fields = ["nlp_engine_name", "models"]
+        missing_fields = [field for field in required_fields if field not in config]
+        if missing_fields:
+            raise ValueError(
+                f"NLP configuration missing required fields: {missing_fields}"
+            )
+
+        # Validate models structure
+        if not isinstance(config["models"], list) or not config["models"]:
+            raise ValueError("Models must be a non-empty list")
+
+        for model in config["models"]:
+            if not isinstance(model, dict):
+                raise ValueError("Each model must be a dictionary")
+            if "lang_code" not in model or "model_name" not in model:
+                raise ValueError("Each model must have 'lang_code' and 'model_name'")
+
+        return config
+
+    @staticmethod
+    def validate_recognizer_registry_configuration(
+        config: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        """Validate recognizer registry configuration using Pydantic models."""
+        try:
+            # Use Pydantic model for validation
+            validated_config = RecognizerRegistryConfig(**config)
+            # Use model_dump() without exclude_unset to include default values
+            return validated_config.model_dump(exclude_unset=False)
+        except ValidationError as e:
+            raise ValueError(f"Invalid recognizer registry configuration: {e}")
+        except ImportError:
+            # Fallback to basic validation if models not available
+            return ConfigurationValidator._validate_recognizer_registry_basic(config)
+
+    @staticmethod
+    def _validate_recognizer_registry_basic(config: Dict[str, Any]) -> Dict[str, Any]:
+        """Validate recognizer registry config."""
+        if not isinstance(config, dict):
+            raise ValueError("Recognizer registry configuration must be a dictionary")
+
+        # Validate supported languages
+        if "supported_languages" in config:
+            ConfigurationValidator.validate_language_codes(
+                config["supported_languages"]
+            )
+
+        # Validate recognizers list
+        if "recognizers" in config and not isinstance(config["recognizers"], list):
+            raise ValueError("Recognizers must be a list")
+
+        return config
+
+    @staticmethod
+    def validate_analyzer_configuration(config: Dict[str, Any]) -> Dict[str, Any]:
+        """Validate analyzer engine validation."""
+        if not isinstance(config, dict):
+            raise ValueError("Analyzer configuration must be a dictionary")
+
+        # Validate supported languages if present
+        if "supported_languages" in config:
+            ConfigurationValidator.validate_language_codes(
+                config["supported_languages"]
+            )
+
+        # Validate score threshold if present
+        if "default_score_threshold" in config:
+            ConfigurationValidator.validate_score_threshold(
+                config["default_score_threshold"]
+            )
+
+        # Validate nested configurations
+        if "nlp_configuration" in config:
+            ConfigurationValidator.validate_nlp_configuration(
+                config["nlp_configuration"]
+            )
+
+        if "recognizer_registry" in config:
+            ConfigurationValidator.validate_recognizer_registry_configuration(
+                config["recognizer_registry"]
+            )
+
+        return config