[백엔드 > FastAPI] 7장. 보안

 

문제

- 애플리케이션과 데이터베이스가 상호 작용할 수 있게 되었지만 인증을 거치지 않은 사용자도 이벤트 플래너 애플리케이션
을 사용해서 이벤트를 생성할 수 있다는 것이 문제

 

주제

- JWT를 사용해 애플리케이션의 보안을 강화

- 즉, 오직 인증된 사용자만 특정 이벤트 처리가 가능하도록 함

 

용어 정의

인증 (Authentication)

- 개체가 전달한 인증 정보를 검증하는 것

 

허가 (Authorization)

- 개체가 특정 처리를 할 수 있도록 권한을 주는 것

 

핵심 내용

- FastAPI의 인증 방식

- OAuth2와 JWT를 사용한 애플리케이션 보안

- 의존성 주입을 사용한 라우트 보호

- 교차 출처 리소스 공유 (CORS)설정

 

목표

- 해시를 사용해 패스워드를 보호

- FastAPI 애플리케이션에 인증 계층 추가

- 허가되지 않은 사용자로부터 라우트를 보호

 

FastAPI는 다양한 인증 방식을 지원함

1. 기본 HTTP 인증

- 사용자 인증 정보( 사용자명과 패스워드)를 Authorization HTTP 헤더를 사용해 전송하는 방식

 

2. 쿠키

- 인증정보 데이터를 클라이언트 측에 저장할 때 사용된다

 

3. bearer 토큰 인증

- 토큰은 Bearer 키워드와 함께 요청의 Authorization 헤더에 포함돼 전송됨

- 가장 많이 사용되는 토큰은 JWT 

- JWT는 사용자 ID와 토큰 만료 기간으로 구성된 딕셔너리 형식이 일반적임

 

이 중 각각 장단점이 있지만 bearer 토큰 인증을 사용하기로 한다

 

의존성 주입

- 인증에 사용되는 메서드는 런타임 시 호출되는 의존 라이브러리로 FastAPI 애플리케이션에 주입됨

- 정의한 인증 메서드는 실제로 사용되기 전까지 휴면 상태에 있음

- 이것을 의존성 주입이라고 함

- 즉, 객체 (함수)가 실행에 필요한 인스턴스 변수를 받는 방식을 의미함 

 

@user_router.post("/signup")
async def sign_user_up(user: User) -> dict:
	user_exist = await User.fine_one(User.email == user.email)

 

- 사용자 모델(User)을 함수에 전달해 email 필드를 추출함
- User 클래스가 의존 라이브러리이며 이를 sign_user_up() 함수에 주입함

- User를 사용자 함수의 인수로 주입해 User 클래스의 속성을 쉽게 추출 가능

 

의존 라이브러리 생성

- FastAPI에서 의존 라이브러리는 함수 또는 클래스로 정의됨

- 함수 내에서 객체를 상속하지 않아도 됨 (기본값, 메서드 접근 가능) 

- 의존성 주입은 반복된 코드 작성을 줄여줌 

 

async def get_user(token: str):
	user = decode_token(token)
    return user

 

- 위 의존 라이브러리는 token을 인수로 사용

- 외부 함수인 decode_token에서 받은 user 매개변수를 반환하는 함수

(라이브러리를 사용하려면 Depends 매개변수를 사용하고자 하는 함수의 인수로 설정해야함) 

 

from fastapi import Depends

@router.get("/user/me")
async get_user_details(user: User = Depends(get_user)):
	return user

 

위 라우트 함수는 get_user 라는 함수에 의존함

- 즉, 라우트를 사용하려면 get_user 의존 라이브러리가 존재해야함

 

Depends 클래스

- 라우트가 실행될 때 인수로 받은 함수를 실행함

- 함수의 반환값을 라우트에 전달함

 

---> 지금까지 의존 라이브러리가 어떻게 생성되고 사용되는지 살펴봄

 

7.2 OAuth2와 JWT를 사용한 애플리케이션 보안

인증된 사용자는 bearer라는 JWT 토큰 정보를 서버에 전송해서 허가를 받아야 특정 작업을 처리할 수 있음

 

JWT

- 페이로드, 시그니처, 알고리즘으로 구성

- 인코딩된 문자열이 제 3자에 의해 해킹되는 것을 방지하기 위해 서버와 클라이언트만 알고 있는 고유한 키로 사인됨

 

 cd auth && touch {__init__,jwt_handler,authenticate,hash_password}.py

 

