GroupMemberViewSet.can_rank() - Code Metrics - Inspection of "Merge pull request #26 from ProjetSigma/refactor/r..." - ProjetSigma/backend - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 9a7f1b...b2a2c0 )

by Camille

created 2016-05-14 16:41 UTC

GroupMemberViewSet.can_rank() C

↳ Parent: GroupMemberViewSet

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
dl	0
loc	18
c	1
b	0
f	0
rs	6.6666
cc	8

from django.http import Http404
from django.db.models import Prefetch, Q

from rest_framework import viewsets, decorators, status, mixins
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated
from rest_framework.filters import BaseFilterBackend

from sigma_core.models.user import User
from sigma_core.models.group import Group, GroupAcknowledgment
from sigma_core.models.group_member import GroupMember
from sigma_core.serializers.group_member import GroupMemberSerializer


class GroupMemberFilterBackend(BaseFilterBackend):
    filter_q = {
        'user': lambda u: Q(user=u),
        'group': lambda g: Q(group=g)
    }

    def filter_queryset(self, request, queryset, view):
        """
        Limits all list requests w.r.t the Normal Rules of Visibility.
        """
        if not request.user.is_sigma_admin():
            invited_to_groups_ids = request.user.invited_to_groups.all().values_list('id', flat=True)
            user_groups_ids = request.user.memberships.filter(perm_rank__gte=1).values_list('group_id', flat=True)
            # I can see a GroupMember if one of the following conditions is met:
            #  - I am member of the group
            #  - I am invited to the group
            #  - (the group is public OR acknowledged by one of my groups) AND I can see the user w.r.t. NRVU
            queryset = queryset.prefetch_related(
                Prefetch('user__memberships', queryset=GroupMember.objects.filter(perm_rank__gte=1)),
                Prefetch('group__group_parents', queryset=GroupAcknowledgment.objects.filter(validated=True))
            ).filter(Q(group_id__in=user_groups_ids) | Q(group_id__in=invited_to_groups_ids) | (
                (Q(group__is_private=False) | Q(group__group_parents__id__in=user_groups_ids)) &
                    Q(user__memberships__group_id__in=user_groups_ids)
            ))

        for (param, q) in self.filter_q.items():
            x = request.query_params.get(param, None)
            if x is not None:
                queryset = queryset.filter(q(x))

        return queryset.distinct()


class GroupMemberViewSet(viewsets.ModelViewSet):
    queryset = GroupMember.objects.select_related('group', 'user')
    serializer_class = GroupMemberSerializer
    permission_classes = [IsAuthenticated, ]
    filter_backends = (GroupMemberFilterBackend, )

    def list(self, request):
        """
        ---
        parameters_strategy:
            query: replace
        parameters:
            - name: user
              type: integer
              paramType: query
            - name: group
              type: integer
              paramType: query
        """
        return super().list(request)

    def create(self, request):
        serializer = GroupMemberSerializer(data=request.data)
        if not serializer.is_valid():
            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

        if request.data.get('user_id', None) != request.user.id and not request.user.is_sigma_admin():
            return Response('You cannot add someone else to a group', status=status.HTTP_403_FORBIDDEN)

        try:
            group = Group.objects.get(pk=request.data.get('group_id', None))
        except Group.DoesNotExist:
            return Response(status=status.HTTP_404_NOT_FOUND)

        if group.default_member_rank < 0:
            return Response('You cannot join this group without an invitation', status=status.HTTP_403_FORBIDDEN)

        mem = serializer.save()
        return Response(serializer.data, status=status.HTTP_201_CREATED)

    def destroy(self, request, pk=None):
        from sigma_core.models.group import Group
        try:
            modified_mship = GroupMember.objects.all().select_related('group').get(pk=pk)
            group = modified_mship.group
            my_mship = GroupMember.objects.all().get(group=group, user=request.user)
        except GroupMember.DoesNotExist:
            raise Http404()

        # You can always quit the Group (ie destroy YOUR membership)
        if my_mship.id != modified_mship.id:
            # Can't modify someone higher than you
            if my_mship.perm_rank <= modified_mship.perm_rank:
                return Response(status=status.HTTP_403_FORBIDDEN)

            # Check permission
            if group.req_rank_kick > my_mship.perm_rank:
                return Response(status=status.HTTP_403_FORBIDDEN)

        modified_mship.delete()
        return Response(status=status.HTTP_204_NO_CONTENT)

    def can_rank(self, request, modified_mship, my_mship, perm_rank_new):
        # You can demote yourself:
        demote_self = modified_mship.id == my_mship.id and perm_rank_new < my_mship.perm_rank
        group = modified_mship.group

        # Can't modify someone higher than you, or set rank to higher than you
        if (my_mship.perm_rank <= modified_mship.perm_rank or my_mship.perm_rank <= perm_rank_new) and not demote_self:
            return False

        # promote
        if perm_rank_new > modified_mship.perm_rank:
            if group.req_rank_promote > my_mship.perm_rank:
                return False
        # demote
        else:
            if group.req_rank_demote > my_mship.perm_rank and not demote_self:
                return False
        return True

    @decorators.detail_route(methods=['put'])
    def rank(self, request, pk=None):
        """
        Change the perm_rank of a member of the group pk.
        ---
        request_serializer: null
        response_serializer: GroupMemberSerializer
        parameters_strategy:
            form: replace
        parameters:
            - name: perm_rank
              type: integer
              required: true
        """
        from sigma_core.models.group import Group
        try:
            modified_mship = GroupMember.objects.all().select_related('group').get(pk=pk)
            my_mship = GroupMember.objects.all().get(group=modified_mship.group, user=request.user)
        except GroupMember.DoesNotExist:
            raise Http404()

        perm_rank_new = request.data.get('perm_rank', None)

        try:
            if perm_rank_new > Group.ADMINISTRATOR_RANK or perm_rank_new < 1 or perm_rank_new == modified_mship.perm_rank:
                return Response(status=status.HTTP_400_BAD_REQUEST)
        except TypeError:
            return Response(status=status.HTTP_400_BAD_REQUEST)

        if not self.can_rank(request, modified_mship, my_mship, perm_rank_new):
            return Response(status=status.HTTP_403_FORBIDDEN)

        modified_mship.perm_rank = perm_rank_new
        modified_mship.save()

        return Response(GroupMemberSerializer(modified_mship).data, status=status.HTTP_200_OK)

    @decorators.detail_route(methods=['put'])
    def accept_join_request(self, request, pk=None):
        """
        Validate a pending membership request.
        ---
        request_serializer: null
        response_serializer: GroupMemberSerializer
        """
        try:
            gm = GroupMember.objects.select_related('group').get(pk=pk)
        except GroupMember.DoesNotExist:
            raise Http404()

        if not request.user.can_accept_join_requests(gm.group):
            return Response(status=status.HTTP_403_FORBIDDEN)

        gm.perm_rank = 1 # default_perm_rank should be 0, so validation is to set perm_rank to 1
        gm.save()

        # TODO: notify user of that change

        s = GroupMemberSerializer(gm)
        return Response(s.data, status=status.HTTP_200_OK)


