FastAPI RBAC权限控制、Token自动配置及后端架构优化问询
问题解决方案
1. 自动将Bearer Token追加到响应头
在登录接口中,通过Response对象手动设置Authorization响应头,同时保留原有JSON响应体:
from fastapi import Response @app.post("/login") def login(response: Response, username: str, password: str): # 用户名密码验证、生成Token的业务逻辑 token = generate_jwt_token(username) # 往响应头添加Bearer Token response.headers["Authorization"] = f"Bearer {token}" # 保持原有JSON返回(可选,方便客户端按需取用) return {"access_token": token, "token_type": "bearer"}
这样客户端请求登录接口后,响应头会自动携带Token,多数HTTP客户端工具(如Postman、浏览器)会自动缓存该头信息,用于后续授权请求。
2. 实现RBAC权限控制
第一步:完善用户模型
在用户数据表中新增role字段(取值如"user"/"admin"),确保认证时能获取用户角色标识:
# SQLAlchemy示例模型 from sqlalchemy import Column, Integer, String from sqlalchemy.ext.declarative import declarative_base Base = declarative_base() class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) username = Column(String, unique=True, index=True) hashed_password = Column(String) role = Column(String, default="user") # 默认普通用户
第二步:自定义权限依赖函数
创建两个基础依赖,分别用于验证登录态和管理员权限:
from fastapi import Depends, HTTPException, status from jose import JWTError, jwt from sqlalchemy.orm import Session # 假设已实现:获取数据库会话、OAuth2认证方案、密钥配置 from .database import get_db from .core.config import SECRET_KEY, ALGORITHM oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login") def get_current_user(db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)): credentials_err = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="凭证验证失败", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) username = payload.get("sub") if not username: raise credentials_err except JWTError: raise credentials_err user = db.query(User).filter(User.username == username).first() if not user: raise credentials_err return user def get_current_admin(current_user: User = Depends(get_current_user)): if current_user.role != "admin": raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="仅管理员可访问此接口" ) return current_user
第三步:给接口绑定权限依赖
- 查看物品接口:仅登录用户可访问,绑定
get_current_user - 创建物品接口:修复未认证问题,仅管理员可访问,绑定
get_current_admin
# 查看物品接口 @app.get("/items") def list_items(current_user: User = Depends(get_current_user), db: Session = Depends(get_db)): return db.query(Item).all() # 创建物品接口(修复未认证漏洞) @app.post("/items") def create_item(item: ItemCreate, current_user: User = Depends(get_current_admin), db: Session = Depends(get_db)): db_item = Item(**item.dict(), owner_id=current_user.id) db.add(db_item) db.commit() db.refresh(db_item) return db_item
第四步:防止IDOR漏洞
操作单个物品(查询/修改/删除)时,校验物品归属:普通用户仅能操作自身物品,管理员不受限制:
@app.get("/items/{item_id}") def get_single_item(item_id: int, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)): item = db.query(Item).filter(Item.id == item_id).first() if not item: raise HTTPException(status_code=404, detail="物品不存在") # 权限校验:管理员通杀,普通用户仅能访问自己的物品 if current_user.role != "admin" and item.owner_id != current_user.id: raise HTTPException(status_code=403, detail="无权限访问此物品") return item
3. 项目架构优化与权限适配
推荐分层架构
采用模块化分层结构,便于权限逻辑的统一维护和扩展:
your_project/ ├── main.py # FastAPI实例入口 ├── api/ # 路由层 │ ├── auth.py # 登录/注册路由 │ └── items.py # 物品相关路由 ├── core/ # 核心配置与工具 │ ├── config.py # 密钥、数据库配置 │ └── security.py # Token生成、权限依赖函数 ├── crud/ # 数据库CRUD操作 │ ├── users.py # 用户CRUD │ └── items.py # 物品CRUD ├── models/ # 数据库ORM模型 │ ├── users.py │ └── items.py ├── schemas/ # Pydantic验证模型 │ ├── users.py │ └── items.py └── database.py # 数据库连接配置
新增功能的权限适配流程
- 如果需要新的权限角色,在
core/security.py中新增对应的依赖函数(如get_current_moderator) - 在路由文件中,给新接口绑定对应的权限依赖
- 在CRUD函数中,根据权限过滤数据(比如普通用户只能查询自己的资源)
关于fastapi_users融合问题
如果现有代码已经有完整的认证逻辑,不建议强行替换为fastapi_users,避免不必要的重构成本。基于现有代码自定义权限控制更高效。如果一定要用fastapi_users,需要:
- 将现有用户模型替换为fastapi_users提供的模型(或继承扩展)
- 用fastapi_users的认证依赖替换自定义的
get_current_user - 扩展用户模型添加
role字段,基于角色实现RBAC逻辑
内容的提问来源于stack exchange,提问作者Sunil Raj
相关产品推荐
相关产品推荐

