super
114 lines
3.5 KiB
Python
114 lines
3.5 KiB
Python
"""
|
|
认证装饰器与 session 校验
|
|
"""
|
|
from functools import wraps
|
|
|
|
from flask import request, redirect, url_for, session, jsonify, g
|
|
|
|
from utils.db import get_db
|
|
|
|
|
|
def is_session_user_valid():
|
|
"""校验 session 中的 user_id 是否仍存在；不存在则清空 session。"""
|
|
uid = session.get('user_id')
|
|
if not uid:
|
|
return False
|
|
cached_user = getattr(g, '_current_user_row', None)
|
|
if cached_user and cached_user.get('id') == uid:
|
|
return True
|
|
try:
|
|
conn = get_db()
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"SELECT id, username, is_admin, role, created_by_id FROM users WHERE id = %s",
|
|
(uid,)
|
|
)
|
|
row = cur.fetchone()
|
|
conn.close()
|
|
if not row:
|
|
session.clear()
|
|
return False
|
|
g._current_user_row = row
|
|
return True
|
|
except Exception:
|
|
session.clear()
|
|
return False
|
|
|
|
|
|
def get_current_admin_role():
|
|
"""返回当前登录用户的管理角色：super_admin / admin / None。"""
|
|
uid = session.get('user_id')
|
|
if not uid:
|
|
return None, None
|
|
try:
|
|
row = getattr(g, '_current_user_row', None)
|
|
if not row or row.get('id') != uid:
|
|
conn = get_db()
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"SELECT id, username, is_admin, role, created_by_id FROM users WHERE id = %s",
|
|
(uid,)
|
|
)
|
|
row = cur.fetchone()
|
|
conn.close()
|
|
if not row:
|
|
return None, None
|
|
g._current_user_row = row
|
|
role = (row.get('role') or '').strip().lower()
|
|
if not role:
|
|
role = 'super_admin' if row.get('is_admin') and row.get('created_by_id') is None else (
|
|
'admin' if row.get('is_admin') else 'normal'
|
|
)
|
|
return role, row
|
|
except Exception:
|
|
return None, None
|
|
|
|
|
|
def is_current_user_admin():
|
|
role, _ = get_current_admin_role()
|
|
return role in ('super_admin', 'admin')
|
|
|
|
|
|
def _is_ajax_request():
|
|
if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
|
|
return True
|
|
if request.path.startswith('/api/'):
|
|
return True
|
|
accept = (request.headers.get('Accept') or '').lower()
|
|
if 'application/json' in accept:
|
|
return True
|
|
return False
|
|
|
|
|
|
def login_required(f):
|
|
@wraps(f)
|
|
def decorated(*args, **kwargs):
|
|
if not session.get('user_id') or not is_session_user_valid():
|
|
if _is_ajax_request():
|
|
return jsonify({'success': False, 'error': '未登录'}), 401
|
|
return redirect(url_for('auth.login'))
|
|
return f(*args, **kwargs)
|
|
return decorated
|
|
|
|
|
|
def admin_required(f):
|
|
@wraps(f)
|
|
def decorated(*args, **kwargs):
|
|
if not session.get('user_id') or not is_session_user_valid():
|
|
if _is_ajax_request():
|
|
return jsonify({'success': False, 'error': '未登录'}), 401
|
|
return redirect(url_for('auth.login'))
|
|
|
|
try:
|
|
role, _ = get_current_admin_role()
|
|
if role not in ('super_admin', 'admin'):
|
|
if _is_ajax_request():
|
|
return jsonify({'success': False, 'error': '需要管理员权限'}), 403
|
|
return redirect(url_for('auth.login'))
|
|
except Exception as exc:
|
|
if _is_ajax_request():
|
|
return jsonify({'success': False, 'error': str(exc)}), 500
|
|
return redirect(url_for('auth.login'))
|
|
return f(*args, **kwargs)
|
|
return decorated
|