怎样用 FastAPI 和 RBAC 打造坚如盘石的安全堡垒?

[复制链接]
发表于 2025-6-27 14:59:57 | 显示全部楼层 |阅读模式

马上注册,结交更多好友,享用更多功能,让你轻松玩转社区。

您需要 登录 才可以下载或查看,没有账号?立即注册

×
扫描二维码
关注大概微信搜一搜:编程智域 前端至全栈交流与成长
发现1000+提升效率与开发的AI工具和实用程序:https://tools.cmdragon.cn/
一、FastAPI 安全认证与 RBAC 体系原理

1.1 RBAC 基础概念

基于脚色的访问控制(RBAC)是保护体系资源的经典模型,其核心包含三个要素:

  • 用户(User):体系操纵主体
  • 脚色(Role):权限的聚集载体(如:管理员、普通用户)
  • 权限(Permission):详细操纵权限(如:商品删除、订单修改)
1.2 FastAPI 安全组件

核心依赖包及版本要求:
  1. fastapi == 0.95
  2. .2
  3. uvicorn == 0.22
  4. .0
  5. python - jose[cryptography] == 3.3
  6. .0
  7. passlib[bcrypt] == 1.7
  8. .4
复制代码
安装命令:
  1. pip install fastapi uvicorn python-jose[cryptography] passlib[bcrypt]
复制代码
二、电商平台 RBAC 体系实现

2.1 数据模型定义

利用 Pydantic 和 SQLAlchemy 构建领域模型:
  1. from pydantic import BaseModel
  2. from sqlalchemy import Column, Integer, String, Table, ForeignKey
  3. class PermissionBase(BaseModel):
  4.     code: str  # 权限编码,如 "order:delete"
  5.     description: str
  6. class RoleCreate(BaseModel):
  7.     name: str
  8.     permission_ids: list[int]
  9. # SQLAlchemy 模型
  10. user_role = Table(
  11.     'user_role', Base.metadata,
  12.     Column('user_id', ForeignKey('users.id')),
  13.     Column('role_id', ForeignKey('roles.id'))
  14. )
  15. role_permission = Table(
  16.     'role_permission', Base.metadata,
  17.     Column('role_id', ForeignKey('roles.id')),
  18.     Column('permission_id', ForeignKey('permissions.id'))
  19. )
复制代码
2.2 认证体系实现

JWT 令牌签发与验证:
  1. from jose import JWTError, jwt
  2. from datetime import datetime, timedelta
  3. SECRET_KEY = "your-secret-key"
  4. ALGORITHM = "HS256"
  5. def create_access_token(data: dict) -> str:
  6.     expire = datetime.utcnow() + timedelta(hours=1)
  7.     return jwt.encode(
  8.         {**data, "exp": expire},
  9.         SECRET_KEY,
  10.         algorithm=ALGORITHM
  11.     )
  12. async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
  13.     try:
  14.         payload = jwt.decode(token, SECRET_KEY, [ALGORITHM])
  15.         return await UserService.get_user(payload["sub"])
  16.     except JWTError:
  17.         raise HTTPException(401, "Invalid token")
复制代码
2.3 权限校验体系

动态权限依赖注入:
  1. def check_permission(permission_code: str):
  2.     async def _permission_checker(
  3.             user: User = Depends(get_current_user)
  4.     ):
  5.         if not any(p.code == permission_code for p in user.permissions):
  6.             raise HTTPException(403, "Permission denied")
  7.     return Depends(_permission_checker)
  8. # 在路由中使用
  9. @app.delete("/orders/{order_id}")
  10. async def delete_order(
  11.         order_id: str,
  12.         _=check_permission("order:delete")
  13. ):
  14.     return OrderService.delete_order(order_id)
复制代码
三、安全防护策略

3.1 敏感数据加密

暗码存储安全处理:
  1. from passlib.context import CryptContext
  2. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  3. def verify_password(plain_password: str, hashed_password: str) -> bool:
  4.     return pwd_context.verify(plain_password, hashed_password)
  5. def get_password_hash(password: str) -> str:
  6.     return pwd_context.hash(password)
复制代码
3.2 哀求速率限制

防止暴力破解攻击:
  1. from fastapi import Request
  2. from fastapi.middleware import Middleware
  3. class RateLimiter:
  4.     def __init__(self, times: int, seconds: int):
  5.         self.times = times
  6.         self.seconds = seconds
  7.     async def __call__(self, request: Request):
  8.         client_ip = request.client.host
  9.         # 使用 Redis 记录请求次数
  10.         current = await redis.incr(client_ip)
  11.         if current > self.times:
  12.             raise HTTPException(429, "Too many requests")
  13.         await redis.expire(client_ip, self.seconds)
复制代码
四、测试与调试

4.1 单位测试示例

权限验证测试用例:
  1. from fastapi.testclient import TestClient
  2. def test_admin_access():
  3.     client = TestClient(app)
  4.     # 普通用户令牌
  5.     token = create_test_token(user_type="user")
  6.     response = client.delete(
  7.         "/products/123",
  8.         headers={"Authorization": f"Bearer {token}"}
  9.     )
  10.     assert response.status_code == 403
  11.     # 管理员令牌
  12.     admin_token = create_test_token(user_type="admin")
  13.     response = client.delete(
  14.         "/products/123",
  15.         headers={"Authorization": f"Bearer {admin_token}"}
  16.     )
  17.     assert response.status_code == 200
复制代码
4.2 常见错误处理

错误 1:401 Unauthorized
  1. {
  2.   "detail": "Could not validate credentials"
  3. }
复制代码
解决方法:

  • 检查哀求头 Authorization 格式是否精确
  • 确认令牌未逾期
  • 验证密钥是否匹配
错误 2:403 Forbidden
  1. {
  2.   "detail": "Permission denied"
  3. }
复制代码
预防发起:

  • 在路由处理器前添加权限检查中心件
  • 定期检察脚色权限分配
课后测验

问题 1: 为什么要在暗码哈希中利用 salt?
答案: Salt 随机值能有效防止彩虹表攻击,即使相同暗码也会生成差别的哈希值
问题 2: 怎样实现动态权限管理?
答案: 应该创建权限管理体系接口,允许管理员动态配置脚色与权限的对应关系
问题 3: JWT 令牌为什么要设置逾期时间?
答案: 缩短令牌有效期可以低沉令牌泄露风险,发起结合 refresh token 利用
版本兼容说明

全部代码在以下环境验证通过:

  • Python 3.8+
  • FastAPI 0.95+
  • SQLAlchemy 1.4+
  • Redis 6.2+
余下文章内容请点击跳转至 个人博客页面 大概 大概微信搜一搜:编程智域 前端至全栈交流与成长
,阅读完备的文章:怎样用 FastAPI 和 RBAC 打造坚如盘石的电商堡垒?
往期文章归档:

回复

使用道具 举报

登录后关闭弹窗

登录参与点评抽奖  加入IT实名职场社区
去登录
快速回复 返回顶部 返回列表