bifroest/services/db-mcp/dbmcp/core/sql.py
2026-10-27 08:34:17 +01:00

23 lines
918 B
Python

"""Prüfung der Abfragen vor der Ausführung (erste von mehreren Sicherungen).
Genau eine lesende Anweisung. Die eigentliche Sicherung sind ein DB-Nutzer mit
reinen Leserechten und eine schreibgeschützte Transaktion im Treiber.
"""
import re
from dbmcp.core import Invalid
COMMENTS = re.compile(r"^\s*(?:--[^\n]*(?:\n|$)|/\*.*?\*/)", re.S)
def check_query(sql: str, starts: tuple[str, ...] = ("select", "with"), max_chars: int = 10000) -> str:
if len(sql) > max_chars:
raise Invalid(f"Abfrage ist länger als {max_chars} Zeichen.")
rest = sql
while (match := COMMENTS.match(rest)) is not None:
rest = rest[match.end():]
first = re.match(r"\s*([A-Za-z]+)", rest)
if first is None or first.group(1).lower() not in starts:
raise Invalid("Erlaubt ist genau eine lesende Abfrage (" + ", ".join(s.upper() for s in starts) + ").")
return sql.strip().rstrip(";").strip()