feat: Implement CSRF protection and PostgreSQL support

- Added CSRF protection mechanism in the backend with utility functions for token management.
- Introduced a new CSRF route to fetch the active CSRF token for SPA bootstrap flows.
- Updated the auth routes to validate CSRF tokens on sensitive operations.
- Configured PostgreSQL as a database option in the environment settings and Docker Compose.
- Created a new SQLite configuration file for local development.
- Enhanced the API client to automatically attach CSRF tokens to requests.
- Updated various frontend components to utilize the new site origin utility for SEO purposes.
- Modified Nginx configuration to improve redirection and SEO headers.
- Added tests for CSRF token handling in the authentication routes.
This commit is contained in:
Your Name
2026-03-17 23:26:32 +02:00
parent 3f24a7ea3e
commit a2824b2132
24 changed files with 332 additions and 319 deletions

View File

@@ -1,7 +1,7 @@
"""Flask Application Factory."""
import os
from flask import Flask
from flask import Flask, jsonify
from config import config
from app.extensions import cors, limiter, talisman, init_celery
@@ -11,6 +11,7 @@ from app.services.ai_cost_service import init_ai_cost_db
from app.services.site_assistant_service import init_site_assistant_db
from app.services.contact_service import init_contact_db
from app.services.stripe_service import init_stripe_db
from app.utils.csrf import CSRFError, apply_csrf_cookie, should_enforce_csrf, validate_csrf_request
def _init_sentry(app):
@@ -48,9 +49,10 @@ def create_app(config_name=None):
# Create upload/output/database directories
os.makedirs(app.config["UPLOAD_FOLDER"], exist_ok=True)
os.makedirs(app.config["OUTPUT_FOLDER"], exist_ok=True)
db_dir = os.path.dirname(app.config["DATABASE_PATH"])
if db_dir:
os.makedirs(db_dir, exist_ok=True)
if not app.config.get("DATABASE_URL"):
db_dir = os.path.dirname(app.config["DATABASE_PATH"])
if db_dir:
os.makedirs(db_dir, exist_ok=True)
# Initialize extensions
cors.init_app(
@@ -97,6 +99,22 @@ def create_app(config_name=None):
force_https=config_name == "production",
)
@app.before_request
def enforce_csrf():
if not should_enforce_csrf():
return None
try:
validate_csrf_request()
except CSRFError as exc:
return jsonify({"error": exc.message}), exc.status_code
return None
@app.after_request
def sync_csrf_cookie(response):
return apply_csrf_cookie(response)
# Initialize Celery
init_celery(app)

View File

@@ -19,6 +19,7 @@ from app.utils.auth import (
login_user_session,
logout_user_session,
)
from app.utils.csrf import get_or_create_csrf_token
auth_bp = Blueprint("auth", __name__)
@@ -105,6 +106,13 @@ def me_route():
return jsonify({"authenticated": True, "user": user}), 200
@auth_bp.route("/csrf", methods=["GET"])
@limiter.limit("240/hour")
def csrf_route():
"""Return the active CSRF token for SPA bootstrap flows."""
return jsonify({"csrf_token": get_or_create_csrf_token()}), 200
@auth_bp.route("/forgot-password", methods=["POST"])
@limiter.limit("5/hour")
def forgot_password_route():

77
backend/app/utils/csrf.py Normal file
View File

@@ -0,0 +1,77 @@
"""Lightweight CSRF protection for browser-originated session requests."""
import secrets
from flask import current_app, request, session
CSRF_SESSION_KEY = "csrf_token"
CSRF_COOKIE_NAME = "csrf_token"
CSRF_HEADER_NAME = "X-CSRF-Token"
_SAFE_METHODS = {"GET", "HEAD", "OPTIONS"}
_EXEMPT_PATHS = {
"/api/stripe/webhook",
}
class CSRFError(Exception):
"""Raised when CSRF validation fails."""
def __init__(self, message: str = "Invalid CSRF token.", status_code: int = 403):
super().__init__(message)
self.message = message
self.status_code = status_code
def get_or_create_csrf_token() -> str:
"""Return the current CSRF token, creating one when missing."""
token = session.get(CSRF_SESSION_KEY)
if not isinstance(token, str) or not token:
token = secrets.token_urlsafe(32)
session[CSRF_SESSION_KEY] = token
return token
def should_enforce_csrf() -> bool:
"""Return whether the current request should pass CSRF validation."""
if request.method.upper() in _SAFE_METHODS:
return False
if not request.path.startswith("/api/"):
return False
if request.path in _EXEMPT_PATHS:
return False
if request.headers.get("X-API-Key", "").strip():
return False
return True
def validate_csrf_request():
"""Validate the current request against the active browser CSRF token."""
session_token = session.get(CSRF_SESSION_KEY)
cookie_token = request.cookies.get(CSRF_COOKIE_NAME, "")
header_token = request.headers.get(CSRF_HEADER_NAME, "").strip()
if not isinstance(session_token, str) or not session_token:
raise CSRFError("CSRF session token is missing.")
if not cookie_token or cookie_token != session_token:
raise CSRFError("CSRF cookie token is missing or invalid.")
if not header_token or header_token != session_token:
raise CSRFError("CSRF header token is missing or invalid.")
def apply_csrf_cookie(response):
"""Persist the active CSRF token into a readable cookie for the SPA."""
token = get_or_create_csrf_token()
response.set_cookie(
CSRF_COOKIE_NAME,
token,
secure=bool(current_app.config.get("SESSION_COOKIE_SECURE", False)),
httponly=False,
samesite=current_app.config.get("SESSION_COOKIE_SAMESITE", "Lax"),
path="/",
)
return response