import os import secrets import click from flask import Flask, Response, jsonify, redirect, request, session from l4d2web.auth import current_user, load_current_user from l4d2web.cli import register_cli from l4d2web.config import load_config from l4d2web.db import init_db from l4d2web.routes.blueprint_routes import bp as blueprint_bp from l4d2web.routes.auth_routes import bp as auth_bp from l4d2web.routes.auth_routes import reset_login_rate_limits from l4d2web.routes.job_routes import bp as job_bp from l4d2web.routes.log_routes import bp as log_bp from l4d2web.routes.overlay_routes import bp as overlay_bp from l4d2web.routes.page_routes import bp as page_bp from l4d2web.routes.server_routes import bp as server_bp from l4d2web.routes.workshop_routes import bp as workshop_bp from l4d2web.services.job_worker import recover_stale_jobs, start_job_workers def _in_flask_cli_context() -> bool: return click.get_current_context(silent=True) is not None def create_app(test_config: dict[str, object] | None = None) -> Flask: app = Flask(__name__) app.config.from_mapping(load_config()) app.config.update( SESSION_COOKIE_HTTPONLY=True, SESSION_COOKIE_SAMESITE="Lax", CSRF_EXEMPT_PATHS={"/login", "/health"}, ) if test_config is not None: app.config.update(test_config) secret_key = app.config.get("SECRET_KEY") if not app.config.get("TESTING") and (not secret_key or secret_key == "dev"): raise RuntimeError("SECRET_KEY must be set to a non-default value outside of testing") secure_env = os.getenv("SESSION_COOKIE_SECURE") if secure_env is not None: app.config["SESSION_COOKIE_SECURE"] = secure_env.lower() not in {"0", "false", "no"} else: app.config["SESSION_COOKIE_SECURE"] = not app.config.get("TESTING", False) with app.app_context(): init_db() @app.before_request def csrf_protect() -> Response | None: if "csrf_token" not in session: session["csrf_token"] = secrets.token_hex(16) if request.method not in {"POST", "PUT", "PATCH", "DELETE"}: return None if request.endpoint is None: return None if request.path in app.config["CSRF_EXEMPT_PATHS"]: return None token = request.headers.get("X-CSRF-Token") or request.form.get("csrf_token") if token != session.get("csrf_token"): return Response("csrf token missing or invalid", status=400) return None app.before_request(load_current_user) app.register_blueprint(auth_bp) app.register_blueprint(overlay_bp) app.register_blueprint(workshop_bp) app.register_blueprint(blueprint_bp) app.register_blueprint(server_bp) app.register_blueprint(job_bp) app.register_blueprint(log_bp) app.register_blueprint(page_bp) register_cli(app) if app.config.get("TESTING"): reset_login_rate_limits() should_start_workers = ( app.config.get("JOB_WORKER_ENABLED") and not app.config.get("TESTING") and not _in_flask_cli_context() ) if should_start_workers: recover_stale_jobs() start_job_workers(app) @app.get("/health") def health(): return jsonify({"status": "ok"}) @app.get("/") def root(): if current_user() is None: return redirect("/login") return redirect("/dashboard") return app