もくもく会アウトプット:Django で JWT を Cookie で処理するやつを理解したい

はじめに

過日、初めてもくもく会というものに参加した際にやったことをアウトプットします。
内容は前回やったことの掘り下げです。

参加したもくもく会のコミュニティ
前回やったこと

内容

とりあえず、まず前回実装した View。

class TokenObtainView(jwt_views.TokenObtainPairView):
    # Token発行
    def post(self, request, *args, **kwargs):
        # 任意のSerializerを引っ張ってくる(今回はTokenObtainPairViewで使われているserializers.TokenObtainPairSerializer)
        serializer = self.get_serializer(data=request.data)
        # 検証
        try:
            # Serialzierのvalidate()が呼び出される
            serializer.is_valid(raise_exception=True)
        # エラーハンドリング
        except jwt_exp.TokenError as e:
            raise jwt_exp.InvalidToken(e.args[0])

        res = Response(serializer.validated_data, status=status.HTTP_200_OK)

        try:
            res.delete_cookie("access_token")
        except Exception as e:
            print(e)

        # CookieヘッダーにTokenをセットする
        res.set_cookie(
            "access_token",
            serializer.validated_data["access"],
            max_age=60 * 60 * 24,
            httponly=True,
        )
        res.set_cookie(
            "refresh_token",
            serializer.validated_data["refresh"],
            max_age=60 * 60 * 24 * 30,
            httponly=True,
        )

        # 最終的にはaccess_tokenとrefresh_tokenを返してもらう
        return res

上記の View の継承元の View(djangorestframework-simplejwt)

# TokenViewBaseを継承したView
class TokenObtainPairView(TokenViewBase):
    """
    Takes a set of user credentials and returns an access and refresh JSON web
    token pair to prove the authentication of those credentials.
    """
    # post()のget_serializer()で呼ばれるserializerはこれ
    serializer_class = serializers.TokenObtainPairSerializer

token_obtain_pair = TokenObtainPairView.as_view()

TokenObtainPairView の継承元(djangorestframework-simplejwt)

class TokenViewBase(generics.GenericAPIView):
    permission_classes = ()
    authentication_classes = ()

    serializer_class = None

    www_authenticate_realm = 'api'

    def get_authenticate_header(self, request):
        return '{0} realm="{1}"'.format(
            AUTH_HEADER_TYPES[0],
            self.www_authenticate_realm,
        )

    def post(self, request, *args, **kwargs):
        serializer = self.get_serializer(data=request.data)

        try:
            serializer.is_valid(raise_exception=True)
        except TokenError as e:
            raise InvalidToken(e.args[0])

        return Response(serializer.validated_data, status=status.HTTP_200_OK)

で、今回継承して利用しているのがpost()なのでこれを辿っていくとまずget_serializer()の処理にあたる。
これは本来serializer_class = ~で定義した serializer が呼び出されるところを、任意の serializer を呼び出したいときに設定するもの。
つまり、serializer を呼び出す処理になるが、じゃあどの serializer が呼び出されるのかというと TokenObtainPairView の TokenObtainPairSerializer ということになる。
じゃあその serializer が何をやるのかというと以下の通り。

class TokenObtainPairSerializer(TokenObtainSerializer):
    @classmethod
    def get_token(cls, user):
        return RefreshToken.for_user(user)

    # TokenObtainSerializerのvalidate()を継承
    def validate(self, attrs):
        data = super().validate(attrs)

        # Token取得
        refresh = self.get_token(self.user)

        # refreshからaccessとrefreshをそれぞれ格納
        data['refresh'] = str(refresh)
        data['access'] = str(refresh.access_token)

        if api_settings.UPDATE_LAST_LOGIN:
            update_last_login(None, self.user)

        # Tokenを返す
        return data

どうやら JWT を取得して返すということはなんとなくわかる。
なのでもう少し掘ってみると validate()で何をやっているかわかれば理解はできそうなので確認してみる。

# usernameとpasswordを受け取る。大抵ログインと関わるはずなのでフォームからRequestに載せてあるはず
def validate(self, attrs):
    authenticate_kwargs = {
        self.username_field: attrs[self.username_field],
        'password': attrs['password'],
    }
    # Requestかどうかの検証
    try:
        authenticate_kwargs['request'] = self.context['request']
    except KeyError:
        pass

    # ここで一回認証処理が行われている
    # ここのauthenticate()はDjango組み込みのそれ(django.contrib.auth)なので、後述のauthenticate()とは違いusernameとpasswordが引数に入る
    self.user = authenticate(**authenticate_kwargs)

    # 最終的に返るのは空のオブジェクト、つまり雛形を作る
    return {}

