1. 개요
FastAPI는 Python으로 작성된 최신 웹 프레임워크로, 매우 빠른 성능과 현대적인 웹 개발의 다양한 요구를 충족하기 위한 강력한 기능들을 제공합니다. 본 문서에서는 FastAPI를 사용하여 OAuth2 및 JSON Web Token(JWT) 인증 방식을 구현하는 방법을 소개하고, 이를 통해 안전한 API를 설계하는 방법을 살펴보겠습니다.
2. 요구 사항
본 강좌를 진행하기 위해 필요한 사항은 다음과 같습니다:
- Python 3.6 이상이 설치된 환경
- FastAPI 라이브러리 및 관련 패키지
- Uvicorn: ASGI 서버
- Pydantic: 데이터 검증 및 설정 관리 라이브러리
3. FastAPI 및 필요한 패키지 설치
콘솔에서 아래의 명령어를 실행하여 FastAPI와 필요한 패키지를 설치합니다:
pip install fastapi uvicorn python-jose passlib[bcrypt]
4. FastAPI 서버 설정
FastAPI 서버를 설정하려면, 간단한 FastAPI 애플리케이션을 만들어 보겠습니다. 다음 코드를 사용하여 기본적인 FastAPI 애플리케이션을 만들어 보세요:
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def read_root():
return {"Hello": "World"}
이 코드를 main.py라는 파일로 저장한 후, 다음 명령어로 서버를 실행합니다:
uvicorn main:app --reload
이제 http://127.0.0.1:8000로 접속하면 “Hello World” 메시지를 확인할 수 있습니다.
5. OAuth2 및 JWT 인증 방식
OAuth2는 인증 및 권한 부여를 위한 프로토콜로, 사용자가 리소스에 접근할 수 있도록 권한을 부여하는 방식입니다. JWT는 세션 정보를 저장할 수 있는 JSON 객체로, 다양한 정보를 담을 수 있으며, 서명을 통해 데이터의 무결성을 확인할 수 있는 장점이 있습니다.
6. 사용자 인증 및 JWT 생성
회원 가입 및 로그인을 구현할 필요가 있습니다. 다음 예제는 사용자 정보를 저장하기 위한 Pydantic 모델과 JWT 생성을 위한 방법을 보여줍니다:
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from datetime import datetime, timedelta
from pydantic import BaseModel
from passlib.context import CryptContext
# 비밀 키와 알고리즘 설정
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
# 비밀번호 암호화 설정
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
# 사용자 모델
class User(BaseModel):
username: str
email: str | None = None
full_name: str | None = None
disabled: bool | None = None
# 로그인 사용자 모델
class UserInDB(User):
hashed_password: str
# OAuth2PasswordBearer 인스턴스 생성
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
# 사용자 데이터 저장소 (예: 데이터베이스 대신 사용)
fake_users_db = {
"johndoe": {
"username": "johndoe",
"full_name": "John Doe",
"email": "johndoe@example.com",
"hashed_password": pwd_context.hash("password"),
"disabled": False,
}
}
# 비밀번호 검증 함수
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
# 사용자 조회 함수
def get_user(db, username: str):
if username in db:
user_dict = db[username]
return UserInDB(**user_dict)
# JWT 생성 함수
def create_access_token(data: dict, expires_delta: timedelta | None = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
# FastAPI 앱 설정
app = FastAPI()
@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
user = get_user(fake_users_db, form_data.username)
if not user or not verify_password(form_data.password, user.hashed_password):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="잘못된 사용자명 또는 비밀번호",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.username}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
위의 코드를 통해 로그인 엔드포인트가 생성되었습니다. 사용자는 사용자명을 입력하고 비밀번호를 제공하여 JWT를 받을 수 있습니다.
7. JWT 인증이 필요한 엔드포인트
인증이 필요한 API 엔드포인트를 만들기 위해 FastAPI의 종속성을 활용할 수 있습니다. 다음은 JWT가 필요한 API 엔드포인트를 만드는 방법입니다:
from fastapi import Security
def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="유효하지 않은 토큰",
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
token_data = User(username=username)
except JWTError:
raise credentials_exception
user = get_user(fake_users_db, username=token_data.username)
if user is None:
raise credentials_exception
return user
@app.get("/users/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_user)):
return current_user
위의 코드에서는 /users/me 엔드포인트를 만들어 권한이 있는 경우에만 사용자가 본인의 정보를 조회할 수 있도록 설정했습니다.
8. 전체 코드
이제까지 구현한 전체 코드를 정리하면 다음과 같습니다:
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from datetime import datetime, timedelta
from pydantic import BaseModel
from passlib.context import CryptContext
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
class User(BaseModel):
username: str
email: str | None = None
full_name: str | None = None
disabled: bool | None = None
class UserInDB(User):
hashed_password: str
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
fake_users_db = {
"johndoe": {
"username": "johndoe",
"full_name": "John Doe",
"email": "johndoe@example.com",
"hashed_password": pwd_context.hash("password"),
"disabled": False,
}
}
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
def get_user(db, username: str):
if username in db:
user_dict = db[username]
return UserInDB(**user_dict)
def create_access_token(data: dict, expires_delta: timedelta | None = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
app = FastAPI()
@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
user = get_user(fake_users_db, form_data.username)
if not user or not verify_password(form_data.password, user.hashed_password):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="잘못된 사용자명 또는 비밀번호",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(data={"sub": user.username}, expires_delta=access_token_expires)
return {"access_token": access_token, "token_type": "bearer"}
def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="유효하지 않은 토큰",
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
token_data = User(username=username)
except JWTError:
raise credentials_exception
user = get_user(fake_users_db, username=token_data.username)
if user is None:
raise credentials_exception
return user
@app.get("/users/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_user)):
return current_user
9. 마무리
이제 FastAPI를 사용하여 OAuth2와 JWT를 활용한 인증 기능을 구현하는 방법을 알아보았습니다. 이 예제를 바탕으로 실제 애플리케이션에 필요한 인증 및 권한 부여 로직을 추가하여, 보다 안전한 API를 설계할 수 있습니다. FastAPI의 뛰어난 성능과 간편한 사용법 덕분에 웹 서비스 개발이 훨씬 수월해질 것입니다.
이 글이 FastAPI와 OAuth2, JWT 인증을 이해하는 데 도움이 되었기를 바랍니다. 더 나아가 FastAPI의 다양한 기능을 탐색하시기를 권장합니다.