Django-Oauth-Toolkit을 커스텀 하는 방법

Backend를 Django로 구성중이고 토큰 기반의 인증에 Django-Oauth-Toolkit을 이용하고 있다. 만들어진 대로 가져다 쓰면 편하지만 개발하다보면 반드시 커스텀이 필요해진다. 로그인을 하다가 패스워드 5회 이상 틀리면 정지를 시킨다거나 사용 정지된 회원은 로그인을 막는 등의 조치들. 이런 커스텀 기능을 입히려면 오버라이딩을 하면 된다.

예를들어 이미 토큰을 획득한, 즉 로그인 한 유저가 또 로그인을 시도하면 에러처리하는 로직을 커스텀하려고 한다.

1.

Django-Oauth-Toolkit 에서 토큰을 획득하는 url은 아래와 같다. o/token 이라는 url을 호출하고 있다.

//출처: https://django-oauth-toolkit.readthedocs.io/en/latest/rest-framework/getting_started.html

curl -X POST -d "grant_type=password&username=<user_name>&password=<password>" -u"<client_id>:<client_secret>" http://localhost:8000/o/token/

2.

깃헙의 Django-Oauth-Toolkit에서 urls.py로 o/token의 경로를 확인해보면 아래와 같이 되어 있다. 토큰 생성을 TokenView 클래스에서 한다.

//출처: https://github.dev/jazzband/django-oauth-toolkit/tree/master/oauth2_provider/views


//DJANGO-OAUTH-TOOLKIT/oauth2_provider/urls.py
base_urlpatterns = [
    ...
    re_path(r"^token/$", views.TokenView.as_view(), name="token"),
    ...
]

3.

TokenView 클래스가 아래 파일에 정의되어 있다.

//출처: https://github.dev/jazzband/django-oauth-toolkit/tree/master/oauth2_provider/views


//DJANGO-OAUTH-TOOLKIT/oauth2_provider/views/base.py 
@method_decorator(csrf_exempt, name="dispatch")
class TokenView(OAuthLibMixin, View):
    """
    Implements an endpoint to provide access tokens

    The endpoint is used in the following flows:
    * Authorization code
    * Password
    * Client credentials
    """

    @method_decorator(sensitive_post_parameters("password"))
    def post(self, request, *args, **kwargs):
        url, headers, body, status = self.create_token_response(request)
        if status == 200:
            access_token = json.loads(body).get("access_token")
            if access_token is not None:
                token = get_access_token_model().objects.get(token=access_token)
                app_authorized.send(sender=self, request=request, token=token)
        response = HttpResponse(content=body, status=status)

        for k, v in headers.items():
            response[k] = v
        return response

4.

별도 파일에 위의 함수를 오버라이딩 해 준다. 아래 예제는 mongodb 기준이라 pymongo로 호출하는 예제이다.

//CustomDjangoOauthToolkit/urls.py
from CustomDjangoOauthToolkit.custom import CustomTokenView

urlpatterns = [
    ...
    path('o/', include('oauth2_provider.urls', namespace='oauth2_provider')),    //기존 url
    path('o/customtoken/', CustomTokenView.as_view(), name='token_obtain_pair'), //커스텀 url
]


//CustomDjangoOauthToolkit/custom.py
from django.http import HttpResponse
from django.utils.decorators import method_decorator
from django.views.decorators.debug import sensitive_post_parameters
from oauth2_provider.models import get_access_token_model
from oauth2_provider.signals import app_authorized
import json
from django.views.decorators.csrf import csrf_exempt
from oauth2_provider.views.base import TokenView
import pymongo

from DrfOauthMongodb.utils import getMongoDBConnectHostFile

MONGODB_HOST_FILE = getMongoDBConnectHostFile()

client = pymongo.MongoClient(MONGODB_HOST_FILE)      //mongodb 커넥션 정보
dbname = client['test_db']                           //데이터베이스
collection = dbname['oauth2_provider_accesstoken']   //access_token이 쌓이는 테이블(=콜렉션)

@method_decorator(csrf_exempt, name="dispatch")
class CustomTokenView(TokenView):
    """
    Implements an endpoint to provide access tokens

    The endpoint is used in the following flows:
    * Authorization code
    * Password
    * Client credentials
    """

    @method_decorator(sensitive_post_parameters("password"))
    def post(self, request, *args, **kwargs):
        token = collection.find_one({'user_id': 3})  //user_id가 3인 사용자가 로그인 한 정보가 있는지 셀렉트, 유저 정보는 리퀘스트 정보를 활용해 불러오면 됨
        if token is not None:                        //로그인 한 정보가 있으면 에러 발생 후 이후 로직 실행하지 않고 종료
            raise Exception('duplicate error') 

        url, headers, body, status = self.create_token_response(request)
        if status == 200:
            access_token = json.loads(body).get("access_token")
            if access_token is not None:
                token = get_access_token_model().objects.get(token=access_token)
                app_authorized.send(sender=self, request=request, token=token)
        response = HttpResponse(content=body, status=status)
        
        for k, v in headers.items():
            response[k] = v
        return response

5.

이제 토큰 획득 url을 호출하면 오버라이딩 한 클래스를 호출한다. 여기에 커스텀 로직들을 추가하면 된다.