from django.contrib.auth import authenticate, login, logout from django.middleware.csrf import get_token from django.views.decorators.csrf import ensure_csrf_cookie from django.utils.decorators import method_decorator from rest_framework import permissions, status, viewsets from rest_framework.response import Response from rest_framework.views import APIView from apps.accounts.models import ApiKey, User from apps.accounts.serializers import ( ApiKeyCreateSerializer, ApiKeySerializer, LoginSerializer, UserSerializer, ) class IsOwnerOrStaff(permissions.BasePermission): def has_object_permission(self, request, view, obj): if request.user.is_staff or request.user.is_superuser: return True return getattr(obj, "user_id", None) == request.user.pk or obj == request.user class IsAdminUser(permissions.IsAdminUser): """Admin access for staff OR superadmin (superadmins may have is_staff=False).""" def has_permission(self, request, view): return bool(request.user and (request.user.is_staff or request.user.is_superuser)) class UserViewSet(viewsets.ModelViewSet): queryset = User.objects.all().order_by("user_name") serializer_class = UserSerializer permission_classes = [IsAdminUser] class ApiKeyViewSet(viewsets.ModelViewSet): http_method_names = ["get", "post", "delete", "head", "options"] def get_queryset(self): qs = ApiKey.objects.select_related("user").order_by("-created_at") if self.request.user.is_staff or self.request.user.is_superuser: return qs return qs.filter(user=self.request.user) def get_serializer_class(self): if self.action == "create": return ApiKeyCreateSerializer return ApiKeySerializer def get_permissions(self): return [permissions.IsAuthenticated()] def create(self, request, *args, **kwargs): serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) name = serializer.validated_data["name"] user = request.user if (request.user.is_staff or request.user.is_superuser) and serializer.validated_data.get( "user_id" ): user = User.objects.get(pk=serializer.validated_data["user_id"]) api_key, raw = ApiKey.generate(user=user, name=name) data = ApiKeySerializer(api_key).data data["raw_key"] = raw return Response(data, status=status.HTTP_201_CREATED) def perform_destroy(self, instance): if ( not self.request.user.is_staff and not self.request.user.is_superuser and instance.user_id != self.request.user.pk ): raise permissions.PermissionDenied() instance.delete() @method_decorator(ensure_csrf_cookie, name="dispatch") class CsrfView(APIView): permission_classes = [permissions.AllowAny] authentication_classes = [] def get(self, request): return Response({"csrfToken": get_token(request)}) class LoginView(APIView): permission_classes = [permissions.AllowAny] authentication_classes = [] def post(self, request): serializer = LoginSerializer(data=request.data) serializer.is_valid(raise_exception=True) user = authenticate( request, username=serializer.validated_data["user_name"], password=serializer.validated_data["password"], ) if user is None: return Response( {"detail": "Invalid credentials"}, status=status.HTTP_401_UNAUTHORIZED, ) login(request, user) return Response(UserSerializer(user).data) class LogoutView(APIView): def post(self, request): logout(request) return Response(status=status.HTTP_204_NO_CONTENT) class MeView(APIView): def get(self, request): return Response(UserSerializer(request.user).data)