WeniVooks

검색

FastAPI 베이스캠프

JWT를 이용한 사용자 인증 구현

1. 라우팅 및 세팅

1.1 URL 정보

이번 챕터의 URL 구성은 아래와 같습니다. 테스트할 순서대로 작성되었습니다.

경로 함수명 메서드 설명
/signup signup POST 회원가입
/token login POST 회원가입된 정보로 토큰 발급
/protected protected_route GET 발급 받은 토큰으로 회원 정보 확인
1.2 기본 세팅

이번 실습 폴더는 04_4_jwt_auth입니다. VSC 터미널에서 사용할 명령어 입니다. 가상환경은 벗어난 상태에서 실행해야 합니다. 만약 터미널 입력창 앞에 (venv)라고 되어 있다면 deactivate 명령어로 가상환경을 나간 상태에서 cd ..으로 상위 폴더로 나와 아래 명령어를 실행해주세요.

mkdir 04_4_jwt_auth
cd 04_4_jwt_auth
python -m venv venv
.\venv\Scripts\activate
pip install fastapi uvicorn python-jose python-multipart
mkdir 04_4_jwt_auth
cd 04_4_jwt_auth
python -m venv venv
.\venv\Scripts\activate
pip install fastapi uvicorn python-jose python-multipart
  • python-jose: JWT를 생성하고 검증하기 위한 라이브러리
  • python-multipart: 멀티파트 요청을 처리하기 위한 라이브러리, 멀티파트 요청이란 파일 업로드와 같이 복수의 데이터를 전송하는 요청을 말합니다.

2. JWT 구현 방식과 기본 세팅

JWT를 구현하기 위해 DB를 사용해야 하지만, 이번 실습에서는 DB 대신 메모리 내 파이썬 데이터 구조를 사용하여 데이터를 저장할 것입니다. 또한 복잡도를 낮추기 위해 리프레시 토큰을 구현하지 않습니다. 이는 과제로 남겨두었습니다. 이는 개념을 간단히 설명하기 위한 것이며, 실제 애플리케이션에서는 보통 앞서 학습한 데이터베이스를 사용합니다.

또 JWT를 구현하기 위해는 User가 있어야 합니다. User는 일반 테이블보다 고려해야 할 사항이 많습니다. 예를 들어, 패스워드를 저장할 때는 해싱을 해야 합니다. 관리자가 패스워드를 DB에서 확인하더라도, 어떤 패스워드인지 알 수 없게 해야 하기 때문입니다. 여기서는 패스워드를 해싱하지 않고 저장하지만, 반드시 실제 애플리케이션에서는 패스워드를 해싱하여 저장해야 합니다.

이번 실습에서는 간단한 User를 구현하고, JWT를 이용하여 사용자 인증을 구현해보겠습니다.

3. 코드 구현

복잡도를 최소화 했기 때문에 코드가 짧습니다. 우선 기본 코드를 작성해놓고 주석을 달아놓은 코드는 뒤에 작성해놓겠습니다.

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
from datetime import datetime, timedelta
from jose import jwt
 
app = FastAPI()
 
# 간단한 설정
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
 
# 간단한 사용자 DB (실제 환경에서는 데이터베이스 사용)
users_db = {}
 
# OAuth2 설정
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 
 
# 모델 정의
class User(BaseModel):
    username: str
    password: str
 
 
class Token(BaseModel):
    access_token: str
    token_type: str
 
 
# JWT 토큰 생성 함수
def create_token(username: str):
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    token_data = {"sub": username, "exp": expire}
    return jwt.encode(token_data, SECRET_KEY, algorithm=ALGORITHM)
 
 
# 현재 사용자 확인
async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=401, detail="Invalid token")
        return username
    except jwt.JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")
 
 
# 회원가입
@app.post("/signup")
async def signup(user: User):
    if user.username in users_db:
        raise HTTPException(status_code=400, detail="Username already exists")
    users_db[user.username] = user.password
    return {"message": "User created successfully"}
 
 
# 로그인 및 토큰 발급
@app.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user_password = users_db.get(form_data.username)
    if not user_password or user_password != form_data.password:
        raise HTTPException(status_code=401, detail="Incorrect username or password")
 
    access_token = create_token(form_data.username)
    return {"access_token": access_token, "token_type": "bearer"}
 
 
# 보호된 엔드포인트 예시
@app.get("/protected")
async def protected_route(current_user: str = Depends(get_current_user)):
    return {"message": f"Hello {current_user}! This is a protected route"}
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
from datetime import datetime, timedelta
from jose import jwt
 
app = FastAPI()
 
# 간단한 설정
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
 
# 간단한 사용자 DB (실제 환경에서는 데이터베이스 사용)
users_db = {}
 
# OAuth2 설정
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 
 
# 모델 정의
class User(BaseModel):
    username: str
    password: str
 
 
class Token(BaseModel):
    access_token: str
    token_type: str
 
 
