FastAPI Users 로 인증 체계 구현

시리즈, 입문

2024. 6. 12. PM 11:43:14

#fastapi, #user, #authentication, #인증, #로그인, #login, #sqlalchemy

FastAPI Users 개요

Add quickly a registration and authentication system to your FastAPI project. FastAPI Users is designed to be as customizable and adaptable as possible.

[[FastAPI Users]]는 FastAPI에 적용할 수 있는 인증 구현체예요. 소개글에 나온 바와 같이 빠르게 적용해 사용할 수 있죠. 게다가 기능을 확장하거나 입맛에 맞추기 좋게 구현되어 있고, HTTPX OAuth를 기반으로 OAuth2 Client를 제공하여 소셜 로그인(Single Sign-On) 구현과 적용하기도 쉬어요.

푸딩캠프도 FastAPI Users를 활용해 인증 체계를 구현했어요. 푸딩캠프는 서버 기반 렌더링을 하는데, FastAPI Users에 서버 기반 렌더링([[Server Side Rendering]])에 접목하는 데 필요한 사항이 있어서 맞춤 구현(Customization)을 했는데, 대체로 무난하더라고요.

아쉬운 점은 사용 사례가 많지 않고, 공식 문서는 기본적인 적용과 활용 위주로만 기술되어 있어 소스 코드를 들여다보는 빈도가 잦은 편이예요. 하지만 자료형 각주([[Type annotation]])가 잘 되어 있고, 구조도 비교적 간단한 편이라 금방 이해하고 활용할 수 있어요.

인증 체계는 한 번 구현하고 나면 이후엔 딱히 손대지 않는데, 잘 만들려고 하면 신경쓸 게 많아요. 그래서 어지간해서는 직접 구현해서는 가성비(?)가 좋지 않은 구현체죠. 그렇다고 중요한 인증 체계를 외부 도구로 구현하는데, 이 도구를 모르는 것도 찝찝하죠. 그래서 FastAPI Users에 대해 살펴보도록 할게요.

FastAPI Users 주요 구성 요소

출처 : FastAPI Users 공식 문서

User Model Protocol

FastAPI Users는 이용자 정보의 데이터베이스를 다루는 모델을 별도로 구현해 제공하지 않아요. 대신 [[Python Protocol]]만 제공하죠. Protocol은 Java나 TypeScript 등 언어의 인터페이스(Interface)라고 볼 수 있어요. 실제 동작에 영향을 미치지 않고, 클래스 구현할 때 참고하는 데 사용하죠.

class UserProtocol(Protocol[ID]):
    """User protocol that ORM model should follow."""

    id: ID
    email: str
    hashed_password: str
    is_active: bool
    is_superuser: bool
    is_verified: bool

User 모델의 기본키([[Primary Key]])는 id필드를 기준이자 전제로 해요. 재밌는 건 id필드의 자료형이 고정되어 있지 않고, 제너릭으로 각주되어 있다는 점이예요. FastAPI Users는 UUID형과 Integer형으로 id필드 자료형을 사용하는 구현체를 제공하는데, 직접 구현하기만 한다면 다른 자료형(가령 str)으로 얼마든지 지정해 사용할 수 있지요.

저는 이 프로젝트에서 Integer형을 ID로 사용할게요. 왜냐하면 공식 문서는 UUID형을 기준으로 설명하거든요.

from pydantic import BaseModel
from fastapi_users.models import UserProtocol
from starlette.authentication import BaseUser


class User(UserProtocol[int], BaseUser, BaseModel):
    id: int
    email: str
    username: str
    hashed_password: str
    is_active: bool
    is_superuser: bool
    is_verified: bool

또 눈길을 끄는 점은 비밀번호 필드가 hashed_password라는 점이예요. 대체로 password필드로 이름짓고, 굳이 hashed_password로 이름 짓지 않더라도 당연히(?) 해시 처리한 문자열을 저장하잖아요. FastAPI Users엔 비밀번호 초기화 등 비밀번호 관리하는 기능이 포함되어 있는데, 만약 이 기능을 그대로 사용한다면 비밀번호 필드는 hashed_passowrd여야 해요. 비밀번호 해시 처리나 초기화 등을 구현하는 게 생각보다 귀찮은 걸 감안하면 hashed_password라고 필드 이름을 짓는 것도 괜찮을 것 같아요.

FastAPI Users는 UserProtocol에 선언된 필드 외엔 관여하지 않으니 기존에 있던 username 필드는 그대로 유지할게요. 아참, 그리고 UserProtocol은 상속받지 않을게요. Pydantic metaclass와 충돌하거든요.

