13099167831

Committed 02 Feb 2025 01:17PM UTC coverage: 83.644% (-0.2%) from 83.876%

Build # 13099167831

Build Type

push

github

Committed by

web-flow

Commit Message

Combine NFC, MXID, and phone number door access logic (#552)

* Refactor NFC, MXID, and phone number door access into one

* Move phone_list next to list

* filter the user with the nfccard

---------

Co-authored-by: Tatu Wikman <tatu.wikman@gmail.com>

Run Details

44 of 44 new or added lines in 1 file covered. (100.0%)

1 existing line in 1 file now uncovered.

2107 of 2519 relevant lines covered (83.64%)

0.84 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.82

/api/views.py

import logging

from django.shortcuts import get_object_or_404

from api.serializers import AccessDataSerializer, UserAccessSerializer
from drfx import config as config
from rest_framework import mixins, status, viewsets
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.throttling import AnonRateThrottle
from rest_framework_tracking.mixins import LoggingMixin
from users.models import CustomUser, ServiceSubscription

from utils.phonenumber import normalize_number

from .models import AccessDevice, DeviceAccessLogEntry
from users.signals import door_access_denied

logger = logging.getLogger(__name__)


class VerySlowThrottle(AnonRateThrottle):
    """
    Throttle for access views
    """

    rate = "10/minute"


class AccessViewSet(LoggingMixin, mixins.ListModelMixin, viewsets.GenericViewSet):
    """
    Access checker api

    contains few sub endpoints like phone, mxid and nfc that can be used to check one user
    with one access method.

    When for example phone endpoint is called with a post it will try to check the incoming
    deviceid and payload against the database and return 200 ok if the user has access.

    Get request to the same phone endpoint will return a list of all active users.

    Also throws erros for all default actions TODO: there is probably a cleaner way to do this
    """

    throttle_classes = [VerySlowThrottle]
    permission_classes = []
    # default to AccessDataSerializer for all methods
    serializer_class = AccessDataSerializer
    # default queryset as none
    queryset = CustomUser.objects.none

    def access_token_abstraction(self, request, format, method):
        logentry = DeviceAccessLogEntry()
        inserializer = AccessDataSerializer(data=request.data)
        inserializer.is_valid(raise_exception=True)

        # check that we know which device this is
        deviceqs = AccessDevice.objects.all()
        deviceid = inserializer.validated_data.get("deviceid")
        device = get_object_or_404(deviceqs, deviceid=deviceid)
        logging.debug(f"found device {device}")

        access_token = inserializer.validated_data.get("payload")
        if method == "phone":
            # phone number comes in payload, but it is in a wrong format
            # the number will most probably start with 00 instead of +
            access_token = normalize_number(access_token)
        users = []
        if method == "phone":
            users = CustomUser.objects.filter(phone=access_token)
        elif method == "nfc":
            users = CustomUser.objects.filter(nfccard__cardid=access_token)
        elif method == "mxid":
            users = CustomUser.objects.filter(mxid=access_token)

        logentry.device = device
        logentry.payload = access_token

        # 0 = success, any other = failure
        response_status = 0

        # nothing found, 480 (NO_CONTENT)
        if users.count() == 0:
            logentry.granted = False
            logentry.save()
            return Response(status=480)

        # planned database scheme says that
        # phone numbers, MXIDs, nfc tags are/will be unique
        user = users.first()

        # user does not have access rights
        if not user.has_door_access():
            response_status = 481

        logentry.granted = response_status == 0
        logentry.save()

        if response_status == 0:
            if method != "phone":
                # uppercase NFC and MXID
                user.log(f"Door opened with {method.upper()}")
            else:
                user.log(f"Door opened with {method}")
            outserializer = UserAccessSerializer(user)
            return Response(outserializer.data)

        if response_status == 481:
            if method != "phone":
                # uppercase NFC and MXID
                user.log(f"Door access denied with {method.upper()}")
            else:
                user.log(f"Door access denied with {method}")
            door_access_denied.send(sender=self.__class__, user=user, method=method)
            outserializer = UserAccessSerializer(user)
            return Response(outserializer.data, status=response_status)

        return Response(status=response_status)

    @action(detail=False, methods=["post"], throttle_classes=[VerySlowThrottle])
    def phone(self, request, format=None):
        """
        Check if the phone number is allowed to access and return some user data
        to caller.

        call with something like this
        http POST http://127.0.0.1:8000/api/v1/access/phone/ deviceid=asdf payload=0440431918

        returns 200 ok with some user data if everything is fine and 4XX for other situations

        users with enough power will also get a list of all users with door access with this endpoint
        """
        return AccessViewSet.access_token_abstraction(self, request, format, "phone")

    @action(detail=False, methods=["post"], throttle_classes=[VerySlowThrottle])
    def nfc(self, request, format=None):
        """
        NFC card access
        """
        return AccessViewSet.access_token_abstraction(self, request, format, "nfc")

    @action(detail=False, methods=["post"], throttle_classes=[VerySlowThrottle])
    def mxid(self, request, format=None):
        """
        Matrix mxid access
        """
        return AccessViewSet.access_token_abstraction(self, request, format, "mxid")

    @phone.mapping.get
    def phone_list(self, request, format=None):
        """
        List all phone access users
        """
        # only for superusers
        if not request.user or not request.user.is_superuser:
            return Response(status=status.HTTP_401_UNAUTHORIZED)

        # collect list of all users that have door access
        users_with_door_access = []
        for ss in (
            ServiceSubscription.objects.select_related("user")
            .filter(service=config.DEFAULT_ACCOUNT_SERVICE)
            .filter(state=ServiceSubscription.ACTIVE)
        ):
            users_with_door_access.append(ss.user)

        # and output it
        outserializer = UserAccessSerializer(users_with_door_access, many=True)
        return Response(outserializer.data)

    def list(self, request):
        return Response(status=status.HTTP_501_NOT_IMPLEMENTED)

1	import logging	1✔
2
3	from django.shortcuts import get_object_or_404	1✔
4
5	from api.serializers import AccessDataSerializer, UserAccessSerializer	1✔
6	from drfx import config as config	1✔
7	from rest_framework import mixins, status, viewsets	1✔
8	from rest_framework.decorators import action	1✔
9	from rest_framework.response import Response	1✔
10	from rest_framework.throttling import AnonRateThrottle	1✔
11	from rest_framework_tracking.mixins import LoggingMixin	1✔
12	from users.models import CustomUser, ServiceSubscription	1✔
13
14	from utils.phonenumber import normalize_number	1✔
15
16	from .models import AccessDevice, DeviceAccessLogEntry	1✔
17	from users.signals import door_access_denied	1✔
18
19	logger = logging.getLogger(__name__)	1✔
20
21
22	class VerySlowThrottle(AnonRateThrottle):	1✔
23	"""
24	Throttle for access views
25	"""
26
27	rate = "10/minute"	1✔
28
29
30	class AccessViewSet(LoggingMixin, mixins.ListModelMixin, viewsets.GenericViewSet):	1✔
31	"""
32	Access checker api
33
34	contains few sub endpoints like phone, mxid and nfc that can be used to check one user
35	with one access method.
36
37	When for example phone endpoint is called with a post it will try to check the incoming
38	deviceid and payload against the database and return 200 ok if the user has access.
39
40	Get request to the same phone endpoint will return a list of all active users.
41
42	Also throws erros for all default actions TODO: there is probably a cleaner way to do this
43	"""
44
45	throttle_classes = [VerySlowThrottle]	1✔
46	permission_classes = []	1✔
47	# default to AccessDataSerializer for all methods
48	serializer_class = AccessDataSerializer	1✔
49	# default queryset as none
50	queryset = CustomUser.objects.none	1✔
51
52	def access_token_abstraction(self, request, format, method):	1✔
53	logentry = DeviceAccessLogEntry()	1✔
54	inserializer = AccessDataSerializer(data=request.data)	1✔
55	inserializer.is_valid(raise_exception=True)	1✔
56
57	# check that we know which device this is
58	deviceqs = AccessDevice.objects.all()	1✔
59	deviceid = inserializer.validated_data.get("deviceid")	1✔
60	device = get_object_or_404(deviceqs, deviceid=deviceid)	1✔
61	logging.debug(f"found device {device}")	1✔
62
63	access_token = inserializer.validated_data.get("payload")	1✔
64	if method == "phone":	1✔
65	# phone number comes in payload, but it is in a wrong format
66	# the number will most probably start with 00 instead of +
67	access_token = normalize_number(access_token)	1✔
68	users = []	1✔
69	if method == "phone":	1✔
70	users = CustomUser.objects.filter(phone=access_token)	1✔
71	elif method == "nfc":	1✔
72	users = CustomUser.objects.filter(nfccard__cardid=access_token)	1✔
73	elif method == "mxid":	1✔
74	users = CustomUser.objects.filter(mxid=access_token)	1✔
75
76	logentry.device = device	1✔
77	logentry.payload = access_token	1✔
78
79	# 0 = success, any other = failure
80	response_status = 0	1✔
81
82	# nothing found, 480 (NO_CONTENT)
83	if users.count() == 0:	1✔
84	logentry.granted = False	1✔
85	logentry.save()	1✔
86	return Response(status=480)	1✔
87
88	# planned database scheme says that
89	# phone numbers, MXIDs, nfc tags are/will be unique
90	user = users.first()	1✔
91
92	# user does not have access rights
93	if not user.has_door_access():	1✔
94	response_status = 481	1✔
95
96	logentry.granted = response_status == 0	1✔
97	logentry.save()	1✔
98
99	if response_status == 0:	1✔
100	if method != "phone":	1✔
101	# uppercase NFC and MXID
102	user.log(f"Door opened with {method.upper()}")	1✔
103	else:
104	user.log(f"Door opened with {method}")	1✔
105	outserializer = UserAccessSerializer(user)	1✔
106	return Response(outserializer.data)	1✔
107
108	if response_status == 481:	1✔
109	if method != "phone":	1✔
110	# uppercase NFC and MXID
111	user.log(f"Door access denied with {method.upper()}")	1✔
112	else:
113	user.log(f"Door access denied with {method}")	1✔
114	door_access_denied.send(sender=self.__class__, user=user, method=method)	1✔
115	outserializer = UserAccessSerializer(user)	1✔
116	return Response(outserializer.data, status=response_status)	1✔
117
UNCOV 118	return Response(status=response_status)	×
119
120	@action(detail=False, methods=["post"], throttle_classes=[VerySlowThrottle])	1✔
121	def phone(self, request, format=None):	1✔
122	"""
123	Check if the phone number is allowed to access and return some user data
124	to caller.
125
126	call with something like this
127	http POST http://127.0.0.1:8000/api/v1/access/phone/ deviceid=asdf payload=0440431918
128
129	returns 200 ok with some user data if everything is fine and 4XX for other situations
130
131	users with enough power will also get a list of all users with door access with this endpoint
132	"""
133	return AccessViewSet.access_token_abstraction(self, request, format, "phone")	1✔
134
135	@action(detail=False, methods=["post"], throttle_classes=[VerySlowThrottle])	1✔
136	def nfc(self, request, format=None):	1✔
137	"""
138	NFC card access
139	"""
140	return AccessViewSet.access_token_abstraction(self, request, format, "nfc")	1✔
141
142	@action(detail=False, methods=["post"], throttle_classes=[VerySlowThrottle])	1✔
143	def mxid(self, request, format=None):	1✔
144	"""
145	Matrix mxid access
146	"""
147	return AccessViewSet.access_token_abstraction(self, request, format, "mxid")	1✔
148
149	@phone.mapping.get	1✔
150	def phone_list(self, request, format=None):	1✔
151	"""
152	List all phone access users
153	"""
154	# only for superusers
155	if not request.user or not request.user.is_superuser:	1✔
156	return Response(status=status.HTTP_401_UNAUTHORIZED)	1✔
157
158	# collect list of all users that have door access
159	users_with_door_access = []	1✔
160	for ss in (	1✔
161	ServiceSubscription.objects.select_related("user")
162	.filter(service=config.DEFAULT_ACCOUNT_SERVICE)
163	.filter(state=ServiceSubscription.ACTIVE)
164	):
165	users_with_door_access.append(ss.user)	1✔
166
167	# and output it
168	outserializer = UserAccessSerializer(users_with_door_access, many=True)	1✔
169	return Response(outserializer.data)	1✔
170
171	def list(self, request):	1✔
172	return Response(status=status.HTTP_501_NOT_IMPLEMENTED)	1✔

TampereHacklab / mulysa / 13099167831

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous