新闻中心

FastAPI 多种认证方式(任选其一)实现指南

2025-11-25
浏览次数:
返回列表

FastAPI 多种认证方式(任选其一)实现指南

本教程详细阐述了如何在 fastapi 中实现多种认证机制(如 basic auth 和 jwt auth),并允许客户端任选其一进行认证。核心方法是修改各个认证依赖项,使其在认证失败时返回 `none` 而非立即抛出异常,从而使一个组合认证依赖能够基于“或”逻辑判断任一认证是否成功,最终实现灵活的多重认证支持。

在构建现代 Web API 时,支持多种认证方式以满足不同客户端或场景的需求是常见的实践。例如,您可能希望同时支持传统的 HTTP Basic 认证和基于令牌的 JWT 认证。然而,在 FastAPI 中直接将多个认证依赖项组合使用时,默认行为可能会导致问题:FastAPI 会强制所有依赖项都通过验证,如果其中一个依赖项在解析阶段抛出 HTTPException,那么后续的依赖项将不会被执行,从而无法实现“任选其一”的逻辑。

理解 FastAPI 依赖注入的默认行为

FastAPI 的依赖注入系统在处理多个 Depends 依赖时,遵循严格的顺序和错误处理机制。当一个依赖项(例如,一个认证函数)内部抛出 HTTPException 时,FastAPI 会立即捕获该异常并返回相应的 HTTP 响应,而不会继续执行后续的依赖项或路由处理函数。这意味着,如果您尝试通过一个自定义的组合依赖函数来包装多个认证依赖,并期望它们能像逻辑“或”一样工作,那么默认情况下是行不通的。

例如,以下尝试实现“或”逻辑的组合认证依赖:

# 假设 jwt_logged_user 和 basic_logged_user 在认证失败时会抛出 HTTPException
def auth_user(jwt_auth: Any = Depends(jwt_logged_user),
              basic_auth: Any = Depends(basic_logged_user)):
    if jwt_auth or basic_auth:
        return jwt_auth or basic_auth
    raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail='Invalid Credentials')

如果 jwt_logged_user 认证失败并抛出异常,那么 basic_auth 依赖将永远不会被解析,auth_user 函数体也永远不会执行,从而无法检查 Basic Auth 是否有效。

核心解决方案:优雅地处理认证失败

要实现“任选其一”的认证逻辑,关键在于修改各个独立的认证依赖项,使其在认证失败时不再立即抛出 HTTPException,而是返回一个指示失败的值(例如 None)。这样,组合认证依赖就可以接收到所有独立认证的结果,并根据这些结果进行逻辑判断。

1. 修改 HTTP Basic 认证依赖

对于 HTTP Basic 认证,FastAPI 提供了 HTTPBasic 类。通过设置 auto_error=False,可以禁用其在认证失败时自动抛出 HTTPException 的行为。

首先,更新 HTTPBasic 实例:

from fastapi.security import HTTPBasic, HTTPBasicCredentials
from typing import Annotated, Optional
import secrets

# ... 其他导入 ...

# 将 auto_error 设置为 False
security = HTTPBasic(auto_error=False)

def basic_logged_user(credentials: Annotated[Optional[HTTPBasicCredentials], Depends(security)]):
    """
    处理 HTTP Basic 认证。
    如果认证失败,返回 None 而非抛出异常。
    """
    if credentials is None:
        # 没有提供 Basic 认证凭据,或者凭据格式不正确 (由 auto_error=False 处理)
        return None

    current_username_bytes = credentials.username.encode("utf8")
    correct_username_bytes = settings.SESSION_LOGIN_USER.encode("utf8")
    is_correct_username = secrets.compare_digest(
        current_username_bytes, correct_username_bytes
    )
    current_password_bytes = credentials.password.encode("utf8")
    correct_password_bytes = settings.SESSION_LOGIN_PASS.encode("utf8")
    is_correct_password = secrets.compare_digest(
        current_password_bytes, correct_password_bytes
    )

    if not (is_correct_username and is_correct_password):
        # 凭据不匹配,返回 None
        return None

    # 认证成功,返回用户名
    return credentials.username

关键点:

美图云修 美图云修

商业级AI影像处理工具

美图云修 50 查看详情 美图云修
  • security = HTTPBasic(auto_error=False):禁用 HTTPBasic 在凭据缺失或格式错误时自动抛出 401 异常。
  • credentials: Annotated[Optional[HTTPBasicCredentials], Depends(security)]:声明 credentials 为 Optional 类型,因为 security 在 auto_error=False 时可能返回 None。
  • if credentials is None::检查是否提供了凭据。
  • if not (is_correct_username and is_correct_password): return None:在自定义逻辑中,如果凭据无效,也返回 None。

2. 修改 JWT 认证依赖

对于 JWT 认证,通常会使用 OAuth2PasswordBearer。同样,通过设置 auto_error=False 来禁用其自动错误处理。此外,您的令牌验证函数 (utils.verify_token) 也应该被修改,以在验证失败时返回 None 或被 try-except 块包裹。

假设 utils.OAuth2_scheme 是 OAuth2PasswordBearer 的实例,并且 utils.verify_token 函数负责验证 JWT。

from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from typing import Annotated, Optional
from jose import JWTError, jwt # 假设您使用 jose 库
from pydantic import BaseModel

# ... 其他导入 ...

# 假设 utils 模块包含 OAuth2_scheme 和 verify_token
# utils.OAuth2_scheme 应该被初始化为 auto_error=False
# 例如:
# class TokenData(BaseModel):
#     username: Optional[str] = None
# class Utils:
#     OAuth2_scheme = OAuth2PasswordBearer(tokenUrl="token", auto_error=False)
#     SECRET_KEY = "your-secret-key" # 替换为实际的密钥
#     ALGORITHM = "HS256"
#     def verify_token(self, token: str) -> dict:
#         try:
#             payload = jwt.decode(token, self.SECRET_KEY, algorithms=[self.ALGORITHM])
#             username: str = payload.get("sub")
#             if username is None:
#                 raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token payload")
#             return {"username": username}
#         except JWTError:
#             raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials")
# utils = Utils()


def jwt_logged_user(token: Annotated[Optional[str], Depends(utils.OAuth2_scheme)],
                    db: Session = Depends(db_session)):
    """
    处理 JWT 认证。
    如果认证失败,返回 None 而非抛出异常。
    """
    if token is None:
        # 没有提供 JWT 令牌,或者令牌格式不正确 (由 auto_error=False 处理)
        return None

    try:
        # 尝试验证令牌,并从 payload 中获取用户名
        # 假设 utils.verify_token 在验证失败时会抛出 HTTPException 或 JWTError
        payload = utils.verify_token(token) # 假设返回一个包含用户信息的字典
        username = payload.get("username") # 根据实际 payload 结构调整

        if username is None:
            return None # 令牌有效但没有找到用户名信息

        # 查询数据库以验证用户是否存在
        user = db.query(User).filter(User.username == username).first()
        if user is None:
            return None # 用户不存在

        # 认证成功,返回用户名
        return user.username
    except HTTPException:
        # 捕获 verify_token 内部抛出的 HTTPException
        return None
    except JWTError: # 捕获 jose 库的 JWT 错误
        return None
    except Exception:
        # 捕获其他未知错误
        return None

关键点:

  • utils.OAuth2_scheme = OAuth2PasswordBearer(tokenUrl="token", auto_error=False):确保 OAuth2PasswordBearer 实例设置了 auto_error=False。
  • token: Annotated[Optional[str], Depends(utils.OAuth2_scheme)]:声明 token 为 Optional 类型。
  • if token is None::检查是否提供了令牌。
  • try...except 块:包裹 utils.verify_token 调用,捕获可能抛出的异常,并在捕获到异常时返回 None。
  • return user.username:认证成功后返回用户名,与 Basic 认证的返回类型保持一致。

3. 实现组合认证逻辑

现在,当 basic_logged_user 和 jwt_logged_user 认证失败时都返回 None,我们可以创建一个新的组合依赖项来检查任一认证是否成功。

from fastapi import HTTPException, status, Depends
from typing import Annotated, Optional, Union

# ... 其他导入 ...

def auth_user(jwt_username: Annotated[Optional[str], Depends(jwt_logged_user)],
              basic_username: Annotated[Optional[str], Depends(basic_logged_user)]) -> str:
    """
    组合认证依赖,允许通过 JWT 或 Basic Auth 中的任一种进行认证。
    如果任一认证成功,返回对应的用户名。
    如果两种认证都失败,则抛出 401 异常。
    """
    if jwt_username:
        # JWT 认证成功
        return jwt_username
    if basic_username:
        # Basic Auth 认证成功
        return basic_username

    # 如果两种认证都失败,则抛出未授权异常
    raise HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail='Invalid Credentials',
        headers={"WWW-Authenticate": "Basic, Bearer"}, # 提示支持的认证方式
    )

# 路由使用组合认证依赖
@router.get("/users/")
async def get_users(db: Session = Depends(db_session), 
                    logged_username: Annotated[str, Depends(auth_user)]):
    """
    获取用户列表,需要通过 JWT 或 Basic Auth 认证。
    """
    # 假设您需要根据用户名进行某些操作,例如权限检查
    print(f"Authenticated user: {logged_username}") 
    query_users = db.query(User).all()
    return query_users

