当前位置：首页 > news >正文

FastAPI系列12：使用JWT 登录认证和RBAC 权限控制

news 来源：原创 2025/5/16 13:43:03

使用JWT 登录认证和RBAC 权限控制

- 1、身份认证（Authentication）与JWT
- - 身份认证（Authentication）的方式
  - JWT（JSON Web Token）的实现原理
- 2、授权（Authorization）与RBAC
- - 授权（Authorization）的方式
  - RBAC的实现原理
- 3、FastAPI 实现 JWT与RBAC
- - 实现思路
  - 完整的实现代码

在开发 Web API 应用时， 安全问题是非常核心的考虑因素。在众多安全问题中， 身份认证（Authentication）与授权（Authorization）是首要的问题，本节我们就在FastAPI中提供一个基于JWT和RBAC的登录认证及权限控制的实现方式。

1、身份认证（Authentication）与JWT

身份认证（Authentication）的方式

在Web应用中，身份认证主要为了确保用户是谁，在目前常见的认证方式包括：

认证方式	描述
Session + Cookie	服务端维护登录状态（Session），前端用 Cookie 自动带上 Session ID
JWT（JSON Web Token）	无状态认证，服务端不存储登录状态，Token 中携带用户信息，前端存储于 localStorage 或 Cookie
OAuth 2.0	用于授权第三方访问用户数据（如 GitHub 登录）
API Key	简单方式，客户端发送固定密钥，适合低安全场景
SAML/OpenID	企业级单点登录协议，较复杂

JWT（JSON Web Token）的实现原理

在上述Web应用认证方式中，JWT实现较为简单，且对前后端分离的开发模式比较友好，应用较为广泛，其原理可以总结为以下几个核心步骤：
1. Token 结构
JWT 由三部分组成，通过 . 分隔，如header.payload.signature。

header：声明签名算法，如 HS256。
payload：实际数据（如用户ID、权限、过期时间等）。
signature：用密钥对 header 和 payload 进行签名，防止篡改。

2. 签发（生成）过程
服务器在用户登录成功后：

构造 header 与 payload；
使用密钥和算法生成 signature；
拼接三个部分生成最终 Token；
将 Token 返回给前端。

3. 验证过程
当客户端带着 Token 访问受保护接口时：

服务器解析 Token，读取并验证 signature；
检查 exp 是否过期；
若验证通过，提取 payload 中的信息用于身份识别和权限判断。

JWT认证过程时序图

2、授权（Authorization）与RBAC

授权（Authorization）的方式

在 Web 应用中，常见的授权方式主要包括以下几种：

方式	描述
RBAC（基于角色）	用户被分配角色，每个角色有一组权限，最常用，易管理。
ABAC（基于属性）	根据用户属性、资源属性、环境等做出细粒度判断（如年龄、时间、IP）。
PBAC（基于策略）	使用策略语言（如 OPA、Rego）动态决策，灵活但复杂度较高。
ACL（访问控制列表）	每个资源列出允许访问的用户列表，适合小型系统。
MAC（强制访问控制）	安全级别固定（如军事系统），不允许随意改变访问权限。

RBAC的实现原理

RBAC（Role-Based Access Control）的实现原理是通过用户-角色-权限三层关系控制资源访问。
核心概念
在RBAC中，有以下三个核心概念：

用户（User）
系统中的实体，如员工、用户账号等。
角色（Role）
权限的集合，例如 admin、editor、viewer。
权限（Permission）
对资源执行操作的能力，如 read:report、delete:user。
在这个三个概念中，一个用户可被分配多个角色，每个角色拥有多个权限，而权限最终控制着具体资源的访问。

授权流程

用户登录，系统识别其绑定角色；
请求接口时，系统根据角色检索权限；
若请求的资源和操作在权限列表中，允许访问；否则拒绝。

3、FastAPI 实现 JWT与RBAC

有了上面关于JWT与RBAC基本理解，下面我们给出在FastAPI 中实现 JWT与RBAC的思路与具体代码示例。

实现思路

JWT 认证实现

创建登录接口，用户提交 username/password。
验证成功后，生成 JWT Token（用 PyJWT 或 python-jose 库），把用户信息（如 user_id, roles）编码进 token。
返回给客户端：access_token（短有效期，如 15 分钟）、refresh_token（长有效期，如 7 天）。
access_token 过期后，客户端使用 refresh_token 调用 /refresh，获得新的 access_token。

