13098608777

Committed 02 Feb 2025 12:09PM UTC coverage: 83.67%. First build

Build # 13098608777

Build Type

Pull #552

github

Committed by

web-flow

Commit Message

Merge a5abbfdef into 336ac0fc0

Pull Request Pull Request #552: Combine NFC, MXID, and phone number door access logic

Run Details

41 of 42 new or added lines in 1 file covered. (97.62%)

2111 of 2523 relevant lines covered (83.67%)

0.84 hits per line

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

98.88

/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, NFCCard, 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 = NFCCard.objects.filter(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
        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 = None
        if method == "nfc":
            user = users.first().user
            logentry.nfccard = users.first()
        else:
            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, NFCCard, 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 = NFCCard.objects.filter(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
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 = None	1✔
91	if method == "nfc":	1✔
92	user = users.first().user	1✔
93	logentry.nfccard = users.first()	1✔
94	else:
95	user = users.first()	1✔
96
97	# user does not have access rights
98	if not user.has_door_access():	1✔
99	response_status = 481	1✔
100
101	logentry.granted = response_status == 0	1✔
102	logentry.save()	1✔
103
104	if response_status == 0:	1✔
105	if method != "phone":	1✔
106	# uppercase NFC and MXID
107	user.log(f"Door opened with {method.upper()}")	1✔
108	else:
109	user.log(f"Door opened with {method}")	1✔
110	outserializer = UserAccessSerializer(user)	1✔
111	return Response(outserializer.data)	1✔
112
113	if response_status == 481:	1✔
114	if method != "phone":	1✔
115	# uppercase NFC and MXID
116	user.log(f"Door access denied with {method.upper()}")	1✔
117	else:
118	user.log(f"Door access denied with {method}")	1✔
119	door_access_denied.send(sender=self.__class__, user=user, method=method)	1✔
120	outserializer = UserAccessSerializer(user)	1✔
121	return Response(outserializer.data, status=response_status)	1✔
122
NEW 123	return Response(status=response_status)	×
124
125	@action(detail=False, methods=["post"], throttle_classes=[VerySlowThrottle])	1✔
126	def phone(self, request, format=None):	1✔
127	"""
128	Check if the phone number is allowed to access and return some user data
129	to caller.
130
131	call with something like this
132	http POST http://127.0.0.1:8000/api/v1/access/phone/ deviceid=asdf payload=0440431918
133
134	returns 200 ok with some user data if everything is fine and 4XX for other situations
135
136	users with enough power will also get a list of all users with door access with this endpoint
137	"""
138	return AccessViewSet.access_token_abstraction(self, request, format, "phone")	1✔
139
140	@action(detail=False, methods=["post"], throttle_classes=[VerySlowThrottle])	1✔
141	def nfc(self, request, format=None):	1✔
142	"""
143	NFC card access
144	"""
145	return AccessViewSet.access_token_abstraction(self, request, format, "nfc")	1✔
146
147	@action(detail=False, methods=["post"], throttle_classes=[VerySlowThrottle])	1✔
148	def mxid(self, request, format=None):	1✔
149	"""
150	Matrix mxid access
151	"""
152	return AccessViewSet.access_token_abstraction(self, request, format, "mxid")	1✔
153
154	@phone.mapping.get	1✔
155	def phone_list(self, request, format=None):	1✔
156	"""
157	List all phone access users
158	"""
159	# only for superusers
160	if not request.user or not request.user.is_superuser:	1✔
161	return Response(status=status.HTTP_401_UNAUTHORIZED)	1✔
162
163	# collect list of all users that have door access
164	users_with_door_access = []	1✔
165	for ss in (	1✔
166	ServiceSubscription.objects.select_related("user")
167	.filter(service=config.DEFAULT_ACCOUNT_SERVICE)
168	.filter(state=ServiceSubscription.ACTIVE)
169	):
170	users_with_door_access.append(ss.user)	1✔
171
172	# and output it
173	outserializer = UserAccessSerializer(users_with_door_access, many=True)	1✔
174	return Response(outserializer.data)	1✔
175
176	def list(self, request):	1✔
177	return Response(status=status.HTTP_501_NOT_IMPLEMENTED)	1✔

TampereHacklab / mulysa / 13098608777

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