{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/opt\/mcp-servers\/mcp-db\/validators\/execute_validator.py",
"content": "\"\"\"Execute Validator - Validierung für DDL\/DML Statements.\"\"\"\nimport re\nimport sys\n\nsys.path.insert(0, \"\/opt\/mcp-servers\/mcp-db\")\nfrom config import Config\n\n\nclass ExecuteValidator:\n \"\"\"Validiert DDL\/DML Statements gegen Sicherheitsregeln.\"\"\"\n\n # Erlaubte Statement-Typen\n ALLOWED_STATEMENTS: list[str] = [\"ALTER\", \"CREATE\", \"DROP\", \"TRUNCATE\"]\n\n # Verbotene Patterns (kritisch!)\n FORBIDDEN_PATTERNS: list[str] = [\n r\"\\bDROP\\s+DATABASE\\b\",\n r\"\\bDROP\\s+SCHEMA\\b\",\n r\"\\bCREATE\\s+DATABASE\\b\",\n r\"\\bCREATE\\s+SCHEMA\\b\",\n r\"\\bGRANT\\b\",\n r\"\\bREVOKE\\b\",\n r\"\\bSHUTDOWN\\b\",\n ]\n\n # Verbotene Datenbanken (System-DBs)\n FORBIDDEN_DATABASES: list[str] = [\n \"information_schema\",\n \"mysql\",\n \"performance_schema\",\n \"sys\",\n ]\n\n @classmethod\n def validate_statement(\n cls, statement: str, database: str\n ) -> tuple[bool, str]:\n \"\"\"\n Validiert ein DDL\/DML Statement.\n\n Args:\n statement: SQL Statement\n database: Zieldatenbank\n\n Returns:\n Tuple (is_valid, error_message)\n \"\"\"\n if not statement or len(statement.strip()) < 5:\n return False, \"Statement must not be empty\"\n\n if len(statement) > Config.MAX_QUERY_LENGTH:\n return False, f\"Statement exceeds max length ({Config.MAX_QUERY_LENGTH})\"\n\n statement_upper = statement.strip().upper()\n\n # Prüfe ob Statement mit erlaubtem Keyword beginnt\n statement_type = cls._get_statement_type(statement_upper)\n if statement_type is None:\n return (\n False,\n f\"Statement type not allowed. Allowed: {', '.join(cls.ALLOWED_STATEMENTS)}\",\n )\n\n # Prüfe verbotene Patterns\n for pattern in cls.FORBIDDEN_PATTERNS:\n if re.search(pattern, statement_upper):\n return False, f\"Forbidden pattern detected: {pattern}\"\n\n # Prüfe Datenbank\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 # Prüfe ob Statement gegen System-DB zielt\n for forbidden_db in cls.FORBIDDEN_DATABASES:\n if forbidden_db.upper() in statement_upper:\n return False, f\"Operations on '{forbidden_db}' are forbidden\"\n\n return True, \"\"\n\n @classmethod\n def _get_statement_type(cls, statement_upper: str) -> str | None:\n \"\"\"Extrahiert den Statement-Typ (ALTER, CREATE, etc.).\"\"\"\n for stmt_type in cls.ALLOWED_STATEMENTS:\n if statement_upper.startswith(stmt_type):\n return stmt_type\n return None\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/opt\/mcp-servers\/mcp-db\/validators\/execute_validator.py",
"content": "\"\"\"Execute Validator - Validierung für DDL\/DML Statements.\"\"\"\nimport re\nimport sys\n\nsys.path.insert(0, \"\/opt\/mcp-servers\/mcp-db\")\nfrom config import Config\n\n\nclass ExecuteValidator:\n \"\"\"Validiert DDL\/DML Statements gegen Sicherheitsregeln.\"\"\"\n\n # Erlaubte Statement-Typen\n ALLOWED_STATEMENTS: list[str] = [\"ALTER\", \"CREATE\", \"DROP\", \"TRUNCATE\"]\n\n # Verbotene Patterns (kritisch!)\n FORBIDDEN_PATTERNS: list[str] = [\n r\"\\bDROP\\s+DATABASE\\b\",\n r\"\\bDROP\\s+SCHEMA\\b\",\n r\"\\bCREATE\\s+DATABASE\\b\",\n r\"\\bCREATE\\s+SCHEMA\\b\",\n r\"\\bGRANT\\b\",\n r\"\\bREVOKE\\b\",\n r\"\\bSHUTDOWN\\b\",\n ]\n\n # Verbotene Datenbanken (System-DBs)\n FORBIDDEN_DATABASES: list[str] = [\n \"information_schema\",\n \"mysql\",\n \"performance_schema\",\n \"sys\",\n ]\n\n @classmethod\n def validate_statement(\n cls, statement: str, database: str\n ) -> tuple[bool, str]:\n \"\"\"\n Validiert ein DDL\/DML Statement.\n\n Args:\n statement: SQL Statement\n database: Zieldatenbank\n\n Returns:\n Tuple (is_valid, error_message)\n \"\"\"\n if not statement or len(statement.strip()) < 5:\n return False, \"Statement must not be empty\"\n\n if len(statement) > Config.MAX_QUERY_LENGTH:\n return False, f\"Statement exceeds max length ({Config.MAX_QUERY_LENGTH})\"\n\n statement_upper = statement.strip().upper()\n\n # Prüfe ob Statement mit erlaubtem Keyword beginnt\n statement_type = cls._get_statement_type(statement_upper)\n if statement_type is None:\n return (\n False,\n f\"Statement type not allowed. Allowed: {', '.join(cls.ALLOWED_STATEMENTS)}\",\n )\n\n # Prüfe verbotene Patterns\n for pattern in cls.FORBIDDEN_PATTERNS:\n if re.search(pattern, statement_upper):\n return False, f\"Forbidden pattern detected: {pattern}\"\n\n # Prüfe Datenbank\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 # Prüfe ob Statement gegen System-DB zielt\n for forbidden_db in cls.FORBIDDEN_DATABASES:\n if forbidden_db.upper() in statement_upper:\n return False, f\"Operations on '{forbidden_db}' are forbidden\"\n\n return True, \"\"\n\n @classmethod\n def _get_statement_type(cls, statement_upper: str) -> str | None:\n \"\"\"Extrahiert den Statement-Typ (ALTER, CREATE, etc.).\"\"\"\n for stmt_type in cls.ALLOWED_STATEMENTS:\n if statement_upper.startswith(stmt_type):\n return stmt_type\n return None\n",
"structuredPatch": [],
"originalFile": null
}
}