Merge pull request #200 from Pseudo-Lab/refactor/auth

refactor(getcloser): authentication and authorization using JWT

Author: hu6r1s

getcloser/backend/app/api/v1/__init__.py

@@ -1,9 +1,20 @@
+from api.v1.auth import auth
 from api.v1.users import users
-from api.v1.challenge import assign_challenge
+from api.v1.challenges import challenges
 from api.v1.teams import teams
-from fastapi import APIRouter
+from core.dependencies import get_current_user
+from fastapi import APIRouter, Depends
+
+private_router = APIRouter(
+  dependencies=[Depends(get_current_user)]
+)
+private_router.include_router(users.router, prefix="/users", tags=["users"])
+private_router.include_router(challenges.router, prefix="/challenges", tags=["challenges"])
+private_router.include_router(teams.router, prefix="/teams", tags=["teams"])
+
+public_router = APIRouter()
+public_router.include_router(auth.router, prefix="/auth", tags=["auth"])
 
 api_router = APIRouter()
-api_router.include_router(users.router, prefix="/users", tags=["users"])
-api_router.include_router(assign_challenge.router, prefix="/challenge", tags=["challenge"])
-api_router.include_router(teams.router, prefix="/teams", tags=["teams"])
+api_router.include_router(public_router)
+api_router.include_router(private_router)

getcloser/backend/app/api/v1/auth/auth.py

@@ -0,0 +1,14 @@
+from core.database import get_db
+from fastapi import APIRouter, Depends
+from models.users import User
+from services.user_service import auth_user
+from sqlalchemy.orm import Session
+from schemas.user_schema import UserAuth, UserResponse
+
+
+router = APIRouter()
+
+@router.post('/', response_model=UserResponse)
+def auth(user_auth: UserAuth, db: Session = Depends(get_db)):
+  token = auth_user(db, user_auth.email)
+  return token

getcloser/backend/app/api/v1/challenge/assign_challenge.py

@@ -1,8 +1,8 @@
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy.orm import Session
 from core.database import get_db
-from schemas.challenge_schema import ChallengeRetryRequest, ChallengeRetryResponse, GoodsRedeemRequest, GoodsRedeemResponse
-from services.challenge_service import redeem_goods, retry_challenge
+from schemas.challenge_schema import ChallengeRequest, ChallengeResponse, ChallengeRetryRequest, ChallengeRetryResponse, GoodsRedeemRequest, GoodsRedeemResponse
+from services.challenge_service import assign_challenges_logic, redeem_goods, retry_challenge
 
 router = APIRouter()
 
@@ -13,3 +13,24 @@ def redeem_goods_controller(request: GoodsRedeemRequest, db: Session = Depends(g
 @router.post("/retry", response_model=ChallengeRetryResponse)
 def challenge_retry_controller(request: ChallengeRetryRequest, db: Session = Depends(get_db)):
   return retry_challenge(db, request.user_id)
+
+@router.post("/assign", response_model=ChallengeResponse)
+def assign_challenges(request: ChallengeRequest, db: Session = Depends(get_db)):
+    try:
+        # my_id = parse_user_id(request.my_id).get("id")
+        # if not my_id:
+        #     raise HTTPException(status_code=400, detail="올바른 형식의 유저 태그가 아닙니다. 예: 김민준#0001")
+        
+        # members_ids = [request.my_id]  # 팀원에 자기 자신 포함
+        # for tag in request.members_ids:
+        #     parsed = parse_user_id(tag)
+        #     if "error" in parsed:
+        #         raise HTTPException(status_code=400, detail=parsed["error"])
+        #     members_ids.append(parsed["id"])
+
+        members_ids = [request.my_id] + request.members_ids
+        assigned = assign_challenges_logic(request.my_id, members_ids, db)
+        return ChallengeResponse(team_id=request.team_id, my_assigned=assigned)
+    
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))

getcloser/backend/app/api/v1/challenges/challenges.py

@@ -1,19 +1,24 @@
 from core.database import get_db
