Python 接口开发中怎么实现 JWT 身份验证与权限控制

一、核心认知：身份验证与权限控制的界限

在动手编写代码之前，必须厘清两个经常被混为一谈的概念：身份验证（Authentication，简称 AuthN）与权限控制（Authorization，简称 AuthZ）。

• 身份验证（Who are you?）：解决“你是谁”的问题。用户提交用户名和密码，系统校验正确后签发 JWT。后续请求中，服务端检查请求头中的 Token 签名是否合法、是否过期，从中恢复用户 ID 等凭证。如果 Token 无效，服务端返回 401 Unauthorized。
• 权限控制（What can you do?）：解决“你能做什么”的问题。在确认了用户是谁之后，系统判定该用户是否有权调用当前接口或操作特定数据（例如普通用户不能执行删除接口）。如果不具备操作权限，服务端返回 403 Forbidden。

JSON Web Token（JWT）本质上是一串由点号（.）分割的三段式无状态加密字符串。它自身不存储于服务端会话内存中，所有的用户声明均内嵌在 Token 内，由数字签名保证其无法被客户端篡改。

二、技术选型与环境依赖准备

在现代 Python 接口开发中，FastAPI 凭借其强大的类型注解和自带的 Depends 依赖注入系统，是实现 JWT 鉴权与 RBAC 的标杆工具。底层加解密库推荐使用维护稳定、符合 RFC 7519 规范的 PyJWT，密码单向哈希加密推荐使用 Passlib (Bcrypt)。

# 在虚拟环境中安装必要依赖
pip install fastapi "uvicorn[standard]" pyjwt "passlib[bcrypt]" python-multipart

需要注意，python-multipart 是处理用户登录表单（OAuth2PasswordRequestForm）的标准前置依赖；如果使用原生 JSON 格式提交登录，可按需选装。

三、第一步：实现 JWT 签发与底层编解码工具模块

首先创建安全工具模块 security.py。它承担三项具体任务：

1. 密码哈希加盐与登录比对；
2. 按标准 Payload 声明（包含主体 sub 与过期时间 exp）签发 Token；
3. 在解码时严谨处理过期与篡改异常，绝不吞掉错误。

# security.py
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, Optional
import jwt
from passlib.context import CryptContext

