3c262b8d35
- auth_routes: 新增内存级登录速率限制（5min/10次），防止暴力破解 - auth_routes: 统一 DEV_MODE 判断收归 config.py 集中管理 - auth_routes: update_user 增加用户存在性校验 - gateway: require_role 装饰器恢复实际角色校验（原为占位实现） - session_routes: get_user_sessions 统一传入 limit=20 参数 - session: get_history SQL 增加 id DESC 排序，修复同秒消息顺序错乱 🤖 Generated with [Qoder][https://qoder.com]
271 lines
7.8 KiB
Python
271 lines
7.8 KiB
Python
"""
|
||
网关认证模块 - 从网关注入的 Header 读取用户信息
|
||
|
||
## 使用方式
|
||
|
||
from auth.gateway import require_gateway_auth, get_current_user
|
||
|
||
@app.route('/protected')
|
||
@require_gateway_auth
|
||
def protected():
|
||
user = request.current_user # {"user_id": ..., "role": ..., ...}
|
||
...
|
||
|
||
## Header 规范（开发模式可选）
|
||
|
||
- X-User-ID: 用户唯一标识 (可选)
|
||
- X-User-Name: 用户名 (可选)
|
||
- X-User-Role: 用户角色 (可选)
|
||
- X-User-Department: 部门 (可选)
|
||
|
||
## 模式说明
|
||
|
||
开发模式 (DEV_MODE=true):
|
||
- 支持 mock token 模拟用户：Authorization: Bearer mock-token-admin
|
||
- 无 Header 时自动使用开发测试用户
|
||
- 适用于前端测试和开发调试
|
||
|
||
生产模式 (DEV_MODE=false，默认):
|
||
- 不需要 Header，直接放行
|
||
- 权限由后端完全控制，通过 collections 参数传入
|
||
- RAG 服务完全无状态，只负责问答检索
|
||
"""
|
||
|
||
from functools import wraps
|
||
from flask import request, jsonify
|
||
from typing import Dict, Optional
|
||
from config import DEV_MODE
|
||
|
||
|
||
# ==================== 模拟用户数据（开发环境）====================
|
||
# 用于前端模拟登录测试，仅 DEV_MODE=true 时生效（需在 .env 中显式开启）
|
||
MOCK_USERS = {
|
||
'admin': {
|
||
'user_id': 'admin001',
|
||
'password': 'admin123',
|
||
'role': 'admin',
|
||
'department': '管理部'
|
||
},
|
||
'admin2': {
|
||
'user_id': 'admin002',
|
||
'password': 'admin456',
|
||
'role': 'admin',
|
||
'department': '技术部'
|
||
},
|
||
'admin3': {
|
||
'user_id': 'admin003',
|
||
'password': 'admin789',
|
||
'role': 'admin',
|
||
'department': '运营部'
|
||
},
|
||
'manager': {
|
||
'user_id': 'manager001',
|
||
'password': 'manager123',
|
||
'role': 'manager',
|
||
'department': '财务部'
|
||
},
|
||
'user': {
|
||
'user_id': 'user001',
|
||
'password': 'test123',
|
||
'role': 'user',
|
||
'department': '技术部'
|
||
}
|
||
}
|
||
|
||
|
||
def require_gateway_auth(f):
|
||
"""
|
||
网关认证装饰器 - 从 Header 读取用户信息
|
||
|
||
开发模式 (DEV_MODE=true):
|
||
- 支持 mock token: Authorization: Bearer mock-token-admin
|
||
- 无 Header 时自动使用开发测试用户（admin 角色）
|
||
|
||
生产模式 (DEV_MODE=false，默认):
|
||
- 不需要 Header，直接放行
|
||
- 用户信息设为默认值
|
||
- 权限由后端通过 collections 参数控制
|
||
"""
|
||
@wraps(f)
|
||
def decorated(*args, **kwargs):
|
||
# 开发模式开关（统一由 config.py 管理）
|
||
dev_mode = DEV_MODE
|
||
|
||
# 开发模式：支持 mock token
|
||
if dev_mode:
|
||
auth_header = request.headers.get('Authorization', '')
|
||
if auth_header.startswith('Bearer mock-token-'):
|
||
username = auth_header.replace('Bearer mock-token-', '')
|
||
mock_user = MOCK_USERS.get(username)
|
||
if mock_user:
|
||
request.current_user = {
|
||
"user_id": mock_user['user_id'],
|
||
"username": username,
|
||
"role": mock_user['role'],
|
||
"department": mock_user['department']
|
||
}
|
||
return f(*args, **kwargs)
|
||
|
||
# 从 Header 读取用户信息（可选）
|
||
user_id = request.headers.get('X-User-ID')
|
||
username = request.headers.get('X-User-Name', '')
|
||
role = request.headers.get('X-User-Role', 'user')
|
||
department = request.headers.get('X-User-Department', '')
|
||
|
||
# 如果有 Header，使用 Header 中的用户信息
|
||
if user_id:
|
||
request.current_user = {
|
||
"user_id": user_id,
|
||
"username": username,
|
||
"role": role,
|
||
"department": department
|
||
}
|
||
return f(*args, **kwargs)
|
||
|
||
# 无 Header 时的默认处理
|
||
if dev_mode:
|
||
# 开发模式：使用默认测试用户
|
||
request.current_user = {
|
||
"user_id": "dev-user",
|
||
"username": "开发测试用户",
|
||
"role": "admin",
|
||
"department": "开发部"
|
||
}
|
||
else:
|
||
# 生产模式：使用默认用户（后端通过 collections 控制权限）
|
||
request.current_user = {
|
||
"user_id": "backend-caller",
|
||
"username": "后端调用",
|
||
"role": "user",
|
||
"department": ""
|
||
}
|
||
|
||
return f(*args, **kwargs)
|
||
|
||
return decorated
|
||
|
||
|
||
def get_current_user() -> Optional[Dict]:
|
||
"""
|
||
获取当前登录用户信息
|
||
|
||
Returns:
|
||
用户信息字典，未认证返回 None
|
||
"""
|
||
return getattr(request, 'current_user', None)
|
||
|
||
|
||
# ==================== 兼容旧代码 ====================
|
||
|
||
# 别名
|
||
require_auth = require_gateway_auth
|
||
|
||
|
||
def get_user_permissions(role: str):
|
||
"""
|
||
兼容旧代码 - 权限由后端管理，此函数仅返回默认值
|
||
"""
|
||
return ['public', 'internal', 'confidential']
|
||
|
||
|
||
def check_collection_permission(role: str, department: str, collection_name: str, operation: str = "read") -> bool:
|
||
"""
|
||
兼容旧代码 - 权限由后端管理，默认返回 True
|
||
"""
|
||
return True
|
||
|
||
|
||
def get_accessible_collections(role: str, department: str, operation: str = "read", all_collections=None):
|
||
"""
|
||
兼容旧代码 - 返回所有向量库
|
||
"""
|
||
if all_collections:
|
||
return all_collections
|
||
return ['public_kb']
|
||
|
||
|
||
def can_create_collection(role: str) -> bool:
|
||
"""兼容旧代码 - 权限由后端管理"""
|
||
return True
|
||
|
||
|
||
def can_delete_collection(role: str) -> bool:
|
||
"""兼容旧代码 - 权限由后端管理"""
|
||
return True
|
||
|
||
|
||
def require_role(*roles):
|
||
"""
|
||
角色验证装饰器（开发和生产环境均生效）
|
||
|
||
需搭配 @require_gateway_auth 使用（先设置 current_user，再验证角色）。
|
||
|
||
开发模式: 检查 mock token 对应用户的角色
|
||
生产模式: 检查网关注入的 X-User-Role Header
|
||
"""
|
||
def decorator(f):
|
||
@wraps(f)
|
||
def decorated(*args, **kwargs):
|
||
if roles:
|
||
user = get_current_user()
|
||
if user is None or user.get('role') not in roles:
|
||
from flask import jsonify
|
||
return jsonify({
|
||
"error": "权限不足，需要角色: {}".format(', '.join(roles)),
|
||
"status": "FORBIDDEN"
|
||
}), 403
|
||
return f(*args, **kwargs)
|
||
return decorated
|
||
return decorator
|
||
|
||
|
||
def require_collection_permission(operation: str):
|
||
"""
|
||
集合权限验证装饰器（开发模式下为占位实现，生产环境权限由网关控制）
|
||
"""
|
||
def decorator(f):
|
||
@wraps(f)
|
||
def decorated(*args, **kwargs):
|
||
# 生产环境下权限由网关/后端统一管控，此处放行
|
||
return f(*args, **kwargs)
|
||
return decorated
|
||
return decorator
|
||
|
||
|
||
def get_auth_manager():
|
||
"""兼容旧代码"""
|
||
return _FakeAuthManager()
|
||
|
||
|
||
class _FakeAuthManager:
|
||
"""兼容旧代码的假 AuthManager"""
|
||
|
||
@staticmethod
|
||
def get_user_permissions(role: str):
|
||
return ['public', 'internal', 'confidential']
|
||
|
||
@staticmethod
|
||
def get_accessible_collections(role: str, department: str):
|
||
return ['public_kb']
|
||
|
||
|
||
def is_admin() -> bool:
|
||
"""检查当前用户是否为管理员"""
|
||
user = get_current_user()
|
||
return user is not None and user.get('role') == 'admin'
|
||
|
||
|
||
def is_manager_or_above() -> bool:
|
||
"""检查当前用户是否为经理或以上级别"""
|
||
user = get_current_user()
|
||
return user is not None and user.get('role') in ('admin', 'manager')
|
||
|
||
|
||
def normalize_department_name(department: str) -> str:
|
||
"""兼容旧代码 - 部门名称标准化"""
|
||
if not department:
|
||
return ""
|
||
if department.replace("_", "").replace("-", "").isalnum() and department.isascii():
|
||
return department.lower()
|
||
return ""
|