FastAPI에서 JWT 인증을 추가하는 방법 – 실용 가이드

FastAPI에서 JWT 인증을 추가하는 방법 – 실용 가이드

 

FastAPI는 Python으로 작성된 현대적이고 빠르며 전투 테스트를 거친 경량 웹 개발 프레임워크입니다. 이 공간에서 인기 있는 다른 옵션은 Django, Flask 및 Bottle입니다.

그리고 새로운 기능이기 때문에 FastAPI에는 장점과 단점이 모두 있습니다.

긍정적인 측면에서 FastAPI는 최신 Python 버전에서 지원하는 기능을 최대한 활용하여 모든 최신 표준을 구현합니다. 비동기 지원 및 유형 힌트가 있습니다. 또한 빠르고(따라서 FastAPI라는 이름이 지정됨), 의견이 없고, 강력하고, 사용하기 쉽습니다.

부정적인 측면에서 FastAPI에는 Django와 함께 제공되는 즉시 사용 가능한 사용자 관리 및 관리자 패널과 같은 몇 가지 복잡한 기능이 없습니다. FastAPI에 대한 커뮤니티 지원은 훌륭하지만 수년 동안 존재했으며 다양한 사용 사례를 위한 수백 또는 수천 개의 오픈 소스 프로젝트가 있는 다른 프레임워크만큼 훌륭하지는 않습니다.

FastAPI에 대한 아주 간단한 소개였습니다. 이 기사에서는 실용적인 예를 들어 FastAPI에서 JWT(JSON 웹 토큰) 인증을 구현하는 방법을 배웁니다.

프로젝트 설정

이 예에서는 다음을 사용할 것입니다. 리플리트 (훌륭한 웹 기반 IDE). 또는 문서를 따라 FastAPI 프로젝트를 로컬로 설정하거나 이 replit 스타터 템플릿을 분기하여 사용할 수 있습니다. 이 템플릿에는 필요한 모든 종속성이 이미 설치되어 있습니다.

로컬 환경에 프로젝트 설정이 있는 경우 JWT 인증을 위해 설치해야 하는 종속성은 다음과 같습니다(FastAPI 프로젝트가 실행 중이라고 가정).

pip install "python-jose[cryptography]" "passlib[bcrypt]" python-multipart

노트: 사용자를 저장하기 위해 replit의 내장 데이터베이스를 사용할 것입니다. 그러나 PostgreSQL, MongoDB 등과 같은 표준 데이터베이스를 사용하는 경우 유사한 작업을 적용할 수 있습니다.

전체 구현을 보려면 프로덕션 준비 FastAPI 응용 프로그램이 가질 수 있는 모든 것을 포함하는 이 전체 비디오 자습서가 있습니다.

JWT 인증을 사용하는 FastAPI 앱

FastAPI로 인증

일반적으로 인증은 암호 해싱 처리 및 토큰 할당에서 각 요청에 대한 토큰 유효성 검사에 이르기까지 많은 부분을 움직일 수 있습니다.

FastAPI는 종속성 주입(소프트웨어 엔지니어링 디자인 패턴)을 활용하여 인증 체계를 처리합니다. 다음은 프로세스의 몇 가지 일반적인 단계 목록입니다.

• 비밀번호 해싱
• JWT 토큰 생성 및 할당
• 사용자 생성
• 인증을 보장하기 위해 각 요청에 대한 토큰 유효성 검사

비밀번호 해싱

사용자 이름과 비밀번호로 사용자를 생성할 때 데이터베이스에 저장하기 전에 비밀번호를 해시해야 합니다. 암호를 쉽게 해시하는 방법을 살펴보겠습니다.

라는 이름의 파일 생성 utils.py 에서 app 디렉토리에 다음 함수를 추가하여 사용자 암호를 해시합니다.

from passlib.context import CryptContext

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


def get_hashed_password(password: str) -> str:
    return password_context.hash(password)


def verify_password(password: str, hashed_pass: str) -> bool:
    return password_context.verify(password, hashed_pass)

우리는 사용하고 있습니다 passlib 암호 해싱을 위한 구성 컨텍스트를 생성합니다. 여기에서 사용하도록 구성하고 있습니다. bcrypt .

그만큼 get_hashed_password 함수는 일반 암호를 사용하여 데이터베이스에 안전하게 저장할 수 있는 해시를 반환합니다. 그만큼 verify_password 함수는 일반 암호와 해시 암호를 사용하여 암호가 일치하는지 여부를 나타내는 부울을 반환합니다.

