FastAPI中构建安全且弹性的权限管控方案

在FastAPI中构建可靠且可变通的权限管控体系

环境配置要求（使用前需安装）

需安装以下依赖：
- fastapi==0.95.0
- uvicorn==0.21.1
- python-multipart==0.0.6
- sqlalchemy==1.4.46
- pydantic==1.10.7
- passlib==1.7.4

1. 权限系统核心原理

权限系统的核心是对请求的过滤机制，FastAPI利用依赖注入体系来实现层级化的验证流程。当请求到来时，会依次经历：
- 身份认证 → 角色验证 → 权限校验 这三级验证
- 每一层级均为独立的依赖项
- 权限数据存储于关系型数据库，支持动态化管理

2. 数据库模型设计

定义了用于存储用户、角色及权限关联信息的数据库模型：

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from databases import Base


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True)
    hashed_password = Column(String(300))
    is_active = Column(Boolean, default=True)
    role_id = Column(Integer, ForeignKey("roles.id"))

    role = relationship("Role", back_populates="users")


class Role(Base):
    __tablename__ = "roles"
    id = Column(Integer, primary_key=True)
    name = Column(String(20), unique=True)
    permissions = Column(String(500))  # 以逗号分隔存储权限标识
    users = relationship("User", back_populates="role")


class PermissionRegistry(Base):
    __tablename__ = "permission_registry"
    id = Column(Integer, primary_key=True)
    endpoint = Column(String(100))  # 路由路径
    method = Column(String(10))  # HTTP方法
    perm_code = Column(String(50))  # 权限标识

3. 权限验证依赖项

创建权限验证器类，在调用时检查当前用户的角色权限是否包含所需权限：

from fastapi import Depends, HTTPException, status
from pydantic import BaseModel


class PermissionValidator:
    def __init__(self, required_perm: str):
        self.required_perm = required_perm

    async def __call__(self,
                       current_user: User = Depends(get_current_user),
                       db: Session = Depends(get_db)):
        # 获取角色关联的权限
        role_perms = current_user.role.permissions.split(",")

        # 验证权限是否存在
        if self.required_perm not in role_perms:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="无访问权限"
            )

        # 记录审计日志（示例）
        audit_log = AuditLog(
            user_id=current_user.id,
            action=f"访问需要 {self.required_perm} 权限的端点"
        )
        db.add(audit_log)
        db.commit()


# 使用示例
@app.get("/admin/dashboard")
async def admin_dashboard(
        perm_check: bool = Depends(PermissionValidator("admin_dashboard"))):
    return {"message": "欢迎来到管理面板"}

4. 动态路由权限注册

通过定义模型接收权限信息并动态注册到数据库：

class PermissionRegistration(BaseModel):
    endpoint: str
    methods: List[str]
    perm_code: str


@app.post("/manage/permissions")
async def register_permission(
        perm_data: PermissionRegistration,
        db: Session = Depends(get_db)
):
    for method in perm_data.methods:
        existing = db.query(PermissionRegistry).filter_by(
            endpoint=perm_data.endpoint,
            method=method
        ).first()

        if not existing:
            new_perm = PermissionRegistry(
                endpoint=perm_data.endpoint,
                method=method,
                perm_code=perm_data.perm_code
            )
            db.add(new_perm)

    db.commit()
    return {"status": "权限注册成功"}

5. 实时权限检查中间件

设置中间件实时检查请求权限，跳过非业务端点：

@app.middleware("http")
async def dynamic_permission_check(request: Request, call_next):
    # 跳过文档相关端点
    if request.url.path.startswith(("/docs", "/redoc")):
        return await call_next(request)

    # 查询权限注册表
    db = SessionLocal()
    perm_record = db.query(PermissionRegistry).filter_by(
        endpoint=request.url.path,
        method=request.method
    ).first()

    if perm_record:
        # 验证用户权限
        current_user = await get_current_user(request)
        if perm_record.perm_code not in current_user.role.permissions.split(","):
            return JSONResponse(
                status_code=403,
                content={"detail": "权限不足"}
            )

    response = await call_next(request)
    return response

课后小测

1. 当用户访问需要特定权限的接口但自身权限不足时，系统会返回什么状态码？
   A) 401 B) 403 C) 404 D) 500
   答案：B) 403。权限验证不通过时会返回403 Forbidden状态码。

2. 如何防止权限注册接口被未授权访问？
   A) 添加JWT认证依赖
   B) 限制仅管理员角色可访问
   C) 同时实现A和B
   D) 无需保护该接口
   答案：C) 需同时进行JWT认证并限制仅管理员角色访问。

常见报错处理

1. 422 Unprocessable Entity
   原因：请求体不符合Pydantic模型验证
   解决：检查字段类型是否正确，补充缺失的必填字段

2. AttributeError: 'NoneType' has no attribute 'permissions'
   原因：用户角色未正确关联
   解决：检查数据库中用户与角色的关联关系，确保每个用户都有对应的角色

3. 数据库连接超时
   预防：配置SQLAlchemy连接池，示例配置：
   python
SQLALCHEMY_DATABASE_URL = "postgresql://user:pass@localhost/dbname?connect_timeout=10"
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
pool_size=20,
max_overflow=10,
pool_timeout=30
)

更多详细内容可跳转至个人博客查看：如何在FastAPI中打造一个既安全又灵活的权限管理系统？ | cmdragon's Blog

往期文章回顾

• FastAPI访问令牌的权限声明与作用域管理：你的API安全真的无懈可击吗？ | cmdragon's Blog
• 如何在FastAPI中构建一个既安全又灵活的多层级权限系统？ | cmdragon's Blog
• 其他相关技术文章可继续在个人博客查阅。