# JWT 토큰 생성 함수
def create_token(username: str):
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    token_data = {"sub": username, "exp": expire}
    return jwt.encode(token_data, SECRET_KEY, algorithm=ALGORITHM)
 
 
# 현재 사용자 확인
async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=401, detail="Invalid token")
        return username
    except jwt.JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")
 
 
# 회원가입
@app.post("/signup")
async def signup(user: User):
    if user.username in users_db:
        raise HTTPException(status_code=400, detail="Username already exists")
    users_db[user.username] = user.password
    return {"message": "User created successfully"}
 
 
# 로그인 및 토큰 발급
@app.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user_password = users_db.get(form_data.username)
    if not user_password or user_password != form_data.password:
        raise HTTPException(status_code=401, detail="Incorrect username or password")
 
    access_token = create_token(form_data.username)
    return {"access_token": access_token, "token_type": "bearer"}
 
 
# 보호된 엔드포인트 예시
@app.get("/protected")
async def protected_route(current_user: str = Depends(get_current_user)):
    return {"message": f"Hello {current_user}! This is a protected route"}

주석이 달린 버전입니다.

# FastAPI 프레임워크와 필요한 의존성들을 임포트
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm  # OAuth2 인증 관련 클래스
from pydantic import BaseModel  # 데이터 검증을 위한 Pydantic 모델
from datetime import datetime, timedelta  # 시간 관련 처리를 위한 클래스
from jose import jwt  # JWT 토큰 생성 및 검증을 위한 라이브러리
 
# FastAPI 인스턴스 생성
app = FastAPI()
 
# JWT 토큰 생성에 필요한 기본 설정값들
SECRET_KEY = "your-secret-key"  # JWT 서명에 사용될 비밀키
ALGORITHM = "HS256"  # JWT 암호화 알고리즘
ACCESS_TOKEN_EXPIRE_MINUTES = 30  # 토큰 유효 기간 (분)
 
# 임시 사용자 저장소 (실제 프로덕션에서는 데이터베이스 사용 필요)
users_db = {}
 
# OAuth2 인증 스키마 설정 - 토큰 엔드포인트 URL 지정
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 
 
# 사용자 데이터 검증을 위한 Pydantic 모델
class User(BaseModel):
    username: str  # 사용자 이름
    password: str  # 비밀번호
 
 
# 토큰 응답 데이터 검증을 위한 Pydantic 모델
class Token(BaseModel):
    access_token: str  # JWT 토큰 문자열
    token_type: str    # 토큰 타입 (bearer)
 
 
# JWT 토큰 생성 함수
def create_token(username: str):
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)  # 만료 시간 설정
    token_data = {"sub": username, "exp": expire}  # 토큰에 포함될 데이터
    return jwt.encode(token_data, SECRET_KEY, algorithm=ALGORITHM)  # JWT 토큰 생성
 
 
# 현재 사용자 인증 함수 - 보호된 라우트에서 사용
async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        # JWT 토큰 디코딩 및 검증
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")  # 토큰에서 사용자 이름 추출
        if username is None:
            raise HTTPException(status_code=401, detail="Invalid token")
        return username
    except jwt.JWTError:  # JWT 토큰 검증 실패시
        raise HTTPException(status_code=401, detail="Invalid token")
 
 
# 회원가입 엔드포인트
@app.post("/signup")
async def signup(user: User):
    if user.username in users_db:  # 이미 존재하는 사용자인지 확인
        raise HTTPException(status_code=400, detail="Username already exists")
    users_db[user.username] = user.password  # 사용자 정보 저장
    return {"message": "User created successfully"}
 
 
# 로그인 및 토큰 발급 엔드포인트
@app.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # 사용자 인증
    user_password = users_db.get(form_data.username)
    if not user_password or user_password != form_data.password:
        raise HTTPException(status_code=401, detail="Incorrect username or password")
 
    # 인증 성공시 JWT 토큰 생성 및 반환
    access_token = create_token(form_data.username)
    return {"access_token": access_token, "token_type": "bearer"}
 
 
# 보호된 라우트 예시 - 인증된 사용자만 접근 가능
@app.get("/protected")
async def protected_route(current_user: str = Depends(get_current_user)):
    return {"message": f"Hello {current_user}! This is a protected route"}
# FastAPI 프레임워크와 필요한 의존성들을 임포트
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm  # OAuth2 인증 관련 클래스
from pydantic import BaseModel  # 데이터 검증을 위한 Pydantic 모델
from datetime import datetime, timedelta  # 시간 관련 처리를 위한 클래스
from jose import jwt  # JWT 토큰 생성 및 검증을 위한 라이브러리
 
# FastAPI 인스턴스 생성
app = FastAPI()
 
# JWT 토큰 생성에 필요한 기본 설정값들
SECRET_KEY = "your-secret-key"  # JWT 서명에 사용될 비밀키
ALGORITHM = "HS256"  # JWT 암호화 알고리즘
ACCESS_TOKEN_EXPIRE_MINUTES = 30  # 토큰 유효 기간 (분)
 
