from attr import dataclass from rest_framework import generics, status from rest_framework.response import Response from django_filters.rest_framework import DjangoFilterBackend from .models import Device from .serializers import ( CreateDeviceSerializer, DeviceSerializer, ReadOnlyDeviceSerializer, BlockDeviceSerializer, ) from api.mixins import StaffEditorPermissionMixin from .filters import DeviceFilter import re import requests from decouple import config class DeviceListCreateAPIView( StaffEditorPermissionMixin, generics.ListCreateAPIView, ): queryset = Device.objects.select_related("user").prefetch_related("payments").all() serializer_class = CreateDeviceSerializer filter_backends = [DjangoFilterBackend] filterset_fields = "__all__" filterset_class = DeviceFilter def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) if not request.user.is_superuser: queryset = queryset.filter(user=request.user) page = self.paginate_queryset(queryset) if page is not None: serializer = self.get_serializer(page, many=True) return self.get_paginated_response(serializer.data) serializer = self.get_serializer(queryset, many=True) return Response(serializer.data) def get_serializer_class(self) -> type: if self.request.method == "POST": return CreateDeviceSerializer return DeviceSerializer def create(self, request, *args, **kwargs): mac = request.data.get("mac", None) MAC_REGEX = re.compile(r"^([0-9A-Fa-f]{2}([.:-]?)){5}[0-9A-Fa-f]{2}$") NORMALIZE_MAC_REGEX = re.compile(r"[^0-9A-Fa-f]") if not isinstance(mac, str) or not MAC_REGEX.match(mac): return Response({"message": "Invalid mac address."}, status=400) if Device.objects.filter(mac=mac).exists(): return Response( {"message": "Device with this mac address already exists."}, status=400 ) mac_details = get_mac_address_details(mac) if mac_details.vendor == "Unknown": return Response({"message": "MAC address vendor not found."}, status=400) mac = re.sub(NORMALIZE_MAC_REGEX, "-", mac).upper() request.data["mac"] = mac return super().create(request, *args, **kwargs) def perform_create(self, serializer): mac_details = get_mac_address_details(serializer.validated_data.get("mac")) serializer.save( user=self.request.user, vendor=mac_details.vendor, ) class DeviceDetailAPIView(StaffEditorPermissionMixin, generics.RetrieveAPIView): queryset = Device.objects.select_related("user").all() serializer_class = ReadOnlyDeviceSerializer lookup_field = "pk" class DeviceUpdateAPIView(StaffEditorPermissionMixin, generics.UpdateAPIView): queryset = Device.objects.all() serializer_class = CreateDeviceSerializer lookup_field = "pk" def update(self, request, **kwargs): instance = self.get_object() user_id = request.user.id if not request.user.is_superuser and instance.user_id != user_id: return Response( {"message": "You are not authorized to update this device."}, status=403, ) # Validate MAC address format mac = request.data.get("mac") if mac and not re.match(r"^([0-9A-Fa-f]{2}([.:-]?)){5}[0-9A-Fa-f]{2}$", mac): return Response({"message": "Invalid MAC address"}, status=400) serializer = self.get_serializer(instance, data=request.data, partial=True) serializer.is_valid(raise_exception=True) self.perform_update(serializer) return Response(serializer.data) class DeviceBlockAPIView(StaffEditorPermissionMixin, generics.UpdateAPIView): queryset = Device.objects.all() serializer_class = BlockDeviceSerializer lookup_field = "pk" def update(self, request, *args, **kwargs): # Pass 'partial=True' to allow partial updates user_id = request.user.id instance = self.get_object() if not request.user.is_superuser and instance.user_id != user_id: return Response( {"message": "You are not authorized to block this device."}, status=403, ) blocked = request.data.get("blocked", None) if blocked is None: return Response({"message": "Blocked field is required."}, status=400) if not isinstance(blocked, bool): return Response({"message": "Blocked field must be a boolean."}, status=400) instance.blocked = blocked instance.save() serializer = self.get_serializer(instance, data=request.data, partial=False) serializer.is_valid(raise_exception=True) self.perform_update(serializer) return Response(serializer.data) class DeviceDestroyAPIView(StaffEditorPermissionMixin, generics.DestroyAPIView): queryset = Device.objects.all() serializer_class = DeviceSerializer lookup_field = "pk" def destroy(self, request, *args, **kwargs): instance = self.get_object() device_name = instance.name self.perform_destroy(instance) return Response( {"message": f"Device '{device_name}' deleted."}, status=status.HTTP_200_OK, ) @dataclass class MacResponse: mac_address: str vendor: str detail: str | None = None def get_mac_address_details(mac: str) -> MacResponse: API_URL = config("MACVENDOR_API_URL") if not API_URL: raise ValueError("MACVENDOR API URL Not set. Please set it.") response = requests.get(f"{API_URL}/lookup/{mac}") json_data = response.json() if response.status_code == 200: return MacResponse( mac_address=json_data.get("mac_address", mac), vendor=json_data.get("vendor", ""), detail=json_data.get("detail"), ) else: return MacResponse(mac_address=mac, vendor="Unknown", detail=None)