UserManager

UserManager는 FastAPI Users에 구현되어 있는 기능들이 이용자 정보를 다룰 때 의존하는 계층이예요. 모델과 API를 이어주는 공통 인터페이스랄까요?

from fastapi_users import BaseUserManager, IntegerIDMixin


class UserManager(IntegerIDMixin, BaseUserManager[User, int]):
    reset_password_token_secret = "뭔가 기밀스러운 시크릿 문자열"
    verification_token_secret = "뭔가 기밀스러운 시크릿 문자열"

UserManager의 메서드를 오버라이딩(Override)하기만 하면 FastAPI Users의 기능들은 UserManager API를 사용해 알아서 척척 동작해요.

기본적으로 두 개 속성은 정의해야 해요. 바로 reset_password_token_secret와 verification_token_secret 속성인데, 각각 비밀번호 초기화 시에 사용할 토큰 시크릿과 E-mail 주소 인증(verification)에 사용할 토큰의 시크릿이죠.

우리의 User 모델은 기본키 필드인 id 필드를 Integer형으로 하므로 IntegerIDMixin 을 함께 상속받아야 해요. 기본키로 데이터베이스에서 데이터르 가져올 때 기본키의 값 자료형을 변환해주는 메서드를 UserManager 에게 제공해야 하거든요.

BaseUserManager 클래스는 두 개 타입을 제너릭(Generic)으로 지정받는데, 첫 번째는 모델 클래스, 두 번째는 기본키 필드의 자료형이죠. 제너릭을 쓰지 않아도 사용엔 문제없지만, 제너릭 자료형을 지정해주면 코드 에디터에서 정확한 지원을 받을 수 있어 좋아요.

class UserManager(IntegerIDMixin, BaseUserManager[User, int]):
    reset_password_token_secret = "뭔가 기밀스러운 시크릿 문자열"
    verification_token_secret = "뭔가 기밀스러운 시크릿 문자열"

    async def on_after_register(self, user: User, request: Optional[Request] = None):
        ...

    async def on_after_forgot_password(
        self, user: User, token: str, request: Optional[Request] = None
    ):
        ...

    async def on_after_request_verify(
        self, user: User, token: str, request: Optional[Request] = None
    ):
        ...

회원가입 처리가 된 직후, 비밀번호 찾기 기능 호출 후, 계정 인증 요청 후에 호출될 메서드를 정의할 수 있어요. 가입 직후에 처리할 조치가 있거나, 계정 사용자에게 비밀번호 분실 요청이나 계정 인증 요청이 들어왔다는 안내를 이 이벤트로 대응해 처리할 수 있는 거죠.

우리는 몇 몇 메서드를 오버라이딩 할 거예요. 우선 FastAPI Users는 계정 식별자를 계정의 기본키 값인 id를 사용해요. 만약 UUID형을 안 쓰고 integer형을 사용한다면 거의 대부분은 자동증가 방식의 integer를 쓰잖아요. 그 말은 제 3자가 사용자 수를 추정해볼 수 있다는 뜻이기도 하고, 사용자 식별자 값이 예측 가능하니 나쁜 공격에 악용하기에도 좋아요. 그래서 우리는 계정 식별자를 username으로 하려 해요. 이건 구현할 때 자세히 알아볼게요.

UserManager는 FastAPI 종단점(Endpoint) 함수가 호출될 때마다 의존성으로 주입받아요. 즉, FastAPI Depends로 의존성을 주입받으며, 종단점 함수가 호출될 때마다 UserManager 객체가 생성되어 전달되어야 하지요.

UserDatabase

FastAPI Users가 데이터베이스 작업을 하는 데 필요한 일종의 어댑터예요. SQLAlchemy, Beanie, 이렇게 두 개 ORM을 지원하죠.

UserManager가 의존하는 객체이며, UserManager와 마찬가지로 FastAPI Depends로 의존성을 주입받게 할 수 있어요.

UserManager와 달리 딱히 손댈 것 없이 그대로 사용해도 되는데요. 우리는 username 필드를 사용할 거라서 손볼 부분이 있어요. 이것도 구현할 때 자세히 알아보죠.

AuthenticationBackend

인증 정보는 후술할 UserManager로부터 받고, 쿠키같은 매개체는 FastAPI 종단점, 즉 HTTP Response가 이뤄지는 계층에 가깝잖아요. 그런 점에서 AuthenticationBackend는 FastAPI 종단점과 UserManager 사이에서 역할을 하는 것으로 보입니다. 물론 동작 흐름상 그렇다는 것이고, AuthenticationBackend는 UserManager와 의존 관계를 맺고 있지 않아요. 위치상으로는 FastAPI 종단점에 더 가까이 있죠.

AuthenticationBackend 클래스는 두 개 객체에 의존해요. AuthenticationBackend 는 자체 비즈니스 로직이 있는 게 아니라 의존하는 두 객체의 구현을 호출해주는 대리자일 뿐이거든요. 이 말은 이 두 객체를 직접 구현해 주입해주기만 하면 인증 동작을 다양한 상황에 맞춰 활용할 수 있다는 의미예요.

Transport

인증 결과를 클라이언트에게 전송하는 역할을 담당해요. 그래서 외부 API의 반환값도 [[HTTP Response]] 객체(fastapi.Response)지요.

쿠키(Cookie)와 Bearer 토큰(Bearer), 이렇게 두 가지가 내장되어 있어요.

Strategy

인증 정보를 어떤 전략으로 다룰 것인지를 담당해요. FastAPI Users에는 크게 세 가지 전략이 기본 제공되는데, 데이터베이스에 저장하는 방식, Redis에 저장하는 방식, 그리고 JWT 방식이예요.

UserSchema

pydantic 모델을 사용하는 User Schema예요. 기본적으로 세 가지 Schema를 정의하죠.

UserRead : 이용자 정보에 접근해 정보를 가져올 때 사용.
UserCreate : 이용자 생성(회원가입 등) 할 때 사용하는 Payload용 Schema.
UserUpdate : 이용자 정보를 변경할 때 사용하는 Payload용 Schema.

FastAPI Users가 제공하는 HTTP API(종단점(Endpoint))를 그대로 사용한다면 UserSchema를 정의해야 하고, 그렇지 않다면 생략해도 돼요.

Router

마지막으로 살펴볼 주요 구성요소는 Router예요. FastAPI Users는 계저 생성, 로그인, 로그아웃, 비밀번호 초기화, 계정 인증(verification), 사용자 정보 관련, OAuth 2 처리(consumer)를 위한 종단점(Endpoint) 함수를 구현해 제공하고 있으며, 이에 대한 API Router도 정의되어 있어요.

예를 들면, 계정 생성, 즉 회원 가입을 위한 HTTP API를 제공하는데, 이 HTTP API를 사용하기 위한 Router를 생성할 때에 UserRead와 UserCreate Schema를 지정하면 이 Schema를 이용해 계정 정보를 검증(validation)해요.

FastAPI Users 설정

자, FastAPI Users를 알아보았으니 이제 사용해볼까요? 먼저 FastAPI Users 인스턴스 객체를 만들게요.

# pudding_todo/authentication.py

from fastapi_users import FastAPIUsers

from .apps.account.models import User


fastapi_users = FastAPIUsers[User, int](
    use_user_manager,
    [auth_backend],
)

FastAPI Users 인스턴스 객체 생성할 때 두 개 인자를 전달하는데, 첫 번째 인자는 UserManager예요. 주의할 점은 FastAPI 종단점 함수가 실행될 때(Runtime)마다 UserManager를 의존성으로 주입해야 한다는 점이예요. 왜냐하면 UserManager는 데이터베이스에 연결하는 동안(session) 트랜잭션을 처리하는 데 UserDatabase를 사용하는데, FastAPI는 비동기로 동작하기 때문에 비동기 맥락(Asynchronous context)에서 벗어나면 안 되거든요. 그래서 호출 가능한 객체를 주입하여 종단점 함수가 호출될 때마다 실행되어 객체로 주입되어야 하죠.

저는 의존성으로 주입될 호출가능한 객체 이름에 use_ 접두사를 사용하여 use_user_manager로 이름 짓겠습니다.

# pudding_todo/authentication.py

async def use_user_manager(user_db: UserDBDep) -> AsyncGenerator[UserManager, None]:
    yield UserManager(user_db, password_helper)

AuthenticationBackend 인스턴스 객체 생성

FastAPI Users 인스턴스 객체 생성에 필요한 두 번째 인자는 AuthentiationBackend 인스턴스 객체예요.

AuthenticationBackend는 Transport 객체와 Strategy 객체에 의존하죠. 우리는 JWT로 인증 정보를 다루고, 클라이언트에겐 쿠키(Cookie)로 전달하겠습니다.

# pudding_todo/authentication.py

from fastapi_users.authentication import CookieTransport, JWTStrategy, AuthenticationBackend


SECRET = "PuddingCamp"


def get_jwt_strategy() -> JWTStrategy:
    return JWTStrategy(secret=SECRET, lifetime_seconds=3600)


