Protokoll #3691

{
    "event": "PreToolUse",
    "tool_name": "Read",
    "tool_input": {
        "file_path": "\/opt\/mcp-servers\/mcp-db\/validators\/query_validator.py"
    }
}

{
    "tool_response": {
        "type": "text",
        "file": {
            "filePath": "\/opt\/mcp-servers\/mcp-db\/validators\/query_validator.py",
            "content": "\"\"\"Query Validator - SRP: Separate Validierungslogik\"\"\"\nimport re\nimport sys\nfrom typing import Tuple\n\nsys.path.insert(0, \"\/opt\/mcp-servers\/mcp-db\")\nfrom config import Config\n\n\nclass QueryValidator:\n    \"\"\"Validiert SQL Queries - SRP: Nur Validierung\"\"\"\n\n    @staticmethod\n    def validate_query(query: str, database: str, max_rows: int) -> Tuple[bool, str]:\n        \"\"\"\n        Validiert eine Query gegen alle Sicherheitsregeln.\n\n        Returns:\n            (is_valid, error_message)\n        \"\"\"\n        # Basis-Validierung\n        if not query or len(query) < 1:\n            return False, \"Query must not be empty\"\n\n        if len(query) > Config.MAX_QUERY_LENGTH:\n            return False, f\"Query must be max {Config.MAX_QUERY_LENGTH} chars\"\n\n        # Nur SELECT erlaubt\n        query_upper = query.strip().upper()\n        if not query_upper.startswith(\"SELECT\"):\n            return False, \"Only SELECT queries allowed\"\n\n        # Dangerous Keyword Blocklist\n        for keyword in Config.BLOCKED_KEYWORDS:\n            # Prüfe Keyword mit Word Boundaries\n            pattern = r\"\\b\" + re.escape(keyword) + r\"\\b\"\n            if re.search(pattern, query_upper):\n                return False, f\"Blocked keyword detected: {keyword}\"\n\n        # Database Allowlist\n        if database not in Config.ALLOWED_DATABASES:\n            return (\n                False,\n                f\"Database '{database}' not allowed. Allowed: {', '.join(Config.ALLOWED_DATABASES)}\",\n            )\n\n        # Max Rows prüfen\n        if max_rows < 1 or max_rows > Config.MAX_ROWS:\n            return False, f\"max_rows must be 1-{Config.MAX_ROWS}\"\n\n        # Table Allowlist (information_schema Schutz)\n        from_tables = QueryValidator._extract_table_names(query_upper)\n        for table in from_tables:\n            # Erlaube information_schema.TABLES für Schema-Tool\n            if \"INFORMATION_SCHEMA\" in table:\n                continue\n            # Prüfe gegen Allowlist\n            table_clean = table.split(\".\")[-1]\n            if table_clean not in [t.upper() for t in Config.ALLOWED_TABLES]:\n                return (\n                    False,\n                    f\"Table '{table}' not allowed. Allowed: {', '.join(Config.ALLOWED_TABLES)}\",\n                )\n\n        return True, \"\"\n\n    @staticmethod\n    def _extract_table_names(query_upper: str) -> list:\n        \"\"\"Extrahiert Tabellennamen aus FROM und JOIN Clauses\"\"\"\n        tables = []\n\n        # FROM table, JOIN table\n        from_match = re.findall(r\"\\bFROM\\s+([a-zA-Z0-9_\\.]+)\", query_upper)\n        join_match = re.findall(r\"\\bJOIN\\s+([a-zA-Z0-9_\\.]+)\", query_upper)\n\n        tables.extend(from_match)\n        tables.extend(join_match)\n\n        return tables\n",
            "numLines": 80,
            "startLine": 1,
            "totalLines": 80
        }
    }
}