FastApi使用jwt-token进行验证
以Depends的方式定义jwt 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"}, ) 需要分发token的url 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 需要token进行访问限制的url 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