cookie_transport = CookieTransport(cookie_max_age=3600)

auth_backend = AuthenticationBackend(
    name="jwt",
    transport=cookie_transport,
    get_strategy=get_jwt_strategy,
)

Strategy도 인스턴스 객체가 아니라 호출 가능한 객체를 전달했는데, 우리는 비록 JWTStrategy를 사용하지만 데이터베이스나 Redis를 사용하는 Strategy를 사용할 수 있고, 따라서 use_user_manager와 마찬가지로 Strategy를 실행 시(Runtime)에 주입받도록 호출가능한 객체를 인자로 전달해야 해요. 이름은 use_jwt_strategy로 하려다가 인자 이름에 맞추었는데, use_ 접두사를 유지할 걸 그랬나? 싶네요. 😅

AuthenticationBackend는 여러 개 사용 가능하므로 각각을 식별할 수 있게 이름을 지어주는데요(name="jwt"). 이 이름은 FastAPI 종단점에 의존성으로 주입되는 Transport의 Schema 값 이름으로도 사용돼요. 그래서 Python의 명명 규칙에 맞춰 FastAPI Users가 이름을 강제로 바꾸기 때문에 AuthenticationBackend의 이름도 Python 명명 규칙을 따르는 게 혼란스럽지 않아요.

auth_backend = AuthenticationBackend(
    name="01-jwt_backend",
    transport=cookie_transport,
    get_strategy=get_jwt_strategy,
)

async def endpoint_func(jwt_backend: str): ...

이름을 01-jwt_backend로 지정했지만, 실제로 종단점으로 전달되는 의존성 이름은 jwt_backend로 변환되는 것처럼요.

UserManager

UserManager 클래스는 앞서 선언한 걸 사용는데, 오버라이딩하여 구현을 확장할 부분은 API에 FastAPI Users를 반영하면서 다루고요. 일단 UserManager의 인스턴스 객체 생성을 살펴보겠습니다.

async def use_user_manager(user_db: UserDBDep):
    yield UserManager(user_db, password_helper)

UserManager는 UserDatabase와 PasswordHelper, 이렇게 두 개를 인자로 전달받아요. PasswordHelper는 이름에서 잘 드러나듯이 비밀번호에 관한 도우미를 제공하는 객체지요.

class PasswordHelperProtocol(Protocol):
    def verify_and_update(
        self, plain_password: str, hashed_password: str
    ) -> Tuple[bool, Union[str, None]]: ...  # pragma: no cover

    def hash(self, password: str) -> str: ...  # pragma: no cover

    def generate(self) -> str: ...  # pragma: no cover

어떤 방식으로 암호화할 것인지, 해시된 비밀번호와 비교할 비밀번호를 검증하는지를 담당하죠. 물론 프로토콜만 있진 않고 실 구현체도 FastAPI Users에 있어요. 그래서 굳이 PasswordHelper 객체를 전달하지 않아도 괜찮아요.

from fastapi_users.password import PasswordHelper
from pwdlib import PasswordHash
from pwdlib.hashers.argon2 import Argon2Hasher


password_hash = PasswordHash((
    Argon2Hasher(),
))
password_helper = PasswordHelper(password_hash)

이 코드는 맞춤 PasswordHelper를 만든 예시예요. FastAPI Users는 Argon2와 Bcrypt 두 개를 해시 알고리즘으로 사용하는데, 다른 알고리즘을 원하는 경우에 별도로 PasswordHelper 객체를 만들어 전달하면 되지요.

UserDatabase

UserDatabase의 자료형 각주인 UserDBDep을 먼저 살펴볼게요.

# pudding_todo/authentication.py

from fastapi_users.db import SQLAlchemyUserDatabase

from pudding_todo.db import DbSessionDep


async def use_user_db(session: DbSessionDep):
    yield SQLAlchemyUserDatabase(session, User)


UserDBDep = Annotated[SQLAlchemyUserDatabase[User, int], Depends(use_user_db)]

우리는 SQLAlchemy를 사용하므로 SQLAlchemyUserDatabase 객체를 사용해요. 이 객체는 두 개 인자를 전달받는데, 첫 번째는 SQLAlchemy Session 객체이고, 두 번째는 User 모델 클래스 자체지요. 두 번째 인자는 좀 더 정확히 표현해서 SQLAlchemy ORM으로써 User 모델이예요.

SQLAlchemy Session을 주입해주는 UserDBDep은 지난 소식지에서 만들었죠.

# pudding_todo/db.py

from typing import Annotated

from fastapi import Depends