JWT 토큰을 생성하는 방법

이 섹션에서는 특정 페이로드로 액세스 및 새로 고침 토큰을 생성하는 두 가지 도우미 함수를 작성합니다. 나중에 이 함수를 사용하여 사용자 관련 페이로드를 전달하여 특정 사용자에 대한 토큰을 생성할 수 있습니다.

내부 app/utils.py 이전에 생성한 파일에 다음 import 문을 추가합니다.

import os
from datetime import datetime, timedelta
from typing import Union, Any
from jose import jwt

액세스 및 새로 고침 토큰 생성을 위한 가져오기

JWT를 생성할 때 전달될 다음 상수를 추가합니다.

ACCESS_TOKEN_EXPIRE_MINUTES = 30  # 30 minutes
REFRESH_TOKEN_EXPIRE_MINUTES = 60 * 24 * 7 # 7 days
ALGORITHM = "HS256"
JWT_SECRET_KEY = os.environ['JWT_SECRET_KEY']   # should be kept secret
JWT_REFRESH_SECRET_KEY = os.environ['JWT_REFRESH_SECRET_KEY']    # should be kept secret

액세스 및 새로 고침 토큰 생성을 위한 상수

JWT_SECRET_KEY 그리고 JWT_REFRESH_SECRET_KEY 모든 문자열이 될 수 있지만 비밀로 유지하고 환경 변수로 설정해야 합니다.

replit.com에서 팔로우하는 경우 다음에서 이러한 환경 변수를 설정할 수 있습니다. Secrets 왼쪽 메뉴 모음의 탭.

끝에 다음 기능을 추가하십시오. app/utils.py 파일:

def create_access_token(subject: Union[str, Any], expires_delta: int = None) -> str:
    if expires_delta is not None:
        expires_delta = datetime.utcnow() + expires_delta
    else:
        expires_delta = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    

to_encode = {"exp": expires_delta, "sub": str(subject)}
encoded_jwt = jwt.encode(to_encode, JWT_SECRET_KEY, ALGORITHM)
return encoded_jwt

def create_refresh_token(subject: Union[str, Any], expires_delta: int = None) -> str:
if expires_delta is not None:
expires_delta = datetime.utcnow() + expires_delta
else:
expires_delta = datetime.utcnow() + timedelta(minutes=REFRESH_TOKEN_EXPIRE_MINUTES)

to_encode = {"exp": expires_delta, "sub": str(subject)}
encoded_jwt = jwt.encode(to_encode, JWT_REFRESH_SECRET_KEY, ALGORITHM)
return encoded_jwt

액세스 및 새로 고침 토큰 생성을 위한 함수

이 두 함수의 유일한 차이점은 새로 고침 토큰의 만료 시간이 액세스 토큰의 만료 시간보다 길다는 것입니다.

함수는 단순히 JWT 내부에 포함할 페이로드를 가져오며 이는 무엇이든 될 수 있습니다. 일반적으로 여기에 USER_ID와 같은 정보를 저장하고 싶지만 이것은 문자열에서 객체/사전까지 무엇이든 될 수 있습니다. 함수는 토큰을 문자열로 반환합니다.

결국 당신의 app/utils.py 파일은 다음과 같아야 합니다.

from passlib.context import CryptContext
import os
from datetime import datetime, timedelta
from typing import Union, Any
from jose import jwt

ACCESS_TOKEN_EXPIRE_MINUTES = 30  # 30 minutes
REFRESH_TOKEN_EXPIRE_MINUTES = 60 * 24 * 7 # 7 days
ALGORITHM = "HS256"
JWT_SECRET_KEY = os.environ['JWT_SECRET_KEY']     # should be kept secret
JWT_REFRESH_SECRET_KEY = os.environ['JWT_REFRESH_SECRET_KEY']      # should be kept secret

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

def get_hashed_password(password: str) -> str:
    return password_context.hash(password)


def verify_password(password: str, hashed_pass: str) -> bool:
    return password_context.verify(password, hashed_pass)


def create_access_token(subject: Union[str, Any], expires_delta: int = None) -> str:
    if expires_delta is not None:
        expires_delta = datetime.utcnow() + expires_delta
    else:
        expires_delta = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    
    to_encode = {"exp": expires_delta, "sub": str(subject)}
    encoded_jwt = jwt.encode(to_encode, JWT_SECRET_KEY, ALGORITHM)
    return encoded_jwt

