fix(sécurité): durcir les accès fichiers et les appels bq

back/build_query/schema_fetcher.py

@@ -14,6 +14,20 @@ def validate_bq_ref(ref: str) -> bool:
     return len(parts) >= 2 and all(_BQ_IDENT_RE.match(p) for p in parts)
 
 
+def _require_bq_component(value: str, label: str) -> str:
+    """Refuse tout composant destiné à la ligne de commande `bq` (projet, dataset,
+    table, billing_project) qui contient autre chose qu'un identifiant BigQuery.
+
+    Le wrapper `bq` est un `.cmd` sous Windows → ses arguments sont re-parsés par
+    cmd.exe même sans ``shell=True`` (classe « BatBadBut »). La validation stricte
+    au point de construction de la commande est donc le vrai garde-fou anti-injection,
+    en complément de ``validate_bq_ref`` côté API (défense en profondeur, audit 2026-07).
+    """
+    if not isinstance(value, str) or not _BQ_IDENT_RE.match(value):
+        raise ValueError(f"Invalid BigQuery {label} (illegal characters): {value!r}")
+    return value
+
+
 def parse_ref(ref: str, billing_project: str) -> tuple[str, str, str]:
     parts = ref.split(".")
     if len(parts) == 2:
@@ -179,21 +193,37 @@ def _is_auth_error(exc: Exception) -> bool:
 async def _run_bq_cli(
     cmd: list[str], billing_project: str, timeout: int = _CLI_TIMEOUT
 ) -> str:
-    """Run a bq CLI command and return stdout, raising on auth errors or non-zero exit."""
+    """Run a bq CLI command and return stdout, raising on auth errors or non-zero exit.
+
+    L'exécutable (``cmd[0]``) est résolu explicitement via ``shutil.which`` et lancé
+    sans ``shell=True`` : plus aucune interprétation de la ligne par cmd.exe/sh, donc
+    les métacaractères d'un argument ne peuvent plus enchaîner une commande (audit
+    sécu 2026-07). Les composants (projet/dataset/table) sont par ailleurs validés
+    par l'appelant (``_require_bq_component``).
+    """
+    import shutil
     import subprocess
 
