{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/mcp-servers\/mcp_db\/tools\/delete_tool.py",
"content": "\"\"\"Delete Tool - Loescht Datensaetze.\"\"\"\n\nimport re\nimport time\nfrom datetime import datetime\n\nfrom constants import LOG_ENTRY_MAX_LENGTH, LOG_QUERY_MAX_LENGTH, MS_PER_SECOND\n\nfrom config import Config\nfrom infrastructure.db_connection import DatabaseConnection\nfrom shared.domain import LogEntry\nfrom shared.infrastructure import get_logger\n\n# Sicherheits-Default fuer LIMIT\nDEFAULT_DELETE_LIMIT = 100\n\n\ndef _validate_identifier(name: str) -> bool:\n \"\"\"Validiert Tabellen-\/Spaltennamen gegen SQL-Injection.\"\"\"\n return bool(re.match(r\"^[a-zA-Z0-9_]+$\", name))\n\n\ndef register_delete_tool(mcp) -> None:\n \"\"\"Registriert db_delete Tool.\"\"\"\n logger = get_logger(\"mcp-db\", Config)\n\n @mcp.tool()\n def db_delete(\n table: str,\n where: dict,\n database: str = \"ki_dev\",\n limit: int | None = None,\n confirm_delete_all: bool = False,\n ) -> dict:\n \"\"\"\n Loescht Datensaetze.\n\n Args:\n table: Zieltabelle\n where: Dict mit Spalte:Wert Paaren (WHERE-Klausel) - PFLICHT!\n Spezial: {\"id\": \">0\"} oder {\"1\": \"1\"} fuer alle Zeilen (mit confirm_delete_all=True)\n database: Zieldatenbank (ki_dev oder ki_content)\n limit: Maximale Anzahl zu loeschender Zeilen (Default: 100)\n confirm_delete_all: True = erlaubt Loeschung aller Zeilen\n\n Returns:\n Dict mit status, deleted_rows, error\n \"\"\"\n start = time.time()\n\n # Validierung: Tabellenname\n if not _validate_identifier(table):\n return {\n \"status\": \"denied\",\n \"error\": \"Invalid table name.\",\n }\n\n # Validierung: Datenbank\n if database not in Config.ALLOWED_DATABASES:\n return {\n \"status\": \"denied\",\n \"error\": f\"Database '{database}' not allowed.\",\n }\n\n # Spezialfall: Alle loeschen mit confirm_delete_all\n delete_all_mode = False\n if where and confirm_delete_all:\n # Check for special patterns like {\"1\": \"1\"} or {\"id\": \">0\"}\n if where == {\"1\": \"1\"} or (len(where) == 1 and list(where.values())[0] == \">0\"):\n delete_all_mode = True\n\n # KRITISCH: WHERE ist Pflicht (ausser delete_all_mode)\n if not where and not delete_all_mode:\n return {\n \"status\": \"denied\",\n \"error\": \"WHERE clause is required. DELETE without WHERE is forbidden.\",\n }\n\n # Validierung: Spaltennamen in where (nicht bei delete_all_mode)\n if not delete_all_mode:\n for col in where:\n if not _validate_identifier(col):\n return {\n \"status\": \"denied\",\n \"error\": f\"Invalid column name in where: {col}\",\n }\n\n # Default LIMIT als Sicherheit (kein Limit bei delete_all_mode)\n if delete_all_mode:\n effective_limit = None\n else:\n effective_limit = limit if limit is not None else DEFAULT_DELETE_LIMIT\n if effective_limit < 1:\n effective_limit = DEFAULT_DELETE_LIMIT\n\n try:\n with DatabaseConnection.get_connection(database) as conn:\n cursor = conn.cursor(buffered=True)\n\n # Build query based on mode\n if delete_all_mode:\n query = f\"DELETE FROM `{table}`\"\n values = ()\n else:\n # Build WHERE clause\n where_parts = [f\"`{col}` = %s\" for col in where]\n where_clause = \" AND \".join(where_parts)\n values = tuple(where.values())\n query = f\"DELETE FROM `{table}` WHERE {where_clause} LIMIT {effective_limit}\"\n\n cursor.execute(query, values)\n deleted_rows = cursor.rowcount\n conn.commit()\n cursor.close()\n\n duration = int((time.time() - start) * MS_PER_SECOND)\n\n # Log\n log_msg = f\"DELETE FROM {table}\" + (\" (ALL)\" if delete_all_mode else f\" LIMIT {effective_limit}\")\n try:\n logger.log(\n LogEntry(\n timestamp=datetime.now(),\n client_name=\"mcp-db\",\n tool_name=\"db_delete\",\n request=log_msg,\n status=\"success\",\n duration_ms=duration,\n )\n )\n except Exception:\n pass\n\n result = {\n \"status\": \"success\",\n \"table\": table,\n \"deleted_rows\": deleted_rows,\n \"execution_ms\": duration,\n }\n if effective_limit:\n result[\"limit_applied\"] = effective_limit\n if delete_all_mode:\n result[\"delete_all\"] = True\n return result\n\n except Exception as e:\n duration = int((time.time() - start) * MS_PER_SECOND)\n\n try:\n logger.log(\n LogEntry(\n timestamp=datetime.now(),\n client_name=\"mcp-db\",\n tool_name=\"db_delete\",\n request=f\"DELETE FROM {table}\",\n status=\"error\",\n duration_ms=duration,\n error_message=str(e)[:LOG_ENTRY_MAX_LENGTH],\n )\n )\n except Exception:\n pass\n\n return {\n \"status\": \"error\",\n \"error\": str(e)[:LOG_QUERY_MAX_LENGTH],\n \"execution_ms\": duration,\n }\n"
}
}
{
"tool_response": {
"type": "update",
"filePath": "\/var\/www\/mcp-servers\/mcp_db\/tools\/delete_tool.py",
"content": "\"\"\"Delete Tool - Loescht Datensaetze.\"\"\"\n\nimport re\nimport time\nfrom datetime import datetime\n\nfrom constants import LOG_ENTRY_MAX_LENGTH, LOG_QUERY_MAX_LENGTH, MS_PER_SECOND\n\nfrom config import Config\nfrom infrastructure.db_connection import DatabaseConnection\nfrom shared.domain import LogEntry\nfrom shared.infrastructure import get_logger\n\n# Sicherheits-Default fuer LIMIT\nDEFAULT_DELETE_LIMIT = 100\n\n\ndef _validate_identifier(name: str) -> bool:\n \"\"\"Validiert Tabellen-\/Spaltennamen gegen SQL-Injection.\"\"\"\n return bool(re.match(r\"^[a-zA-Z0-9_]+$\", name))\n\n\ndef register_delete_tool(mcp) -> None:\n \"\"\"Registriert db_delete Tool.\"\"\"\n logger = get_logger(\"mcp-db\", Config)\n\n @mcp.tool()\n def db_delete(\n table: str,\n where: dict,\n database: str = \"ki_dev\",\n limit: int | None = None,\n confirm_delete_all: bool = False,\n ) -> dict:\n \"\"\"\n Loescht Datensaetze.\n\n Args:\n table: Zieltabelle\n where: Dict mit Spalte:Wert Paaren (WHERE-Klausel) - PFLICHT!\n Spezial: {\"id\": \">0\"} oder {\"1\": \"1\"} fuer alle Zeilen (mit confirm_delete_all=True)\n database: Zieldatenbank (ki_dev oder ki_content)\n limit: Maximale Anzahl zu loeschender Zeilen (Default: 100)\n confirm_delete_all: True = erlaubt Loeschung aller Zeilen\n\n Returns:\n Dict mit status, deleted_rows, error\n \"\"\"\n start = time.time()\n\n # Validierung: Tabellenname\n if not _validate_identifier(table):\n return {\n \"status\": \"denied\",\n \"error\": \"Invalid table name.\",\n }\n\n # Validierung: Datenbank\n if database not in Config.ALLOWED_DATABASES:\n return {\n \"status\": \"denied\",\n \"error\": f\"Database '{database}' not allowed.\",\n }\n\n # Spezialfall: Alle loeschen mit confirm_delete_all\n delete_all_mode = False\n if where and confirm_delete_all:\n # Check for special patterns like {\"1\": \"1\"} or {\"id\": \">0\"}\n if where == {\"1\": \"1\"} or (len(where) == 1 and list(where.values())[0] == \">0\"):\n delete_all_mode = True\n\n # KRITISCH: WHERE ist Pflicht (ausser delete_all_mode)\n if not where and not delete_all_mode:\n return {\n \"status\": \"denied\",\n \"error\": \"WHERE clause is required. DELETE without WHERE is forbidden.\",\n }\n\n # Validierung: Spaltennamen in where (nicht bei delete_all_mode)\n if not delete_all_mode:\n for col in where:\n if not _validate_identifier(col):\n return {\n \"status\": \"denied\",\n \"error\": f\"Invalid column name in where: {col}\",\n }\n\n # Default LIMIT als Sicherheit (kein Limit bei delete_all_mode)\n if delete_all_mode:\n effective_limit = None\n else:\n effective_limit = limit if limit is not None else DEFAULT_DELETE_LIMIT\n if effective_limit < 1:\n effective_limit = DEFAULT_DELETE_LIMIT\n\n try:\n with DatabaseConnection.get_connection(database) as conn:\n cursor = conn.cursor(buffered=True)\n\n # Build query based on mode\n if delete_all_mode:\n query = f\"DELETE FROM `{table}`\"\n values = ()\n else:\n # Build WHERE clause\n where_parts = [f\"`{col}` = %s\" for col in where]\n where_clause = \" AND \".join(where_parts)\n values = tuple(where.values())\n query = f\"DELETE FROM `{table}` WHERE {where_clause} LIMIT {effective_limit}\"\n\n cursor.execute(query, values)\n deleted_rows = cursor.rowcount\n conn.commit()\n cursor.close()\n\n duration = int((time.time() - start) * MS_PER_SECOND)\n\n # Log\n log_msg = f\"DELETE FROM {table}\" + (\" (ALL)\" if delete_all_mode else f\" LIMIT {effective_limit}\")\n try:\n logger.log(\n LogEntry(\n timestamp=datetime.now(),\n client_name=\"mcp-db\",\n tool_name=\"db_delete\",\n request=log_msg,\n status=\"success\",\n duration_ms=duration,\n )\n )\n except Exception:\n pass\n\n result = {\n \"status\": \"success\",\n \"table\": table,\n \"deleted_rows\": deleted_rows,\n \"execution_ms\": duration,\n }\n if effective_limit:\n result[\"limit_applied\"] = effective_limit\n if delete_all_mode:\n result[\"delete_all\"] = True\n return result\n\n except Exception as e:\n duration = int((time.time() - start) * MS_PER_SECOND)\n\n try:\n logger.log(\n LogEntry(\n timestamp=datetime.now(),\n client_name=\"mcp-db\",\n tool_name=\"db_delete\",\n request=f\"DELETE FROM {table}\",\n status=\"error\",\n duration_ms=duration,\n error_message=str(e)[:LOG_ENTRY_MAX_LENGTH],\n )\n )\n except Exception:\n pass\n\n return {\n \"status\": \"error\",\n \"error\": str(e)[:LOG_QUERY_MAX_LENGTH],\n \"execution_ms\": duration,\n }\n",
"structuredPatch": [
{
"oldStart": 1,
"oldLines": 19,
"newStart": 1,
"newLines": 17,
"lines": [
"-\"\"\"Delete Tool - Löscht Datensätze.\"\"\"",
"+\"\"\"Delete Tool - Loescht Datensaetze.\"\"\"",
"+",
" import re",
"-import sys",
" import time",
"+from datetime import datetime",
" ",
"-sys.path.insert(0, \"\/var\/www\/mcp-servers\/mcp_db\")",
"-sys.path.insert(0, \"\/var\/www\/mcp-servers\/shared\")",
"-",
" from constants import LOG_ENTRY_MAX_LENGTH, LOG_QUERY_MAX_LENGTH, MS_PER_SECOND",
" ",
" from config import Config",
"-from domain.log_contract import LogEntry",
" from infrastructure.db_connection import DatabaseConnection",
"-from infrastructure.protokoll_logger import ProtokollLogger",
"+from shared.domain import LogEntry",
"+from shared.infrastructure import get_logger",
" ",
"-# Sicherheits-Default für LIMIT",
"+# Sicherheits-Default fuer LIMIT",
" DEFAULT_DELETE_LIMIT = 100",
" ",
" "
]
},
{
"oldStart": 24,
"oldLines": 7,
"newStart": 22,
"newLines": 7,
"lines": [
" ",
" def register_delete_tool(mcp) -> None:",
" \"\"\"Registriert db_delete Tool.\"\"\"",
"- logger = ProtokollLogger()",
"+ logger = get_logger(\"mcp-db\", Config)",
" ",
" @mcp.tool()",
" def db_delete("
]
},
{
"oldStart": 35,
"oldLines": 15,
"newStart": 33,
"newLines": 15,
"lines": [
" confirm_delete_all: bool = False,",
" ) -> dict:",
" \"\"\"",
"- Löscht Datensätze.",
"+ Loescht Datensaetze.",
" ",
" Args:",
" table: Zieltabelle",
" where: Dict mit Spalte:Wert Paaren (WHERE-Klausel) - PFLICHT!",
"- Spezial: {\"id\": \">0\"} oder {\"1\": \"1\"} für alle Zeilen (mit confirm_delete_all=True)",
"+ Spezial: {\"id\": \">0\"} oder {\"1\": \"1\"} fuer alle Zeilen (mit confirm_delete_all=True)",
" database: Zieldatenbank (ki_dev oder ki_content)",
"- limit: Maximale Anzahl zu löschender Zeilen (Default: 100)",
"- confirm_delete_all: True = erlaubt Löschung aller Zeilen",
"+ limit: Maximale Anzahl zu loeschender Zeilen (Default: 100)",
"+ confirm_delete_all: True = erlaubt Loeschung aller Zeilen",
" ",
" Returns:",
" Dict mit status, deleted_rows, error"
]
},
{
"oldStart": 64,
"oldLines": 14,
"newStart": 62,
"newLines": 14,
"lines": [
" \"error\": f\"Database '{database}' not allowed.\",",
" }",
" ",
"- # Spezialfall: Alle löschen mit confirm_delete_all",
"+ # Spezialfall: Alle loeschen mit confirm_delete_all",
" delete_all_mode = False",
" if where and confirm_delete_all:",
" # Check for special patterns like {\"1\": \"1\"} or {\"id\": \">0\"}",
" if where == {\"1\": \"1\"} or (len(where) == 1 and list(where.values())[0] == \">0\"):",
" delete_all_mode = True",
" ",
"- # KRITISCH: WHERE ist Pflicht (außer delete_all_mode)",
"+ # KRITISCH: WHERE ist Pflicht (ausser delete_all_mode)",
" if not where and not delete_all_mode:",
" return {",
" \"status\": \"denied\","
]
},
{
"oldStart": 122,
"oldLines": 6,
"newStart": 120,
"newLines": 9,
"lines": [
" try:",
" logger.log(",
" LogEntry(",
"+ timestamp=datetime.now(),",
"+ client_name=\"mcp-db\",",
"+ tool_name=\"db_delete\",",
" request=log_msg,",
" status=\"success\",",
" duration_ms=duration,"
]
},
{
"oldStart": 148,
"oldLines": 6,
"newStart": 149,
"newLines": 9,
"lines": [
" try:",
" logger.log(",
" LogEntry(",
"+ timestamp=datetime.now(),",
"+ client_name=\"mcp-db\",",
"+ tool_name=\"db_delete\",",
" request=f\"DELETE FROM {table}\",",
" status=\"error\",",
" duration_ms=duration,"
]
}
],
"originalFile": "\"\"\"Delete Tool - Löscht Datensätze.\"\"\"\nimport re\nimport sys\nimport time\n\nsys.path.insert(0, \"\/var\/www\/mcp-servers\/mcp_db\")\nsys.path.insert(0, \"\/var\/www\/mcp-servers\/shared\")\n\nfrom constants import LOG_ENTRY_MAX_LENGTH, LOG_QUERY_MAX_LENGTH, MS_PER_SECOND\n\nfrom config import Config\nfrom domain.log_contract import LogEntry\nfrom infrastructure.db_connection import DatabaseConnection\nfrom infrastructure.protokoll_logger import ProtokollLogger\n\n# Sicherheits-Default für LIMIT\nDEFAULT_DELETE_LIMIT = 100\n\n\ndef _validate_identifier(name: str) -> bool:\n \"\"\"Validiert Tabellen-\/Spaltennamen gegen SQL-Injection.\"\"\"\n return bool(re.match(r\"^[a-zA-Z0-9_]+$\", name))\n\n\ndef register_delete_tool(mcp) -> None:\n \"\"\"Registriert db_delete Tool.\"\"\"\n logger = ProtokollLogger()\n\n @mcp.tool()\n def db_delete(\n table: str,\n where: dict,\n database: str = \"ki_dev\",\n limit: int | None = None,\n confirm_delete_all: bool = False,\n ) -> dict:\n \"\"\"\n Löscht Datensätze.\n\n Args:\n table: Zieltabelle\n where: Dict mit Spalte:Wert Paaren (WHERE-Klausel) - PFLICHT!\n Spezial: {\"id\": \">0\"} oder {\"1\": \"1\"} für alle Zeilen (mit confirm_delete_all=True)\n database: Zieldatenbank (ki_dev oder ki_content)\n limit: Maximale Anzahl zu löschender Zeilen (Default: 100)\n confirm_delete_all: True = erlaubt Löschung aller Zeilen\n\n Returns:\n Dict mit status, deleted_rows, error\n \"\"\"\n start = time.time()\n\n # Validierung: Tabellenname\n if not _validate_identifier(table):\n return {\n \"status\": \"denied\",\n \"error\": \"Invalid table name.\",\n }\n\n # Validierung: Datenbank\n if database not in Config.ALLOWED_DATABASES:\n return {\n \"status\": \"denied\",\n \"error\": f\"Database '{database}' not allowed.\",\n }\n\n # Spezialfall: Alle löschen mit confirm_delete_all\n delete_all_mode = False\n if where and confirm_delete_all:\n # Check for special patterns like {\"1\": \"1\"} or {\"id\": \">0\"}\n if where == {\"1\": \"1\"} or (len(where) == 1 and list(where.values())[0] == \">0\"):\n delete_all_mode = True\n\n # KRITISCH: WHERE ist Pflicht (außer delete_all_mode)\n if not where and not delete_all_mode:\n return {\n \"status\": \"denied\",\n \"error\": \"WHERE clause is required. DELETE without WHERE is forbidden.\",\n }\n\n # Validierung: Spaltennamen in where (nicht bei delete_all_mode)\n if not delete_all_mode:\n for col in where:\n if not _validate_identifier(col):\n return {\n \"status\": \"denied\",\n \"error\": f\"Invalid column name in where: {col}\",\n }\n\n # Default LIMIT als Sicherheit (kein Limit bei delete_all_mode)\n if delete_all_mode:\n effective_limit = None\n else:\n effective_limit = limit if limit is not None else DEFAULT_DELETE_LIMIT\n if effective_limit < 1:\n effective_limit = DEFAULT_DELETE_LIMIT\n\n try:\n with DatabaseConnection.get_connection(database) as conn:\n cursor = conn.cursor(buffered=True)\n\n # Build query based on mode\n if delete_all_mode:\n query = f\"DELETE FROM `{table}`\"\n values = ()\n else:\n # Build WHERE clause\n where_parts = [f\"`{col}` = %s\" for col in where]\n where_clause = \" AND \".join(where_parts)\n values = tuple(where.values())\n query = f\"DELETE FROM `{table}` WHERE {where_clause} LIMIT {effective_limit}\"\n\n cursor.execute(query, values)\n deleted_rows = cursor.rowcount\n conn.commit()\n cursor.close()\n\n duration = int((time.time() - start) * MS_PER_SECOND)\n\n # Log\n log_msg = f\"DELETE FROM {table}\" + (\" (ALL)\" if delete_all_mode else f\" LIMIT {effective_limit}\")\n try:\n logger.log(\n LogEntry(\n request=log_msg,\n status=\"success\",\n duration_ms=duration,\n )\n )\n except Exception:\n pass\n\n result = {\n \"status\": \"success\",\n \"table\": table,\n \"deleted_rows\": deleted_rows,\n \"execution_ms\": duration,\n }\n if effective_limit:\n result[\"limit_applied\"] = effective_limit\n if delete_all_mode:\n result[\"delete_all\"] = True\n return result\n\n except Exception as e:\n duration = int((time.time() - start) * MS_PER_SECOND)\n\n try:\n logger.log(\n LogEntry(\n request=f\"DELETE FROM {table}\",\n status=\"error\",\n duration_ms=duration,\n error_message=str(e)[:LOG_ENTRY_MAX_LENGTH],\n )\n )\n except Exception:\n pass\n\n return {\n \"status\": \"error\",\n \"error\": str(e)[:LOG_QUERY_MAX_LENGTH],\n \"execution_ms\": duration,\n }\n"
}
}