45 lines
1.6 KiB
Python
45 lines
1.6 KiB
Python
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)
|