如何在FastAPI中刷新令牌和注销时销毁令牌？

Question

我使用此代码进行身份验证。

#1. auth_handler.py

import time

from decouple import config
import jwt

JWT_SECRET = config("JWT_SECRET")
JWT_ALGORITHM = config("JWT_ALGORITHM")


def response_token(token: str) -> dict:
    """
        docstring
    """
    return {
        "access_token": token
    }


def sign_jwt(user_id: str or int, validity: int) -> dict:
    """
        docstring
    """
    payload = {
        "user_id": user_id,
        "expiry": time.time() + validity
    }

    token = jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)

    return response_token(
        {
            "access_token": token
        }
    )


def decode_jwt(token: str) -> dict:
    """
        docstring
    """
    try:
        decoded_token = jwt.decode(
            token,
            JWT_SECRET,
            algorithms=[JWT_ALGORITHM]
        )
        return decoded_token if decoded_token["expiry"] >= time.time() else {}
    except:
        return {}

和

#2. auth_bearer.py

from fastapi import Request, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

from . import auth_handler


class JWTBearer(HTTPBearer):
    """
        docstring
    """

    def __init__(self, auto_error: bool = True):
        """
            docstring
        """
        super(JWTBearer, self).__init__(auto_error=auto_error)

    async def __call__(self, request: Request):
        """
            docstring
        """
        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, jwt_token: str) -> bool:
        """
            docstring
        """
        isTokenValid: bool = False

        try:
            payload = auth_handler.decode_jwt(jwt_token)
        except:
            payload = None
        if payload:
            isTokenValid = True
        return isTokenValid

为了保护路由，我将通过 FastAPI 的 Depends 利用依赖注入。

from fastapi import FastAPI, Body, Depends
from schemas import UserSchema, UserLoginSchema
from auth_bearer import JWTBearer
from auth_handler import signJWT

app = FastAPI()

def check_user(data: UserLoginSchema):
    for user in users:
        if user.email == data.email and user.password == data.password:
            return True
    return False

@app.post("/user/signup", tags=["user"])
async def create_user(user: UserSchema = Body(...)):
    users.append(user) # replace with db call, making sure to hash the password first
    return signJWT(user.email)


@app.post("/user/login", tags=["user"])
async def user_login(user: UserLoginSchema = Body(...)):
    if check_user(user):
        return signJWT(user.email)
    return {
        "error": "Wrong login details!"
    }

@app.post("/posts", dependencies=[Depends(JWTBearer())], tags=["posts"])
async def add_post(post: PostSchema) -> dict:
    post.id = len(posts) + 1
    posts.append(post.dict())
    return {
        "data": "post added."
    }

此代码适用于登录和正确授权。

但我想在用户注销时添加刷新令牌功能和刻录令牌功能。

如何实现这些功能？

Answer 1

为此，您需要在数据库中声明

Token

模型（表）并在那里保持登录令牌。

Token

模型必须包含：

user_id

、

access_token

、

refresh_token

注意（

refresh_token

过期时间必须长于

access_token

）

1. 你应该只返回

   access_token

   给客户端，在

   JWTBearer()

   类中你必须检查

   access_token

   是否过期并且refresh_token过期，如果仅

   access_token

   过期则返回

   refresh_token

   而不是

   access_token

   并刷新

   refresh_token

   新生成的令牌并用

   access_token

    更新 

   refresh_token

2）你可以创建“/check”api，在这里你可以检查access_token是否过期