00:00
音乐列表-选一首爱听的吧！

文章目录

加载目录中...

JWT认证与FastAPI利用JWT进行用户认证的最佳实践

作者：依然会有dream
发布时间：2025-11-12
分类：FastAPI
评论：0

11.用户认证

11.1 登录认证概述

登录是很多系统的基本功能, 有些页面(如用户信息页面)需要登录之后才能进行访问. 实现这一功能的方案大体为:

  1. 首先进行登录, 登录成功后, 给前端(浏览器)返回一个值"xxxx"(session或者token)
  2. 前端(浏览器)去访问需要登录的页面(如用户信息页面)时, 会带上上面值"xxxx"
  3. (后端)服务器根据传入的值"xxxx"获取到这个值对应的用户是哪一个, 那么就返回这个用户的信息

上面的方案再根据 返回值"xxx" 的方式不同, 可以细分为两种:

  • session:
  1. 用户登录后, 在服务器这一方会生成一个随机值session_id, 把session_id和用户的唯一标识(如user_id)映射起来保存, 可以保存在数据库中或者内存中, 然后给前端返回这个session_id
  2. 前端下次请求时, 带上这个session_id, 后端可以根据这个session_id找到对应的user_id, 即可知道请求的是哪个用户的信息了

但是这种方式由于需要服务器端保存这个session_id, 因此会衍生出一些问题, 如:

a. 保存session_id需要耗费服务器资源

b. 当业务量比较大需要多台机器进行负载均衡, 统一提供服务时, 多台机器的session_id需要进行同步, 否则跨机器访问时就获取不到用户了

于是就慢慢发展出了第二种方式: token

  • token:
  1. 用户登录后, 服务器端根据user_id和秘钥(盐值)进行签名加密, 直接返回给前端(浏览器), 不进行保存操作
  2. 前端下次请求时, 带上这个token, 服务器端对这个token进行解密, 获取到解密后的user_id, 即可知道请求的是哪个用户的信息了

基于token的登录认证解决了基于session方式带来的问题, 且扩张性更强, 已经是现在的首选方案了

11.2 OAuth2

既然采用了上述的token方案, 那么可以考虑的再具体一点, 如何进行token的加密和解密? 后端返回token的格式是什么? 前端访问时token是放在url参数中, 还是请求头中或者请求体中? 前端携带token访问时的格式是什么?

OAuth2就是对上述具体问题的一套规范的解决方案, 当然它不只是解决上面的问题, 也可以解决第三方应用的授权问题等.

在FastAPI中, 提供了多种认证解决方案工具, 其中也包括了OAuth2, 可以使用OAuth2PasswordBearer类来实现OAuth2的功能, 使用的是OAuth2中的一种认证方案, 通过bearer token来携带token, 具体做法就是:

在请求头中添加参数Authorization, 其值为Bearer和token中间使用空格连接形成的字符串, 如Bearer your_token_string, 注意, Authorization和Bearer都是规范中固定的写法, 不可修改。

这个token的生成规则和验证规则, 不属于OAuth2的范畴, 我们一般使用JWT

11.3 JWT认证

整个流程如图：

JWT全称是Json Web Token，在python中我们一般使用PyJWT这个包实现JWT的编解码操作, Fastapi官网使用的是python-jose这个包, 因为它提供了 PyJWT 的所有功能，以及之后与其他工具进行集成时你可能需要的一些其他功能。

安装依赖：

pip install PyJWT

a.FastAPI + JWT 认证核心概念

  1. JWT 组成（token字符串长什么样） Header.Payload.Signature
    • Header：声明类型与签名算法（如 HS256）
    • Payload：业务声明（sub、exp、iat、自定义字段等）
    • Signature：用服务端密钥对「Header+Payload」签名，防篡改
  2. 典型认证流程（无刷新令牌场景） ① 用户拿用户名/密码 → ② 服务端校验 → ③ 签发 AccessToken → ④ 客户端在后续请求 Header 中携带 Authorization: Bearer
  3. 刷新令牌（RefreshToken）场景 AccessToken 有效期短（如 30 min），RefreshToken 长（如 7 天）。 当 AccessToken 过期，客户端用 RefreshToken 请求 /refresh，服务端验证后颁发新 AccessToken，无需再次登录。
  4. FastAPI 官方生态
    • python-jose：JWT 编解码（编码：jwt.encode() 解码：jwt.decode()）
    • passlib：密码哈希（如 bcrypt）
    • 中间件/依赖注入：把「解析并验证 JWT」封装成可复用的依赖函数，路由加依赖即可

b.基本使用

步骤一：首先在.env文件中设置jwt配置参数和密钥

注意，HS256不要加引号，加了后本地没问题，若使用docker容器运行则会报算法不支持错误，原因在于docker环境变量设置问题，会将双引号读取，期望得到的HS256,实际得到的"HS256"，所以报错。踩过坑。

# JWT配置
SECRET_KEY=miyao # 密钥，应为至少32位随机字符串
ALGORITHM=HS256 # 加密算法
ACCESS_TOKEN_EXPIRE_MINUTES=240 # 过期时间

步骤二：在security.py中编写生成与校验jwt令牌的函数

