v6ole
f1f8518985
feat(auth): 添加用户权限获取接口并完善JWT令牌角色信息 - 在JWT令牌中添加用户角色信息 - 新增get_my_permissions接口用于获取当前用户权限码列表 - 重构认证回调逻辑，增加错误日志记录 - 更新用户信息获取接口使用Authorization头验证 ```
97 lines
3.1 KiB
Python
97 lines
3.1 KiB
Python
"""权限检查中间件（数据库驱动 + Redis 缓存）"""
|
||
import json
|
||
from fastapi import HTTPException, Depends, Header
|
||
from sqlalchemy.orm import Session
|
||
from sqlalchemy import text
|
||
import redis
|
||
|
||
from app.core.database import get_db
|
||
from app.core.security import verify_token
|
||
from app.core.config import settings
|
||
|
||
_redis_client = None
|
||
|
||
|
||
def get_redis() -> redis.Redis:
|
||
global _redis_client
|
||
if _redis_client is None:
|
||
_redis_client = redis.from_url(settings.REDIS_URL, decode_responses=True)
|
||
return _redis_client
|
||
|
||
|
||
def get_role_permissions(role: str, db: Session) -> list:
|
||
"""从数据库加载角色权限，结果缓存到 Redis（TTL 5分钟）"""
|
||
if role == 'admin':
|
||
return ['*']
|
||
|
||
r = get_redis()
|
||
cache_key = f"permissions:role:{role}"
|
||
cached = r.get(cache_key)
|
||
if cached:
|
||
return json.loads(cached)
|
||
|
||
rows = db.execute(
|
||
text("""
|
||
SELECT p.code FROM permissions p
|
||
JOIN role_permissions rp ON rp.permission_id = p.id
|
||
WHERE rp.role = :role
|
||
"""),
|
||
{"role": role}
|
||
).fetchall()
|
||
perms = [row[0] for row in rows]
|
||
|
||
r.setex(cache_key, 300, json.dumps(perms))
|
||
return perms
|
||
|
||
|
||
def invalidate_role_cache(role: str) -> None:
|
||
"""修改角色权限后清除缓存"""
|
||
get_redis().delete(f"permissions:role:{role}")
|
||
|
||
|
||
def require_permission(permission: str):
|
||
"""FastAPI Depends 工厂，检查 Bearer token 中的角色是否拥有指定权限"""
|
||
def dependency(
|
||
authorization: str = Header(..., alias="Authorization"),
|
||
db: Session = Depends(get_db),
|
||
) -> dict:
|
||
if not authorization.startswith("Bearer "):
|
||
raise HTTPException(status_code=401, detail="未授权")
|
||
token = authorization[7:]
|
||
payload = verify_token(token)
|
||
if not payload:
|
||
raise HTTPException(status_code=401, detail="令牌无效或已过期")
|
||
|
||
role = payload.get('role', 'user')
|
||
perms = get_role_permissions(role, db)
|
||
|
||
if '*' not in perms and permission not in perms:
|
||
raise HTTPException(status_code=403, detail="权限不足")
|
||
|
||
# 附加用户的区域/学校分配信息，供数据范围过滤使用
|
||
user_id = payload.get('sub')
|
||
if user_id and role in ('area_admin', 'school_admin'):
|
||
from app.models.user import User
|
||
user = db.query(User).filter(User.id == int(user_id)).first()
|
||
if user:
|
||
payload['assigned_area'] = user.assigned_area
|
||
payload['assigned_school'] = user.assigned_school
|
||
|
||
return payload
|
||
|
||
return dependency
|
||
|
||
|
||
def get_current_user(
|
||
authorization: str = Header(..., alias="Authorization"),
|
||
db: Session = Depends(get_db),
|
||
) -> dict:
|
||
"""仅验证登录状态，不检查具体权限"""
|
||
if not authorization.startswith("Bearer "):
|
||
raise HTTPException(status_code=401, detail="未授权")
|
||
token = authorization[7:]
|
||
payload = verify_token(token)
|
||
if not payload:
|
||
raise HTTPException(status_code=401, detail="令牌无效或已过期")
|
||
return payload
|