Feature/bennt updates

.gitignore

@@ -61,6 +61,8 @@ backup/
 history.json
 
 # Temporary files
+.pytest_cache/
+.pytest-tmp/
 *.tmp
 *.bak
 *_backup.*

README.md

@@ -35,6 +35,16 @@ flask --app app create-admin
 python app.py
 # Visit: http://localhost:8084
 ```
+
+For a fresh or unmanaged installation, `init-db` is safe to run again: it
+creates missing tables without deleting existing data.
+
+For an installation already managed by Alembic, apply schema updates before
+restarting:
+
+```bash
+flask --app app db upgrade
+```
 
 ## Dependencies
 

app.py

@@ -1,6 +1,5 @@
 import argparse
 import datetime
-import json
 import logging
 import os
 from pathlib import Path
@@ -10,13 +9,16 @@
 from flask import Flask, render_template
 from flask_login import LoginManager
 from flask_migrate import Migrate
+from flask_wtf.csrf import CSRFError, CSRFProtect
 
 from models import User, db, initialize_postgres_extensions
 from utils.email_service import init_mail
+from utils.model_catalog import load_model_catalog
 
 
 BASE_DIR = Path(__file__).resolve().parent
 load_dotenv(BASE_DIR / ".env")
+csrf = CSRFProtect()
 
 
 def _is_production() -> bool:
@@ -47,16 +49,7 @@ def _secret_key() -> str:
 
 
 def _load_model_database() -> dict:
-    model_db_path = BASE_DIR / "data" / "model_database.json"
-    if not model_db_path.is_file():
-        raise RuntimeError(f"Required model database is missing: {model_db_path}")
-
-    with model_db_path.open(encoding="utf-8") as model_db_file:
-        models_data = json.load(model_db_file)
-
-    if not isinstance(models_data, dict) or not models_data:
-        raise RuntimeError("The model database must be a non-empty JSON object.")
-    return models_data
+    return load_model_catalog(BASE_DIR)
 
 
 def _register_cli_commands(app: Flask) -> None:
@@ -108,6 +101,16 @@ def create_app() -> Flask:
         SQLALCHEMY_TRACK_MODIFICATIONS=False,
         SQLALCHEMY_ENGINE_OPTIONS={"pool_pre_ping": True},
         MAX_CONTENT_LENGTH=4 * 1024 * 1024,
+        SESSION_COOKIE_HTTPONLY=True,
+        SESSION_COOKIE_SAMESITE="Lax",
+        SESSION_COOKIE_SECURE=_is_production(),
+        REMEMBER_COOKIE_HTTPONLY=True,
+        REMEMBER_COOKIE_SAMESITE="Lax",
+        REMEMBER_COOKIE_SECURE=_is_production(),
+        PERMANENT_SESSION_LIFETIME=datetime.timedelta(days=7),
+        WTF_CSRF_ENABLED=os.environ.get(
+            "WTF_CSRF_ENABLED", "true"
+        ).lower() == "true",
         MAIL_SERVER=os.environ.get("MAIL_SERVER", "smtp.gmail.com"),
         MAIL_PORT=int(os.environ.get("MAIL_PORT", 587)),
         MAIL_USE_TLS=os.environ.get("MAIL_USE_TLS", "true").lower() == "true",
@@ -131,6 +134,7 @@ def create_app() -> Flask:
 
     db.init_app(app)
     Migrate(app, db)
+    csrf.init_app(app)
     init_mail(app)
 
     login_manager = LoginManager()
@@ -182,6 +186,14 @@ def upload_too_large(_error):
             "error.html", error="Uploads must be smaller than 4 MB."
         ), 413
 
+    @app.errorhandler(CSRFError)
+    def handle_csrf_error(error):
+        logger.warning("CSRF validation failed: %s", error.description)
+        return render_template(
+            "error.html",
+            error="This form expired or could not be verified. Please try again.",
+        ), 400
+
     @app.errorhandler(500)
     def internal_server_error(error):
         logger.exception("Unhandled application error: %s", error)

migrations/versions/8f4e3d2c1b0a_add_questionnaire_drafts.py

@@ -0,0 +1,55 @@
+"""Add server-side questionnaire drafts.
+
+Revision ID: 8f4e3d2c1b0a
+Revises: 5c442376367c
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "8f4e3d2c1b0a"
+down_revision = "5c442376367c"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        "questionnaire_drafts",
+        sa.Column("id", sa.String(length=64), nullable=False),
+        sa.Column("user_id", sa.Integer(), nullable=True),
+        sa.Column("content", sa.JSON(), nullable=False),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["user_id"],
+            ["users.id"],
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        "ix_questionnaire_drafts_user_id",
+        "questionnaire_drafts",
+        ["user_id"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_questionnaire_drafts_updated_at",
+        "questionnaire_drafts",
+        ["updated_at"],
+        unique=False,
+    )
+
+
+def downgrade():
+    op.drop_index(
+        "ix_questionnaire_drafts_updated_at",
+        table_name="questionnaire_drafts",
+    )
+    op.drop_index(
+        "ix_questionnaire_drafts_user_id",
+        table_name="questionnaire_drafts",
+    )
+    op.drop_table("questionnaire_drafts")

models.py

@@ -1,6 +1,6 @@
 from flask_login import UserMixin
 from flask_sqlalchemy import SQLAlchemy
-from datetime import datetime
+from datetime import datetime
 import json
 import os
 from werkzeug.security import generate_password_hash, check_password_hash
@@ -42,8 +42,26 @@ def is_admin(self):
         return self._is_admin
 
     @property
-    def is_expert(self):
-        return self._is_expert and self.is_approved_expert
+    def is_expert(self):
+        return self._is_expert and self.is_approved_expert
+
+    @property
+    def role(self):
+        """Return the user's effective role for templates and older callers."""
+        if self.is_admin:
+            return 'admin'
+        if self.is_expert:
+            return 'expert'
+        return 'user'
+
+    @property
+    def expertise(self):
+        """Backward-compatible alias for the canonical expertise field."""
+        return self.areas_of_expertise
+
+    @expertise.setter
+    def expertise(self, value):
+        self.areas_of_expertise = value
 
     def __repr__(self):
         return f'<User {self.username}>'
@@ -142,7 +160,7 @@ class Consultation(db.Model):
     def __repr__(self):
         return f'<Consultation {self.id}: {self.title}>'
 
-class Questionnaire(db.Model):
+class Questionnaire(db.Model):
     __tablename__ = 'questionnaires'
 
     id = db.Column(db.Integer, primary_key=True)
@@ -163,8 +181,31 @@ class Questionnaire(db.Model):
               postgresql_ops={'title': 'gin_trgm_ops', 'topic': 'gin_trgm_ops'}),
     )
 
-    def __repr__(self):
-        return f'<Questionnaire {self.id}: {self.title}>'
+    def __repr__(self):
+        return f'<Questionnaire {self.id}: {self.title}>'
+
+
+class QuestionnaireDraft(db.Model):
+    """Server-side working copy for guest and authenticated questionnaires."""
+
+    __tablename__ = 'questionnaire_drafts'
+
+    id = db.Column(db.String(64), primary_key=True)
+    user_id = db.Column(
+        db.Integer,
+        db.ForeignKey('users.id', ondelete='CASCADE'),
+        nullable=True,
+        index=True,
+    )
+    content = db.Column(db.JSON, nullable=False)
+    created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
+    updated_at = db.Column(
+        db.DateTime,
+        default=datetime.utcnow,
+        onupdate=datetime.utcnow,
+        nullable=False,
+        index=True,
+    )
 
 # PostgreSQL-specific function to initialize extensions
 def initialize_postgres_extensions(app):
@@ -188,14 +229,13 @@ def initialize_postgres_extensions(app):
             # Don't fail the entire app if extensions can't be created
             db.session.rollback()
 
-def get_model_details(model_name):
-    try:
-        import os
-        model_db_path = os.path.join(os.path.dirname(__file__), 'data', 'model_database.json')
-        with open(model_db_path, 'r') as f:
-            models = json.load(f)
-
-        # Get the model directly from the dictionary
+def get_model_details(model_name):
+    try:
+        from utils.model_catalog import load_model_catalog
+
+        models = load_model_catalog(os.path.dirname(__file__))
+
+        # Get the model directly from the dictionary
         if model_name in models:
             return models[model_name]
         return None

public/static/js/chatbot.js

@@ -125,9 +125,12 @@ class ChatBot {
             // Call API to get response
             const response = await fetch('/chatbot/ask', {
                 method: 'POST',
-                headers: {
-                    'Content-Type': 'application/json',
-                },
+                headers: {
+                    'Content-Type': 'application/json',
+                    'X-CSRFToken': document.querySelector(
+                        'meta[name="csrf-token"]'
+                    )?.content || '',
+                },
                 body: JSON.stringify({
                     question: userMessage,
                     context: this.pageContext

requirements.txt

@@ -5,6 +5,7 @@ flask-login>=0.6.0,<1.0.0
 flask-sqlalchemy>=3.1.0,<4.0.0
 flask-mail>=0.9.0,<1.0.0
 flask-migrate>=4.0.0,<5.0.0
+flask-wtf>=1.2.0,<2.0.0
 
 SQLAlchemy>=2.0.0,<3.0.0
 alembic>=1.16.0,<2.0.0

routes/admin_routes.py

@@ -17,6 +17,11 @@
     ai_usage_storage_ready,
     initialize_ai_usage_storage,
 )
+from utils.validation import (
+    is_valid_email,
+    is_valid_username,
+    normalize_email,
+)
 from sqlalchemy.exc import SQLAlchemyError
 import logging
 import re
@@ -109,16 +114,22 @@ def edit_user(user_id):
     user = User.query.get_or_404(user_id)
     if request.method == 'POST':
         # Update user information
-        user.username = request.form.get('username')
-        user.email = request.form.get('email')
-        # Update roles
-        is_admin = request.form.get('is_admin') == 'on'
-        is_expert = request.form.get('is_expert') == 'on'
-        user._is_admin = is_admin
-        if is_expert:
+        username = request.form.get('username', '').strip()
+        email = normalize_email(request.form.get('email'))
+        if not is_valid_username(username) or not is_valid_email(email):
+            flash('Provide a valid username and email address.', 'danger')
+            return render_template('admin/edit_user.html', user=user), 400
+        user.username = username
+        user.email = email
+        role = request.form.get('role', 'user')
+        user._is_admin = role == 'admin'
+        if role == 'expert':
             user._is_expert = True
             user.is_approved_expert = request.form.get('is_approved_expert') == 'on'
-            user.areas_of_expertise = request.form.get('areas_of_expertise')
+            user.areas_of_expertise = request.form.get(
+                'areas_of_expertise',
+                request.form.get('expertise', ''),
+            )
             user.institution = request.form.get('institution')
             user.bio = request.form.get('bio')
         else: