FastAPI实现多重可选认证：同时支持Basic Auth与JWT Auth

from fastapi.security import HTTPBasic, HTTPBasicCredentials
from fastapi import Depends, HTTPException, status
from typing import Optional, Annotated
import secrets # 用于安全地比较字符串

# 假设 settings.SESSION_LOGIN_USER 和 settings.SESSION_LOGIN_PASS 存储了正确的用户名和密码
# 假设 router 和 db_session 已定义
# 假设 User 模型已定义

# 1. 初始化 HTTPBasic，设置 auto_error=False
security = HTTPBasic(auto_error=False)

# 2. 创建 Basic Auth 依赖函数
def basic_logged_user(credentials: Annotated[Optional[HTTPBasicCredentials], Depends(security)]):
    """
    Basic Auth 认证依赖。
    如果认证失败，不抛出异常，而是返回 None。
    """
    if credentials is None:
        # 没有提供 Basic Auth 凭据
        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

from fastapi.security import OAuth2PasswordBearer
# ... 其他导入 ...
# 假设 utils.verify_token 是一个验证JWT并返回用户数据的函数
# 假设 db_session 和 User 模型已定义

# 1. 初始化 OAuth2PasswordBearer，设置 auto_error=False
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token", auto_error=False) # tokenUrl 根据你的实际情况设置

# 2. 创建 JWT Auth 依赖函数
def jwt_logged_user(token: Annotated[Optional[str], Depends(oauth2_scheme)], 
                    db: Session = Depends(db_session)):
    """
    JWT Auth 认证依赖。
    如果认证失败，不抛出异常，而是返回 None。
    """
    if token is None:
        # 没有提供 JWT token
        return None

    try:
        # 假设 utils.verify_token 会验证 token 并返回一个包含用户信息的对象
        # 如果 token 无效，utils.verify_token 内部应该捕获异常或返回 None
        # 这里我们假设它在验证失败时会抛出异常，我们捕获它
        payload = utils.verify_token(token) # 假设 verify_token 成功返回 payload，失败抛异常
        username = payload.get("sub") # 假设 sub 字段存储用户名
        if not username:
            return None

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

        return user # 认证成功，返回用户对象
    except Exception:
        # token 验证失败（过期、无效签名等）
        return None

from fastapi import HTTPException, status
# ... 其他导入 ...

def auth_user(jwt_auth_result: Annotated[Optional[User], Depends(jwt_logged_user)], 
              basic_auth_result: Annotated[Optional[str], Depends(basic_logged_user)]):
    """
    组合认证依赖，支持 Basic Auth 或 JWT Auth。
    如果任一认证方式成功，则返回相应的认证用户。
    如果两种方式都失败，则抛出 401 异常。
    """
    if jwt_auth_result:
        # JWT 认证成功
        return jwt_auth_result

    if basic_auth_result:
        # Basic Auth 认证成功
        # 注意：这里返回的类型需要与 jwt_auth_result 的类型保持一致，
        # 或者根据你的业务逻辑决定返回哪个。
        # 为了简化，这里假设可以返回一个字符串作为认证成功的标识。
        # 如果需要返回统一的用户对象，则需要从 basic_auth_result 中获取用户对象。
        return basic_auth_result

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

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
# ... 其他导入 ...

router = APIRouter() # 假设你有一个 APIRouter 实例

@router.get("/users")
async def get_users(db: Session = Depends(db_session), 
                    logged_user: Annotated[Union[User, str], Depends(auth_user)]):
    """
    获取用户列表，需要 Basic Auth 或 JWT Auth 认证。
    """
    query_users = db.query(User).all()
    return query_users