# SPDX-License-Identifier: BUSL-1.1 # Copyright (c) 2024-2026 Ali Sadeghi Aghili """Bridge between the Flask app and the repo's NLQ engine. Imports the parent project (``api.runner.run_query``) by adding the repo root to ``sys.path`` and loading its ``.env``, so the web app runs the exact same generation → validation → execution pipeline as the CLI / API. Every successful query also exports the result rows to ``webapp/outputs/output_.csv`` (UTF-8 with BOM so Excel opens Persian text correctly). """ from __future__ import annotations import csv import logging import sys import time from datetime import datetime from pathlib import Path WEBAPP_DIR = Path(__file__).resolve().parent REPO_ROOT = WEBAPP_DIR.parent if str(REPO_ROOT) not in sys.path: sys.path.append(str(REPO_ROOT)) from dotenv import load_dotenv # noqa: E402 load_dotenv(REPO_ROOT / ".env") from api.errors import NLQError # noqa: E402 from api.runner import run_query # noqa: E402 from appdb.key_store import _maximally_restrictive_denied_columns # noqa: E402 from exporters.sanitize import defuse_formula # noqa: E402 from knowledge.config_loader import resolve_system_prompt_path # noqa: E402 from security.auth import Principal # noqa: E402 SYSTEM_PROMPT_PATH = resolve_system_prompt_path() OUTPUT_DIR = WEBAPP_DIR / "outputs" logger = logging.getLogger(__name__) _system_prompt: str | None = None #: Findings 13/14 — this module predates the Phase 8 principal model and #: was never connected to it (see tests/security_audit/ #: test_webapp_principal_and_acl.py's module docstring): every question #: asked through the Flask app reached ``run_query`` with no principal at #: all, and ``api.runner.run_query`` treats "no principal" as "no column #: restriction" -- deliberately, for the REPL and the test suite, which #: *are* legitimate principal-less callers. The Flask app is not one of #: those: it has real, named, logged-in users (``webapp/db.py``'s users #: table) and is obliged to say who is asking. When a logged-in username #: cannot be mapped to a configured Phase-8 principal (webapp/app.py has #: no such mapping today, and may never have every user enrolled), the #: fail-closed answer is this fallback -- built from the exact same #: column list a freshly issued, not-yet-widened API key gets #: (``appdb.key_store._maximally_restrictive_denied_columns``), reused #: rather than re-derived so the two "I don't know who you are yet" #: postures in this codebase stay identical by construction instead of by #: two people remembering to update both. def _restrictive_fallback_principal(identity: str | None) -> Principal: """Build the fail-closed ``Principal`` for a caller this app cannot place: every column denied, no administrative capability. See the module-level comment above this function for why an *empty* ``denied_columns`` (the bug the audit found) is not an acceptable default here.""" return Principal( id=identity or "webapp-unmapped", name=identity or "unmapped webapp user", denied_columns=tuple(_maximally_restrictive_denied_columns()), ) def principal_for_username(username: str | None) -> Principal: """Resolve a Flask login (``webapp/db.py``'s users table) to the Phase-8 ``Principal`` that should govern its queries. The two identity systems are not formally linked -- a Flask account and an API-key principal are created through entirely different flows (``app.py create-user`` vs. ``scripts.issue_api_key`` / the admin panel) -- but they share one namespace by convention (``docs/fa/getting-started.md`` issues both as e.g. ``analyst-1``), so a deployment that wants its Flask users to carry the SAME column ACL the admin panel manages only has to issue a key with a matching ``principal_id``. Looked up defensively: the application database that ``appdb.key_store`` reads from is not something this historically DB-agnostic web app has ever depended on, so any failure to reach it (unset ``APP_DB_URL``, an unmigrated table, a network blip) must fall through to the restrictive default rather than surface as a 500 or, worse, silently grant unrestricted access. """ if username: try: from appdb.key_store import get_active_principals for principal in get_active_principals().values(): if principal.id == username: return principal except Exception: # noqa: BLE001 - see docstring: never let this widen access logger.warning( "principal_for_username(%r): appdb lookup failed, falling back " "to the restrictive default", username, exc_info=True, ) return _restrictive_fallback_principal(username) def system_prompt() -> str: """Load and cache the repo's system prompt. Raises ------ RuntimeError If ``SYSTEM_PROMPT_PATH`` does not exist -- named explicitly here rather than surfacing a bare, unhelpful ``FileNotFoundError``, the same "refuse to start with a clear message naming the expected path" contract ``api/server.py``'s ``lifespan`` and ``app.py``'s ``_load_system_prompt`` already give their own callers. """ global _system_prompt if _system_prompt is None: if not SYSTEM_PROMPT_PATH.exists(): raise RuntimeError(f"System prompt not found: {SYSTEM_PROMPT_PATH}") _system_prompt = SYSTEM_PROMPT_PATH.read_text(encoding="utf-8") return _system_prompt def _export_csv(rows: list[dict]) -> str: """Write result rows to outputs/ with a current-datetime filename. Finding 15 (formula injection): this file is written UTF-8-with-BOM specifically so Excel opens it -- see the module docstring -- which makes it exactly the file format that treats a cell starting with ``=``/``+``/``-``/``@`` as a formula to execute rather than text to display. CSV's own field quoting (what ``csv.DictWriter`` already does) escapes embedded quotes; it does nothing about a cell that simply *begins* with one of those characters, which is what a warehouse value the analyst never typed can do. Every cell is passed through :func:`exporters.sanitize.defuse_formula` for the same reason ``exporters/excel_exporter.py`` does -- one shared rule, not two copies of it that can drift. """ OUTPUT_DIR.mkdir(parents=True, exist_ok=True) columns = list(rows[0].keys()) stamp = datetime.now().strftime("%Y%m%d_%H%M%S") path = OUTPUT_DIR / f"output_{stamp}.csv" suffix = 1 while path.exists(): path = OUTPUT_DIR / f"output_{stamp}_{suffix}.csv" suffix += 1 defused_rows = [ {key: defuse_formula(value) for key, value in row.items()} for row in rows ] with path.open("w", newline="", encoding="utf-8-sig") as fh: writer = csv.DictWriter(fh, fieldnames=columns) writer.writeheader() writer.writerows(defused_rows) return str(path) def answer_question( question: str, interpret: bool = False, principal: Principal | None = None, ) -> dict: """Run one question through the full pipeline; never raises. Uses the shared OpenAI-compatible LLM backend. Parameters ---------- interpret: Whether result rows are sent to the model for a plain-language summary. Defaults to ``False`` -- Finding 14: every other entry point in this project (``AskTurnRequest.interpret``, ``QueryRequest.interpret``) defaults this the same way, because it is a data-governance decision (up to twenty rows of real results leaving the process) and not something a caller should get without asking for it. principal: The authenticated Flask user, as resolved by :func:`principal_for_username` -- or ``None`` for a caller that has not identified one (a direct script, a test). Finding 13: this module used to call ``run_query`` with no principal at all, which ``api.runner.run_query`` reads as "no restriction" -- a deliberate default for its OTHER legitimate principal-less callers (the REPL, the test suite), but wrong here, because this path has real logins. ``None`` is therefore never forwarded as-is: it is replaced with the same maximally-restrictive fallback an unmapped username gets, so "I forgot to pass a principal" and "I could not identify this caller" fail exactly the same way -- closed. Returns a dict with ``status`` ("SUCCESS" or "ERROR"), the generated SQL, result rows/columns, plain-language interpretation, output file path and timing. On failure ``error_message`` holds the reason. """ start = time.perf_counter() if principal is None: principal = _restrictive_fallback_principal(None) try: resp = run_query( question, system_prompt(), mode="full", interpret=interpret, principal=principal, ) rows = resp.result or [] output_file = _export_csv(rows) if rows else None return { "status": "SUCCESS", "question": question, "sql": resp.sql, "interpretation": resp.interpretation, "rows": rows, "columns": list(rows[0].keys()) if rows else [], "row_count": len(rows), "model": resp.model, "output_file": output_file, "error_message": None, "elapsed_seconds": round(time.perf_counter() - start, 3), } except NLQError as exc: return _error_result(str(exc), start, question) except Exception as exc: # noqa: BLE001 - surface anything to the UI return _error_result(f"Unexpected error: {exc}", start, question) def _error_result(message: str, start: float, question: str = "") -> dict: return { "status": "ERROR", "question": question, "sql": None, "interpretation": None, "rows": [], "columns": [], "row_count": 0, "model": None, "output_file": None, "error_message": message, "elapsed_seconds": round(time.perf_counter() - start, 3), }