def create_refresh_token(subject: Union[str, Any], expires_delta: int = None) -> str:
    if expires_delta is not None:
        expires_delta = datetime.utcnow() + expires_delta
    else:
        expires_delta = datetime.utcnow() + timedelta(minutes=REFRESH_TOKEN_EXPIRE_MINUTES)
    
    to_encode = {"exp": expires_delta, "sub": str(subject)}
    encoded_jwt = jwt.encode(to_encode, JWT_REFRESH_SECRET_KEY, ALGORITHM)
    return encoded_jwt

사용자 가입을 처리하는 방법

내부 app/app.py 파일에서 사용자 등록을 처리하기 위한 다른 끝점을 만듭니다. 엔드포인트는 사용자 이름/이메일 및 비밀번호를 데이터로 가져와야 합니다. 그런 다음 이메일/사용자 이름을 가진 다른 계정이 존재하지 않는지 확인합니다. 그런 다음 사용자를 생성하고 데이터베이스에 저장합니다.

~ 안에 app/app.py다음 핸들러 함수를 추가하십시오.

from fastapi import FastAPI, status, HTTPException
from fastapi.responses import RedirectResponse
from app.schemas import UserOut, UserAuth
from replit import db
from app.utils import get_hashed_password
from uuid import uuid4

@app.post('/signup', summary="Create new user", response_model=UserOut)
async def create_user(data: UserAuth):
    # querying database to check if user already exist
    user = db.get(data.email, None)
    if user is not None:
            raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="User with this email already exist"
        )
    user = {
        'email': data.email,
        'password': get_hashed_password(data.password),
        'id': str(uuid4())
    }
    db[data.email] = user    # saving user to database
    return user

로그인 처리 방법

FastAPI에는 OpenAPI 표준을 준수하기 위해 로그인을 처리하는 표준 방법이 있습니다. 이렇게 하면 추가 구성 없이 swagger 문서에 인증이 자동으로 추가됩니다.

사용자 로그인을 위한 다음 핸들러 함수를 추가하고 각 사용자 액세스 및 새로 고침 토큰을 할당합니다. 수입품을 포함하는 것을 잊지 마십시오.

from fastapi import FastAPI, status, HTTPException, Depends
from fastapi.security import OAuth2PasswordRequestForm
from fastapi.responses import RedirectResponse
from app.schemas import UserOut, UserAuth, TokenSchema
from replit import db
from app.utils import (
    get_hashed_password,
    create_access_token,
    create_refresh_token,
    verify_password
)
from uuid import uuid4

@app.post('/login', summary="Create access and refresh tokens for user", response_model=TokenSchema)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = db.get(form_data.username, None)
    if user is None:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Incorrect email or password"
        )

    hashed_pass = user['password']
    if not verify_password(form_data.password, hashed_pass):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Incorrect email or password"
        )
    
    return {
        "access_token": create_access_token(user['email']),
        "refresh_token": create_refresh_token(user['email']),
    }

이 끝점은 들어오는 데이터를 필터링하기 위한 스키마를 정의한 다른 사후 끝점과 약간 다릅니다.

로그인 엔드포인트의 경우 다음을 사용합니다. OAuth2PasswordRequestForm 종속성으로. 이렇게 하면 요청에서 데이터를 추출하고 다음과 같이 전달합니다. form_data 주장 login 핸들러 함수. python-multipart 양식 데이터를 추출하는 데 사용됩니다. 따라서 설치했는지 확인하십시오.

끝점은 사용자 이름 및 암호에 대한 입력과 함께 swagger 문서에 반영됩니다.

응답이 성공하면 다음과 같은 토큰을 받게 됩니다.

보호된 경로를 추가하는 방법

이제 로그인 및 등록에 대한 지원이 추가되었으므로 보호된 끝점을 추가할 수 있습니다. FastAPI에서 보호된 끝점은 종속성 주입을 사용하여 처리되며 FastAPI는 OpenAPI 스키마에서 이를 유추하고 swagger 문서에 반영할 수 있습니다.

의존성 주입의 힘을 봅시다. 이 시점에서 문서에서 인증할 수 있는 방법은 없습니다. 이는 현재 보호된 엔드포인트가 없기 때문에 OpenAPI 스키마에 우리가 사용하는 로그인 전략에 대한 정보가 충분하지 않기 때문입니다.

