🛡️ Input Validation¶

def __post_init__(self) -> None:
    """Post-initialization for Default-valuee."""
    if self.errors is None:
        self.errors = []
    if self.warnings is None:
        self.warnings = []
    if self.metadata is None:
        self.metadata = {}

def add_error(
    self, message: str, severity: ValidationSeverity = ValidationSeverity.ERROR
) -> None:
    """Fügt Valitherungsfehler hinto.

    Args:
        message: errormeldung
        severity: Schweregrad of the errors
    """
    self.errors.append(message)
    self.valid = False

    # Log securitysereignis on kritischen errorn
    if severity == ValidationSeverity.CRITICAL:
        _logger.log_security_event(
            event_type="input_validation_critical",
            severity="high",
            description=message,
            validation_error=message,
        )

def add_warning(self, message: str) -> None:
    """Fügt Valitherungswarnung hinto.

    Args:
        message: Warnmeldung
    """
    self.warnings.append(message)

@abstractmethod
def validate(self, value: Any) -> ValidationResult:
    """Validates Input-value.

    Args:
        value: To valitherenthe value

    Returns:
        Valitherungsergebnis
    """
    pass

def validate(self, value: Any) -> ValidationResult:
    """Validates string-Input."""
    # Prüfe Required
    required_result = self._check_required(value)
    if required_result:
        return required_result

    # Konvertiere to string
    if value is None:
        str_value = ""
    else:
        str_value = str(value)

    result = ValidationResult(valid=True, sanitized_value=str_value)

    # Längen-Valitherung
    if self.min_length is not None and len(str_value) < self.min_length:
        result.add_error(f"string to kurz: {len(str_value)} < {self.min_length}")

    if self.max_length is not None and len(str_value) > self.max_length:
        result.add_error(f"string to latg: {len(str_value)} > {self.max_length}")

    # Pattern-Valitherung
    if self.pattern and not self.pattern.match(str_value):
        result.add_error(
            f"string entspricht not the Pattern: {self.pattern.pattern}"
        )

    # Erlaubte Zeichen
    if self.allowed_chars:
        invalid_chars = set(str_value) - set(self.allowed_chars)
        if invalid_chars:
            result.add_error(f"Unerlaubte Zeichen: {invalid_chars}")

    # Verbotene Patterns
    for forbidden_pattern in self.forbidden_patterns:
        if forbidden_pattern.search(str_value):
            result.add_error(
                f"Verbotenes Pattern gefunden: {forbidden_pattern.pattern}",
                ValidationSeverity.CRITICAL,
            )

    # Satitization
    sanitized = str_value

    if self.satitize_html:
        sanitized = html.escape(sanitized)
        if sanitized != str_value:
            result.add_warning("HTML-Zeichen wurthe escaped")

    if self.satitize_sql:
        # Afache SQL-Injection-Prävention
        sql_patterns = [
            r"(\b(SELECT|INSERT|UPDATE|DELETE|DROP|CREATE|ALTER|EXEC|UNION)\b)",
            r"(--|#|/\*|\*/)",
            r"(\b(OR|AND)\s+\d+\s*=\s*\d+)",
            r"(\'\s*(OR|AND)\s+\'\w+\'\s*=\s*\'\w+\')",
        ]

        for pattern in sql_patterns:
            if re.search(pattern, sanitized, re.IGNORECASE):
                result.add_error(
                    "Potenzielle SQL-Injection erkatnt", ValidationSeverity.CRITICAL
                )
                break

    result.sanitized_value = sanitized
    return result

def validate(self, value: Any) -> ValidationResult:
    """Validates naroattheischen Input."""
    # Prüfe Required
    required_result = self._check_required(value)
    if required_result:
        return required_result

    result = ValidationResult(valid=True, sanitized_value=value)

    # Konvertiere to Zahl
    try:
        if self.integer_only:
            naroatd_value = int(value)
        else:
            naroatd_value = float(value)
    except (ValueError, TypeError):
        result.add_error(f"Ungültiger naroattheischer value: {value}")
        return result

    result.sanitized_value = naroatd_value

    # Bereichs-Valitherung
    if self.min_value is not None and naroatd_value < self.min_value:
        result.add_error(f"value to kla: {naroatd_value} < {self.min_value}")

    if self.max_value is not None and naroatd_value > self.max_value:
        result.add_error(f"value to groß: {naroatd_value} > {self.max_value}")

    # Positive-Only Valitherung
    if self.positive_only and naroatd_value <= 0:
        result.add_error(f"value must positiv sa: {naroatd_value}")

    return result

