AUTHENTICATION

Tham khảo thêm: OAuth2 scopes - FastAPI  

Cài 2 thư viện:

pip install pyjwt

pip install "passlib[bcrypt]"

from fastapi import FastAPI, UploadFile, File, Depends, Query, HTTPException, status

import jwt

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

from pydantic import BaseModel, ValidationError

from jwt.exceptions import InvalidTokenError

from passlib.context import CryptContext

# to get a string like this run:

# openssl rand -hex 32

SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"

ALGORITHM = "HS256"

ACCESS_TOKEN_EXPIRE_MINUTES = 30

class Token(BaseModel):

    access_token: str

    token_type: str

class TokenData(BaseModel):

    username: str | None = None

    scopes: list[str] = []

class Customer(BaseModel):

    username: str

    email: str | None = None

    full_name: str | None = None

    disabled: bool | None = None

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

oauth2_scheme = OAuth2PasswordBearer(

    tokenUrl="token",

    scopes={"me": "Read information about the current user.", "items": "Read items."},

)

Hàm verify password (so khớp password clear text với password đã hash)

def verify_password(plain_password, hashed_password):

    return pwd_context.verify(plain_password, hashed_password)

Hàm hash password:

def get_password_hash(password):

    return pwd_context.hash(password)

Chỉnh sửa API POST để lưu trong database là password đã hash

@app.post("/users/", tags=["user"])

def create_user(model: User, session: SessionDep) -> User:

    model.password = get_password_hash(model.password)

    print('Sau mã hóa', model)

    session.add(model)

    session.commit()

    session.refresh(model)

    return model

Định nghĩa hàm xác thực người dùng:

def authenticate_user(session: SessionDep, username: str, password: str):

    users = session.exec(select(User)).all() # optimize

    # Check username

    user = None

    for u in users:

        if u.username == username:

            user = u

            break

    if not user:

        return False

    if not verify_password(password, user.password):

        return False

    return user

Định nghĩa hàm tạo token:

def create_access_token(data: dict, expires_delta: timedelta | None = None):

    to_encode = data.copy()

    if expires_delta:

        expire = datetime.now(timezone.utc) + expires_delta

    else:

        expire = datetime.now(timezone.utc) + timedelta(minutes=15)

    to_encode.update({"exp": expire})

    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

    return encoded_jwt

Nhớ thêm thư viện:

from datetime import datetime, timedelta, timezone

Viết API authenticate:

@app.post("/token")

async def login_for_access_token(

    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],

    session: SessionDep

) -> Token:

    user = authenticate_user(

        session, form_data.username, form_data.password

    )

    if not user:

        raise HTTPException(

            status_code=status.HTTP_401_UNAUTHORIZED,

            detail="Incorrect username or password",

            headers={"WWW-Authenticate": "Bearer"},

        )

    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)

    access_token = create_access_token(

        data={"sub": user.username}, expires_delta=access_token_expires

    )

    return Token(access_token=access_token, token_type="bearer")