SourceShareOSV1/server/app_factory.py

from pathlib import Path

from flask import Flask
from flask import jsonify, request
import secrets

from .config import load_config
from .db import ensure_default_admin, ensure_default_plans, init_db
from .routes import register_routes


def create_app() -> Flask:
    project_root = Path(__file__).resolve().parent.parent
    app = Flask(__name__, static_folder=str(project_root / "static"), template_folder=str(project_root / "templates"))
    config = load_config()
    app.config["APP_CONFIG"] = config
    app.secret_key = config.secret_key
    app.config.setdefault("SESSION_COOKIE_SAMESITE", "Lax")
    app.config.setdefault("SESSION_COOKIE_HTTPONLY", True)

    @app.before_request
    def _attach_config_to_g() -> None:
        from flask import g

        g.app_config = config

    @app.before_request
    def _csrf_protect() -> tuple[dict, int] | None:
        if request.method in {"GET", "HEAD", "OPTIONS", "TRACE"}:
            return None
        if request.path == "/pay/callback":
            return None
        if request.method != "POST":
            return None
        if (request.mimetype or "").lower() == "application/json":
            return None
        csrf_cookie = (request.cookies.get("csrf_token") or "").strip()
        csrf_header = (request.headers.get("X-CSRF-Token") or request.headers.get("X-Csrf-Token") or "").strip()
        if not csrf_cookie or not csrf_header or csrf_cookie != csrf_header:
            return jsonify({"error": "csrf_failed"}), 403
        return None

    @app.before_request
    def _block_writes_during_migration() -> tuple[dict, int] | None:
        if request.method in {"GET", "HEAD", "OPTIONS", "TRACE"}:
            return None
        from .db import is_migrating

        if not is_migrating():
            return None
        allow = {"/admin/db/switch", "/admin/db/status", "/admin/mysql/test"}
        if request.path in allow:
            return None
        return jsonify({"error": "readonly_migrating"}), 503

    @app.after_request
    def _set_csrf_cookie(resp):
        token = (request.cookies.get("csrf_token") or "").strip()
        if not token:
            token = secrets.token_urlsafe(32)
            resp.set_cookie("csrf_token", token, httponly=False, samesite="Lax", secure=bool(request.is_secure))
        return resp

    @app.teardown_appcontext
    def _close_db(_exc) -> None:
        from flask import g
        from .db import close_db

        db = g.pop("db", None)
        if db is not None:
            close_db(db)
        ctl_db = g.pop("ctl_db", None)
        if ctl_db is not None:
            try:
                ctl_db.close()
            except Exception:
                pass

    with app.app_context():
        init_db()
        ensure_default_admin()
        ensure_default_plans()

    register_routes(app)
    return app