1			from django.http import Http404
2			from django.db.models import Prefetch, Q
3
4			from rest_framework import viewsets, decorators, status, mixins
5			from rest_framework.response import Response
6			from rest_framework.permissions import IsAuthenticated
7			from rest_framework.filters import BaseFilterBackend
8
9			from sigma_core.models.user import User
10			from sigma_core.models.group import Group, GroupAcknowledgment
11			from sigma_core.models.group_member import GroupMember
12			from sigma_core.serializers.group_member import GroupMemberSerializer
13
14
15			class GroupMemberFilterBackend(BaseFilterBackend):
16			filter_q = {
17			'user': lambda u: Q(user=u),
18			'group': lambda g: Q(group=g)
19			}
20
21			def filter_queryset(self, request, queryset, view):
22			"""
23			Limits all list requests w.r.t the Normal Rules of Visibility.
24			"""
25			if not request.user.is_sigma_admin():
26			invited_to_groups_ids = request.user.invited_to_groups.all().values_list('id', flat=True)
27			user_groups_ids = request.user.memberships.filter(perm_rank__gte=1).values_list('group_id', flat=True)
28			# I can see a GroupMember if one of the following conditions is met:
29			# - I am member of the group
30			# - I am invited to the group
31			# - (the group is public OR acknowledged by one of my groups) AND I can see the user w.r.t. NRVU
32			queryset = queryset.prefetch_related(
33			Prefetch('user__memberships', queryset=GroupMember.objects.filter(perm_rank__gte=1)),
34			Prefetch('group__group_parents', queryset=GroupAcknowledgment.objects.filter(validated=True))
35			).filter(Q(group_id__in=user_groups_ids) \| Q(group_id__in=invited_to_groups_ids) \| (
36			(Q(group__is_private=False) \| Q(group__group_parents__id__in=user_groups_ids)) &
37			Q(user__memberships__group_id__in=user_groups_ids)
38			))
39
40			for (param, q) in self.filter_q.items():
41			x = request.query_params.get(param, None)
42			if x is not None:
43			queryset = queryset.filter(q(x))
44
45			return queryset.distinct()
46
47
48			class GroupMemberViewSet(viewsets.ModelViewSet):
49			queryset = GroupMember.objects.select_related('group', 'user')
50			serializer_class = GroupMemberSerializer
51			permission_classes = [IsAuthenticated, ]
52			filter_backends = (GroupMemberFilterBackend, )
53
54			def list(self, request):
55			"""
56			---
57			parameters_strategy:
58			query: replace
59			parameters:
60			- name: user
61			type: integer
62			paramType: query
63			- name: group
64			type: integer
65			paramType: query
66			"""
67			return super().list(request)
68
69			def create(self, request):
70			serializer = GroupMemberSerializer(data=request.data)
71			if not serializer.is_valid():
72			return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
73
74			if request.data.get('user_id', None) != request.user.id and not request.user.is_sigma_admin():
75			return Response('You cannot add someone else to a group', status=status.HTTP_403_FORBIDDEN)
76
77			try:
78			group = Group.objects.get(pk=request.data.get('group_id', None))
79			except Group.DoesNotExist:
80			return Response(status=status.HTTP_404_NOT_FOUND)
81
82			if group.default_member_rank < 0:
83			return Response('You cannot join this group without an invitation', status=status.HTTP_403_FORBIDDEN)
84
85			mem = serializer.save()
86			return Response(serializer.data, status=status.HTTP_201_CREATED)
87
88			def destroy(self, request, pk=None):
89			from sigma_core.models.group import Group
90			try:
91			modified_mship = GroupMember.objects.all().select_related('group').get(pk=pk)
92			group = modified_mship.group
93			my_mship = GroupMember.objects.all().get(group=group, user=request.user)
94			except GroupMember.DoesNotExist:
95			raise Http404()
96
97			# You can always quit the Group (ie destroy YOUR membership)
98			if my_mship.id != modified_mship.id:
99			# Can't modify someone higher than you
100			if my_mship.perm_rank <= modified_mship.perm_rank:
101			return Response(status=status.HTTP_403_FORBIDDEN)
102
103			# Check permission
104			if group.req_rank_kick > my_mship.perm_rank:
105			return Response(status=status.HTTP_403_FORBIDDEN)
106
107			modified_mship.delete()
108			return Response(status=status.HTTP_204_NO_CONTENT)
109
110			def can_rank(self, request, modified_mship, my_mship, perm_rank_new):
111			# You can demote yourself:
112			demote_self = modified_mship.id == my_mship.id and perm_rank_new < my_mship.perm_rank
113			group = modified_mship.group
114
115			# Can't modify someone higher than you, or set rank to higher than you
116			if (my_mship.perm_rank <= modified_mship.perm_rank or my_mship.perm_rank <= perm_rank_new) and not demote_self:
117			return False
118
119			# promote
120			if perm_rank_new > modified_mship.perm_rank:
121			if group.req_rank_promote > my_mship.perm_rank:
122			return False
123			# demote
124			else:
125			if group.req_rank_demote > my_mship.perm_rank and not demote_self:
126			return False
127			return True
128
129			@decorators.detail_route(methods=['put'])
130			def rank(self, request, pk=None):
131			"""
132			Change the perm_rank of a member of the group pk.
133			---
134			request_serializer: null
135			response_serializer: GroupMemberSerializer
136			parameters_strategy:
137			form: replace
138			parameters:
139			- name: perm_rank
140			type: integer
141			required: true
142			"""
143			from sigma_core.models.group import Group
144			try:
145			modified_mship = GroupMember.objects.all().select_related('group').get(pk=pk)
146			my_mship = GroupMember.objects.all().get(group=modified_mship.group, user=request.user)
147			except GroupMember.DoesNotExist:
148			raise Http404()
149
150			perm_rank_new = request.data.get('perm_rank', None)
151
152			try:
153			if perm_rank_new > Group.ADMINISTRATOR_RANK or perm_rank_new < 1 or perm_rank_new == modified_mship.perm_rank:
154			return Response(status=status.HTTP_400_BAD_REQUEST)
155			except TypeError:
156			return Response(status=status.HTTP_400_BAD_REQUEST)
157
158			if not self.can_rank(request, modified_mship, my_mship, perm_rank_new):
159			return Response(status=status.HTTP_403_FORBIDDEN)
160
161			modified_mship.perm_rank = perm_rank_new
162			modified_mship.save()
163
164			return Response(GroupMemberSerializer(modified_mship).data, status=status.HTTP_200_OK)
165
166			@decorators.detail_route(methods=['put'])
167			def accept_join_request(self, request, pk=None):
168			"""
169			Validate a pending membership request.
170			---
171			request_serializer: null
172			response_serializer: GroupMemberSerializer
173			"""
174			try:
175			gm = GroupMember.objects.select_related('group').get(pk=pk)
176			except GroupMember.DoesNotExist:
177			raise Http404()
178
179			if not request.user.can_accept_join_requests(gm.group):
180			return Response(status=status.HTTP_403_FORBIDDEN)
181
182			gm.perm_rank = 1 # default_perm_rank should be 0, so validation is to set perm_rank to 1
183			gm.save()
184
185			# TODO: notify user of that change
186
187			s = GroupMemberSerializer(gm)
188			return Response(s.data, status=status.HTTP_200_OK)
189

ProjetSigma / backend

Push — master ( 9a7f1b...b2a2c0 )

GroupMemberViewSet.can_rank() C

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like