Django REST Framework認証機構の実装と内部動作

認証コンポーネントの基礎

DRFの認証機能を利用する際の手順は以下の通り。

  1. BaseAuthenticationを継承したクラスを作成
  2. authenticate(self, request)メソッドを実装(request引数を受け取る)
  3. 検証成功時はNoneまたは2要素のタプルを返却

ビューでの適用方法(クラス属性、括弧不要):

authentication_classes = [MyAuth]

設定の優先順位

ビュー単位(優先)

authentication_classes = [MyAuth]

プロジェクト全体(settings.py)

REST_FRAMEWORK = {
    "DEFAULT_AUTHENTICATION_CLASSES": ["myapp.auth.MyAuth"]
}

グローバル設定の個別無効化

authentication_classes = []

実装サンプル:トークン認証

モデル定義

from django.db import models

class Account(models.Model):
    ROLE_CHOICES = (
        (1, 'standard'),
        (2, 'premium'),
        (3, 'enterprise'),
    )
    role = models.IntegerField(choices=ROLE_CHOICES, default=1)
    name = models.CharField(max_length=32)
    secret = models.CharField(max_length=64)

class Session(models.Model):
    account = models.OneToOneField(Account, on_delete=models.CASCADE)
    key = models.CharField(max_length=64)

ログインエンドポイント

import hashlib
import time
from django.http import JsonResponse
from rest_framework.views import APIView
from myapp import models

def generate_digest(identifier):
    timestamp = str(time.time())
    hasher = hashlib.sha256(identifier.encode('utf-8'))
    hasher.update(timestamp.encode('utf-8'))
    return hasher.hexdigest()

class SignInView(APIView):
    def post(self, request, *args, **kwargs):
        result = {'status': 200, 'detail': None}
        try:
            username = request.data.get('name')
            password = request.data.get('secret')
            profile = models.Account.objects.filter(
                name=username, secret=password
            ).first()
            if not profile:
                result['status'] = 401
                result['detail'] = '認証情報が無効です'
                return JsonResponse(result)
            
            session_key = generate_digest(username)
            models.Session.objects.update_or_create(
                account=profile, defaults={'key': session_key}
            )
            result['key'] = session_key
        except Exception as err:
            result['status'] = 500
            result['detail'] = 'サーバーエラー'
        return JsonResponse(result)

カスタム認証クラス

from rest_framework import exceptions
from rest_framework.authentication import BaseAuthentication
from myapp import models

class TokenValidator(BaseAuthentication):
    def authenticate(self, request):
        session_key = request.query_params.get('key')
        record = models.Session.objects.filter(key=session_key).first()
        if not record:
            raise exceptions.AuthenticationFailed('セッションが無効です')
        return (record.account, record)

    def authenticate_header(self, request):
        return 'Token'

保護対象ビュー

from django.http import JsonResponse
from rest_framework.views import APIView

CATALOG = {
    1: {'item': 'headphones', 'cost': 299},
    2: {'item': 'keyboard', 'cost': 149},
}

class ProductView(APIView):
    authentication_classes = [TokenValidator]

    def get(self, request, *args, **kwargs):
        # request.user  → 認証済みアカウント
        # request.auth  → セッションレコード
        return JsonResponse({
            'status': 200,
            'inventory': CATALOG
        })

内部処理の解明

リクエストの初期化

APIView.dispatch()が呼び出される際、まずinitialize_request()でリクエストをラップする。

def initialize_request(self, request, *args, **kwargs):
    return Request(
        request,
        parsers=self.get_parsers(),
        authenticators=self.get_authenticators(),  # 認証クラスのインスタンス化
        negotiator=self.get_content_negotiator(),
        parser_context=parser_context
    )

get_authenticators()は各認証クラスをインスタンス化する。

def get_authenticators(self):
    return [auth() for auth in self.authentication_classes]

認証チェーンの実行

dispatch() → initial() → perform_authentication() と呼び出される。

def perform_authentication(self, request):
    request.user  # プロパティアクセスで認証が実行される

Request.userプロパティは遅延評価で、初回アクセス時に_authenticate()を呼び出す。

@property
def user(self):
    if not hasattr(self, '_user'):
        self._authenticate()
    return self._user

認証ロジックの中核

def _authenticate(self):
    for validator in self.authenticators:
        try:
            credential = validator.authenticate(self)
        except exceptions.APIException:
            self._not_authenticated()
            raise
        
        if credential is not None:
            self._authenticator = validator
            self.user, self.auth = credential
            return
    
    self._not_authenticated()

各認証クラスのauthenticate()の戻り値:

  • タプル (user, auth) → 認証成功、request.userとrequest.authに設定
  • None → 次の認証クラスに委譲
  • 例外発生 → 認証失敗、AuthenticationFailedがスローされる

未認証時のフォールバック

def _not_authenticated(self):
    self._authenticator = None
    self.user = api_settings.UNAUTHENTICATED_USER()   # AnonymousUser
    self.auth = api_settings.UNAUTHENTICATED_TOKEN()  # None

BaseAuthenticationの継承

DRF標準のBaseAuthenticationを継承することで、インターフェースを明確にできる。

from rest_framework.authentication import BaseAuthentication

class TokenValidator(BaseAuthentication):
    def authenticate(self, request):
        # 必須:認証ロジックを実装
        pass

    def authenticate_header(self, request):
        # オプション:401レスポンス時のWWW-Authenticateヘッダー
        pass

設定のベストプラクティス

共通モジュールへの切り出し:

# myapp/security/authn.py
from rest_framework import exceptions
from myapp import models

class SignatureAuth(BaseAuthentication):
    def authenticate(self, request):
        signature = request.META.get('HTTP_X_SIGNATURE')
        if not signature:
            return None
        
        entry = models.Session.objects.select_related('account').filter(
            key=signature
        ).first()
        if not entry:
            raise exceptions.AuthenticationFailed('署名が無効です')
        
        return (entry.account, entry)

グローバル適用:

# settings.py
REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'myapp.security.authn.SignatureAuth',
    ]
}

特定ビューでの無効化:

class PublicView(APIView):
    authentication_classes = []  # 空リストで認証をスキップ
    
    def get(self, request):
        return JsonResponse({'message': '認証不要の公開情報'})

タグ: Django REST Framework Token Authentication BaseAuthentication APIView Request

10月21日 21:34 投稿