+    resolved = shutil.which(cmd[0])
+    if not resolved:
+        raise RuntimeError(
+            f"'{cmd[0]}' introuvable dans le PATH — installer le Google Cloud SDK "
+            "(bq) ou utiliser l'API BigQuery."
+        )
+    argv = [resolved, *cmd[1:]]
+
     extra: dict = {}
     if os.name == "nt":
         extra["creationflags"] = subprocess.CREATE_NO_WINDOW
     env = os.environ.copy()
     env["CLOUDSDK_CORE_DISABLE_PROMPTS"] = "1"
     try:
         result = await asyncio.to_thread(
-            lambda c=cmd: subprocess.run(
+            lambda c=argv: subprocess.run(  # noqa: S603 exe résolu (shutil.which), composants validés
                 c,
                 capture_output=True,
                 text=True,
-                shell=True,
+                shell=False,
                 stdin=subprocess.DEVNULL,
                 timeout=timeout,
                 env=env,
@@ -219,6 +249,10 @@ async def _fetch_table_via_cli(
     ``bq show --schema`` so that timePartitioning info is available.
     """
     proj, dataset, table = parse_ref(ref, billing_project)
+    _require_bq_component(billing_project, "billing_project")
+    _require_bq_component(proj, "project")
+    _require_bq_component(dataset, "dataset")
+    _require_bq_component(table, "table")
     bq_ref = f"{proj}:{dataset}.{table}"
     cmd = [
         "bq",
@@ -276,6 +310,10 @@ async def _fetch_partition_values_cli(
 ) -> tuple[list[str], bool]:
     """Fetch the last *limit* partition IDs via ``bq query`` on INFORMATION_SCHEMA."""
     proj, dataset, table = parse_ref(ref, billing_project)
+    _require_bq_component(billing_project, "billing_project")
+    _require_bq_component(proj, "project")
+    _require_bq_component(dataset, "dataset")
+    _require_bq_component(table, "table")
     sql = (
         f"SELECT partition_id "
         f"FROM `{proj}.{dataset}.INFORMATION_SCHEMA.PARTITIONS` "

back/cli/main.py

@@ -1128,7 +1128,9 @@ def _open_when_ready() -> None:
 
         threading.Thread(target=_open_when_ready, daemon=True).start()
 
-    uvicorn.run(server_module, host="0.0.0.0", port=port)
+    # TODO(sécu audit 2026-07, finding #1) : binder 127.0.0.1 par défaut + token si
+    # exposition réseau voulue. Bind 0.0.0.0 conservé le temps de la décision produit.
+    uvicorn.run(server_module, host="0.0.0.0", port=port)  # noqa: S104
 
 
 def main() -> None:

back/pyproject.toml

@@ -80,6 +80,24 @@ pytest-asyncio = "^1.4.0"
 pytest-mock = "^3.15.1"
 httpx = "^0.28.1"
 
+[tool.ruff.lint]
+# Défauts ruff (E4/E7/E9 + F) + flake8-bandit (S) pour la sécurité. Les règles S
+# à fort bruit et inhérentes au domaine (asserts de tests, SQL en f-string au cœur
+# d'un outil de test SQL) sont désactivées ; on garde le signal utile : subprocess,
+# bind-all-interfaces, etc. Audit sécu 2026-07.
+select = ["E4", "E7", "E9", "F", "S"]
+ignore = [
+    "S101",  # assert — omniprésent et légitime (tests + gardes internes)
+    "S608",  # SQL en f-string — construction interne de requêtes DuckDB (par design)
+    "S110",  # try/except/pass — nettoyages best-effort assumés
+    "S112",  # try/except/continue — idem
+    "S311",  # random non-crypto — génération de données de test, pas de la crypto
+]
+
+[tool.ruff.lint.per-file-ignores]
+# Les tests utilisent assert + subprocess + données factices → S non pertinent.
+"tests/**" = ["S"]
+
 [tool.pytest.ini_options]
 asyncio_mode = "auto"
 testpaths = ["tests"]

back/server.py

@@ -101,16 +101,23 @@ async def root():
 async def serve_spa(full_path: str, request: Request):
     if request.url.path.startswith("/api"):
         raise HTTPException(status_code=404, detail="Not Found")
-    # si le fichier existe physiquement, on le renvoie
-    file_path = get_static_dir() / full_path
-    if file_path.is_file():
+    # si le fichier existe physiquement, on le renvoie — mais uniquement s'il reste
+    # sous le dossier statique (un `../` ne doit pas servir back/.env, audit sécu
+    # 2026-07). Tout chemin hors racine retombe sur l'index SPA.
+    from utils.path_guard import safe_join
+
+    static_dir = get_static_dir()
+    file_path = safe_join(static_dir, full_path)
+    if file_path is not None and file_path.is_file():
         return FileResponse(str(file_path))
     # sinon SPA
-    return FileResponse(str(get_static_dir() / "index.html"))
+    return FileResponse(str(static_dir / "index.html"))
 
 
 # ─── 9) Uvicorn – exécution directe ─────────────────
 if __name__ == "__main__":
     import uvicorn
 
-    uvicorn.run(app, host="0.0.0.0", port=int(os.getenv("PORT", 8080)), reload=False)
+    # TODO(sécu audit 2026-07, finding #1) : binder 127.0.0.1 par défaut + token si
+    # exposition réseau voulue. Bind 0.0.0.0 conservé le temps de la décision produit.
+    uvicorn.run(app, host="0.0.0.0", port=int(os.getenv("PORT", 8080)), reload=False)  # noqa: S104

back/storage/context_loader.py

@@ -1,6 +1,7 @@
 from pathlib import Path
 
 from storage.config import get_models_path
+from utils.path_guard import safe_join
 
 
 def load_model_context(model_name: str) -> str:
@@ -19,6 +20,11 @@ def load_model_context(model_name: str) -> str:
     models_path = get_models_path()
     parts = Path(model_name).parts  # e.g. ("finance", "revenue")
 
+    # Garde traversal : `model_name` vient de l'appelant HTTP — un `../` ne doit pas
+    # aspirer un `.md` hors de models_path (audit sécu 2026-07).
+    if safe_join(models_path, model_name, suffix=".md") is None:
+        return ""
+
     fragments: list[str] = []
 
     # Walk from models_path root down to the file's parent directory

back/storage/test_repository.py

@@ -15,6 +15,7 @@
     load_preprocessor_fn,
 )
 from storage.test_files import cache_path_for, read_test_doc, write_test_doc
+from utils.path_guard import safe_join
 
 _UUID_RE = re.compile(
     r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", re.I
@@ -23,6 +24,11 @@
 # Namespace fixe pour dériver un test_id stable d'un model_name (uuid5).
 _TEST_ID_NAMESPACE = uuid.UUID("6f9619ff-8b86-d011-b42d-00cf4fc964ff")
 
+# Un SHA de commit git = hex. Valider avant de l'injecter dans un argv git empêche
+# qu'un `source_sha` commençant par `-` soit interprété comme une option git (audit
+# sécu 2026-07).
+_GIT_SHA_RE = re.compile(r"^[0-9a-fA-F]{7,64}$")
+
 # Tests créés mais pas encore persistés (en attente de validation réussie).
 _pending_tests: Dict[str, Dict[str, Any]] = {}
 
@@ -39,12 +45,24 @@ def _tests_root() -> Path:
 
 
 def _test_path(model_name: str) -> Path:
-    """Single file per model: .mocksql/tests/{model_id}.json (supports nested paths)."""
-    p = _tests_root() / f"{model_name}.json"
+    """Single file per model: .mocksql/tests/{model_id}.json (supports nested paths).
+
+    ``model_name`` provient des endpoints HTTP → passer par ``safe_join`` pour qu'un
+    `../` ne fasse pas écrire/supprimer hors de la racine tests (audit sécu 2026-07).
+    """
+    p = safe_join(_tests_root(), model_name, suffix=".json")
+    if p is None:
+        raise ValueError(f"Invalid model name (path traversal): {model_name!r}")
     p.parent.mkdir(parents=True, exist_ok=True)
     return p
 
 
+def _model_sql_path(model_name: str) -> Optional[Path]:
+    """Chemin du fichier `.sql` d'un modèle sous ``models_path``, ou ``None`` si
+    ``model_name`` tente de s'échapper de la racine (traversal)."""
+    return safe_join(get_models_path(), model_name, suffix=".sql")
+
+
 def _derive_model_name_from_path(p: Path) -> Optional[str]:
     """`.../tests/<rel>.json` → `<rel>` (posix), ou None si hors de la racine tests."""
     try:
@@ -159,8 +177,8 @@ def read_model_sql(model_name: str) -> Optional[str]:
     if dbt_project and dbt_project.is_dbt_model(model_name):
         return dbt_project.compiled_sql_for_model(model_name)
 
-    sql_file = get_models_path() / f"{model_name}.sql"
-    if not sql_file.exists():
+    sql_file = _model_sql_path(model_name)
+    if sql_file is None or not sql_file.exists():
         return None
     raw_sql = sql_file.read_text(encoding="utf-8")
     fn_ref = get_preprocessor_fn()
@@ -172,12 +190,12 @@ def read_model_sql(model_name: str) -> Optional[str]:
 
 def get_model_file_git_sha(model_name: str) -> Optional[str]:
     """Return the SHA of the last commit that touched the model's .sql file, or None."""
-    sql_file = get_models_path() / f"{model_name}.sql"
-    if not sql_file.exists():
+    sql_file = _model_sql_path(model_name)
+    if sql_file is None or not sql_file.exists():
         return None
     try:
-        result = subprocess.run(
-            ["git", "log", "-1", "--format=%H", "--", str(sql_file)],
+        result = subprocess.run(  # noqa: S603 argv fixe, git en PATH, chemin validé
+            ["git", "log", "-1", "--format=%H", "--", str(sql_file)],  # noqa: S607
             capture_output=True,
             text=True,
             timeout=5,
@@ -191,8 +209,8 @@ def get_model_file_git_sha(model_name: str) -> Optional[str]:
 
 def get_model_file_hash(model_name: str) -> Optional[str]:
     """Return a short SHA-256 hash of the model file's current content, or None."""
-    sql_file = get_models_path() / f"{model_name}.sql"
-    if not sql_file.exists():
+    sql_file = _model_sql_path(model_name)
+    if sql_file is None or not sql_file.exists():
         return None
     try:
         content = sql_file.read_bytes()
@@ -203,12 +221,14 @@ def get_model_file_hash(model_name: str) -> Optional[str]:
 
 def get_commits_since_sha(model_name: str, source_sha: str) -> int:
     """Return the number of commits that touched the model file since source_sha."""
-    sql_file = get_models_path() / f"{model_name}.sql"
-    if not sql_file.exists():
+    sql_file = _model_sql_path(model_name)
+    if sql_file is None or not sql_file.exists():
+        return 0
+    if not _GIT_SHA_RE.match(source_sha or ""):
         return 0
     try:
-        result = subprocess.run(
-            ["git", "rev-list", "--count", f"{source_sha}..HEAD", "--", str(sql_file)],
+        result = subprocess.run(  # noqa: S603 argv fixe, git en PATH, sha/chemin validés
+            ["git", "rev-list", "--count", f"{source_sha}..HEAD", "--", str(sql_file)],  # noqa: S607
             capture_output=True,
             text=True,
             timeout=5,