FastApi使用jwt-token进行验证

from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi.security import OAuth2PasswordBearer
from fastapi import Depends, HTTPException, status
from db import Curd
from typing import Optional
from datetime import datetime
from config import settings
from utils import logger
from .db import get_db
from datetime import timedelta


pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_schema = OAuth2PasswordBearer(tokenUrl="/users/login")


def hash_password(plain_password: str) -> str:
    return pwd_context.hash(plain_password)


def verity_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)


async def jwt_authenticate_user(username: str, password: str):
    db = get_db()
    user = await Curd.get_user_by_name(await db.__anext__(), username)
    if not user:
        return False
    if not verity_password(password, user.password):
        return False
    return user


def created_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=settings.config.ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(claims=to_encode, key=settings.default.SECRET_KEY, algorithm="HS256")
    return encoded_jwt


async def jwt_verity(token: str = Depends(oauth2_schema)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        playload = jwt.decode(token=token, key=settings.default.SECRET_KEY, algorithms=["HS256"])
    except JWTError as e:
        logger.error("验证出现错误:\n", e)
        raise credentials_exception
    else:
        return playload


async def jwt_get_current_user(playload: dict = Depends(jwt_verity)):
    db = get_db()
    user_id = int(playload.get("sub"))
    user = await Curd.get_user_by_id(await db.__anext__(), user_id=user_id)
    return user


async def jwt_verity_admin(playload: dict = Depends(jwt_verity)):
    db = get_db()
    user_id = int(playload.get("sub"))
    is_admin = playload.get("is_admin")
    if is_admin:
        user = await Curd.get_user_by_id(await db.__anext__(), user_id=user_id)
        if user.is_admin:
            return
    raise HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )

router = APIRouter(
    prefix="/users",
    tags=["user"],
)

# it defind at dependencies.jwt
# oauth2_schema = OAuth2PasswordBearer(tokenUrl="/users/token")


@router.post("/register", response_model=UserToken)
async def user_register(user: OAuth2PasswordRequestForm = Depends(), db: AsyncSession = Depends(get_db)):
    if await Curd.is_user_name_exists(db, user.username):
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
                            detail="username already exists")
    hashed_password = hash_password(user.password)
    new_user = UserCreate(id=worker.get_id(), name=user.username, password=hashed_password)
    await Curd.register_user(db, new_user)


@router.post("/login", response_model=UserToken)
async def user_login(user: OAuth2PasswordRequestForm = Depends()):
    user = await jwt_authenticate_user(username=user.username, password=user.password)
    if not user:
        raise HTTPException(
            status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = created_access_token(data={"sub": str(user.id)})
    user_token = UserToken(username=user.name,
                           access_token=access_token,
                           token_type="bearer")
    return user_token

router = APIRouter(
    prefix="/works",
    tags=["works"],
)


@router.get("/get", response_model=Works)
async def test(
        order: int,
        nsfw: bool,
        page: int,
        sort: int,
        user: User = Depends(jwt_get_current_user),
        db: AsyncSession = Depends(get_db),
        ):
    works = await Curd.get_works(db, user.id, order, nsfw, page, sort)
    return works