• __init__.py : 해당 폴더에 있는 파일들이 모듈로 사용된다는 것을 명시함
• jwt_handler.py : JWT 문자열을 인코딩, 디코딩하는 함수가 포함됨
• authenticate.py : 의존 라이브러리가 포함되며 인증 및 권한을 위해 라우트에 주입됨
• hash_password.py : 패스워드르 암호화하는 함수가 포함됨. 이 함수는 계정을 등록할 때
                                      또는 로그인 시 패스워드를 비교할 때 사용됨

 

1. 사용자를 해싱(hashing)하는 기능부터 만들어본다.

 

패스워드 해싱

- 사용자 패스워드를 일반 텍스트로 저장하는 것은 API를 구축할 때 매우 나쁜 습관이다 

- 패스워드는 적절한 라이브러리를 사용해서 반드시 암호화(해싱이라고함) 해야한다

- bcrypt 알고리즘을 사용해보자

 

1) passlib 라이브러리를 설치한다.

- 이 라이브러리는 패스워드를 해싱하는 bcrypt 알고리즘을 제공함

pip install passlib

 

2) hash_password.py 파일에 패스워드를 해싱하는 함수를 작성함

 

from passlib.context import CryptContext

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

class HashPassword:
    def create_hash(self, password: str):
        return pwd_context.hash(password)

    def verify_hash(self, plain_password:str, hashed_password: str):
        return pwd_context.verify(plain_password, hashed_password)

 

• create_hash() : 문자열을 해싱한 값을 반환함
• verify_hash() : 일반 텍스트 패스워드와 해싱한 패스워드를 인수로 받아 두 값이 일치하는지 비교함
                           일치 여부에 따라 불린 값을 반환함

3) 패스워드를 해싱해서 데이터베이스에 저장하도록 routes/user.py 의 사용자 등록 라우트를 수정하자

 

에러1 내용 및 해결방안

- 참고 링크: https://sonsazang.tistory.com/31

- 방법은 두가지 

1) 데이터베이스 이름을 따로 설정해 주어서 쉽게 해결 ( self.DATABASE_NAME 부분을 통해 데이터베이스의 이름이 설정되어서 에러가 해결됨 )

2) AsyncIOMotorClient() 메서드를 제공하는 Motor (Async MongoDB 드라이버) 공식 문서에 있는 2.0 마이그레이션 가이드 부분에 있는 방식

[FastAPI, MongoDB] No default database name defined or provided 버그 수정

< 기존 코드 >

async def initialize_database(self):
    client = AsyncIOMotorClient(self.DATABASE_URL)
    await init_beanie(database=client.get_default_database(),
                      document_models=[Event, User])

 

<변경된 코드 >

    async def initialize_database(self):
        client = AsyncIOMotorClient(self.DATABASE_URL)
        db = client.get_database("mydatabase")
        await init_beanie(database=db,
                          document_models=[Event, User])

 

에러2 내용 및 해결방안

- pydantic 버전이 beanie 버전과 일치하지 않았음

- pip install "pydantic<2" 을 통해서 pydantic 1.10.14 버전, beanie 1.13.1 버전을 설치하여 해결함

 

< 에러 내용> 

beanie 1.24.0 로 사용하면 아래 에러가 발생함  

- get_model_type 메서드가 없는 것 같다

 

< 에러 로그>

