79 lines
2.7 KiB
Python
79 lines
2.7 KiB
Python
"""开放平台认证与授权：
|
|
APIKeyAuthentication 支持第三方系统通过 Header:
|
|
X-API-Key: dh_xxxx.yyyy
|
|
或
|
|
Authorization: Api-Key dh_xxxx.yyyy
|
|
安全访问开放接口。
|
|
"""
|
|
|
|
from rest_framework import authentication, exceptions, permissions
|
|
from django.utils import timezone
|
|
from asgiref.sync import sync_to_async
|
|
from .models import APIKey
|
|
|
|
|
|
class APIKeyAuthentication(authentication.BaseAuthentication):
|
|
"""基于 API Key 的第三方系统鉴权类。"""
|
|
|
|
def authenticate(self, request):
|
|
auth_header = request.META.get("HTTP_AUTHORIZATION", "")
|
|
api_key_header = request.META.get("HTTP_X_API_KEY", "")
|
|
|
|
raw_key = None
|
|
if api_key_header:
|
|
raw_key = api_key_header.strip()
|
|
elif auth_header.startswith("Api-Key "):
|
|
raw_key = auth_header[8:].strip()
|
|
|
|
if not raw_key:
|
|
return None
|
|
|
|
if "." not in raw_key:
|
|
raise exceptions.AuthenticationFailed("API Key 格式无效（必须包含前缀）")
|
|
|
|
prefix = raw_key.split(".")[0]
|
|
key_obj = APIKey.objects.filter(prefix=prefix, is_active=True).select_related("tenant").first()
|
|
if not key_obj:
|
|
raise exceptions.AuthenticationFailed("API Key 不存在或已被禁用")
|
|
|
|
if key_obj.expires_at and key_obj.expires_at < timezone.now():
|
|
raise exceptions.AuthenticationFailed("API Key 已过期")
|
|
|
|
if not key_obj.verify_key(raw_key):
|
|
raise exceptions.AuthenticationFailed("API Key 签名无效")
|
|
|
|
# P0-5：把纸面的 rate_limit 接进认证链（分钟窗口，超限 429）。
|
|
from apps.core.ratelimit import check_apikey_rate_limit
|
|
|
|
check_apikey_rate_limit(key_obj.prefix, key_obj.rate_limit)
|
|
|
|
# 记录调用时间并绑定 tenant
|
|
key_obj.last_used_at = timezone.now()
|
|
key_obj.save(update_fields=["last_used_at"])
|
|
|
|
# 注入租户上下文
|
|
request.tenant_obj = key_obj.tenant
|
|
if hasattr(request, "_request"):
|
|
request._request.tenant_obj = key_obj.tenant
|
|
|
|
# 绑定系统用户
|
|
user = key_obj.created_by
|
|
if not user:
|
|
from django.contrib.auth import get_user_model
|
|
user = get_user_model().objects.filter(is_superuser=True).first()
|
|
|
|
return (user, key_obj)
|
|
|
|
|
|
def require_scope(scope_name: str):
|
|
"""用于视图权限检查的 Scope 装饰辅助类。"""
|
|
|
|
class ScopePermission(permissions.BasePermission):
|
|
def has_permission(self, request, view):
|
|
api_key = getattr(request, "auth", None)
|
|
if not isinstance(api_key, APIKey):
|
|
return False # 未携带有效 APIKey 则拒绝访问
|
|
return scope_name in api_key.scopes or "*" in api_key.scopes
|
|
|
|
return ScopePermission
|