| """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}") |
|
|