ERROR:    Traceback (most recent call last):
  File "/Users/chriks/PycharmProjects/planner/3.9venv/lib/python3.9/site-packages/starlette/routing.py", line 705, in lifespan
    async with self.lifespan_context(app) as maybe_state:
  File "/Users/chriks/PycharmProjects/planner/3.9venv/lib/python3.9/site-packages/starlette/routing.py", line 584, in __aenter__
    await self._router.startup()
  File "/Users/chriks/PycharmProjects/planner/3.9venv/lib/python3.9/site-packages/starlette/routing.py", line 682, in startup
    await handler()
  File "/Users/chriks/PycharmProjects/planner/main.py", line 17, in init_db
    await settings.initialize_database()
  File "/Users/chriks/PycharmProjects/planner/database/connection.py", line 55, in initialize_database
    await init_beanie(database=client[self.DATABASE_NAME],
  File "/Users/chriks/PycharmProjects/planner/3.9venv/lib/python3.9/site-packages/beanie/odm/utils/init.py", line 754, in init_beanie
    await Initializer(
  File "/Users/chriks/PycharmProjects/planner/3.9venv/lib/python3.9/site-packages/beanie/odm/utils/init.py", line 123, in __init__
    self.document_models.sort(
  File "/Users/chriks/PycharmProjects/planner/3.9venv/lib/python3.9/site-packages/beanie/odm/utils/init.py", line 124, in <lambda>
    key=lambda val: sort_order[val.get_model_type()]
  File "/Users/chriks/PycharmProjects/planner/3.9venv/lib/python3.9/site-packages/pydantic/_internal/_model_construction.py", line 215, in __getattr__
    raise AttributeError(item)
AttributeError: get_model_type

 

액세스 토큰 생성과 검증

- 토큰의 페이로드는 사용자ID와 만료 시간으로 구성됨 (하나의 긴문자열로 인코딩됨)

- database/connection.py 파일의 Settings 클래스

class Settings(BaseSettings):
    SECRET_KEY: Optional[str] = None

- 환경 파일인 .env에 비밀키를 저장할 SECRET_KEY 변수를 설정

SECRET_KEY=HI5HL3V3L$3CR3T

- jwt_handler.py 파일을 변경

import time
from datetime import datetime

from fastapi import HTTPException, status
from jose import jwt, JWTError
from database.connection import Settings

 

- 아래 명령어를 통해 jose 설치

pip install python-jose python-multipart

 

- SECRET_KEY 변수를 추출할 수 있도록 Settings 클래스의 인스턴스를 만들고 토큰 생성용 함수를 정의함

- expires 값(만료 시간)은 생성시점에서 한 시간 후로 설정됨

settings = Settings()

def create_access_token(user: str):
    payload = {
        "user": user,
        "expires": time.time() + 3600
    }
    
    token = jwt.encode(payload, settings.SECRET_KEY, algorithm="HS256")
    return token

- encode()는 세 개의 인수를 받고 payload를 암호화함

• 페이로드 : 값이 저장된 딕셔너리로, 인코딩할 대상임
• 키: 페이로드를 사인하기 위한 키
• 알고리즘: 페이로드를 사인 및 암호화하는 알고리즘으로, 기본값인 HS256 알고리즘이 가장 많이 사용됨

애플리케이션으로 전달된 토큰을 검증하는 함수

- auth/jwt_handler.py

def verify_access_token(token: str):
    try:
        data = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
        expire = data.get("expires")

        if expire is None:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="No access token supplied"
            )
        if datetime.utcnow() > datetime.utcformtimestamp(expire):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Token expired"
            )
        return data

    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Invalid token"
        )

 

1) 토큰의 만료 시간이 존재하는지 여부

2) 토큰이 유효한지(만료 시간이 지나지 않았는지) 여부

3) except 블록에서 JWT 요청 자체에 오류가 있는지 확인

 

사용자 인증

사용자 인증 검증, 의존 라이브리로 사용할 함수를 만든다

- 의존 함수를 구현해서 이벤트 라우트에 주입

- 이 함수는 호라성 세션에 존재하는 사용자 정보를 추출하는 역할을 함

- auth/authenticate.py

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from auth.jwt_handler import verify_access_token

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/user/signin")

async def authenticate(token: str = Depends(oauth2_scheme)) -> str:
    if not token:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Sign in for access"
        )
    decoded_token = verify_access_token(token)
    return decoded_token["user"]

 

- 위 코드는 다음과 같은 의존 라이브러리를 임포트함

• Depends: oauth2_scheme을 의존 라이브러리 함수에 주입한다
• OAuth2PasswordBearer: 보안 로직이 존재한다는 것을 애플리케이션에 알려준다
• verify_access_token: 앞서 정의한 토큰 생성 및 검증 함수로, 토큰의 유효성을 확인한다

- 토큰이 유효하면 토큰을 디코딩한 후 페이로드의 사용자 필드를 반환하고 유효하지 않으면,
   verify_access_token() 함수에 정의된 오류 메시지를 반환함

7.3 애플리케이션 변경

- 라우트를 수정해서 인증 모델을 적용

- 이벤트 추가용 POST 라우트를 변경해서 사용자 레코드에 이벤트 필드 추가