+from core.dependencies import get_current_user
 from fastapi import APIRouter, Depends
-from services.user_service import auth_user, get_user, parse_user_id
+from models.users import User
+from services.user_service import  get_user, parse_user_id
 from sqlalchemy.orm import Session
-from schemas.user_schema import UserAuth, UserResponse
 
 router = APIRouter()
 
-@router.post('/auth', response_model=UserResponse)
-def auth(user_auth: UserAuth, db: Session = Depends(get_db)):
-  user = auth_user(db, user_auth.email)
-  return user
-
+@router.get("/me")
+def get_my_info(
+  current_user: User = Depends(get_current_user),
+):
+  return current_user
+  
 
 @router.get("/{id}")
-def get_user_name(id: int, db: Session = Depends(get_db)):
+def get_user_name(
+  id: int,
+  db: Session = Depends(get_db)
+  ):
   user = get_user(db, id)
   return {"data": f"{user.name}#{str(user.id).zfill(4)}"}
 
@@ -24,4 +29,4 @@ def get_user_id(tag: str):
     '이름#0001' 형식의 문자열을 받아서 ID(int)로 변환합니다.
     """
     response = parse_user_id(tag)
-    return response
+    return response

getcloser/backend/app/api/v1/users/users.py

@@ -6,8 +6,8 @@ class Settings(BaseSettings):
     """
     JWT 안쓸 것 같아 일단 주석 처리하고 추후 확정 시 삭제
     """
-    # SECRET_KEY: str = os.getenv("SECRET_KEY", "change-me-in-prod")
-    # ALGORITHM: str = "HS256"
-    # ACCESS_TOKEN_EXPIRE_MINUTES: int = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "60"))
+    SECRET_KEY: str = os.getenv("SECRET_KEY", "change-me-in-prod")
+    ALGORITHM: str = "HS256"
+    ACCESS_TOKEN_EXPIRE_MINUTES: int = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "60"))
 
 settings = Settings()

getcloser/backend/app/core/config.py

@@ -0,0 +1,31 @@
+from core.database import get_db
+from fastapi import Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer
+from jose import jwt, JWTError
+from sqlalchemy.orm import Session
+
+from core.config import settings
+from models.users import User
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth")
+
+def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Invalid authentication credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+
+    try:
+        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
+        user_id: str = payload.get("sub")
+        if user_id is None:
+            raise credentials_exception
+    except JWTError:
+        raise credentials_exception
+
+    # user = db.query(User).filter(User.id == user_id).first()
+    # if user is None:
+    #     raise credentials_exception
+
+    return payload

getcloser/backend/app/core/dependencies.py

@@ -0,0 +1,20 @@
+from datetime import datetime, timedelta
+from jose import JWTError, jwt
+from core.config import settings
+
+
+def create_access_token(data: dict, expires_delta: timedelta | None = None):
+    to_encode = data.copy()
+    expire = datetime.utcnow() + (
+        expires_delta or timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
+    )
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
+    return encoded_jwt
+
+def verify_access_token(token: str):
+    try:
+        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
+        return payload
+    except JWTError:
+        return None

getcloser/backend/app/core/security.py

@@ -4,9 +4,7 @@ class UserAuth(BaseModel):
   email: EmailStr
 
 class UserResponse(BaseModel):
-  id: int
-  email: str
-  name: str
+  accessToken: str
 
 class Config:
   orm_mode = True

getcloser/backend/app/schemas/user_schema.py

@@ -1,3 +1,4 @@
+from core.security import create_access_token
 from exceptions.user_exceptions import UserNotFoundException
 from models.users import User
 from sqlalchemy.orm import Session
@@ -7,7 +8,13 @@ def auth_user(db: Session, email: str):
   user = db.query(User).filter(User.email == email).first()
   if not user:
     raise UserNotFoundException(email)
-  return user
+  
+  token = create_access_token({
+    "sub": str(user.id),
+    "email": user.email,
+    "name": user.name
+    })
+  return {"accessToken": token}
 
 
 def get_user(db: Session, id: int):