# 임시 사용자 저장소 (실제 프로덕션에서는 데이터베이스 사용 필요)
users_db = {}
 
# OAuth2 인증 스키마 설정 - 토큰 엔드포인트 URL 지정
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 
 
# 사용자 데이터 검증을 위한 Pydantic 모델
class User(BaseModel):
    username: str  # 사용자 이름
    password: str  # 비밀번호
 
 
# 토큰 응답 데이터 검증을 위한 Pydantic 모델
class Token(BaseModel):
    access_token: str  # JWT 토큰 문자열
    token_type: str    # 토큰 타입 (bearer)
 
 
# JWT 토큰 생성 함수
def create_token(username: str):
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)  # 만료 시간 설정
    token_data = {"sub": username, "exp": expire}  # 토큰에 포함될 데이터
    return jwt.encode(token_data, SECRET_KEY, algorithm=ALGORITHM)  # JWT 토큰 생성
 
 
# 현재 사용자 인증 함수 - 보호된 라우트에서 사용
async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        # JWT 토큰 디코딩 및 검증
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")  # 토큰에서 사용자 이름 추출
        if username is None:
            raise HTTPException(status_code=401, detail="Invalid token")
        return username
    except jwt.JWTError:  # JWT 토큰 검증 실패시
        raise HTTPException(status_code=401, detail="Invalid token")
 
 
# 회원가입 엔드포인트
@app.post("/signup")
async def signup(user: User):
    if user.username in users_db:  # 이미 존재하는 사용자인지 확인
        raise HTTPException(status_code=400, detail="Username already exists")
    users_db[user.username] = user.password  # 사용자 정보 저장
    return {"message": "User created successfully"}
 
 
# 로그인 및 토큰 발급 엔드포인트
@app.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # 사용자 인증
    user_password = users_db.get(form_data.username)
    if not user_password or user_password != form_data.password:
        raise HTTPException(status_code=401, detail="Incorrect username or password")
 
    # 인증 성공시 JWT 토큰 생성 및 반환
    access_token = create_token(form_data.username)
    return {"access_token": access_token, "token_type": "bearer"}
 
 
# 보호된 라우트 예시 - 인증된 사용자만 접근 가능
@app.get("/protected")
async def protected_route(current_user: str = Depends(get_current_user)):
    return {"message": f"Hello {current_user}! This is a protected route"}

여기서 OAuth2PasswordBearer를 설명하기 위해 OAuth2를 설명할 필요가 있습니다. OAuth2는 인증 및 권한 부여를 위한 업계 표준 프로토콜입니다. 주요 목적은 사용자의 비밀번호를 공유하지 않고도 인증을 할 수 있게 하는 것에 목적이 있습니다. 여러 특징이 있지만 가장 중요한 특징으로는 해더에 "Bearer {token}" 형식으로 토큰을 담아 보낸다는 특징이 있습니다. 여기서 OAuth2PasswordBearer는 FastAPI에서 OAuth2를 사용하여 토큰을 처리합니다. 이렇게 하면 썬더클라이언트에서는 아래와 같이 응답을 받을 수 있습니다.

썬더 클라이언트가 아니라 프론트엔드에서는 아래와 같이 처리해야 합니다. 다만 여기서는 CORS 애러가 발생하므로 테스트할 때는 썬더클라이언트로만 진행해주세요. CORS 애러를 해결하기 위한 미들웨어 설정은 부록에서 다룹니다.

fetch('http://localhost:8000/protected/', {
  method: 'GET', // or 'POST', 'PUT', 'DELETE'
  headers: {
    'Authorization': 'Bearer ' + token
  }
})
.then(response => response.json())
.then(data => {
  console.log('Success:', data);
})
fetch('http://localhost:8000/protected/', {
  method: 'GET', // or 'POST', 'PUT', 'DELETE'
  headers: {
    'Authorization': 'Bearer ' + token
  }
})
.then(response => response.json())
.then(data => {
  console.log('Success:', data);
})

4. 애플리케이션 실행

터미널에서 다음 명령어를 실행하여 애플리케이션을 시작합니다:

uvicorn main:app --reload
uvicorn main:app --reload

5. API 테스트

5.1 새 사용자 생성
  • URL: 127.0.0.1:8000/signup
  • Method: POST
  • Body: 
      {
      "username": "licat",
      "password": "1234"
  }
      {
      "username": "licat",
      "password": "1234"
  }
5.2 로그인 및 토큰 발급
  • URL: 127.0.0.1:8000/token
  • Method: POST
  • Body (form-data):
    • username: licat
    • password: 1234
5.3 사용자 정보 조회
  • URL: 127.0.0.1:8000/protected
  • Method: GET
  • Headers:
    • Authorization: Bearer {your_access_token}

연습 문제

  1. 리프레시 토큰을 추가하여 토큰 갱신 기능을 구현해보세요.
  2. DB를 사용하여 사용자 정보를 저장하도록 코드를 수정해보세요.
4.3 JWT 토큰이란5장 블로그 기능 구현