File size: 3,730 Bytes
45b7714 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | """
API authentication decorators.
"""
import logging
from functools import wraps
from django.conf import settings
from django.core.cache import cache
from django.http import JsonResponse
from django.utils import timezone
from core.models import DataStoreAPIToken
logger = logging.getLogger(__name__)
# Rate limiting settings
API_RATE_LIMIT = getattr(settings, "API_RATE_LIMIT", 60) # requests per minute
API_RATE_WINDOW = 60 # seconds
def api_token_required(view_func):
"""
Decorator to require API token authentication.
Validates the token, checks expiration, enforces rate limiting,
and attaches the token to the request for view access.
Token can be provided via:
- Authorization: Bearer <token>
- X-API-Key: <token>
"""
@wraps(view_func)
def wrapper(request, *args, **kwargs):
# Extract token from headers
token = _extract_token(request)
if not token:
return JsonResponse(
{"error": {"code": "UNAUTHORIZED", "message": "API token required"}},
status=401,
)
# Validate token
try:
api_token = DataStoreAPIToken.objects.select_related("datastore").get(
token=token,
is_active=True,
)
except DataStoreAPIToken.DoesNotExist:
logger.warning(f"API request with invalid token: {token[:8]}...")
return JsonResponse(
{"error": {"code": "UNAUTHORIZED", "message": "Invalid API token"}},
status=401,
)
# Check expiration
if api_token.expires_at and api_token.expires_at < timezone.now():
logger.info(f"API request with expired token: {api_token.name}")
return JsonResponse(
{"error": {"code": "UNAUTHORIZED", "message": "API token has expired"}},
status=401,
)
# Rate limiting by token
rate_key = f"api_rate_{api_token.id}"
requests_count = cache.get(rate_key, 0)
if requests_count >= API_RATE_LIMIT:
logger.warning(f"API rate limit exceeded for token: {api_token.name}")
return JsonResponse(
{"error": {"code": "RATE_LIMITED", "message": "Rate limit exceeded. Try again later."}},
status=429,
)
cache.set(rate_key, requests_count + 1, API_RATE_WINDOW)
# Update last used timestamp (async-safe, won't block)
DataStoreAPIToken.objects.filter(id=api_token.id).update(
last_used_at=timezone.now()
)
# Attach token to request for view access
request.api_token = api_token
return view_func(request, *args, **kwargs)
return wrapper
def _extract_token(request):
"""
Extract API token from request headers.
Supports:
- Authorization: Bearer <token>
- X-API-Key: <token>
"""
# Try Authorization: Bearer <token>
auth_header = request.headers.get("Authorization", "")
if auth_header.startswith("Bearer "):
return auth_header[7:]
# Try X-API-Key: <token>
return request.headers.get("X-API-Key")
def add_cors_headers(response):
"""
Add CORS headers to API response.
For self-hosted deployments, we allow all origins by default.
This can be configured via API_CORS_ORIGINS setting.
"""
cors_origins = getattr(settings, "API_CORS_ORIGINS", "*")
response["Access-Control-Allow-Origin"] = cors_origins
response["Access-Control-Allow-Methods"] = "GET, OPTIONS"
response["Access-Control-Allow-Headers"] = "Authorization, X-API-Key, Content-Type"
response["Access-Control-Max-Age"] = "86400"
return response
|