from django.contrib.auth.hashers import check_password from django.utils import timezone from rest_framework.authentication import BaseAuthentication from rest_framework.exceptions import AuthenticationFailed from ..models import APIKey class SystemUser: """Anonymous system user with scopes as permissions and a bound tenant.""" def __init__(self, scopes=None, tenant=None): self.scopes = scopes or [] self.tenant = tenant self.tenant_id = tenant.pk if tenant is not None else None self.is_authenticated = True self.is_superuser = False self.pk = None self.id = None @property def is_anonymous(self): return False def __str__(self): return f"SystemUser(scopes={self.scopes})" class APIKeyAuthentication(BaseAuthentication): """Service-to-service authentication via X-API-Key header (tenant-scoped).""" keyword = "X-API-Key" def authenticate(self, request): raw_key = request.META.get(f"HTTP_{self.keyword.replace('-', '_').upper()}") if not raw_key: return None try: prefix, secret = raw_key.split(".", 1) except ValueError: raise AuthenticationFailed("Invalid API key format.") # Keys are tenant-owned; look up across tenants and bind the tenant. api_key = APIKey.all_objects.filter(prefix=prefix, is_active=True).select_related("tenant").first() if not api_key: raise AuthenticationFailed("API key not found.") if not check_password(secret, api_key.key_hash): raise AuthenticationFailed("Invalid API key.") if api_key.expires_at and api_key.expires_at < timezone.now(): raise AuthenticationFailed("API key expired.") api_key.last_used_at = timezone.now() api_key.save(update_fields=["last_used_at"]) from infrasynth.tenancy.context import set_current_tenant if api_key.tenant is not None: set_current_tenant(api_key.tenant) return (SystemUser(scopes=api_key.scopes, tenant=api_key.tenant), api_key)