def validate(self, value: Any) -> ValidationResult:
    """Validates JSON-Input."""
    # Prüfe Required
    required_result = self._check_required(value)
    if required_result:
        return required_result

    result = ValidationResult(valid=True, sanitized_value=value)

    # Konvertiere string to JSON
    if isinstance(value, str):
        # Größen-Check
        if len(value.encode("utf-8")) > self.max_size:
            result.add_error(f"JSON to groß: {len(value)} > {self.max_size} Bytes")
            return result

        try:
            json_value = json.loads(value)
        except json.JSONDecodeError as e:
            result.add_error(f"Ungültiges JSON: {str(e)}")
            return result
    else:
        json_value = value

    # Tiefe-Valitherung
    depth = self._calculate_depth(json_value)
    if depth > self.max_depth:
        result.add_error(f"JSON to tief verschachtelt: {depth} > {self.max_depth}")

    # type-Valitherung
    if not self._validate_types(json_value):
        result.add_error("JSON enthält unerlaubte typeen")

    result.sanitized_value = json_value
    return result

def add_field(self, field_name: str, validator: BaseValidator) -> None:
    """Fügt Feld-Validator hinto.

    Args:
        field_name: Name of the Felof the
        validator: Validator for the Feld
    """
    self.field_validators[field_name] = validator

def validate(self, data: Dict[str, Any]) -> ValidationResult:
    """Validates komplexes object."""
    if not isinstance(data, dict):
        result = ValidationResult(valid=False, sanitized_value=data)
        result.add_error("Input must a dictionary sa")
        return result

    result = ValidationResult(valid=True, sanitized_value={})
    all_valid = True

    # Valithere jeof the Feld
    for field_name, validator in self.field_validators.items():
        field_value = data.get(field_name)
        field_result = validator.validate(field_value)

        if not field_result.valid:
            all_valid = False
            for error in field_result.errors:
                result.add_error(f"{field_name}: {error}")

        for warning in field_result.warnings:
            result.add_warning(f"{field_name}: {warning}")

        result.sanitized_value[field_name] = field_result.sanitized_value

    # Prüfe on unbekatnte Felthe
    unknown_fields = set(data.keys()) - set(self.field_validators.keys())
    if unknown_fields:
        result.add_warning(f"Unbekatnte Felthe ignoriert: {unknown_fields}")

    result.valid = all_valid
    return result

def register_validator(self, name: str, validator: BaseValidator) -> None:
    """Regisers Validator.

    Args:
        name: Name of the Validators
        validator: Validator-instatce
    """
    self.validators[name] = validator

def validate(self, validator_name: str, value: Any) -> ValidationResult:
    """Validates value with specificm Validator.

    Args:
        validator_name: Name of the Validators
        value: To valitherenthe value

    Returns:
        Valitherungsergebnis

    Raises:
        ValidationError: If Validator not gefatthe
    """
    if validator_name not in self.validators:
        raise ValidationError(f"Validator not gefatthe: {validator_name}")

    validator = self.validators[validator_name]
    result = validator.validate(value)

    # Log Valitherungsergebnis
    _logger.debug(
        f"Input-Valitherung: {validator_name}",
        validator=validator_name,
        valid=result.valid,
        errors=len(result.errors),
        warnings=len(result.warnings),
    )

    return result

def validate_agent_operation(
    self, operation: str, data: Dict[str, Any]
) -> ValidationResult:
    """Validates agent operation-Input.

    Args:
        operation: operation name
        data: operation data

    Returns:
        Valitherungsergebnis
    """
    # Statdard-Valitherung for agent operationen
    composite = CompositeValidator(f"agent_operation_{operation}")

    # Basis-Felthe
    composite.add_field(
        "operation",
        stringValidator(
            "operation",
            min_length=1,
            max_length=100,
            pattern=r"^[a-zA-Z_][a-zA-Z0-9_]*$",
        ),
    )

    # operation-specific Valitherung
    if operation == "plat":
        composite.add_field(
            "objective",
            stringValidator(
                "objective",
                min_length=1,
                max_length=1000,
                satitize_html=True,
                satitize_sql=True,
            ),
        )
        composite.add_field(
            "context", JSONValidator("context", required=False, max_depth=5)
        )
    elif operation == "act":
        composite.add_field(
            "action", stringValidator("action", min_length=1, max_length=100)
        )
        composite.add_field(
            "parameters", JSONValidator("parameters", required=False, max_depth=5)
        )

    return composite.validate(data)