webapp/db.py

Rama main · réplica de lectura del repositorio original.

# SPDX-License-Identifier: BUSL-1.1
# Copyright (c) 2024-2026 Ali Sadeghi Aghili
"""SQLite persistence for the web app: users + query logs.

The database file lives next to this module (``webapp/app.db``) and is
created automatically on first use.  Passwords are stored as salted
hashes (werkzeug), never in plain text.
"""

from __future__ import annotations

import sqlite3
import sys
from contextlib import closing
from datetime import datetime
from pathlib import Path
from typing import Any

from werkzeug.security import check_password_hash, generate_password_hash

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from core.fileperms import restrict_file  # noqa: E402

DB_PATH = Path(__file__).resolve().parent / "app.db"

_SCHEMA = """
CREATE TABLE IF NOT EXISTS users (
    id            INTEGER PRIMARY KEY AUTOINCREMENT,
    username      TEXT NOT NULL UNIQUE,
    password_hash TEXT NOT NULL,
    created_at    TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS logs (
    id              INTEGER PRIMARY KEY AUTOINCREMENT,
    username        TEXT NOT NULL,
    question        TEXT NOT NULL,
    generated_sql   TEXT,
    interpretation  TEXT,
    output_file     TEXT,
    row_count       INTEGER NOT NULL DEFAULT 0,
    status          TEXT NOT NULL,
    error_message   TEXT,
    elapsed_seconds REAL,
    created_at      TEXT NOT NULL
);
"""


def _connect() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn


def init_db() -> None:
    """Create tables if they do not exist yet.  Safe to call repeatedly.

    Finding 18 (2026 audit): this file holds every Flask user's password
    hash and the full log of questions asked through the web UI, and was
    previously created at the ambient process umask (world-readable on a
    typical Linux host). ``restrict_file`` runs after every call rather
    than only the one that first creates ``app.db`` -- this function is
    called from ``create_user``/``verify_user``/``log_query`` on every
    request, not just at start-up, so gating the call on "did this call
    just create the file" would need a second existence check of its own
    for no real benefit: the helper is idempotent and a plain ``chmod``
    on an already-correct file is cheap.
    """
    with closing(_connect()) as conn, conn:
        conn.executescript(_SCHEMA)
    restrict_file(DB_PATH)


# ---------------------------------------------------------------------------
# Users
# ---------------------------------------------------------------------------

def create_user(username: str, password: str) -> None:
    """Insert a new user.  Raises sqlite3.IntegrityError on duplicate name."""
    init_db()
    with closing(_connect()) as conn, conn:
        conn.execute(
            "INSERT INTO users (username, password_hash, created_at) VALUES (?, ?, ?)",
            (
                username,
                generate_password_hash(password),
                datetime.now().isoformat(timespec="seconds"),
            ),
        )


def verify_user(username: str, password: str) -> dict[str, Any] | None:
    """Return the user row as a dict if credentials match, else None."""
    init_db()
    with closing(_connect()) as conn:
        row = conn.execute(
            "SELECT * FROM users WHERE username = ?", (username,)
        ).fetchone()
    if row is None or not check_password_hash(row["password_hash"], password):
        return None
    return dict(row)


# ---------------------------------------------------------------------------
# Query logs
# ---------------------------------------------------------------------------

def log_query(
    username: str,
    question: str,
    status: str,
    generated_sql: str | None = None,
    interpretation: str | None = None,
    output_file: str | None = None,
    row_count: int = 0,
    error_message: str | None = None,
    elapsed_seconds: float | None = None,
) -> None:
    """Record one submitted question and its outcome."""
    init_db()
    with closing(_connect()) as conn, conn:
        conn.execute(
            """INSERT INTO logs (
                   username, question, generated_sql, interpretation,
                   output_file, row_count, status, error_message,
                   elapsed_seconds, created_at
               ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
            (
                username,
                question,
                generated_sql,
                interpretation,
                output_file,
                row_count,
                status,
                error_message,
                elapsed_seconds,
                datetime.now().isoformat(timespec="seconds"),
            ),
        )