206 lines
7.5 KiB
Python
206 lines
7.5 KiB
Python
"""JWT 认证中间件：校验 Token，注入 user_id。
|
||
|
||
支持两种模式：
|
||
1. 登录模式：通过 JWT Token 认证
|
||
2. 访客模式：未携带 Token 时自动生成匿名用户（GUEST_MODE=true 时启用）
|
||
"""
|
||
import json
|
||
import os
|
||
import jwt
|
||
import uuid
|
||
from functools import wraps
|
||
from flask import request, jsonify, current_app
|
||
|
||
|
||
def _is_guest_mode_enabled() -> bool:
|
||
"""判断是否开启访客模式。"""
|
||
env_value = os.environ.get("GUEST_MODE", "")
|
||
if env_value:
|
||
return env_value.lower() == "true"
|
||
|
||
config_value = current_app.config.get("GUEST_MODE", False)
|
||
if isinstance(config_value, bool):
|
||
return config_value
|
||
return str(config_value).lower() == "true"
|
||
|
||
|
||
def jwt_required(f):
|
||
"""装饰器：校验 JWT Token，将 user_id 注入 request。
|
||
|
||
如果 GUEST_MODE=true，未携带 Token 时自动使用匿名用户。
|
||
"""
|
||
@wraps(f)
|
||
def decorated(*args, **kwargs):
|
||
auth_header = request.headers.get("Authorization", "")
|
||
|
||
# 没有 Token
|
||
if not auth_header.startswith("Bearer "):
|
||
# 访客模式：使用固定的匿名用户 ID（基于 cookie 或 session）
|
||
if _is_guest_mode_enabled():
|
||
# 使用固定的 guest ID（基于 cookie）
|
||
guest_id = request.cookies.get("guest_id")
|
||
if not guest_id:
|
||
guest_id = uuid.uuid4().hex[:8]
|
||
request.user_id = f"guest_{guest_id}"
|
||
request.user_role = "client"
|
||
response = f(*args, **kwargs)
|
||
# 设置 cookie，确保后续请求使用相同的 guest ID
|
||
from flask import make_response
|
||
if isinstance(response, tuple):
|
||
resp, status_code = response
|
||
resp = make_response(resp, status_code)
|
||
elif hasattr(response, 'set_cookie'):
|
||
resp = response
|
||
else:
|
||
resp = make_response(response)
|
||
resp.set_cookie("guest_id", guest_id, max_age=365*24*60*60) # 1年
|
||
return resp
|
||
return jsonify({"code": 1002, "message": "未授权", "data": None}), 401
|
||
|
||
token = auth_header[7:]
|
||
|
||
# 访客模式：前端发送 Bearer guest_<id> 格式的 Token
|
||
if token.startswith("guest_"):
|
||
if _is_guest_mode_enabled():
|
||
request.user_id = token # 使用完整的 guest_xxx 作为 user_id
|
||
request.user_role = "client"
|
||
request.user_department = ""
|
||
return f(*args, **kwargs)
|
||
return jsonify({"code": 1002, "message": "未授权", "data": None}), 401
|
||
|
||
# 检查 Token 黑名单
|
||
try:
|
||
from insurance.db.compat import redis_client
|
||
if redis_client.get(f"token:blacklist:{token}"):
|
||
return jsonify({"code": 1002, "message": "Token 已失效", "data": None}), 401
|
||
except Exception as e:
|
||
import logging
|
||
logging.warning(f"Token 黑名单检查失败（Redis 可能不可用）: {e}")
|
||
|
||
# 校验 Token
|
||
secret = current_app.config.get("JWT_SECRET", "change-this")
|
||
try:
|
||
payload = jwt.decode(token, secret, algorithms=["HS256"])
|
||
request.user_id = payload["user_id"]
|
||
request.user_role = payload.get("role", "")
|
||
request.user_department = payload.get("department", "")
|
||
except jwt.ExpiredSignatureError:
|
||
return jsonify({"code": 1003, "message": "Token 已过期", "data": None}), 401
|
||
except jwt.InvalidTokenError:
|
||
return jsonify({"code": 1002, "message": "Token 无效", "data": None}), 401
|
||
|
||
return f(*args, **kwargs)
|
||
return decorated
|
||
|
||
|
||
def admin_required(f):
|
||
"""装饰器：要求管理员或超级管理员权限。"""
|
||
@wraps(f)
|
||
@jwt_required
|
||
def decorated(*args, **kwargs):
|
||
if request.user_role not in ("super_admin", "admin"):
|
||
return jsonify({"code": 1004, "message": "权限不足", "data": None}), 403
|
||
return f(*args, **kwargs)
|
||
return decorated
|
||
|
||
|
||
def super_admin_required(f):
|
||
"""装饰器：要求超级管理员权限。"""
|
||
@wraps(f)
|
||
@jwt_required
|
||
def decorated(*args, **kwargs):
|
||
if request.user_role != "super_admin":
|
||
return jsonify({"code": 1004, "message": "权限不足", "data": None}), 403
|
||
return f(*args, **kwargs)
|
||
return decorated
|
||
|
||
|
||
def permission_required(permission: str):
|
||
"""权限校验装饰器：查询 insurance_roles 表校验权限。
|
||
|
||
用法:
|
||
@app.route("/admin/users")
|
||
@permission_required("user_manage")
|
||
def list_users():
|
||
...
|
||
"""
|
||
def decorator(f):
|
||
@wraps(f)
|
||
@jwt_required
|
||
def decorated(*args, **kwargs):
|
||
role_name = request.user_role
|
||
if not role_name:
|
||
return jsonify({"code": 1004, "message": "权限不足", "data": None}), 403
|
||
|
||
# super_admin 和 admin 拥有所有权限
|
||
if role_name in ("super_admin", "admin"):
|
||
return f(*args, **kwargs)
|
||
|
||
# 从数据库查询角色权限
|
||
try:
|
||
from insurance.models.role import Role
|
||
from insurance.db.compat import db
|
||
role = db.session.query(Role).filter_by(name=role_name).first()
|
||
if role:
|
||
perms = json.loads(role.permissions) if role.permissions else []
|
||
if "*" in perms or permission in perms:
|
||
return f(*args, **kwargs)
|
||
except Exception as e:
|
||
import logging
|
||
logging.warning(f"权限校验查询失败: {e}")
|
||
|
||
return jsonify({"code": 1004, "message": "权限不足", "data": None}), 403
|
||
return decorated
|
||
return decorator
|
||
|
||
|
||
def get_user_permissions(role_name: str) -> list:
|
||
"""获取指定角色的权限列表。
|
||
|
||
返回:
|
||
list: 权限字符串列表，如 ["chat", "kb_manage"]
|
||
"""
|
||
if role_name in ("super_admin", "admin"):
|
||
return ["*"]
|
||
try:
|
||
from insurance.models.role import Role
|
||
from insurance.db.compat import db
|
||
role = db.session.query(Role).filter_by(name=role_name).first()
|
||
if role:
|
||
return json.loads(role.permissions) if role.permissions else []
|
||
except Exception:
|
||
pass
|
||
return []
|
||
|
||
|
||
def get_data_scope():
|
||
"""获取当前用户的数据权限范围。
|
||
|
||
返回:
|
||
dict: 包含 scope (all/team/self) 和 department 字段
|
||
"""
|
||
role = getattr(request, "user_role", "")
|
||
user_id = getattr(request, "user_id", "")
|
||
|
||
# 超级管理员和管理员：查看所有数据
|
||
if role in ("super_admin", "admin"):
|
||
return {"scope": "all", "department": None, "user_id": None}
|
||
|
||
# 销售主管：查看本部门数据
|
||
if role == "manager":
|
||
# 从JWT或数据库获取department
|
||
department = getattr(request, "user_department", None)
|
||
if not department:
|
||
try:
|
||
from insurance.models.wecom_user import WeComUserMapping
|
||
from insurance.db.compat import db
|
||
user = db.session.query(WeComUserMapping).filter_by(username=user_id).first()
|
||
if user:
|
||
department = user.department
|
||
except Exception:
|
||
pass
|
||
return {"scope": "team", "department": department, "user_id": None}
|
||
|
||
# 销售人员和其他角色：只看自己的数据
|
||
return {"scope": "self", "department": None, "user_id": user_id}
|