bvandeusen
b33aa25fb7
Add session_version to users table. Sessions now carry session_version alongside user_id; @login_required rejects any session where the version doesn't match the DB value. - migration 0024: session_version INTEGER NOT NULL DEFAULT 1 - models/user.py: session_version Mapped[int] column - auth.py: version mismatch → session.clear() + 401 - services/auth.py: change_password and reset_password_with_token both increment session_version, evicting all other live sessions - routes/auth.py: login/register/oauth_callback store session_version; update_password rehydrates current session with new version so the user who changed their password stays logged in Existing sessions will require one re-login after the migration runs. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
38 lines
1.1 KiB
Python
38 lines
1.1 KiB
Python
import functools
|
|
|
|
from quart import g, jsonify, session
|
|
|
|
from fabledassistant.services.auth import get_user_by_id
|
|
|
|
|
|
def _check_auth(f, required_role: str | None = None):
|
|
@functools.wraps(f)
|
|
async def decorated(*args, **kwargs):
|
|
user_id = session.get("user_id")
|
|
if not user_id:
|
|
return jsonify({"error": "Authentication required"}), 401
|
|
user = await get_user_by_id(user_id)
|
|
if not user:
|
|
session.clear()
|
|
return jsonify({"error": "Authentication required"}), 401
|
|
if session.get("session_version") != user.session_version:
|
|
session.clear()
|
|
return jsonify({"error": "Session expired. Please log in again."}), 401
|
|
if required_role and user.role != required_role:
|
|
return jsonify({"error": "Admin access required"}), 403
|
|
g.user = user
|
|
return await f(*args, **kwargs)
|
|
return decorated
|
|
|
|
|
|
def login_required(f):
|
|
return _check_auth(f)
|
|
|
|
|
|
def admin_required(f):
|
|
return _check_auth(f, required_role="admin")
|
|
|
|
|
|
def get_current_user_id() -> int:
|
|
return g.user.id
|