강의 9: 사용자 인증 및 권한 부여

강의 9: 사용자 인증 및 권한 부여

강의 목표

• OAuth2와 JWT를 사용하여 사용자 인증 및 권한 부여 방법을 이해합니다.
• 인증이 필요한 API 엔드포인트를 생성합니다.

준비물

• 이전 강의에서 설정한 FastAPI 프로젝트

실습 예제

1. auth.py 파일을 생성하고 OAuth2 및 JWT 설정을 추가합니다.

    from fastapi import Depends, HTTPException, status
    from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
    from jose import JWTError, jwt
    from passlib.context import CryptContext
    from sqlalchemy.orm import Session
   
    from . import models, schemas, crud
    from .database import SessionLocal
   
    SECRET_KEY = "secret"
    ALGORITHM = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES = 30
   
    pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
    oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
   
    def verify_password(plain_password, hashed_password):
        return pwd_context.verify(plain_password, hashed_password)
   
    def get_password_hash(password):
        return pwd_context.hash(password)
   
    def authenticate_user(db: Session, username: str, password: str):
        user = crud.get_user_by_email(db, email=username)
        if not user:
            return False
        if not verify_password(password, user.hashed_password):
            return False
        return user
   
    def create_access_token(data: dict, expires_delta: timedelta = None):
        to_encode = data.copy()
        if expires_delta:
            expire = datetime.utcnow() + expires_delta
        else:
            expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
        to_encode.update({"exp": expire})
        encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
        return encoded_jwt
   
    async def get_current_user(db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)):
        credentials_exception = HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
        try:
            payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
            username: str = payload.get("sub")
            if username is None:
                raise credentials_exception
        except JWTError:
            raise credentials_exception
        user = crud.get_user_by_email(db, email=username)
        if user is None:
            raise credentials_exception
        return user

2. crud.py 파일에 get_user_by_email 함수를 수정합니다.

    def get_user_by_email(db: Session, email: str):
        return db.query(models.User).filter(models.User.email == email).first()

3. main.py 파일에서 인증 엔드포인트를 추가합니다.

    @app.post("/token", response_model=schemas.Token)
    def login_for_access_token(db: Session = Depends(get_db), form_data: OAuth2PasswordRequestForm = Depends()):
        user = auth.authenticate_user(db, 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=auth.ACCESS_TOKEN_EXPIRE_MINUTES)
        access_token = auth.create_access_token(
            data={"sub": user.email}, expires_delta=access_token_expires
        )
        return {"access_token": access_token, "token_type": "bearer"}
   
    @app.get("/users/me/", response_model=schemas.User)
    def read_users_me(current_user: models.User = Depends(auth.get_current_user)):
        return current_user

4. schemas.py 파일에 Token 모델을 추가합니다.

    class Token(BaseModel):
        access_token: str
        token_type: str

추가 실습

• 사용자 권한 부여를 추가하여 특정 역할만 접근할 수 있는 엔드포인트를 만들어 보세요.