from datetime import timedelta from django.contrib.auth.models import AnonymousUser from django.utils import timezone from rest_framework import exceptions from rest_framework.authentication import BaseAuthentication from rest_framework.permissions import BasePermission from apps.system.models import ApiKey class ApiKeyAuthentication(BaseAuthentication): """ API Key认证: 取请求头 X-API-Key 或查询参数 api_key 通过后 request.auth 为 ApiKey 实例, request.user 为匿名用户 未携带Key时返回None, 交给后续认证类处理 """ def authenticate(self, request): raw_key = request.META.get('HTTP_X_API_KEY') or request.GET.get('api_key') if not raw_key: return None try: apikey = ApiKey.objects.get(key=raw_key) except ApiKey.DoesNotExist: raise exceptions.AuthenticationFailed('无效的API Key') if not apikey.is_active: raise exceptions.AuthenticationFailed('API Key已禁用') if apikey.expires_at and apikey.expires_at <= timezone.now(): raise exceptions.AuthenticationFailed('API Key已过期') now = timezone.now() # last_used_at节流更新, 避免每次请求都写库 if apikey.last_used_at is None or now - apikey.last_used_at > timedelta(minutes=5): ApiKey.objects.filter(pk=apikey.pk).update(last_used_at=now) return (AnonymousUser(), apikey) class HasApiKey(BasePermission): """ 持有效API Key即放行 """ def has_permission(self, request, view): return isinstance(request.auth, ApiKey)