File size: 1,433 Bytes
98aa813 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 | """Validation SQL en lecture seule (règles Rag_crag.md)."""
from __future__ import annotations
import re
FORBIDDEN_SQL = {
"INSERT", "UPDATE", "DELETE", "DROP",
"ALTER", "TRUNCATE", "CREATE", "EXEC",
"EXECUTE", "GRANT", "REVOKE", "MERGE",
"ATTACH", "DETACH", "PRAGMA", "VACUUM",
"REPLACE", "INTO",
}
ALLOWED_TABLES = {"sales"}
ALLOWED_COLUMNS = {
"client", "produit", "region", "mois", "trimestre",
"chiffre_affaires", "marge",
}
ALLOWED_AGGREGATIONS = {"SUM", "MAX", "MIN", "AVG", "COUNT"}
MAX_ROWS = 100
def is_safe_sql(query: str) -> bool:
"""Refuse tout SQL non-SELECT ou contenant un mot-clé interdit."""
normalized = query.upper().strip()
if not normalized.startswith("SELECT"):
return False
if ";" in normalized.rstrip(";"):
return False
tokens = set(re.findall(r"[A-Z_]+", normalized))
return not any(keyword in tokens for keyword in FORBIDDEN_SQL)
def assert_whitelisted_identifiers(
table: str,
columns: list[str],
aggregation: str | None = None,
) -> None:
if table not in ALLOWED_TABLES:
raise ValueError(f"Table non autorisée: {table}")
for col in columns:
if col not in ALLOWED_COLUMNS:
raise ValueError(f"Colonne non autorisée: {col}")
if aggregation is not None and aggregation.upper() not in ALLOWED_AGGREGATIONS:
raise ValueError(f"Agrégation non autorisée: {aggregation}")
|