Feat: 로그인 기능 추가 (#115)

* 가입할 때 username, password 필드를 추가했습니다.
* 토큰 방식의 인증을 추가했습니다.
* 현재 유저를 얻을 때 토큰을 사용하도록 했습니다.

Author: MunJaeyoung

.env.test

@@ -4,4 +4,5 @@ DB_NAME=testdb
 DB_HOST=127.0.0.1
 DB_PORT=3307
 S3_PORTFOLIO_BUCKET_NAME=wacruit-portfolio-test
+TOKEN_SECRET=test
 SLACK_API_TOKEN=""

poetry.lock

@@ -28,6 +28,8 @@ pytest-mock = "3.8.2"
 moto = {version = "4.2.7", extras = ["all", "ec2", "s3"]}
 tenacity = "^9.0.0"
 pyyaml = "6.0.2"
+argon2-cffi = "^25.1.0"
+authlib = "^1.6.5"
 
 
 [tool.poetry.group.dev.dependencies]

pyproject.toml

@@ -0,0 +1 @@
+from .views import v3_router as router

wacruit/src/apps/auth/__init__.py

@@ -0,0 +1,11 @@
+from wacruit.src.apps.common.exceptions import WacruitException
+
+
+class UserNotFoundException(WacruitException):
+    def __init__(self):
+        super().__init__(status_code=404, detail="해당하는 계정이 존재하지 않습니다.")
+
+
+class InvalidTokenException(WacruitException):
+    def __init__(self):
+        super().__init__(status_code=401, detail="잘못된 토큰입니다.")

wacruit/src/apps/auth/exceptions.py

@@ -0,0 +1,12 @@
+from sqlalchemy.orm import Mapped
+
+from wacruit.src.database.base import DeclarativeBase
+from wacruit.src.database.base import intpk
+from wacruit.src.database.base import str255
+
+
+class BlockedToken(DeclarativeBase):
+    __tablename__ = "blocked_token"
+
+    id: Mapped[intpk]
+    token: Mapped[str255]

wacruit/src/apps/auth/models.py

@@ -0,0 +1,39 @@
+from typing import Annotated
+
+from fastapi import Depends
+from sqlalchemy.orm import Session
+
+from wacruit.src.apps.auth.models import BlockedToken
+from wacruit.src.apps.user.models import User
+from wacruit.src.database.connection import get_db_session
+from wacruit.src.database.connection import Transaction
+
+
+class AuthRepository:
+    def __init__(
+        self,
+        session: Annotated[Session, Depends(get_db_session)],
+        transaction: Annotated[Transaction, Depends()],
+    ) -> None:
+        self.session = session
+        self.transaction = transaction
+
+    def get_user_by_username(self, username: str) -> User | None:
+        return self.session.query(User).where(User.username == username).first()
+
+    def get_user_by_id(self, user_id: int) -> User | None:
+        return self.session.query(User).where(User.id == user_id).first()
+
+    def is_blocked_token(self, token: str) -> bool:
+        result = (
+            self.session.query(BlockedToken).where(BlockedToken.token == token).first()
+        )
+        if result:
+            return True
+        return False
+
+    def block_token(self, token: str) -> bool:
+        to_block = BlockedToken(token=token)
+        with self.transaction:
+            self.session.add(to_block)
+        return True

wacruit/src/apps/auth/repositories.py

@@ -0,0 +1,11 @@
+from pydantic import BaseModel
+
+
+class LoginRequest(BaseModel):
+    username: str
+    password: str
+
+
+class TokenResponse(BaseModel):
+    access_token: str
+    refresh_token: str

wacruit/src/apps/auth/schemas.py

@@ -0,0 +1,84 @@
+from datetime import datetime
+from datetime import timedelta
+from typing import Annotated
+
+from authlib.jose import jwt
+from authlib.jose import JWTClaims
+from authlib.jose.errors import JoseError
+from fastapi import Depends
+
+from wacruit.src.apps.auth.exceptions import InvalidTokenException
+from wacruit.src.apps.auth.exceptions import UserNotFoundException
+from wacruit.src.apps.auth.repositories import AuthRepository
+from wacruit.src.apps.common.security import get_token_secret
+from wacruit.src.apps.common.security import PasswordService
+from wacruit.src.apps.user.models import User
+
+
+class AuthService:
+    def __init__(
+        self,
+        auth_repository: Annotated[AuthRepository, Depends()],
+        token_secret: Annotated[str, Depends(get_token_secret)],
+    ) -> None:
+        self.auth_repository = auth_repository
+        self.token_secret = token_secret
+
+    def get_user_by_id(self, user_id: int) -> User | None:
+        return self.auth_repository.get_user_by_id(user_id)
+
+    def login(self, username: str, password: str) -> tuple[str, str]:
+        user = self.auth_repository.get_user_by_username(username)
+
+        if user is None:
+            raise UserNotFoundException()
+        if user.password is None:
+            raise UserNotFoundException()
+
+        if PasswordService.verify_password(password, user.password):
+            access_token = self.issue_token(user.id, 24, "access")
+            refresh_token = self.issue_token(user.id, 24 * 7, "refresh")
+            return (access_token, refresh_token)
+        raise UserNotFoundException()
+
+    def refresh_token(self, refresh_token: str) -> tuple[str, str]:
+        decoded_token = self.decode_token(refresh_token)
+        if decoded_token["token_type"] != "refresh":
+            raise UserNotFoundException()
+
+        user_id = decoded_token["sub"]
+        user = self.auth_repository.get_user_by_id(user_id)
+
+        if self.auth_repository.is_blocked_token(refresh_token):
+            raise UserNotFoundException()
+        if user is None:
+            raise UserNotFoundException()
+
+        self.block_token(refresh_token)
+        access_token = self.issue_token(user.id, 24, "access")
+        new_refresh_token = self.issue_token(user.id, 24 * 7, "refresh")
+        return (access_token, new_refresh_token)
+
+    def block_token(self, token: str) -> None:
+        if self.auth_repository.is_blocked_token(token):
+            raise InvalidTokenException()
+
+        self.auth_repository.block_token(token)
+
+    def decode_token(self, token: str) -> JWTClaims:
+        try:
+            claims = jwt.decode(token, key=self.token_secret)
+            claims.validate()
+            return claims
+        except JoseError as e:
+            raise InvalidTokenException() from e
+
+    def issue_token(self, user_id: int, expiration_hour: int, token_type: str) -> str:
+        header = {"alg": "HS256"}
+        payload = {
+            "sub": user_id,
+            "exp": int((datetime.now() + timedelta(hours=expiration_hour)).timestamp()),
+            "token_type": token_type,
+        }
+
+        return jwt.encode(header, payload, key=self.token_secret).decode("utf-8")

wacruit/src/apps/auth/services.py

@@ -0,0 +1,34 @@
+from typing import Annotated
+
+from fastapi import APIRouter
+from fastapi import Depends
+from fastapi import Security
+from fastapi.security import HTTPAuthorizationCredentials
+from fastapi.security import HTTPBearer
+
+from wacruit.src.apps.auth.schemas import LoginRequest
+from wacruit.src.apps.auth.schemas import TokenResponse
+from wacruit.src.apps.auth.services import AuthService
+
+v3_router = APIRouter(prefix="/v3/auth", tags=["auth"])
+
+security = HTTPBearer(scheme_name="refresh_token", description="토큰 갱신을 위한 Bearer 토큰")
+
+
+@v3_router.post("/login")
+def login(
+    req: LoginRequest, auth_service: Annotated[AuthService, Depends()]
+) -> TokenResponse:
+    res = auth_service.login(req.username, req.password)
+
+    return TokenResponse(access_token=res[0], refresh_token=res[1])
+
+
+@v3_router.post("/refresh")
+def refresh_token(
+    refresh_credentials: Annotated[HTTPAuthorizationCredentials, Security(security)],
+    auth_service: Annotated[AuthService, Depends()],
+) -> TokenResponse:
+    res = auth_service.refresh_token(refresh_credentials.credentials)
+
+    return TokenResponse(access_token=res[0], refresh_token=res[1])