# 生产环境中务必从系统环境变量中读取，长度建议至少 32 字符
SECRET_KEY = "your-fallback-safe-secret-key-replace-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """验证明文密码与哈希摘要是否匹配"""
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    """为明文密码生成 bcrypt 安全哈希"""
    return pwd_context.hash(password)

def create_access_token(data: Dict[str, Any], expires_delta: Optional[timedelta] = None) -> str:
    """生成带有过期时间的 JWT Access Token"""
    to_encode = data.copy()
    now_utc = datetime.now(timezone.utc)
    if expires_delta:
        expire = now_utc + expires_delta
    else:
        expire = now_utc + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    
    # 注入标准声明 exp (UNIX timestamp) 和 iat (签发时间)
    to_encode.update({"exp": expire, "iat": now_utc})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

def decode_access_token(token: str) -> Dict[str, Any]:
    """
    解码并校验 JWT
    若签名错误或已过期，将抛出 jwt.PyJWTError 子类异常
    """
    return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])

四、第二步：利用依赖注入提取身份凭证（身份验证）

客户端在发起受保护接口的 HTTP 请求时，通常将 Token 放置于请求头中：

Authorization: Bearer <your_jwt_token_here>

我们使用 FastAPI 的 HTTPBearer 或 OAuth2PasswordBearer 来自动提取该 Header，并编写依赖项 get_current_user 将解密出的用户模型注入路由。

编写核心依赖模块 dependencies.py：

# dependencies.py
from typing import Optional, List
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
from pydantic import BaseModel
from security import decode_access_token

# 使用 HTTPBearer 自动提取请求头中的 Authorization: Bearer 
security_scheme = HTTPBearer()

class UserModel(BaseModel):
    user_id: int
    username: str
    roles: List[str] = []
    is_active: bool = True

# 模拟数据库或缓存中的用户查询，实际开发中替换为 SQLAlchemy 或 Tortoise ORM 查询
FAKE_USER_DB = {
    1: {"user_id": 1, "username": "alice", "roles": ["admin", "editor"], "is_active": True},
    2: {"user_id": 2, "username": "bob", "roles": ["viewer"], "is_active": True},
    3: {"user_id": 3, "username": "charlie", "roles": ["viewer"], "is_active": False},
}

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security_scheme)
) -> UserModel:
    """从 Token 解码获取当前已认证用户"""
    token = credentials.credentials
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭据或凭据已损坏",
        headers={"WWW-Authenticate": "Bearer"},
    )
    
    try:
        payload = decode_access_token(token)
        user_id_str: Optional[str] = payload.get("sub")
        if user_id_str is None:
            raise credentials_exception
        user_id = int(user_id_str)
    except jwt.ExpiredSignatureError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token 已经过期，请重新登录",
            headers={"WWW-Authenticate": "Bearer"},
        )
    except (jwt.PyJWTError, ValueError):
        raise credentials_exception

    user_dict = FAKE_USER_DB.get(user_id)
    if not user_dict:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户不存在",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    user = UserModel(**user_dict)
    if not user.is_active:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="该用户账户已被冻结",
        )
    return user

五、第三步：实现细粒度权限控制（RBAC 守卫）

通过身份验证后，系统明确了请求者的真实身份。接下来需要限制接口的访问权限。业界普遍采用 基于角色的访问控制（RBAC, Role-Based Access Control） 模型。

在 Python 中，最佳实践并非在每个接口内部写 if "admin" not in user.roles: raise ...，而是编写一个可复用的依赖项工厂（Callable Dependency Factory）或装饰器。

编写角色权限守卫工厂：

# permissions.py
from typing import List, Callable
from fastapi import Depends, HTTPException, status
from dependencies import get_current_user, UserModel

class RequireRole:
    """角色守卫依赖工厂类"""
    def __init__(self, allowed_roles: List[str]):
        self.allowed_roles = set(allowed_roles)

    def __call__(self, user: UserModel = Depends(get_current_user)) -> UserModel:
        user_roles = set(user.roles)
        # 如果用户角色与允许访问的角色集合没有交集，则拒绝访问
        if not self.allowed_roles.intersection(user_roles):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"权限不足：当前操作需要具备以下角色之一：{list(self.allowed_roles)}"
            )
        return user

六、第四步：完整的 API 接口整合示例

将上述模块串联，创建完整的 FastAPI 应用 main.py。包含用户登录签发 Token、获取个人信息（仅需身份验证）以及删除系统文章（需要管理员权限）三个典型接口。

# main.py
from fastapi import FastAPI, Depends, HTTPException, status
from pydantic import BaseModel
from security import create_access_token, get_password_hash, verify_password
from dependencies import get_current_user, UserModel
from permissions import RequireRole

app = FastAPI(title="JWT 认证与权限控制演示接口")

# 模拟用户登录存储（Alice 密码为 secret123）
ACCOUNTS_DB = {
    "alice": {
        "user_id": 1,
        "username": "alice",
        "hashed_password": get_password_hash("secret123"),
        "roles": ["admin"]
    },
    "bob": {
        "user_id": 2,
        "username": "bob",
        "hashed_password": get_password_hash("password456"),
        "roles": ["viewer"]
    }
}

class LoginRequest(BaseModel):
    username: str
    password: str

class TokenResponse(BaseModel):
    access_token: str
    token_type: str
    expires_in_minutes: int

@app.post("/auth/login", response_model=TokenResponse, summary="用户登录并获取 JWT")
async def login(req: LoginRequest):
    user_record = ACCOUNTS_DB.get(req.username)
    if not user_record or not verify_password(req.password, user_record["hashed_password"]):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误"
        )
    
    # 签发 JWT：sub 存入字符串形式的用户唯一 ID
    token_payload = {
        "sub": str(user_record["user_id"]),
        "username": user_record["username"],
        "roles": user_record["roles"]
    }
    token = create_access_token(data=token_payload)
    return TokenResponse(
        access_token=token,
        token_type="bearer",
        expires_in_minutes=60
    )

@app.get("/api/user/profile", summary="获取当前登录用户的个人资料（需登录）")
async def get_profile(current_user: UserModel = Depends(get_current_user)):
    return {
        "message": "验证成功",
        "user_id": current_user.user_id,
        "username": current_user.username,
        "roles": current_user.roles
    }

@app.delete("/api/articles/{article_id}", summary="删除指定文章（仅限管理员角色）")
async def delete_article(
    article_id: int,
    # 注入权限依赖：必须拥有 admin 角色
    admin_user: UserModel = Depends(RequireRole(["admin"]))
):
    return {
        "message": f"文章 {article_id} 已由管理员 {admin_user.username} 成功删除",
        "status": "success"
    }

接口执行效果验证

七、生产级工程实践与安全进阶

在实际生产部署中，单纯依靠基础 JWT 仍然会面临两个安全挑战：“Token 签发后无法中途吊销”以及“Token 有效期太长存在被截获风险”。推荐采用“双 Token 轮转”与“Redis 黑名单”架构。

常见排查与错误处理

• Signature verification failed（签名校验失败）：通常是因为解码时使用的 SECRET_KEY 或 ALGORITHM 与签发时不一致；也有可能是反向代理（如 Nginx）修改了 Header 中的部分编码。
• 时钟同步偏差引发的异常：在多台分布式机器间，如果服务器硬件时钟有微小不同步，刚签发的 Token 在其他服务器上解码可能会触发 ImmatureSignatureError 或提前过期。PyJWT 解码时可传入 leeway=10 允许 10 秒以内的微同时钟偏差。
• CORS 跨域请求丢失 Authorization：前端跨域访问时，如果服务端未在 CORS 中间件中配置允许 Authorization 头，浏览器预检请求（OPTIONS）可能会阻断携带 Token 的实际请求。