fix admin user persm
Build and Push Docker Images / Build and Push Docker Images (push) Failing after 5s

This commit is contained in:
2026-08-03 21:44:55 +05:00
parent 192f987125
commit 9d8b5cd26f
7 changed files with 106 additions and 32 deletions
+38
View File
@@ -1,6 +1,44 @@
from rest_framework import permissions from rest_framework import permissions
def user_is_admin(user) -> bool:
"""Single source of truth for "is this user an administrator?".
Any of the three flags grants admin access: the app-specific ``is_admin``
flag, Django's ``is_staff``, or ``is_superuser``. Both the permission
classes and the per-view authorization checks use this so the three flags
behave identically everywhere.
"""
return bool(
user
and user.is_authenticated
and (
getattr(user, "is_admin", False)
or getattr(user, "is_staff", False)
or getattr(user, "is_superuser", False)
)
)
class IsAdminOrStaffPermission(permissions.BasePermission):
"""Admin gate for admin-only endpoints and for views that have no
model/queryset of their own (e.g. proxy endpoints).
``IsStaffEditorPermission`` can't be used on model-less views because it
derives the required permission from ``view.queryset.model`` (``None``
there); it also requires granular Django model permissions that admin
accounts are not necessarily granted. This gate keys off admin status
instead.
"""
message = {
"message": "You do not have permission to perform this action.",
}
def has_permission(self, request, view):
return user_is_admin(request.user)
class IsStaffEditorPermission(permissions.DjangoModelPermissions): class IsStaffEditorPermission(permissions.DjangoModelPermissions):
perms_map = { perms_map = {
"GET": ["%(app_label)s.view_%(model_name)s"], "GET": ["%(app_label)s.view_%(model_name)s"],
+1
View File
@@ -94,6 +94,7 @@ class CustomUserSerializer(serializers.ModelSerializer):
"email", "email",
"last_login", "last_login",
"date_joined", "date_joined",
"is_staff",
"is_superuser", "is_superuser",
) )
+23 -27
View File
@@ -7,6 +7,7 @@ from rest_framework import generics, permissions
from rest_framework.authtoken.serializers import AuthTokenSerializer from rest_framework.authtoken.serializers import AuthTokenSerializer
from api.filters import UserFilter from api.filters import UserFilter
from api.mixins import StaffEditorPermissionMixin from api.mixins import StaffEditorPermissionMixin
from api.permissions import IsAdminOrStaffPermission, user_is_admin
from api.models import User, Atoll, Island, TemporaryUser from api.models import User, Atoll, Island, TemporaryUser
from api.notifications import send_sms from api.notifications import send_sms
from rest_framework.response import Response from rest_framework.response import Response
@@ -256,7 +257,8 @@ class UserprofileAPIView(generics.RetrieveUpdateAPIView):
return self.request.user return self.request.user
class UserUpdateAPIView(StaffEditorPermissionMixin, generics.UpdateAPIView): class UserUpdateAPIView(generics.UpdateAPIView):
permission_classes = [IsAdminOrStaffPermission]
serializer_class = UserUpdateSerializer serializer_class = UserUpdateSerializer
queryset = User.objects.all() queryset = User.objects.all()
lookup_field = "pk" lookup_field = "pk"
@@ -269,10 +271,7 @@ class UserUpdateAPIView(StaffEditorPermissionMixin, generics.UpdateAPIView):
{"message": "You cannot update a superuser."}, {"message": "You cannot update a superuser."},
status=status.HTTP_403_FORBIDDEN, status=status.HTTP_403_FORBIDDEN,
) )
if request.user != user and ( if request.user != user and not user_is_admin(request.user):
not request.user.is_authenticated
or not getattr(request.user, "is_admin", False)
):
return Response( return Response(
{"message": "You are not authorized to update this user."}, {"message": "You are not authorized to update this user."},
status=status.HTTP_403_FORBIDDEN, status=status.HTTP_403_FORBIDDEN,
@@ -287,7 +286,8 @@ class UserUpdateAPIView(StaffEditorPermissionMixin, generics.UpdateAPIView):
return super().update(request, *args, **kwargs) return super().update(request, *args, **kwargs)
class AgreementUpdateAPIView(StaffEditorPermissionMixin, generics.UpdateAPIView): class AgreementUpdateAPIView(generics.UpdateAPIView):
permission_classes = [IsAdminOrStaffPermission]
serializer_class = UserAgreementSerializer serializer_class = UserAgreementSerializer
queryset = User.objects.all() queryset = User.objects.all()
lookup_field = "pk" lookup_field = "pk"
@@ -300,10 +300,7 @@ class AgreementUpdateAPIView(StaffEditorPermissionMixin, generics.UpdateAPIView)
{"message": "You cannot update a superuser."}, {"message": "You cannot update a superuser."},
status=status.HTTP_403_FORBIDDEN, status=status.HTTP_403_FORBIDDEN,
) )
if request.user != user and ( if request.user != user and not user_is_admin(request.user):
not request.user.is_authenticated
or not getattr(request.user, "is_admin", False)
):
return Response( return Response(
{"message": "You are not authorized to update this user."}, {"message": "You are not authorized to update this user."},
status=status.HTTP_403_FORBIDDEN, status=status.HTTP_403_FORBIDDEN,
@@ -372,7 +369,8 @@ class ListUserView(StaffEditorPermissionMixin, generics.ListAPIView):
return User.objects.none() return User.objects.none()
class UserVerifyAPIView(StaffEditorPermissionMixin, generics.UpdateAPIView): class UserVerifyAPIView(generics.UpdateAPIView):
permission_classes = [IsAdminOrStaffPermission]
serializer_class = CustomUserSerializer serializer_class = CustomUserSerializer
queryset = User.objects.all() queryset = User.objects.all()
lookup_field = "pk" lookup_field = "pk"
@@ -380,10 +378,7 @@ class UserVerifyAPIView(StaffEditorPermissionMixin, generics.UpdateAPIView):
def update(self, request, *args, **kwargs): def update(self, request, *args, **kwargs):
user_id = kwargs.get("pk") user_id = kwargs.get("pk")
user = get_object_or_404(User, pk=user_id) user = get_object_or_404(User, pk=user_id)
if request.user != user and ( if request.user != user and not user_is_admin(request.user):
not request.user.is_authenticated
or not getattr(request.user, "is_admin", False)
):
return Response( return Response(
{"message": "You are not authorized to update this user."}, {"message": "You are not authorized to update this user."},
status=status.HTTP_403_FORBIDDEN, status=status.HTTP_403_FORBIDDEN,
@@ -421,7 +416,8 @@ class UserVerifyAPIView(StaffEditorPermissionMixin, generics.UpdateAPIView):
return Response({"message": "User successfully verified."}) return Response({"message": "User successfully verified."})
class UserRejectAPIView(StaffEditorPermissionMixin, generics.DestroyAPIView): class UserRejectAPIView(generics.DestroyAPIView):
permission_classes = [IsAdminOrStaffPermission]
serializer_class = CustomUserSerializer serializer_class = CustomUserSerializer
queryset = User.objects.all() queryset = User.objects.all()
lookup_field = "pk" lookup_field = "pk"
@@ -446,10 +442,7 @@ class UserRejectAPIView(StaffEditorPermissionMixin, generics.DestroyAPIView):
{"message": "You cannot remove a superuser."}, {"message": "You cannot remove a superuser."},
status=status.HTTP_403_FORBIDDEN, status=status.HTTP_403_FORBIDDEN,
) )
if request.user != user and ( if request.user != user and not user_is_admin(request.user):
not request.user.is_authenticated
or not getattr(request.user, "is_admin", False)
):
return Response( return Response(
{"message": "You are not authorized to reject this user."}, {"message": "You are not authorized to reject this user."},
status=status.HTTP_403_FORBIDDEN, status=status.HTTP_403_FORBIDDEN,
@@ -524,7 +517,8 @@ def filter_temporary_user(request):
) )
class UserDetailAPIView(StaffEditorPermissionMixin, generics.RetrieveAPIView): class UserDetailAPIView(generics.RetrieveAPIView):
permission_classes = [IsAdminOrStaffPermission]
queryset = User.objects.all() queryset = User.objects.all()
serializer_class = CustomReadOnlyUserSerializer serializer_class = CustomReadOnlyUserSerializer
lookup_field = "pk" lookup_field = "pk"
@@ -532,11 +526,7 @@ class UserDetailAPIView(StaffEditorPermissionMixin, generics.RetrieveAPIView):
def retrieve(self, request, *args, **kwargs): def retrieve(self, request, *args, **kwargs):
instance = self.get_object() instance = self.get_object()
user = request.user user = request.user
if ( if user != instance and not user_is_admin(user):
user != instance
and not getattr(user, "is_admin", False)
and not user.is_superuser #type: ignore
):
return Response( return Response(
{"message": "You are not authorized to view this user's details."}, {"message": "You are not authorized to view this user's details."},
status=status.HTTP_403_FORBIDDEN, status=status.HTTP_403_FORBIDDEN,
@@ -617,7 +607,7 @@ class RetrieveUpdateDestroyIslandView(
return super().update(request, *args, **kwargs) return super().update(request, *args, **kwargs)
class PersonVerifyAPIView(StaffEditorPermissionMixin, generics.GenericAPIView): class PersonVerifyAPIView(generics.GenericAPIView):
""" """
Admin-gated proxy to the external Person verification API. Admin-gated proxy to the external Person verification API.
@@ -625,9 +615,15 @@ class PersonVerifyAPIView(StaffEditorPermissionMixin, generics.GenericAPIView):
directly (it would leak an internal infra host to the browser), so the directly (it would leak an internal infra host to the browser), so the
backend owns the integration. Returns the upstream JSON as-is. backend owns the integration. Returns the upstream JSON as-is.
This view has no model of its own, so it uses ``IsAdminOrStaffPermission``
rather than the model-based ``StaffEditorPermissionMixin`` (which would
crash dereferencing ``view.queryset.model``).
GET /api/auth/person/<id_card>/ GET /api/auth/person/<id_card>/
""" """
permission_classes = [IsAdminOrStaffPermission]
def get(self, request, id_card: str, *args, **kwargs): def get(self, request, id_card: str, *args, **kwargs):
import requests import requests
from decouple import config from decouple import config
@@ -0,0 +1,22 @@
# Generated by Django 5.2 on 2026-08-03 15:59
import devices.models
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("devices", "0008_alter_device_blocked_by"),
]
operations = [
migrations.AlterField(
model_name="device",
name="mac",
field=models.CharField(
max_length=255,
unique=True,
validators=[devices.models.validate_mac_address],
),
),
]
+9
View File
@@ -15,10 +15,19 @@ def validate_mac_address(value):
return value return value
def normalize_mac(value):
"""Canonicalize any accepted MAC format to upper-case, dash-separated
form (e.g. "aa:bb:cc:dd:ee:ff" and "aabbccddeeff" -> "AA-BB-CC-DD-EE-FF").
Used so uniqueness is enforced on one canonical representation."""
hex_only = re.sub(r"[^0-9A-Fa-f]", "", value or "").upper()
return "-".join(hex_only[i : i + 2] for i in range(0, len(hex_only), 2))
class Device(models.Model): class Device(models.Model):
name = models.CharField(max_length=255) name = models.CharField(max_length=255)
mac = models.CharField( mac = models.CharField(
max_length=255, max_length=255,
unique=True,
validators=[ validators=[
validate_mac_address, validate_mac_address,
], ],
+4 -1
View File
@@ -1,5 +1,5 @@
from rest_framework import serializers from rest_framework import serializers
from .models import Device from .models import Device, normalize_mac
from api.serializers import CustomReadOnlyUserSerializer from api.serializers import CustomReadOnlyUserSerializer
from billing.models import Payment # Import the Payment model from billing.models import Payment # Import the Payment model
@@ -8,6 +8,9 @@ class CreateDeviceSerializer(serializers.ModelSerializer):
name = serializers.CharField(required=True) name = serializers.CharField(required=True)
mac = serializers.CharField(required=True) mac = serializers.CharField(required=True)
def validate_mac(self, value):
return normalize_mac(value)
class Meta: # type: ignore class Meta: # type: ignore
model = Device model = Device
fields = [ fields = [
+9 -4
View File
@@ -4,7 +4,7 @@ from rest_framework import generics, status
from rest_framework.response import Response from rest_framework.response import Response
from django_filters.rest_framework import DjangoFilterBackend from django_filters.rest_framework import DjangoFilterBackend
from billing.models import Payment from billing.models import Payment
from .models import Device from .models import Device, normalize_mac
from django.db.models import Prefetch from django.db.models import Prefetch
from .serializers import ( from .serializers import (
CreateDeviceSerializer, CreateDeviceSerializer,
@@ -79,9 +79,13 @@ class DeviceListCreateAPIView(
raw_mac = request.data.get("mac", None) raw_mac = request.data.get("mac", None)
mac = raw_mac.strip() if raw_mac else None mac = raw_mac.strip() if raw_mac else None
MAC_REGEX = re.compile(r"^([0-9A-Fa-f]{2}([.:-]?)){5}[0-9A-Fa-f]{2}$") 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): if not isinstance(mac, str) or not MAC_REGEX.match(mac):
return Response({"message": "Invalid mac address."}, status=400) return Response({"message": "Invalid mac address."}, status=400)
# Canonicalize BEFORE checking uniqueness and before saving, so that
# the same physical MAC in different formats/cases is treated as one.
mac = normalize_mac(mac)
if Device.objects.filter(mac=mac).exists(): if Device.objects.filter(mac=mac).exists():
return Response( return Response(
{"message": "Device with this mac address already exists."}, status=400 {"message": "Device with this mac address already exists."}, status=400
@@ -90,8 +94,9 @@ class DeviceListCreateAPIView(
if not mac_details.ok: if not mac_details.ok:
return Response({"message": "MAC address vendor not found."}, status=400) return Response({"message": "MAC address vendor not found."}, status=400)
mac = re.sub(NORMALIZE_MAC_REGEX, "-", mac).upper() # The serializer canonicalizes the MAC again on save (see
# CreateDeviceSerializer.validate_mac), so the stored value matches
# what we checked above regardless of the raw request format.
return super().create(request, *args, **kwargs) return super().create(request, *args, **kwargs)
def perform_create(self, serializer): def perform_create(self, serializer):