centralized input validation with 8 functions: validate_isin, validate_exchange, validate_date, validate_int_range, sanitize_sql_string, validate_isin_list, validate_exchange_list, validate_table_name

src/utils/validation.py

@@ -0,0 +1,124 @@
+"""
+Zentrale Validierungs- und Sanitisierungsfunktionen fuer SQL-Queries.
+
+Da QuestDB's HTTP API keine parametrisierten Queries unterstuetzt,
+muessen alle Werte vor der Interpolation validiert/bereinigt werden.
+"""
+
+import re
+import datetime
+
+
+# Gueltige Exchange-Namen (alle registrierten Boersen)
+VALID_EXCHANGES = {
+    "EIX", "LS", "XETRA", "FRA", "QUOTRIX", "GETTEX", "STU",
+    "DUSA", "DUSB", "DUSC", "DUSD",
+    "HAMA", "HAMB", "HANA", "HANB",
+    "NONE",  # Platzhalter fuer leere Analytics-Eintraege
+}
+
+# Gueltige Analytics-Tabellennamen
+VALID_TABLES = {
+    "analytics_custom", "analytics_exchange_daily",
+    "analytics_daily_summary", "analytics_volume_changes",
+    "analytics_stock_trends", "trades", "metadata",
+}
+
+# ISIN-Format: 2 Buchstaben Laendercode + 9 alphanumerische Zeichen + 1 Pruefziffer
+_ISIN_PATTERN = re.compile(r'^[A-Z]{2}[A-Z0-9]{9}[0-9]$')
+
+# Datumsformat: YYYY-MM-DD
+_DATE_PATTERN = re.compile(r'^\d{4}-\d{2}-\d{2}$')
+
+# Gefaehrliche SQL-Fragmente
+_SQL_DANGEROUS = re.compile(r'(--|/\*|\*/|;)')
+
+
+def validate_isin(value: str) -> str:
+    """Validiert einen ISIN-Wert. Wirft ValueError bei ungueltigem Format."""
+    if not isinstance(value, str):
+        raise ValueError(f"ISIN muss ein String sein, nicht {type(value).__name__}")
+    value = value.strip().upper()
+    if not _ISIN_PATTERN.match(value):
+        raise ValueError(f"Ungueltiges ISIN-Format: {value!r}")
+    return value
+
+
+def validate_exchange(value: str) -> str:
+    """Validiert einen Exchange-Namen gegen die Whitelist."""
+    if not isinstance(value, str):
+        raise ValueError(f"Exchange muss ein String sein, nicht {type(value).__name__}")
+    value = value.strip().upper()
+    if value not in VALID_EXCHANGES:
+        raise ValueError(f"Unbekannte Exchange: {value!r}")
+    return value
+
+
+def validate_date(value: str) -> str:
+    """Validiert ein Datum im Format YYYY-MM-DD."""
+    if not isinstance(value, str):
+        raise ValueError(f"Datum muss ein String sein, nicht {type(value).__name__}")
+    value = value.strip()
+    if not _DATE_PATTERN.match(value):
+        raise ValueError(f"Ungueltiges Datumsformat: {value!r} (erwartet: YYYY-MM-DD)")
+    # Pruefe ob es ein gueltiges Datum ist
+    try:
+        datetime.date.fromisoformat(value)
+    except ValueError:
+        raise ValueError(f"Ungueltiges Datum: {value!r}")
+    return value
+
+
+def validate_int_range(value: int, min_val: int = 0, max_val: int = 10000) -> int:
+    """Validiert einen Integer-Wert innerhalb eines Bereichs."""
+    value = int(value)
+    if value < min_val or value > max_val:
+        raise ValueError(f"Wert {value} ausserhalb des erlaubten Bereichs [{min_val}, {max_val}]")
+    return value
+
+
+def sanitize_sql_string(value: str, max_length: int = 200) -> str:
+    """
+    Bereinigt einen String fuer die Verwendung in SQL-Queries.
+    Escapet Single-Quotes und entfernt gefaehrliche SQL-Fragmente.
+    """
+    if not isinstance(value, str):
+        raise ValueError(f"Wert muss ein String sein, nicht {type(value).__name__}")
+    value = value[:max_length]
+    # Single-Quotes escapen (SQL-Standard)
+    value = value.replace("'", "''")
+    # Gefaehrliche SQL-Fragmente entfernen
+    value = _SQL_DANGEROUS.sub('', value)
+    # Backslashes entfernen
+    value = value.replace('\\', '')
+    return value
+
+
+def validate_isin_list(value: str) -> list:
+    """Validiert eine komma-separierte Liste von ISINs."""
+    if not isinstance(value, str):
+        raise ValueError("ISIN-Liste muss ein String sein")
+    items = [item.strip() for item in value.split(",") if item.strip()]
+    if not items:
+        raise ValueError("ISIN-Liste ist leer")
+    return [validate_isin(item) for item in items]
+
+
+def validate_exchange_list(value: str) -> list:
+    """Validiert eine komma-separierte Liste von Exchange-Namen."""
+    if not isinstance(value, str):
+        raise ValueError("Exchange-Liste muss ein String sein")
+    items = [item.strip() for item in value.split(",") if item.strip()]
+    if not items:
+        raise ValueError("Exchange-Liste ist leer")
+    return [validate_exchange(item) for item in items]
+
+
+def validate_table_name(value: str) -> str:
+    """Validiert einen Tabellennamen gegen die Whitelist."""
+    if not isinstance(value, str):
+        raise ValueError(f"Tabellenname muss ein String sein, nicht {type(value).__name__}")
+    value = value.strip().lower()
+    if value not in VALID_TABLES:
+        raise ValueError(f"Unbekannter Tabellenname: {value!r}")
+    return value