left4me/l4d2web/l4d2web/app.py

import hmac
import os
import secrets

import click
from flask import Flask, Response, g, 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.console_routes import bp as console_bp
from l4d2web.routes.files_routes import bp as files_bp
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.profile_routes import bp as profile_bp
from l4d2web.routes.profile_routes import reset_profile_password_rate_limits
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,
    start_state_poller,
)
from l4d2web.services.live_state_poller import start_live_state_poller
from l4d2web.services.log_listener import start_log_listener
from l4d2web.services.timeago import format_time_html


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.add_template_filter(format_time_html, "timeago")

    @app.context_processor
    def inject_base_layout() -> dict[str, str]:
        from flask import has_request_context

        is_modal = has_request_context() and request.headers.get("HX-Modal") == "1"
        return {"base_layout": "_modal_partial.html" if is_modal else "base.html"}

    @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")
        expected = session.get("csrf_token")
        if not token or not expected or not hmac.compare_digest(token, expected):
            return Response("csrf token missing or invalid", status=400)
        return None

    @app.before_request
    def assign_csp_nonce() -> None:
        g.csp_nonce = secrets.token_urlsafe(16)

    @app.after_request
    def set_security_headers(response: Response) -> Response:
        response.headers.setdefault("X-Content-Type-Options", "nosniff")
        response.headers.setdefault("X-Frame-Options", "DENY")
        response.headers.setdefault("Referrer-Policy", "strict-origin-when-cross-origin")
        nonce = getattr(g, "csp_nonce", "")
        # script-src nonce blocks inline XSS; style 'unsafe-inline' kept for
        # htmx's auto-injected indicator styles. img/data: for SVG icons.
        # *.steamstatic.com covers Steam avatar hosts; *.steamusercontent.com
        # serves Workshop item preview thumbnails.
        response.headers.setdefault(
            "Content-Security-Policy",
            "default-src 'self'; "
            f"script-src 'self' 'nonce-{nonce}'; "
            "style-src 'self' 'unsafe-inline'; "
            "img-src 'self' data: https://*.steamstatic.com https://*.steamusercontent.com; "
            "connect-src 'self'; "
            "frame-ancestors 'none'; "
            "base-uri 'self'; "
            "form-action 'self'",
        )
        if not app.config.get("TESTING") and request.is_secure:
            response.headers.setdefault(
                "Strict-Transport-Security",
                "max-age=31536000; includeSubDomains",
            )
        return response

    @app.errorhandler(404)
    def not_found(_err):
        return Response("not found", status=404, mimetype="text/plain")

    @app.errorhandler(500)
    def internal_error(_err):
        return Response("internal server error", status=500, mimetype="text/plain")

    app.before_request(load_current_user)
    app.register_blueprint(auth_bp)
    app.register_blueprint(overlay_bp)
    app.register_blueprint(files_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(console_bp)
    app.register_blueprint(page_bp)
    app.register_blueprint(profile_bp)
    register_cli(app)
    if app.config.get("TESTING"):
        reset_login_rate_limits()
        reset_profile_password_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)
        start_state_poller(app)
    if not app.config.get("TESTING"):
        start_live_state_poller(app)
        start_log_listener(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