# src.app.core.security
# 导入的包.....
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt


oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/login")


def create_access_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.now(UTC) + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    # JWT编码，函数参数分别为data字典，密钥，加密算法
    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return encoded_jwt

# 将oauth2_scheme作为依赖注入，自动获取请求头中的令牌
def verify_token(token: str = Depends(oauth2_scheme)):
    """验证 JWT 令牌"""
    try:
        # 进行解码，拿到data字典
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        return payload
    except JWTError:
        return None # 若出现错误返回None,这回导致get_current_user依赖注入函数因拿不到用户id报错

关于oauth2_scheme

在 FastAPI 里，oauth2_scheme 是一个 OAuth2PasswordBearer类的 实例，它干了两件核心的事情：

  1. 告诉框架： “本接口（乃至整个依赖链）期望请求在 Authorization: Bearer 头里带一只 JWT。”
  2. 自动把 <token> 提取出来并返回给依赖函数，如果没带或格式不对，直接抛 401 + WWW-Authenticate 头，省得自己再写一堆校验。
  3. 代码层面长这样：
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")   # tokenUrl 可以随便写，仅用于 OpenAPI 文档
  • 参数 tokenUrl="login" 并不会真正发请求，它只会在 Swagger/OpenAPI 页面 里生成一个「Authorize」按钮，告诉前端“去这个路径换 token”。
  • 真正运行时，oauth2_scheme 就是一个 callable（依赖函数），FastAPI 会把它注入到路径操作函数或者嵌套依赖里：
async def verify_token(token: str = Depends(oauth2_scheme)):
    ...

执行顺序：

  1. 请求进来 →
  2. FastAPI 先调用 oauth2_scheme(request) →
  3. 它检查请求头有没有 Authorization: Bearer xxxxx
    • 有：把 xxxxx 原样返回（字符串）
    • 没有 / 格式不对：立即抛 HTTPException(401, headers={"WWW-Authenticate": "Bearer"})，不会进入你的业务代码。

因此你在 verify_token 里拿到的 token 参数，一定是字符串且非空，否则根本进不了函数体。

一句话总结： oauth2_scheme 是 FastAPI 提供的「Bearer 令牌提取器 + 401 守门员」，让你专注写 JWT 解码逻辑，而不用自己反复解析请求头和做错误响应。

步骤三：在登录视图中，当账号密码校验通过后create_access_token()函数生成jwt令牌并以json数据形式返回

from src.app.core.security import verify_password, create_access_token

def login(db, data):
    user = crud.get_user_by_phone(db, data)
    if user:
        hashed_password = user.password
        # 调用verify_password函数校验密码
        if verify_password(data.password, hashed_password):
            data = {"sub": user.user_id} # 将user_id作为sub的值来进行加密，注意：sub的值只能是字符串，不可为整型
            token = create_access_token(data=data)
            return {
                "token": token, # 在json中返回给客户端
                "userId": user.user_id,
                "phoneNumber": user.phone_number,
            }
        else:
            raise HTTPException(status_code=400, detail="用户名或密码错误")
    else:
        raise HTTPException(status_code=400, detail="用户名或密码错误")

步骤四：在get_current_user()中通过依赖注入使用verify_token()函数

from src.app.core.security import verify_token

# 全局认证依赖，通过依赖注入注入verify_token函数，解码jwt令牌并拿到解码得到的payload
def get_current_user(payload=Depends(verify_token), db: Session = Depends(get_db)):
    """
    通过验证 Authorization header 中的 token 获取当前用户。
    """
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        user_id = payload.get("sub") # 从payload中拿到sub对应的user_id值
        if user_id is None:
            raise credentials_exception
        user = get_user(db, user_id) # 通过拿到的user_id值进行数据库查询，拿到当前登录用户对象
        if user is None:
            raise credentials_exception
        return user
    except Exception:
        raise credentials_exception

步骤五：在路由中添加全局认证依赖来保护接口

from fastapi import APIRouter, Depends
from src.app.core.database import get_db
from sqlalchemy.orm import Session
from src.app.core.dependencies import get_current_user
from typing import Annotated

person = APIRouter()


@person.get('/persons/self')
# 通过依赖注入将get_current_user函数注入，current_user是get_current_user函数返回的对象
def get_person_self(current_user: Annotated[User, Depends(get_current_user)], db: Session = Depends(get_db)):
    return service.get_person_self(current_user, db)

整体流程是这样的：

  1. 访问受保护的接口的请求到来
  2. 会先执行注入的get_current_user函数，因为get_current_user函数依赖了verify_token函数，所以会先拿到请求头中的token进行验证。
  3. 若验证未通过，直接报错，不会执行业务函数，若通过，在get_current_user函数中根据解码得到的user_id进行数据库查询拿到对应的用户对象。注入了get_current_user函数的函数就可以直接使用current_user这个对象进行一系列操作了（返回用户信息等）。

使用postman进行测试：

未携带令牌（即未携带Authorization请求头）：

携带过期令牌（距发放令牌时间已超过设置的时间）：

携带正确未过期令牌：

携带错误令牌（令牌被篡改）：

返回文章列表

评论区 0