ちなみに前回独自に以下のauthentication.pyを実装したが、

from rest_framework_simplejwt.authentication import JWTAuthentication
from rest_framework.response import Response
from django.conf import settings


class CookieHandlerJWTAuthentication(JWTAuthentication):
    def authenticate(self, request):
        # Cookieヘッダーからaccess_tokenを取得
        access_token = request.COOKIES.get('access_token')
        if not access_token:
            Response({"message": 'no Token'})
        else:
            Response(access_token)

        if access_token:
            request.META['HTTP_AUTHORIZATION'] = '{header_type} {access_token}'.format(
                header_type=settings.SIMPLE_JWT['AUTH_HEADER_TYPES'][0], access_token=access_token)

        # JWTAuthenticationのauthenticate()
        return super().authenticate(request)

ここで使うauthenticate()は以下の処理。

    def authenticate(self, request):
        header = self.get_header(request)
        if header is None:
            return None

        raw_token = self.get_raw_token(header)
        if raw_token is None:
            return None

        validated_token = self.get_validated_token(raw_token)

        return self.get_user(validated_token), validated_token

        # 認証失敗
        if not api_settings.USER_AUTHENTICATION_RULE(self.user):
            raise exceptions.AuthenticationFailed(
                self.error_messages['no_active_account'],
                'no_active_account',


# 上記に関わる処理

    # Headerからsettings.AUTH_HEADER_NAMEで設定しているAUTH_HEDERを取り出して返す
    def get_header(self, request):
        """
        Extracts the header containing the JSON web token from the given
        request.
        """
        header = request.META.get(api_settings.AUTH_HEADER_NAME)

        if isinstance(header, str):
            # Work around django test client oddness
            header = header.encode(HTTP_HEADER_ENCODING)

        return header

    # get_header()の返り値からJWTがあるかどうか確認する
    def get_raw_token(self, header):
        """
        Extracts an unvalidated JSON web token from the given "Authorization"
        header value.
        """
        parts = header.split()

        if len(parts) == 0:
            # Empty AUTHORIZATION header sent
            return None

        if parts[0] not in AUTH_HEADER_TYPE_BYTES:
            # Assume the header does not contain a JSON web token
            return None

        if len(parts) != 2:
            raise AuthenticationFailed(
                _('Authorization header must contain two space-delimited values'),
                code='bad_authorization_header',
            )

        return parts[1]

    # JWTを検証する
    def get_validated_token(self, raw_token):
        """
        Validates an encoded JSON web token and returns a validated token
        wrapper object.
        """
        messages = []
        for AuthToken in api_settings.AUTH_TOKEN_CLASSES:
            try:
                return AuthToken(raw_token)
            except TokenError as e:
                messages.append({'token_class': AuthToken.__name__,
                                 'token_type': AuthToken.token_type,
                                 'message': e.args[0]})

        raise InvalidToken({
            'detail': _('Given token not valid for any token type'),
            'messages': messages,
        })

閑話休題、TokenObtainPairSerializer に戻ると以下の通りになる。

class TokenObtainPairSerializer(TokenObtainSerializer):
    @classmethod
    def get_token(cls, user):
        return RefreshToken.for_user(user)

    # TokenObtainSerializerのvalidate()を継承
    def validate(self, attrs):
        # エラーでなければdata = {}
        data = super().validate(attrs)

        # Token取得
        refresh = self.get_token(self.user)

        # refreshからaccessとrefreshをそれぞれ格納
        data['refresh'] = str(refresh)
        data['access'] = str(refresh.access_token)

        if api_settings.UPDATE_LAST_LOGIN:
            update_last_login(None, self.user)

        # Tokenを返す
        return data

あとはここで返ってきた Token をResponse()で返してset_cookie()でセットしていたということになる。

まとめ

初参加のもくもく会ということで勝手がわからず、急遽ソースをたどって処理を理解するということをやってみた。
authenticate()がどのauthenticate()を用いているのか非常に分かりづらく、あれ？ 初回の認証の処理中なのに Token 確認するの？ と混乱もしたが同じ認証でも初回の認証と、Token がある場合の認証とでわかれていることが理解できた。