FastAPI应用验证Django生成JWT时出现签名验证失败

我正在开发一个基于FastAPI+Python的应用，需要对接Django开发的通用认证系统生成的JWT令牌。之前在其他项目里已经成功实现过相同的认证逻辑，但当前新项目里一直报Signature Verification Failed错误。

相关代码 #

auth_utils.py #

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi import Request
from fastapi.exceptions import HTTPException
import jwt
import base64
from jwt.exceptions import InvalidTokenError
import json
from loguru import logger
import os


ALGORITHM = os.getenv('ALGORITHM')
JWT_SECRET_KEY = os.getenv('SECRET')

def decodeJWT(jwtoken: str):
    jwt_options = {
        'verify_signature': True,
        'verify_exp': True,
        'verify_nbf': False,
        'verify_iat': True,
        'verify_aud': False
    }
    try:
        payload = json.loads(json.dumps(jwt.decode(jwt=jwtoken,
                              key=JWT_SECRET_KEY,
                              algorithms=ALGORITHM,
                              options=jwt_options)))
                    
        print(payload)
        return payload
    except InvalidTokenError as e:
        logger.error(f'{e}')
        return None

class JWTBearer(HTTPBearer):
    def __init__(self, auto_error: bool = True):
        super(JWTBearer, self).__init__(auto_error=auto_error)

    async def __call__(self, request: Request):
        credentials: HTTPAuthorizationCredentials = await super(JWTBearer, self).__call__(request)
        if credentials:
            if not credentials.scheme == 'Bearer':
                raise HTTPException(status_code=403, detail="Invalid authentication scheme.")
            if not self.verify_jwt(credentials.credentials):
                raise HTTPException(status_code=403, detail="Invalid token or expired token.")

            return credentials.credentials
        else: 
            raise HTTPException(status_code=403, detail="Invalid authorization code.")
        
        
    def verify_jwt(self, jwtoken: str) -> bool:
        isTokenValid: bool = False

        try:
            payload = decodeJWT(jwtoken)
        except:
            payload = None
        if payload:
            isTokenValid = True
        return isTokenValid
    

jwt_bearer = JWTBearer()

def format_jwt(token):
    
    token_decode = token.split(".")[1]

    padding = '=' * (4 - len(token_decode) % 4)
    token_decode += padding

    decoded_bytes = base64.urlsafe_b64decode(token_decode)
    decoded_str = decoded_bytes.decode('utf-8')

    decoded_payload = json.loads(decoded_str)

    return decoded_payload

示例端点代码 #

from api.v1.apps.theme.service.services import insert, get_all, get_one, update, remove
from fastapi import APIRouter, HTTPException, status, Depends
from sqlalchemy.exc import SQLAlchemyError, IntegrityError
from api.v1.apps.theme.schemas.schemas import ThemeSchema
from helpers.auth_utils import JWTBearer, format_jwt
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm.exc import NoResultFound
from db.session import get_async_session
from loguru import logger
import os

router = APIRouter()

SUPER_ADMIN = os.getenv('SUPER_ADMIN')
ADMIN = os.getenv('ADMIN')
VIEWER = os.getenv('VIEWER')

@router.post('/create-theme/')
async def create_theme(theme: ThemeSchema, dependencies=Depends(JWTBearer()), session: AsyncSession = Depends(get_async_session)):
    """Endpoint para criação de temas"""

    try:
        decoded_payload = format_jwt(dependencies)
        usuario = decoded_payload.get('role')
        
        if usuario == SUPER_ADMIN or usuario == ADMIN:

            return await insert(description=theme.description)
    
    except IntegrityError as e:
        logger.error(f"Integrity error when saving theme: {e}")
        await session.rollback()
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Integrity violation when saving theme")
    except SQLAlchemyError as e:
        logger.error(f"Database error when saving theme: {e}")
        await session.rollback()
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error when saving theme")
    except Exception as e:
        logger.error(f"Unexpected error when saving theme: {e}")
        await session.rollback()
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Error saving theme")

错误日志 #

2024-06-21T16:16:36.975431-0300 Signature verification failed

解决方案 #

• 核对环境变量与Django配置一致性
  确认ALGORITHM和SECRET完全匹配Django端：

  • Django默认JWT算法为HS256，确保FastAPI这边的ALGORITHM值大小写一致（不要写成hs256）
  • SECRET必须和Django项目的SECRET_KEY完全一致，包括所有特殊字符、空格，避免多余的引号或换行符

• 简化JWT解码逻辑
  当前decodeJWT函数中的json.loads(json.dumps(...))属于冗余操作，直接返回jwt.decode的结果即可：

  def decodeJWT(jwtoken: str):
      jwt_options = {
          'verify_signature': True,
          'verify_exp': True,
          'verify_nbf': False,
          'verify_iat': True,
          'verify_aud': False
      }
      try:
          payload = jwt.decode(
              jwt=jwtoken,
              key=JWT_SECRET_KEY,
              algorithms=ALGORITHM,
              options=jwt_options
          )
          print(payload)
          return payload
      except InvalidTokenError as e:
          logger.error(f'{e}')
          return None
  

• 验证令牌与算法匹配性
  手动解析令牌确认：

  • 令牌头部的alg字段和你配置的ALGORITHM一致
  • 如果Django使用RS256等非对称加密算法，需要用对应的公钥而非Django的SECRET_KEY来验证签名

• 修复端点的payload获取逻辑
  原端点中使用format_jwt手动解码payload绕过了签名验证，存在安全风险，应替换为已验证的decodeJWT结果：

  # 替换create_theme中的payload解码代码
  decoded_payload = decodeJWT(dependencies)
  if not decoded_payload:
      raise HTTPException(status_code=403, detail="Invalid token")
  usuario = decoded_payload.get('role')
  

• 检查请求令牌格式
  确认请求头Authorization的格式为Bearer <token>，无多余空格或截断，令牌完整无误

内容的提问来源于stack exchange，提问作者Ivan Campelo