如何正确对比加密后的JWT刷新令牌？FastAPI实践问题

背景与登录实现逻辑 #

我在FastAPI应用中使用JWT做用户认证，用户登录时会生成access令牌与refresh令牌，同时为用户在数据库创建会话，将refresh令牌通过cryptography库的Fernet工具加密后存储。

登录逻辑代码如下：

@router.post("/login", response_model=schemas.AccessToken)
async def post_login(
    request: Request,
    db_session: db_session_dependency,
    _ = Depends(allow_only_unauthorized_user),
    logged_in_user: User = Depends(validate_user_login),
    user_agent: str = Header(None, alias='User-Agent'),
):
    # create session uuid
    session_id = auth_utils.create_user_session_uuid()

    # get utc now
    now = datetime.now(UTC)

    # create jwt tokens to login user
    access_expire_timedelta = timedelta(
        minutes=settings.auth.access_token_expire_minutes
    )
    access_token = auth_utils.create_token(
        auth_utils.ACCESS_TOKEN_TYPE,
        now, now + access_expire_timedelta,
        sub=str(logged_in_user.id),
        jti=str(session_id),
        user_id=logged_in_user.id,
    )

    refresh_expire_timedelta = timedelta(
        minutes=settings.auth.access_token_expire_minutes
    )
    refresh_token = auth_utils.create_token(
        auth_utils.REFRESH_TOKEN_TYPE,
        now, now + refresh_expire_timedelta,
        sub=str(logged_in_user.id),
        jti=str(session_id),
        user_id=logged_in_user.id,
    )

    # create login user session in db
    encrypted_refresh_token = auth_utils.get_encrypted_refresh_token(
        refresh_token
    )
    session_ip = request.client.host
    await db.create_user_session(
        db_session, 
        user_session_id=session_id,
        user_id=logged_in_user.id,
        encrypted_refresh_token=encrypted_refresh_token, 
        expires_at=now + refresh_expire_timedelta,
        ip=session_ip,
        user_agent=user_agent
    )

    # return tokens and other data as model
    return schemas.AccessToken(
        session_id=str(session_id),
        access_token=access_token,
        refresh_token=refresh_token,
    )

刷新令牌加密逻辑 #

from cryptography.fernet import Fernet

fernet = Fernet(settings.auth.refresh_tokens_encrypt_key)

def get_encrypted_refresh_token(
    refresh_token: str
) -> str:
    return fernet.encrypt(refresh_token.encode()).decode()

当前问题与需求 #

当需要删除用户会话或刷新令牌时，我通过OAuth2PasswordBearer从请求头获取refresh令牌，用上述相同的加密函数处理后，尝试在数据库查询中匹配加密后的令牌，但同一refresh令牌每次加密都会生成不同的字符串，无法用==运算符直接匹配，导致无法实现「无需先查询UserSession记录，直接通过一次数据库查询完成删除操作」的目标。

当前的删除会话代码：

async def delete_user_session(
    session: AsyncSession,
    user_session_id: uuid.UUID,
    user_id: int, 
    encrypted_refresh_token: str,
) -> bool:
    query = (
        delete(UserSession)
        .where(and_(
            UserSession.id==user_session_id,
            UserSession.user_id==user_id,
            UserSession.encrypted_refresh_token==encrypted_refresh_token
        ))
    )

    res = await session.execute(query)
    await session.commit()
    return res.rowcount and res.rowcount > 0

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