关键点:

  • jwt_username: Annotated[Optional[str], Depends(jwt_logged_user)] 和 basic_username: Annotated[Optional[str], Depends(basic_logged_user)]:接收两个子认证依赖的结果,它们现在可以是 None。
  • if jwt_username: return jwt_username:如果 JWT 认证成功(返回了用户名),则直接返回该用户名。
  • if basic_username: return basic_username:如果 JWT 认证失败但 Basic Auth 成功,则返回 Basic Auth 的用户名。
  • raise HTTPException(...):只有当所有子认证都失败时,才由 auth_user 统一抛出 401 异常。这样确保了只有在没有任何有效凭据的情况下才拒绝访问。
  • headers={"WWW-Authenticate": "Basic, Bearer"}:在 401 响应头中明确告知客户端支持的认证方式。

完整示例代码结构

为了更好地理解,以下是一个整合了上述修改的 FastAPI 应用片段:

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials, OAuth2PasswordBearer
from sqlalchemy.orm import Session
from typing import Annotated, Optional, Union
import secrets
from jose import JWTError, jwt # 假设使用 jose 库

# 假设这些是您的配置和模型
class Settings:
    SESSION_LOGIN_USER = "admin"
    SESSION_LOGIN_PASS = "securepassword"
    JWT_SECRET_KEY = "your-super-secret-jwt-key" # 替换为实际的密钥
    JWT_ALGORITHM = "HS256"

settings = Settings()

# 假设 db_session 是您的数据库会话依赖
# 假设 User 是您的 SQLAlchemy 用户模型
class User: # 简化示例
    def __init__(self, username):
        self.username = username
    def __repr__(self):
        return f"<User {self.username}>"

class MockDBSession: # 模拟数据库会话
    def query(self, model):
        return self
    def filter(self, *args, **kwargs):
        return self
    def first(self):
        return User("testuser") # 模拟返回一个用户
    def all(self):
        return [User("user1"), User("user2")]

def get_db():
    db = MockDBSession()
    try:
        yield db
    finally:
        pass # 实际应用中会关闭会话

db_session = Depends(get_db)

# JWT 相关的工具类
class Utils:
    OAuth2_scheme = OAuth2PasswordBearer(tokenUrl="/token", auto_error=False)
    SECRET_KEY = settings.JWT_SECRET_KEY
    ALGORITHM = settings.JWT_ALGORITHM

    def verify_token(self, token: str) -> dict:
        try:
            payload = jwt.decode(token, self.SECRET_KEY, algorithms=[self.ALGORITHM])
            username: str = payload.get("sub")
            if username is None:
                raise JWTError("Invalid token payload: no subject")
            return {"username": username}
        except JWTError as e:
            # 在这里可以记录错误,但不再抛出 HTTPException
            raise e # 重新抛出 JWTError,由 jwt_logged_user 捕获
        except Exception as e:
            raise e

utils = Utils()

router = APIRouter()

# --- 独立认证依赖项 ---

# Basic Auth
security = HTTPBasic(auto_error=False)

def basic_logged_user(credentials: Annotated[Optional[HTTPBasicCredentials], Depends(security)]) -> Optional[str]:
    if credentials is None:
        return None

    current_username_bytes = credentials.username.encode("utf8")
    correct_username_bytes = settings.SESSION_LOGIN_USER.encode("utf8")
    is_correct_username = secrets.compare_digest(
        current_username_bytes, correct_username_bytes
    )
    current_password_bytes = credentials.password.encode("utf8")
    correct_password_bytes = settings.SESSION_LOGIN_PASS.encode("utf8")
    is_correct_password = secrets.compare_digest(
        current_password_bytes, correct_password_bytes
    )

    if not (is_correct_username and is_correct_password):
        return None

    return credentials.username

# JWT Auth
def jwt_logged_user(token: Annotated[Optional[str], Depends(utils.OAuth2_scheme)],
                    db: Session = Depends(db_session)) -> Optional[str]:
    if token is None:
        return None

    try:
        payload = utils.verify_token(token)
        username = payload.get("username")

        if username is None:
            return None

        user = db.query(User).filter(User.username == username).first()
        if user is None:
            return None

        return user.username
    except JWTError:
        return None
    except Exception:
        return None

# --- 组合认证依赖项 ---

