webapp/app.py
Rama main · réplica de lectura del repositorio original.
# SPDX-License-Identifier: BUSL-1.1
# Copyright (c) 2024-2026 Ali Sadeghi Aghili
"""Simple web front-end for the local SQL agent.
Usage (from this folder)::
py -3.13 app.py create-user alice # prompts for password
py -3.13 app.py create-user alice s3cret # password on the command line
py -3.13 app.py # dev server on http://127.0.0.1:5000
Users are stored in ``app.db`` (SQLite) with hashed passwords. Every
submitted question is logged to the ``logs`` table with its result.
"""
from __future__ import annotations
import getpass
import os
import re
import secrets
import sqlite3
import sys
import threading
import time
from pathlib import Path
from flask import (
Flask,
abort,
current_app,
flash,
redirect,
render_template,
request,
send_from_directory,
session,
url_for,
)
import db
import i18n
from agent import OUTPUT_DIR, answer_question, principal_for_username
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from core.console import use_utf8_console # noqa: E402
from core.provenance import log_startup_notice # noqa: E402
WEBAPP_DIR = Path(__file__).resolve().parent
SECRET_KEY_FILE = WEBAPP_DIR / ".secret_key"
def _require_admin_user() -> str:
"""Resolve the one account permitted to create new users (``/register``).
Finding 16: this used to fall back to a literal username baked into
source -- a real employee's, published in a repository that was
public at audit time (finding 1), naming exactly the one account this
app treats as privileged. There is no safe default for "the identity
of a privileged account": a guessed or
hardcoded one is either wrong (silently breaking a deployment that
meant to set its own) or, worse, silently *right* for this one
deployment and then copied as boilerplate into the next one's config.
An unset ``ADMIN_USER`` is a misconfiguration, not something to guess
past, so this raises rather than falling back to anything -- read
fresh on every call (not cached in a module-level constant) so a
process that reloads this module after fixing the environment
variable does not keep failing on a stale value from before the fix.
"""
admin_user = os.getenv("ADMIN_USER")
if not admin_user:
raise RuntimeError(
"ADMIN_USER is not set. This app requires an explicit "
"administrator username -- the only account permitted to "
"create new accounts via /register -- with no default. Set "
"the ADMIN_USER environment variable before starting the app."
)
return admin_user
def _flash(text: str, category: str = "message", **kwargs) -> None:
"""Flash ``text`` translated into the session's current language."""
flash(i18n.translate(text, i18n.get_lang(), **kwargs), category)
def _secret_key() -> str:
"""Env var wins; otherwise persist a random key so sessions survive restarts.
Finding 18: the persisted file is restricted to owner-only permissions
immediately after being written. This file holds the Flask session-
signing key -- anyone else with a shell on the host (another service
account, an operator, a compromised process) who can read it can
forge a session cookie for any user, including the ``ADMIN_USER``
account, with no password and nothing in the auth logs. Guarded for
Windows (``os.name != "nt"``): file mode bits are a POSIX concept and
the deployment target this finding describes is a Linux server, but
the ``chmod`` call itself must stay unconditional in the source so a
Linux deployment gets it regardless of what platform last edited
this file.
"""
env_key = os.getenv("SECRET_KEY")
if env_key:
return env_key
if SECRET_KEY_FILE.exists():
return SECRET_KEY_FILE.read_text().strip()
key = secrets.token_hex(32)
SECRET_KEY_FILE.write_text(key)
if os.name != "nt":
os.chmod(SECRET_KEY_FILE, 0o600)
return key
# ---------------------------------------------------------------------------
# Finding 17a: per-IP login throttling
# ---------------------------------------------------------------------------
# In-process and per-app (see _login_failures' own note for why not
# module-level), matching the FastAPI side's own bucketing shape
# (api/middleware.py's RateLimitMiddleware) -- a fixed threshold of
# consecutive failures per source IP, the account name on the other end
# never being part of the key. The published ADMIN_USER account (finding
# 16) is exactly the target unlimited guessing would go after; capping
# guesses per IP rather than per attempted username also stops the
# obvious workaround of guessing many different usernames from one
# source. Cleared on a successful login from that IP -- this defends
# against a *stranger* guessing, not against an operator who mistyped
# their own password several times in a row.
#
# Round 2 remediation: the counter used to have no expiry at all --
# `_login_failure_counts[ip]` only ever went up, or was reset to zero by a
# *successful* login from that IP. An IP that never succeeds (because it
# belongs to a shared NAT/proxy/office network the real account holder is
# behind, or because the attacker keeps guessing past the real user's
# actual password too) stays throttled forever. That turns a control meant
# to slow down a guesser into a permanent denial of service against
# whoever else is behind that address -- worse than the attack it defends
# against, since the attacker only had to spend thirty wrong guesses to
# inflict it, and needs no further access to keep it in effect. Each
# entry now carries the time its failure window started as well as its
# count: a window older than ``_LOGIN_FAILURE_WINDOW_SECONDS`` is treated
# as stale and discarded (by the very next check or failure from that
# IP) rather than compounding forever, and throttling lifts on its own
# once the window rolls past, even with no successful login at all. This
# does not weaken the defence against sustained guessing: an attacker who
# keeps trying stays throttled for the same reason a stale window is
# discarded -- it is exactly the fresh failure that starts the next one.
_LOGIN_FAILURE_THRESHOLD = 30
#: How long a throttled IP stays throttled with no further failures, and
#: the span of consecutive failures the threshold above is measured over.
#: Fifteen minutes is long enough that a scripted guesser gains nothing by
#: waiting it out repeatedly, short enough that a real user sharing an
#: address with whoever tripped it is not locked out for the rest of the
#: day.
_LOGIN_FAILURE_WINDOW_SECONDS = 15 * 60
#: ``{ip: (failure_count, window_started_at)}`` -- ``window_started_at`` is
#: a ``time.monotonic()`` reading, never wall-clock time, so this is
#: immune to the system clock being adjusted backwards mid-window.
#:
#: Deliberately owned by the Flask app object rather than by this module.
#: A process serves exactly one app, so in production the two are the same
#: thing -- but a module-level dict is shared by every ``create_app()`` a
#: process ever builds, which means one caller's failures follow the next
#: app into existence. That surfaced first as a test that passed alone and
#: failed in a full run (the throttle test's forty failures left the next
#: test's address already locked out), and a test that depends on
#: collection order is the visible half of a real defect: anything that
#: builds a second app in one process -- a test suite, a WSGI server
#: reloading, an embedding harness -- inherits state it never created.
#: :func:`_failure_state` reads it off ``current_app`` instead.
_LOGIN_FAILURE_STATE_KEY = "_login_failures"
def _failure_state() -> tuple[dict[str, tuple[int, float]], threading.Lock]:
"""The current app's own failure map and the lock guarding it."""
extensions = current_app.extensions.setdefault(_LOGIN_FAILURE_STATE_KEY, {})
if "map" not in extensions:
extensions["map"] = {}
extensions["lock"] = threading.Lock()
return extensions["map"], extensions["lock"]
def _client_ip() -> str:
return request.remote_addr or "unknown"
def _login_throttled(ip: str) -> bool:
"""Whether *ip* has hit the failure threshold within its current window.
A window whose start is more than :data:`_LOGIN_FAILURE_WINDOW_SECONDS`
in the past is stale -- the failures it counted no longer describe an
ongoing attempt -- and is discarded here rather than only by a future
:func:`_record_login_failure` call, so a throttled IP that simply stops
trying is not left reporting itself as throttled indefinitely.
"""
now = time.monotonic()
failures, lock = _failure_state()
with lock:
entry = failures.get(ip)
if entry is None:
return False
count, window_started_at = entry
if now - window_started_at >= _LOGIN_FAILURE_WINDOW_SECONDS:
del failures[ip]
return False
return count >= _LOGIN_FAILURE_THRESHOLD
def _record_login_failure(ip: str) -> None:
"""Count one failure for *ip*, starting a fresh window if the last one
has expired (see :func:`_login_throttled`) or none exists yet."""
now = time.monotonic()
failures, lock = _failure_state()
with lock:
entry = failures.get(ip)
if entry is None or now - entry[1] >= _LOGIN_FAILURE_WINDOW_SECONDS:
failures[ip] = (1, now)
else:
count, window_started_at = entry
failures[ip] = (count + 1, window_started_at)
def _clear_login_failures(ip: str) -> None:
failures, lock = _failure_state()
with lock:
failures.pop(ip, None)
# ---------------------------------------------------------------------------
# Finding 17b: CSRF -- a session-bound token, enforced on POST
# ---------------------------------------------------------------------------
# No new dependency (Flask-WTF is the obvious alternative -- see
# tests/security_audit/test_flask_hardening.py's module docstring): a
# random token is minted into the session the first time a form that
# needs one is rendered, embedded as a hidden field, and compared with
# constant-time equality against the form submission.
#
# Round 2 remediation note: /login used to be listed here as "deliberately
# exempt", on the reasoning that no session exists yet to bind a token to.
# That reasoning does not hold up -- writing into `session[...]` inside
# `_csrf_token()` (called by every template that renders the hidden field,
# login.html included) makes Flask send a Set-Cookie for that session on
# the GET response, before any credential is ever submitted. The token is
# therefore already bound to a real, if pre-authentication, session by the
# time a form using it can be filled in -- exactly the "issue a pre-session
# token" shape this class of fix takes. Leaving the field in login.html
# rendered but unchecked (the state this comment used to describe) was
# strictly worse than either enforcing it or removing it: a control that
# looks present protects nobody and invites the next reader to assume it
# does. login is no longer listed in _CSRF_EXEMPT_ENDPOINTS.
#
# This does not weaken _LOGIN_FAILURE_THRESHOLD above. A scripted
# credential-stuffing client that skips the GET and posts credentials
# directly is refused with 400 before the throttle counter is ever
# touched, which is a strictly earlier and cheaper refusal, not a bypass
# of it -- and a client that does fetch a token first (one GET, reused
# across every guess, exactly how a real browser and any minimally
# competent script both behave) still hits the same per-IP counter on
# every attempt after. What actually changes is that submitting this
# specific form now requires a token minted by this server for this
# session -- which is precisely what stops a third-party page from
# silently auto-submitting a victim's browser into a login it never
# intended (the attack this project's own audit calls "login CSRF": an
# attacker's page logs the victim into the *attacker's* account, then the
# victim's own subsequent input -- files uploaded, data entered -- lands
# somewhere the attacker can read it back from).
_CSRF_SESSION_KEY = "_csrf_token"
_CSRF_EXEMPT_ENDPOINTS = {"static"}
def _csrf_token() -> str:
token = session.get(_CSRF_SESSION_KEY)
if not token:
token = secrets.token_hex(16)
session[_CSRF_SESSION_KEY] = token
return token
def create_app() -> Flask:
log_startup_notice()
db.init_db()
admin_user = _require_admin_user()
app = Flask(__name__)
app.secret_key = _secret_key()
app.config["PERMANENT_SESSION_LIFETIME"] = 60 * 60 * 8 # 8h login
# Finding 17c: cookie hardening. Set explicitly rather than left to
# Flask's own defaults or a browser's -- see this module's docstring
# in tests/security_audit/test_flask_hardening.py for why relying on
# either is "a decision someone else can change". SESSION_COOKIE_SECURE
# means the cookie is withheld over plain HTTP: a local dev server run
# over http:// will look like login silently does not "stick" --
# documented in webapp/WebApp.md rather than left to be rediscovered.
app.config.update(
SESSION_COOKIE_HTTPONLY=True,
SESSION_COOKIE_SECURE=True,
SESSION_COOKIE_SAMESITE="Strict",
)
@app.context_processor
def _inject_auth() -> dict[str, str | None]:
return {"current_user": session.get("username"), "admin_user": admin_user}
@app.context_processor
def _inject_i18n() -> dict:
lang = i18n.get_lang()
return {
"current_lang": lang,
"langs": [(code, code.upper()) for code in i18n.LANGS],
"_": lambda text, **kwargs: i18n.translate(text, lang, **kwargs),
"csrf_token": _csrf_token,
}
@app.before_request
def _check_csrf():
"""Reject a token-less POST to a form that needs one (finding 17b).
Runs before every view, so a request never reaches a route's own
logic (including a route's "redirect to /login if not signed in",
which is a 302, not the 400/403 this defence is meant to produce)
without a valid token first. See the module-level comment above
``_CSRF_EXEMPT_ENDPOINTS`` for why ``login`` is excluded.
"""
if request.method != "POST" or request.endpoint in _CSRF_EXEMPT_ENDPOINTS:
return None
submitted = request.form.get("csrf_token", "")
expected = session.get(_CSRF_SESSION_KEY, "")
if not expected or not secrets.compare_digest(submitted, expected):
abort(400)
return None
@app.route("/lang/<lang>")
def set_lang(lang):
if lang in i18n.LANGS:
session["lang"] = lang
referrer = request.referrer or ""
if referrer.startswith(request.host_url):
return redirect(referrer)
return redirect(url_for("index"))
@app.route("/login", methods=["GET", "POST"])
def login():
if "username" in session:
return redirect(url_for("index"))
if request.method == "POST":
ip = _client_ip()
# Finding 17a: checked BEFORE touching the database at all --
# once an IP is throttled, a guess costs it nothing further to
# keep guessing against, which is the point.
if _login_throttled(ip):
return ("Too many failed login attempts. Try again later.", 429)
username = request.form.get("username", "").strip()
password = request.form.get("password", "")
if db.verify_user(username, password) is None:
_record_login_failure(ip)
_flash("Invalid username or password.", "error")
else:
_clear_login_failures(ip)
session["username"] = username
session.permanent = True
return redirect(url_for("index"))
return render_template("login.html")
@app.route("/register", methods=["GET", "POST"])
def register():
username = session.get("username")
if username is None:
return redirect(url_for("login"))
if username != admin_user:
_flash("Only an administrator can create accounts.", "error")
return redirect(url_for("index"))
if request.method == "POST":
new_username = request.form.get("username", "").strip()
password = request.form.get("password", "")
confirm = request.form.get("confirm", "")
if not new_username or not password:
_flash("Username and password are required.", "error")
elif password != confirm:
_flash("Passwords do not match.", "error")
else:
try:
db.create_user(new_username, password)
except sqlite3.IntegrityError:
_flash("Username '%(name)s' is already taken.", "error", name=new_username)
else:
_flash("Account '%(name)s' created.", "success", name=new_username)
return redirect(url_for("register"))
return render_template("register.html", username=username)
@app.route("/logout")
def logout():
session.clear()
return redirect(url_for("login"))
@app.route("/download/<filename>")
def download(filename):
if "username" not in session:
return redirect(url_for("login"))
# Only serve files the agent itself generated, never arbitrary paths.
if not re.fullmatch(r"output_[0-9]{8}_[0-9]{6}(?:_[0-9]+)?\.csv", filename):
abort(404)
return send_from_directory(OUTPUT_DIR, filename, as_attachment=True)
@app.route("/", methods=["GET", "POST"])
def index():
if "username" not in session:
return redirect(url_for("login"))
result = None
if request.method == "POST":
question = request.form.get("question", "").strip()
if not question:
_flash("Please enter a question.", "error")
else:
# Finding 13: the caller must identify itself so the column
# ACL (appdb.key_store / the admin panel's "ستونها" action)
# actually applies to this session -- see agent.py's
# principal_for_username and its module-level comment.
# Finding 14: the checkbox's own value is honoured exactly
# as submitted, never overridden -- answer_question's
# default covers only the caller who omits the argument.
interpret = request.form.get("interpret") == "on"
result = answer_question(
question,
interpret=interpret,
principal=principal_for_username(session["username"]),
)
db.log_query(
username=session["username"],
question=question,
status=result["status"],
generated_sql=result["sql"],
interpretation=result["interpretation"],
output_file=result["output_file"],
row_count=result["row_count"],
error_message=result["error_message"],
elapsed_seconds=result["elapsed_seconds"],
)
if result["status"] != "SUCCESS":
flash(result["error_message"], "error")
return render_template(
"index.html",
username=session["username"],
result=result,
output_filename=(
Path(result["output_file"]).name
if result and result["output_file"]
else None
),
)
return app
app = create_app()
def _cli_create_user(argv: list[str]) -> int:
"""create-user <username> [password] — password prompts if omitted."""
if len(argv) < 2:
print("Usage: app.py create-user <username> [password]")
return 2
username = argv[1].strip()
password = argv[2] if len(argv) > 2 else getpass.getpass("Password: ")
if not username or not password:
print("Username and password must not be empty.")
return 2
try:
db.create_user(username, password)
except Exception as exc: # noqa: BLE001 - sqlite3.IntegrityError and friends
print(f"Failed to create user: {exc}")
return 1
print(f"User '{username}' created.")
return 0
def main() -> int:
argv = sys.argv[1:]
if argv and argv[0] == "create-user":
return _cli_create_user(argv)
app.run(host="127.0.0.1", port=5000, debug=False)
return 0
if __name__ == "__main__":
use_utf8_console()
sys.exit(main())