async def use_session():
    session_class = async_sessionmaker(
        engine, class_=AsyncSession, expire_on_commit=False, autoflush=False,
    )
    async with session_class() as session:
        yield session


DbSessionDep = Annotated[AsyncSession, Depends(use_session)]

FastAPI 종단점 함수가 호출될 때 SQLAlchemy Session 객체를 생성하여 주입받지요. 이 말은 UserDatabase도 FastAPI 종단점 함수에서 의존성 주입([[Dependency Injection]])이 된다는 걸 뜻해요. 그런데 UserDatabase에 의존하는 건 UserManager잖아요. 의존 관계를 그리면 다음과 같아요.

Endpoint
⬆ UserManager
⬆ UserDatabase
⬆ Database Session

FastAPI의 장점인 의존 방향을 단순하게 표현할 수 있다는 점이 잘 드러나죠.

이렇게 해서 FastAPI Users를 사용하는 설정을 마쳤어요. 여기까지 진행한 코드는 7f3ae79 커밋에 있어요.

FastAPI User Manager

설정을 마쳤으니 적용해볼까요?

UserService에 의존성 주입

로그인 API인 login() 종단점 함수를 잠시 보면요.

@router.post("/login", name="login")
async def login(payload: LoginSchema, service: UserServiceDep) -> User:
    user = await service.authenticate(payload.username, payload.password)
    if not user.is_authenticated:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    
    return user

이와 같이 UserService에 의존해 인증 처리를 하고 있어요. 그래서 UserService에 UserManager를 주입해줘야 해요.

# pudding_todo/apps/account/services.py

from fastapi import Depends

from pudding_todo.authentication import use_user_manager
from pudding_todo.db import DbSessionDep

from .models import User

class UserService:
    session: AsyncSession

    def __init__(self, session: DbSessionDep, user_manager=Depends(use_user_manager)):
        self.session = session
        self.user_manager = user_manager

근데 SQLAlchemy Session과 마찬가지로 UserManager 역시 종단점 함수가 동작하는 맥락 상에서 주입되는 객체잖아요. 현 코드와 같이 user_manager인자를 기본값 인자로 지정하는 건 의도에 맞지 않아요. 그래서 DBSessionDep처럼 인자에 자료형 각주하여 FastAPI가 의존성을 주입하도록 변경하겠습니다.

# pudding_todo/db.py

async def use_user_manager(user_db: UserDBDep):
    yield UserManager(user_db, password_helper)

UserManagerDep = Annotated[UserManager, Depends(use_user_manager)]

이 각주를 UserService에 반영합니다.

# pudding_todo/apps/account/services.py

from pudding_todo.authentication import UserManagerDep
from pudding_todo.db import DbSessionDep

from .models import User

class UserService:
    session: AsyncSession

    def __init__(self, session: DbSessionDep, user_manager: UserManagerDep):
        self.session = session
        self.user_manager = user_manager

종단점 함수는 변경할 내용이 없어요. 의존 관계 상 종단점 함수는 UserService에 의존하고, UserService는 UserManager를 의존하며, 이를 FastAPI가 의존성을 주입하도록 했기 때문이죠.

UserManager를 이용해 User 정보 가져오기

UserService의 get_by_username() 메서드는 계정명(username)을 인자로 받아 계정 정보를 가져옵니다.

class UserService:
    ...

    async def get_by_username(self, username: str) -> User | None:
        stmt = select(User).where(User.username == username)            
        result = await self.session.execute(stmt)
        return result.scalars().one_or_none()

그런데 FastAPI Users에 구현된 여러 기능도 사용자를 가져오는 API를 사용해요. 바로 UserManager에 있는 get() 메서드죠. 이 메서드는 기본키값을 인자로 받는데, 우리는 기본키값을 외부에 노출하지 않고, 그 대신 username을 사용하기로 했잖아요. 그래서 get() 메서드를 오버라이딩하여 동작을 변경할 필요가 있어요. 그럼 UserService의 get_by_username() 메서드도 UserManager의 get() 메서드를 사용하도록 하여 계정 정보를 가져오는 API를 통일할 수 있죠.

# pudding_todo/authentication.py

class UserManager(IntegerIDMixin, BaseUserManager[User, int]):
    ...

    async def get(self, username: int | str) -> Optional[User]:
        if isinstance(username, int):
            return await super().get(username)
        return await self.get_by_username(username)

    async def get_by_username(self, username: str) -> Optional[User]:
        statement = select(self.user_db.user_table).where(
            self.user_db.user_table.username == username,
        )
        return await self.user_db._get_user(statement)