RBAC 权限控制实现

在用户 Token 中加上角色（roles）字段。
每个接口定义需要的权限角色。
写一个权限校验的依赖项，比如：

from fastapi import Depends, HTTPException, statusdef role_required(required_roles: list):def checker(user = Depends(get_current_user)):if not any(role in user.roles for role in required_roles):raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Forbidden")return userreturn checker

然后在路由上使用：

@app.get("/admin/dashboard", dependencies=[Depends(role_required(["admin"]))])
async def admin_dashboard():return {"message": "Welcome, Admin!"}

JWT 认证与RBAC 权限控制的时序图

完整的实现代码

项目目录结构

 fastapi_jwt_rbac/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── auth.py
│   ├── deps.py
│   └── models.py
├── requirements.txt

app/models.py

from pydantic import BaseModel
from typing import Listclass User(BaseModel):username: strhashed_password: strroles: List[str]

app/auth.py

from jose import jwt, JWTError
from passlib.context import CryptContext
from datetime import datetime, timedelta
from app.models import UserSECRET_KEY = "your_access_secret"
REFRESH_SECRET_KEY = "your_refresh_secret"
ALGORITHM = "HS256"
ACCESS_EXPIRE_MINUTES = 15
REFRESH_EXPIRE_DAYS = 7pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")fake_user_db = {"alice": {"username": "alice","hashed_password": pwd_context.hash("password123"),"roles": ["admin"]},"bob": {"username": "bob","hashed_password": pwd_context.hash("password456"),"roles": ["user"]}
}def verify_password(plain, hashed): return pwd_context.verify(plain, hashed)def authenticate_user(username, password):user = fake_user_db.get(username)if user and verify_password(password, user["hashed_password"]):return User(**user)def create_access_token(data: dict):to_encode = data.copy()expire = datetime.utcnow() + timedelta(minutes=ACCESS_EXPIRE_MINUTES)to_encode.update({"exp": expire})return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)def create_refresh_token(data: dict):to_encode = data.copy()expire = datetime.utcnow() + timedelta(days=REFRESH_EXPIRE_DAYS)to_encode.update({"exp": expire})return jwt.encode(to_encode, REFRESH_SECRET_KEY, algorithm=ALGORITHM)def decode_token(token: str):return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])def decode_refresh_token(token: str):return jwt.decode(token, REFRESH_SECRET_KEY, algorithms=[ALGORITHM])

app/deps.py

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from app.auth import decode_token
from app.models import Useroauth2_scheme = OAuth2PasswordBearer(tokenUrl="/login")def get_current_user(token: str = Depends(oauth2_scheme)) -> User:try:payload = decode_token(token)return User(username=payload["sub"], hashed_password="", roles=payload["roles"])except:raise HTTPException(status_code=401, detail="Invalid token")def role_required(roles: list):def checker(user: User = Depends(get_current_user)):if not any(role in user.roles for role in roles):raise HTTPException(status_code=403, detail="Insufficient role")return userreturn checker

app/main.py

from fastapi import FastAPI, Depends, HTTPException, status, Body
from fastapi.security import OAuth2PasswordRequestForm
from app.auth import authenticate_user, create_access_token, create_refresh_token, decode_refresh_token
from app.deps import get_current_user, role_requiredapp = FastAPI()@app.post("/login")
def login(form_data: OAuth2PasswordRequestForm = Depends()):user = authenticate_user(form_data.username, form_data.password)if not user:raise HTTPException(status_code=401, detail="Invalid credentials")return {"access_token": create_access_token({"sub": user.username, "roles": user.roles}),"refresh_token": create_refresh_token({"sub": user.username, "roles": user.roles}),"token_type": "bearer"}@app.post("/refresh")
def refresh(refresh_token: str = Body(...)):try:payload = decode_refresh_token(refresh_token)return {"access_token": create_access_token({"sub": payload["sub"], "roles": payload["roles"]})}except:raise HTTPException(status_code=401, detail="Invalid refresh token")@app.get("/user")
def user_area(user = Depends(get_current_user)):return {"msg": f"Hello {user.username}"}@app.get("/admin")
def admin_area(user = Depends(role_required(["admin"]))):return {"msg": f"Welcome Admin {user.username}"}

requirements.txt

fastapi
uvicorn
python-jose
passlib[bcrypt]