Swagger 문서에는 로그인할 수 있는 버튼이 없습니다.

사용자 지정 종속성을 생성해 보겠습니다. 핸들러 함수에 전달된 인수를 가져오기 위해 실제 핸들러 함수보다 먼저 실행되는 함수일 뿐입니다. 실용적인 예를 들어 보겠습니다.

다른 파일 만들기 app/deps.py 다음 기능을 추가하십시오.

from typing import Union, Any
from datetime import datetime
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from .utils import (
    ALGORITHM,
    JWT_SECRET_KEY
)

from jose import jwt
from pydantic import ValidationError
from app.schemas import TokenPayload, SystemUser
from replit import db

reuseable_oauth = OAuth2PasswordBearer(
    tokenUrl="/login",
    scheme_name="JWT"
)


async def get_current_user(token: str = Depends(reuseable_oauth)) -> SystemUser:
    try:
        payload = jwt.decode(
            token, JWT_SECRET_KEY, algorithms=[ALGORITHM]
        )
        token_data = TokenPayload(**payload)
        
        if datetime.fromtimestamp(token_data.exp) < datetime.now():
            raise HTTPException(
                status_code = status.HTTP_401_UNAUTHORIZED,
                detail="Token expired",
                headers={"WWW-Authenticate": "Bearer"},
            )
    except(jwt.JWTError, ValidationError):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Could not validate credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
        
    user: Union[dict[str, Any], None] = db.get(token_data.sub, None)
    
    
    if user is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Could not find user",
        )
    
    return SystemUser(**user)

여기에서 우리는 정의하고 있습니다 get_current_user 차례로 인스턴스를 취하는 종속성 기능 OAuth2PasswordBearer 종속성으로.

reuseable_oauth = OAuth2PasswordBearer(
    tokenUrl="/login",
    scheme_name="JWT"
)

OAuth2PasswordBearer 두 개의 필수 매개변수를 취합니다. tokenUrl 사용자 로그인 및 반환 토큰을 처리하는 애플리케이션의 URL입니다. scheme_name 로 설정 JWT 프론트엔드 swagger 문서가 호출하도록 허용합니다. tokenUrl 프론트엔드에서 토큰을 메모리에 저장합니다. 그런 다음 보호된 끝점에 대한 각 후속 요청에는 토큰이 다음과 같이 전송됩니다. Authorization 헤더 그래서 OAuth2PasswordBearer 파싱할 수 있습니다.

이제 응답으로 사용자 계정 정보를 반환하는 보호된 끝점을 추가해 보겠습니다. 이를 위해 사용자는 로그인해야 하며 엔드포인트는 현재 로그인한 사용자에 대한 정보로 응답합니다.

~ 안에 app/app.py 다른 핸들러 함수를 만듭니다. 수입품도 포함해야 합니다.

from app.deps import get_current_user

@app.get('/me', summary='Get details of currently logged in user', response_model=UserOut)
async def get_me(user: User = Depends(get_current_user)):
    return user

이 끝점을 추가하는 즉시 볼 수 있습니다. Authorize Swagger 문서의 버튼과 보호된 끝점 앞의 🔒 아이콘 /me.

이것은 의존성 주입의 힘이자 자동 OpenAPI 스키마를 생성하는 FastAPI의 능력입니다.

클릭 Authorize 버튼을 누르면 로그인에 필요한 필드가 있는 승인 양식이 열립니다. 응답이 성공하면 토큰이 저장되고 헤더의 후속 요청으로 전송됩니다.

Swagger 통합 로그인 양식

성공적으로 로그인

이 시점에서 보호된 모든 엔드포인트에 액세스할 수 있습니다. 엔드포인트를 보호하려면 다음을 추가하기만 하면 됩니다. get_current_user 종속성으로 기능합니다. 그게 다야!

결론

따라했다면 JWT 인증을 사용하는 작동하는 FastAPI 애플리케이션이 있어야 합니다. 그렇지 않은 경우 항상 이 repl을 실행하고 이를 가지고 놀거나 이 배포된 버전을 방문할 수 있습니다. 이 프로젝트에 대한 GitHub 코드는 여기에서 찾을 수 있습니다.

이 기사가 도움이 되셨다면 트위터 @abdadeel_. 그리고 실제 사례와 함께 자세한 설명을 보려면 이 비디오를 항상 볼 수 있다는 것을 잊지 마십시오.

감사 ;)