UserService에 있던 get_by_username() 메서드를 거의 그대로 옮겼어요. self.user_db는 UserManager에 주입한 UserDatabase예요. UserDatabase 인스턴스 객체를 생성할 때 두 번째 인자로 User 모델을 전달했는데, 이 모델이 self.user_db.user_table에 할당되어 있어요. 따라서 select(self.user_db.user_table는 select(User)와 동일해요.

UserDatabase의 _get_user() 메서드는 BaseUserDatabase 클래스에는 없고, SQLAlchemyUserDatabase에 구현되어 있어요. SQLAlchemy Statement를 인자로 전달하면 Query를 실행한 후 데이터 하나를 반환하거나 None을 반환하죠.

UserManager에 사용자를 가져오는 메서드를 보완했으니 UserService에서 이 메서드를 사용하도록 변경할 차례예요.

# pudding_todo/apps/account/services.py

class UserService:
    ...

    async def get_by_username(self, username: str) -> User | None:
        return await self.user_manager.get_by_username(username)

적용 완료! 실은 UserDatabase에 get_by_username() 메서드를 구현하여 이 메서드를 UserManager에서 호출하는 게 맞지만, 우리는 현재 SQLAlchemyUserDatabase를 그대로 사용하고 있고, UserDatabase의 인터페이스는 BaseUserDatabase 클래스여서 모두 손보기 번거로워서 UserManager에서 데이터베이스에 바로 접근하도록 구현한 것입니다.

이 부분을 구현한 코드는 0fe2cb5 커밋에서 확인할 수 있어요.

비밀번호 검증 구현

UserManager에 지정한 사용자에 대해 비밀번호가 일치하는지 검증하는 메서드가 있어요. 그런데 구현이 되어 있지 않죠.

class BaseUserManager(Generic[models.UP, models.ID]):
    ...
    
    async def validate_password(
        self, password: str, user: Union[schemas.UC, models.UP]
    ) -> None:
        """
        Validate a password.

        *You should overload this method to add your own validation logic.*

        :param password: The password to validate.
        :param user: The user associated to this password.
        :raises InvalidPasswordException: The password is invalid.
        :return: None if the password is valid.
        """
        return  # pragma: no cover

우리의 UserManager 클래스에 구현할게요.

# pudding_todo/authentication.py

class UserManager(IntegerIDMixin, BaseUserManager[User, int]):
    ...
    
    async def validate_password(self, password: str, user: User) -> None:
        result, _ = self.password_helper.verify_and_update(
            password,
            user.hashed_password,
		)
        if not result:
            raise InvalidPasswordException("Password is incorrect.")

UserManager에 PasswordHelper가 있으니 이 Helper를 이용하는 것 뿐입니다. 간단하죠? 주의해야할 점은 이 메서드가 비동기로 동작한다는 점이예요. 이 점을 유의하며 UserService에 있는 인증 메서드에 UserManager를 반영할게요.

# pudding_todo/apps/account/services.py


class UserService:
    ...

    async def authenticate(self, username: str, password: SecretStr | str):
        if isinstance(password, SecretStr):
            password = password.get_secret_value()

        user = await self.get_by_username(username)
        try:
            await self.user_manager.validate_password(password, user)
        except InvalidPasswordException:
            return UnauthenticatedUser()

        return user

test_auth_endpoint.py 테스트를 수행하면 구현이 변경됐지만 동일하게 동작한다는 걸 볼 수 있어요.

test_user_service.py 는 테스트를 실패하는데, UserService의 초기화 인자가 달라졌고, 실제 데이터베이스에 있는 이용자로 인증을 하므로 conftest.py에서 생성하는 valid_user Fixture를 사용하도록 테스트 코드를 수정하면 돼요. 이 부분은 e3a5cc7 커밋에서 확인하세요.

인증 정보를 쿠키로 굽기

AuthenticationBackend를 이용해 인증 정보 전송

인증 처리는 잘 동작하니, 인증 후 JWT를 쿠키에 구워 심는 구현을 할 차례예요. 물론 FastAPI Users의 CookieTransport에 구현이 되어 있는데요. 이 구현은 HTTP API로 접근하는 걸 전제로 해요. 인증 후에는 HTTP Status 204 응답을 하거든요. 우리는 서버측 렌더링(Server Side Rendering)을 할 것이므로 204 응답을 반으면 사용자는 텅빈 화면을 응답받게 돼요.

사용자 상호작용은 나중에 구현하기로 하고, 이번 편에서는 직접 쿠키를 구워볼게요. 다시 한 번 상기시켜 드리자면, 이 구현은 우리 프로젝트가 서버측 렌더링을 하기 때문에 필요한 것이지, RESTful/HTTP API로 구현하여 Front-end 에서 로그인 후 조치를 취한다면 이 구현은 생략해도 돼요.

사용자 인증 처리는 UserManager가 담당한다면, 인증 정보를 무엇에 보관하고 어떻게 클라이언트에게 전송할 것인지는 AuthenticationBackend가 담당해요. AuthenticationBackend 객체를 FastAPI 종단점에서 사용해야 하니 이 객체를 의존성 주입해주도록 할게요.

# pudding_todo/authentication.py

auth_backend = AuthenticationBackend(
    name="jwt",
    transport=cookie_transport,
    get_strategy=get_jwt_strategy,
)

AuthBackendDep = Annotated[AuthenticationBackend[User, int], Depends(lambda: auth_backend)]

먼저 FastAPI가 의존성 주입에 참조할 자료형 각주를 만들고요.

# pudding_todo/apps/account/endpoints.py

from fastapi import APIRouter, HTTPException, status, Response

from pudding_todo.authentication import AuthBackendDep

@router.post("/login", name="login")
async def login(payload: LoginSchema, service: UserServiceDep, auth_backend: AuthBackendDep) -> Response:
    user = await service.authenticate(payload.username, payload.password)
    if not user.is_authenticated:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    
    strategy = auth_backend.get_strategy()
    response = await auth_backend.login(strategy, user)
    return response

로그인 종단점인 login() 함수에 의존성으로 주입시킨 후, AuthenticationBackend 객체를 이용해 로그인 완료에 대한 HTTP Response 객체를 만듭니다.

쿠키가 잘 구워지는지 테스트 해볼게요.

# tests/apps/account/test_auth_endpoints.py

def test_login_successfully(client: TestClient, valid_user: User):
    payload = {"username": valid_user.username, "password": "PuddingCamp2024"}
    res = client.post(app.router.url_path_for("login"), json=payload)
    assert res.status_code == status.HTTP_204_NO_CONTENT
    assert auth_backend.transport.cookie_name in res.cookies
    assert not not res.cookies[auth_backend.transport.cookie_name]

HTTP 응답에 인증에 대한 쿠키가 잘 심어지는 걸 확인했습니다. 여기까지 코드는 669e042 커밋에서 확인하실 수 있어요.

한 가지 의아한 점

그런데 FastAPI Users의 구현이 다소 아리송한 부분이 있는데, FastAPI Users 인스턴스 객체는 복수 개 AuthenticationBackend를 지원하는데, FastAPI Users에 기본 제공되는 API(종단점 함수)는 한 번에 한 AuthenticationBackend만 사용해요. 그래서 Bearer 방식용 HTTP API와 쿠키 방식용 HTTP API를 분리해야 하죠. 이 부분은 어떤 의도와 맥락이 있는지 아직 잘 모르겠어요. 🥲 암시적이지 않고 명시적이긴 한데...

여러분은 어떻게 생각하시나요? 🙋‍♀️🙋‍♂️

사용자 정보 의존성 주입

사용자 객체 의존성 주입을 위한 각주

드디어 마지막 구현이예요! 바로 현 HTTP 접속(세션)이 누구인지, 즉 사용자 정보를 가져오는 구현이죠. 크게 두 가지 방식이 있어요. 하나는 FastAPI 미들웨어에서 인증 처리를 하여 request 객체(HTTP Request)의 user 속성에 정보를 할당하는 거고요. 다른 하나는 종단점에 사용자 객체를 의존성으로 주입하는 방식이예요. 둘 다 구현할 건데, 분량이 너무 많아지니 이번 편에서는 두 번째 방식만 구현할게요.

구현은 아주 간단해요. FastAPI Users에서 기능을 제공하거든요.

# pudding_todo/authentication.py

fastapi_users = FastAPIUsers[User, int](
    use_user_manager,
    [auth_backend],
)

use_current_active_user = fastapi_users.current_user(
	active=True,
	optional=False,
)
use_current_active_user_optional = fastapi_users.current_user(
	active=True,
	optional=True,
)

CurrentUserDep = Annotated[User, Depends(use_current_active_user)]

CurrentUserOptionalDep = Annotated[User, Depends(use_current_active_user_optional)]

FastAPI Users 인스턴스 객체에 current_user() 메서드가 있는데요. 이 메서드를 호출하면 현 세션의 이용자 객체를 의존성으로 주입하는 함수가 반환돼요. FastAPI Users에 Authenticator 라는 구현체가 겉으로 드러나지 않게 구현되어 있는데, FastAPI Users의 구성 요소들을 활용해 FastAPI 종단점에서 인증 처리를 사용할 수 있게 해주지요. 우리는 그동안 Authenticator가 동작하는 데 필요한 설정을 한 셈입니다. 😁

current_user() 메서드를 이용해 사용자 각주를 두 개 만들었는데요. CurrentUserDep은 현재 이용자가 로그인 상태가 아니면 HTTP Status 401 응답을 해요. FastAPI의 HTTPException을 HTTP Status 401 값으로 일으키는(raise) 동작이죠. CurrentUserOptionalDep은 로그인 상태가 아니면 None이 할당돼요.

User 모델에 is_active 속성 추가

FastAPI Users의 현 세션 이용자 정보를 가져오는 의존성을 사용하려면 이용자 모델의 프로토콜을 맞춰줘야 해요. current_user() 메서드에 전달한 active 인자와 관련된 거죠.

# pudding_todo/apps/account/models.py

class User(BaseUser, SQLModel, table=True):
    ...
    is_active: bool = Field(default=True)

데이터베이스에 is_active 열(column)의 값으로 참조하는 게 아니라 인증 토큰에서 가져온 사용자 정보에서 is_active를 참조하는 거라서 Python property로 정의해도 무방해요.

class User(BaseUser, SQLModel, table=True):
    ...

    @property
    def is_active(self):
        return True

종단점에 사용자 객체 의존성 주입

로그인 여부를 구분하는 FastAPI 종단점 함수를 두 개 만들어 잘 동작하는지 확인해보겠습니다.

# pudding_todo/apps/account/endpoints.py

from pudding_todo.authentication import CurrentUserDep

@router.get("/only-auth-user", name="only-auth-user", response_model=User)
async def only_auth_user(user: CurrentUserDep):
    return user

간단하죠? 테스트 해보겠습니다.

# tests/apps/account/test_auth_endpoints.py

async def test_only_auth_user_without_auth(client: TestClient):
    res = client.get(app.router.url_path_for("only-auth-user"))
    
    assert res.status_code == status.HTTP_401_UNAUTHORIZED

테스트를 통과합니다. 이번엔 로그인 상태 이용자로 접근해보죠.

# tests/apps/account/test_auth_endpoints.py

async def test_only_auth_user_with_auth(client: TestClient, valid_user: User):
    token = await auth_backend.get_strategy().write_token(valid_user)
    res = client.get(
        app.router.url_path_for("only-auth-user"),
        cookies={auth_backend.transport.cookie_name: token},
    )
    
    assert res.status_code == status.HTTP_200_OK
    data = res.json()
    assert data["username"] == valid_user.username

역시 잘 동작하여 테스트를 통과합니다.

이번엔 CurrentUserDep을 적용해볼까요?

# pudding_todo/apps/account/endpoints.py

from pudding_todo.authentication import CurrentUserOptionalDep

@router.get("/any-user", name="any-user")
async def any_user(user: CurrentUserOptionalDep) -> dict:
    if not user:
        return {"detail": "Anonymous"}
    return {"detail": "Authenticated"}

간단한 구현입니다. user가 Falsy하면, 즉 로그인 상태가 아니어서 None인 경우와 아닌 경우에 따라 응답이 달라져요. 테스트 해보겠습니다.

def test_any_user_without_auth(client: TestClient):
    res = client.get(app.router.url_path_for("any-user"))
    assert res.status_code == status.HTTP_200_OK
    data = res.json()
    assert data == {"detail": "Anonymous"}


async def test_any_user_with_auth(client: TestClient, valid_user: User):
    token = await auth_backend.get_strategy().write_token(valid_user)
    res = client.get(
        app.router.url_path_for("any-user"),
        cookies={auth_backend.transport.cookie_name: token},
	)
    
    assert res.status_code == status.HTTP_200_OK
    data = res.json()
    assert data == {"detail": "Authenticated"}

마찬가지로 테스트를 통과합니다. 이번 구현은 5c7a1bb 커밋에 작성되어 있어요.

학습이 이뤄지고 있는지 확신이 안 서나요?
사수가 없나요?
효과적이고 효율적으로 학습하며 성장하는
자기만의 학습과 성장 체계를 만들 수 있습니다.
연사자들의 각양각색 학습과 성장 스토리로
여러분의 학습과 성장을 키워보세요.

푸딩캠프 뉴스레터를 구독하면 학습과 성장, 기술에 관해 요약된 컨텐츠를 매주 편하게 받아보실 수 있습니다.

뉴스레터 구독하기

이전 컨텐츠

다음 컨텐츠