def auth_user(jwt_username: Annotated[Optional[str], Depends(jwt_logged_user)],
              basic_username: Annotated[Optional[str], Depends(basic_logged_user)]) -> str:
    if jwt_username:
        return jwt_username
    if basic_username:
        return basic_username

    raise HTTPException(
        status_code=status.

以上就是FastAPI 多种认证方式(任选其一)实现指南的详细内容,更多请关注其它相关文章!


# 而非  # 成都创意营销推广  # 营销推广怎么做到  # 兰州皋兰县网站建设平台  # 网络推广公司怎么做营销  # 淮北seo公司怎么选  # seo1加密线路1  # 越秀seo优化公司排名  # 承德抖音搜索seo  # 咸宁网站建设大全推广  # 忻州网站建设网络推广  # 自定义  # 客户端  # 两种  # word  # 多个  # 美图  # 文档  # 您的  # 令牌  # 抛出  # asic  # red  # 路由  # ai  # session  # 工具  # go 


相关栏目: 【 科技资讯46185 】 【 网络学院92790


相关推荐: 网站内容防复制粘贴的实现策略与局限性  b站怎么取消点赞_b站点赞取消操作方法  React/Next.js中实现列表项的动态移动与状态管理:兼论唯一键的重要性  电脑IP地址怎么查 查看本机IP地址的几种方法  如何使用Node.js csv 包按条件移除含空字段的CSV记录  向日葵客户端怎么进行远程CentOS控制_向日葵客户端远程CentOS控制操作教程  聚水潭ERP登录页面入口 聚水潭ERP官网登录界面  正确连接J*aScript到HTML实现可点击图片与自定义事件处理  React项目中导航栏Logo自适应布局:避免裁剪与布局溢出  J*aScript生成器_j*ascript异步迭代  Safari自带网页翻译功能怎么用 无需插件轻松看懂外文网站【方法】  护手霜蹭到袖口上了如何清洗? 怎样避免留下一圈油印?  J*aScript中高效管理与清空动态列表:避免循环陷阱  Linux如何构建多环境配置管理_Linux多环境配置方案  离线运行Go语言之旅:本地部署与GOPATH配置指南  2025AO3夸克浏览器通道_AO3手机HTTPS安全入口分享  b站怎么看视频的弹幕数量_b站弹幕数量查看方法  微信怎么把收藏的内容分类管理 微信收藏内容标签分类方法  Golang如何实现简单的Web表单_Golang表单提交与验证处理方法  拼多多购物车商品数量无法修改如何处理 拼多多购物车操作优化方法  在Typer应用中优雅地处理和重组任意命令行参数  vivo手机参数配置怎么增强信号_vivo手机参数配置信号增强方法  机构:以往存储涨价周期小米利润率实际上有所改善 能转嫁给消费者等  谷歌浏览器如何快速清除某个网站的数据_Chrome网站缓存清理方法  TikTok网页版直接登录 TikTok网页端官方平台入口  谷歌google账号怎么注册账号 谷歌账号注册官方流程  如何使用spryker/configurable-bundles-products-resource-relationship模块解决复杂产品捆绑关系难题  在FastAPI中利用lifespan与依赖注入高效管理Redis连接池  Excel文件在线转换快速入口 Excel在线格式转换网站  零跑汽车11月交付量达70327台 实现连续9个月正增长  zookeeper 都有哪些功能?  MAC如何安全彻底地删除文件_MAC使用终端命令确保文件无法被恢复  Django模型中自动计算可用余额的实现方法  在J*aScript中复现SciPy的B样条拟合与求值:关键考量  C++ vector二维数组定义_C++ vector of vector用法  漫蛙网页登录入口 漫蛙漫画官方授权网址  J*a里如何实现订单支付与库存同步功能_支付库存同步项目开发方法说明  《刺客信条4:黑旗》重制版新细节曝光:无缝加载 地图更细致!  composer的"require-dev"部分是用来做什么的?  火锅吃太多会怎样 火锅吃太多会上火吗  ExcelARRAYTOTEXT函数怎么自定义分隔符输出数组文本_ARRAYTOTEXT实现动态生成SQL语句  如何高效处理PHP中的Excel数据导入导出?PortPHP/Spreadsheet助你轻松搞定!  夸克浏览器图书入口 夸克手机浏览器阅读入口  PySpark中从现有列右侧提取可变长度字符创建新列的教程  腾讯视频怎么使用多账号家庭管理_腾讯视频家庭多账号统一管理与权限分配教程  HuggingFaceEmbeddings中向量嵌入维度调整的限制与理解  AO3中文官网链接_AO3网页版稳定镜像站  解决Tabulator日期时间排序问题的专业指南  WordPress插件开发:正确注册卸载钩子与避免常见陷阱  在Go Martini框架中高效服务动态生成图像的实践指南 

搜索