24964072477

Committed 26 Apr 2026 12:25PM UTC coverage: 86.106% (+0.4%) from 85.665%

Build # 24964072477

Build Type

push

github

Committed by

rolandgeider

Commit Message

Restrict trainer login access

Coverage Stats

17 of 18 new or added lines in 2 files covered. (94.44%)

91 existing lines in 9 files now uncovered.

17495 of 20318 relevant lines covered (86.11%)

0.86 hits per line

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

81.88

/wger/core/api/views.py

# -*- coding: utf-8 -*-

# This file is part of wger Workout Manager.
#
# wger Workout Manager is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# wger Workout Manager is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with Workout Manager.  If not, see <http://www.gnu.org/licenses/>.

# Standard Library
import logging

# Django
from django.conf import settings
from django.contrib.auth.models import User
from django.utils.decorators import method_decorator
from django.views.decorators.cache import cache_page

# Third Party
from allauth.account.models import EmailAddress
from drf_spectacular.types import OpenApiTypes
from drf_spectacular.utils import (
    OpenApiParameter,
    OpenApiResponse,
    extend_schema,
    inline_serializer,
)
from rest_framework import (
    status,
    viewsets,
)
from rest_framework.decorators import (
    action,
    api_view,
)
from rest_framework.fields import (
    BooleanField,
    CharField,
)
from rest_framework.permissions import (
    AllowAny,
    IsAuthenticated,
)
from rest_framework.response import Response

# wger
from wger.core.api.serializers import (
    LanguageCheckSerializer,
    LanguageSerializer,
    LicenseSerializer,
    RepetitionUnitSerializer,
    RoutineWeightUnitSerializer,
    UserLoginSerializer,
    UserprofileSerializer,
    UserRegistrationSerializer,
)
from wger.core.forms import UserLoginForm
from wger.core.models import (
    Language,
    License,
    RepetitionUnit,
    UserProfile,
    WeightUnit,
)
from wger.utils.api_token import create_token
from wger.utils.permissions import WgerPermission
from wger.version import (
    MIN_APP_VERSION,
    MIN_SERVER_VERSION,
    get_version,
)


logger = logging.getLogger(__name__)


class UserProfileViewSet(viewsets.ModelViewSet):
    """
    API endpoint for the user profile

    This endpoint works somewhat differently than the others since it always
    returns the data for the currently logged-in user's profile. To update
    the profile, use a POST request with the new data, not a PATCH.
    """

    serializer_class = UserprofileSerializer
    permission_classes = (
        IsAuthenticated,
        WgerPermission,
    )

    def get_queryset(self):
        """
        Only allow access to appropriate objects
        """
        # REST API generation
        if getattr(self, 'swagger_fake_view', False):
            return UserProfile.objects.none()

        return UserProfile.objects.filter(user=self.request.user)

    def get_owner_objects(self):
        """
        Return objects to check for ownership permission
        """
        return [(User, 'user')]

    def list(self, request, *args, **kwargs):
        """
        Customized list view, that returns only the current user's data
        """
        queryset = self.get_queryset()
        serializer = self.serializer_class(queryset.first(), many=False)

        return Response(serializer.data)

    def retrieve(self, request, *args, **kwargs):
        return self.list(request, *args, **kwargs)

    def create(self, request, *args, **kwargs):
        serializer = self.get_serializer(request.user.userprofile, data=request.data)
        serializer.is_valid(raise_exception=True)
        serializer.save()
        return Response(serializer.data)

    def update(self, request, *args, **kwargs):
        return self.create(request, *args, **kwargs)

    def partial_update(self, request, *args, **kwargs):
        return self.create(request, *args, **kwargs)

    def destroy(self, request, *args, **kwargs):
        return Response(status=status.HTTP_405_METHOD_NOT_ALLOWED)

    @action(detail=False, url_name='verify-email', url_path='verify-email')
    def verify_email(self, request):
        """
        Verify the user's email address
        """
        email_obj = request.user.userprofile.get_allauth_email

        if email_obj is None:
            return Response({'result': 'not sent', 'message': 'The user has no associated email'})

        if email_obj.verified:
            return Response({'status': 'verified', 'message': 'This email is already verified'})

        email_obj.send_confirmation(request)
        return Response(
            {'status': 'sent', 'message': f'A verification email was sent to {request.user.email}'}
        )


class ApplicationVersionView(viewsets.ViewSet):
    """
    Returns the application's version
    """

    permission_classes = (AllowAny,)

    @staticmethod
    @extend_schema(
        parameters=[],
        responses={
            200: OpenApiTypes.STR,
        },
    )
    def get(request):
        return Response(get_version())


class PermissionView(viewsets.ViewSet):
    """
    Checks whether the user has a django permission
    """

    permission_classes = (AllowAny,)

    @staticmethod
    @extend_schema(
        parameters=[
            OpenApiParameter(
                'permission',
                OpenApiTypes.STR,
                OpenApiParameter.QUERY,
                description='The name of the django permission such as "exercises.change_muscle"',
            ),
        ],
        responses={
            201: inline_serializer(
                name='PermissionResponse',
                fields={
                    'result': BooleanField(),
                },
            ),
            400: OpenApiResponse(
                description="Please pass a permission name in the 'permission' parameter"
            ),
        },
    )
    def get(request):
        permission = request.query_params.get('permission')

        if permission is None:
            return Response(
                "Please pass a permission name in the 'permission' parameter",
                status=status.HTTP_400_BAD_REQUEST,
            )

        if request.user.is_anonymous:
            return Response({'result': False})

        return Response({'result': request.user.has_perm(permission)})


class RequiredApplicationVersionView(viewsets.ViewSet):
    """
    Returns the minimum required version of flutter app to access this server.
    """

    permission_classes = (AllowAny,)

    @staticmethod
    @extend_schema(
        parameters=[],
        responses={
            200: OpenApiTypes.STR,
        },
    )
    def get(request):
        return Response(str(MIN_APP_VERSION))


class RequiredServerVersionView(viewsets.ViewSet):
    """
    Returns the minimum required version of the server to perform sync requests
    """

    permission_classes = (AllowAny,)

    @staticmethod
    @extend_schema(
        parameters=[],
        responses={
            200: OpenApiTypes.STR,
        },
    )
    def get(request):
        return Response(str(MIN_SERVER_VERSION))


class UserAPILoginView(viewsets.ViewSet):
    """
    API login endpoint. Returns a token that can subsequently passed in the
    header.

    Note that it is recommended to use token authorization instead.
    """

    permission_classes = (AllowAny,)
    queryset = User.objects.all()
    serializer_class = UserLoginSerializer
    throttle_scope = 'login'

    def get(self, request):
        return Response(data={'message': "You must send a 'username' and 'password' via POST"})

    @extend_schema(
        parameters=[],
        responses={
            status.HTTP_200_OK: inline_serializer(
                name='loginSerializer',
                fields={'token': CharField()},
            ),
        },
    )
    def post(self, request):
        serializer = self.serializer_class(data=request.data, request=request)
        serializer.is_valid(raise_exception=True)

        # This is a bit hacky, but saving the email or username as the username
        # allows us to simply use the helpers.EmailAuthBackend backend which also
        # uses emails
        username = serializer.data.get('username', serializer.data.get('email', None))
        data = {'username': username, 'password': serializer.data['password']}
        form = UserLoginForm(data=data, authenticate_on_clean=False)

        if not form.is_valid():
            logger.info(f"Tried logging via API with unknown user : '{username}'")
            return Response(
                {'detail': 'Username or password unknown'},
                status=status.HTTP_401_UNAUTHORIZED,
            )

        form.authenticate(request)
        token = create_token(form.get_user())
        return Response(
            data={'token': token.key},
            status=status.HTTP_200_OK,
            headers={
                'Deprecation': 'Sat, 01 Oct 2022 23:59:59 GMT',
            },
        )


class UserAPIRegistrationViewSet(viewsets.ViewSet):
    """
    API endpoint
    """

    # permission_classes = (AllowRegisterUser,)
    serializer_class = UserRegistrationSerializer
    throttle_scope = 'registration'

    def get_queryset(self):
        """
        Only allow access to appropriate objects
        """
        return UserProfile.objects.filter(user=self.request.user)

    @extend_schema(
        parameters=[],
        responses={
            status.HTTP_200_OK: inline_serializer(
                name='loginSerializer',
                fields={'token': CharField()},
            ),
        },
    )
    def post(self, request):
        if not settings.WGER_SETTINGS['ALLOW_REGISTRATION']:
            return Response(
                {'message': 'Registration is not allowed on this instance'},
                status=status.HTTP_200_OK,
            )

        data = request.data
        serializer = self.serializer_class(data=data)
        serializer.is_valid(raise_exception=True)
        user = serializer.save()
        # user.userprofile.added_by = request.user
        user.userprofile.save()
        token = create_token(user)

        EmailAddress.objects.add_email(request, user, user.email, confirm=True)

        return Response(
            {'message': 'api user successfully registered', 'token': token.key},
            status=status.HTTP_201_CREATED,
        )


class LanguageViewSet(viewsets.ReadOnlyModelViewSet):
    """
    API endpoint for the languages used in the application
    """

    queryset = Language.objects.all()
    serializer_class = LanguageSerializer
    ordering_fields = '__all__'
    filterset_fields = ('full_name', 'short_name')

    @method_decorator(cache_page(settings.WGER_SETTINGS['EXERCISE_CACHE_TTL']))
    def dispatch(self, request, *args, **kwargs):
        return super().dispatch(request, *args, **kwargs)


class LicenseViewSet(viewsets.ReadOnlyModelViewSet):
    """
    API endpoint for license objects
    """

    queryset = License.objects.all()
    serializer_class = LicenseSerializer
    ordering_fields = '__all__'
    filterset_fields = (
        'full_name',
        'short_name',
        'url',
    )


class RepetitionUnitViewSet(viewsets.ReadOnlyModelViewSet):
    """
    API endpoint for repetition units objects
    """

    queryset = RepetitionUnit.objects.all()
    serializer_class = RepetitionUnitSerializer
    ordering_fields = '__all__'
    filterset_fields = ('name',)


class RoutineWeightUnitViewSet(viewsets.ReadOnlyModelViewSet):
    """
    API endpoint for weight units objects
    """

    queryset = WeightUnit.objects.all()
    serializer_class = RoutineWeightUnitSerializer
    ordering_fields = '__all__'
    filterset_fields = ('name',)


@api_view(['POST'])
def check_language(request):
    """
    Checks the language of a string
    """
    serializer = LanguageCheckSerializer(data=request.data)
    serializer.is_valid(raise_exception=True)

    return Response({'result': True})

1	# -- coding: utf-8 --
2
3	# This file is part of wger Workout Manager.
4	#
5	# wger Workout Manager is free software: you can redistribute it and/or modify
6	# it under the terms of the GNU Affero General Public License as published by
7	# the Free Software Foundation, either version 3 of the License, or
8	# (at your option) any later version.
9	#
10	# wger Workout Manager is distributed in the hope that it will be useful,
11	# but WITHOUT ANY WARRANTY; without even the implied warranty of
12	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13	# GNU General Public License for more details.
14	#
15	# You should have received a copy of the GNU Affero General Public License
16	# along with Workout Manager. If not, see <http://www.gnu.org/licenses/>.
17
18	# Standard Library
19	import logging	1✔
20
21	# Django
22	from django.conf import settings	1✔
23	from django.contrib.auth.models import User	1✔
24	from django.utils.decorators import method_decorator	1✔
25	from django.views.decorators.cache import cache_page	1✔
26
27	# Third Party
28	from allauth.account.models import EmailAddress	1✔
29	from drf_spectacular.types import OpenApiTypes	1✔
30	from drf_spectacular.utils import (	1✔
31	OpenApiParameter,
32	OpenApiResponse,
33	extend_schema,
34	inline_serializer,
35	)
36	from rest_framework import (	1✔
37	status,
38	viewsets,
39	)
40	from rest_framework.decorators import (	1✔
41	action,
42	api_view,
43	)
44	from rest_framework.fields import (	1✔
45	BooleanField,
46	CharField,
47	)
48	from rest_framework.permissions import (	1✔
49	AllowAny,
50	IsAuthenticated,
51	)
52	from rest_framework.response import Response	1✔
53
54	# wger
55	from wger.core.api.serializers import (	1✔
56	LanguageCheckSerializer,
57	LanguageSerializer,
58	LicenseSerializer,
59	RepetitionUnitSerializer,
60	RoutineWeightUnitSerializer,
61	UserLoginSerializer,
62	UserprofileSerializer,
63	UserRegistrationSerializer,
64	)
65	from wger.core.forms import UserLoginForm	1✔
66	from wger.core.models import (	1✔
67	Language,
68	License,
69	RepetitionUnit,
70	UserProfile,
71	WeightUnit,
72	)
73	from wger.utils.api_token import create_token	1✔
74	from wger.utils.permissions import WgerPermission	1✔
75	from wger.version import (	1✔
76	MIN_APP_VERSION,
77	MIN_SERVER_VERSION,
78	get_version,
79	)
80
81
82	logger = logging.getLogger(__name__)	1✔
83
84
85	class UserProfileViewSet(viewsets.ModelViewSet):	1✔
86	"""
87	API endpoint for the user profile
88
89	This endpoint works somewhat differently than the others since it always
90	returns the data for the currently logged-in user's profile. To update
91	the profile, use a POST request with the new data, not a PATCH.
92	"""
93
94	serializer_class = UserprofileSerializer	1✔
95	permission_classes = (	1✔
96	IsAuthenticated,
97	WgerPermission,
98	)
99
100	def get_queryset(self):	1✔
101	"""
102	Only allow access to appropriate objects
103	"""
104	# REST API generation
105	if getattr(self, 'swagger_fake_view', False):	×
106	return UserProfile.objects.none()	×
107
108	return UserProfile.objects.filter(user=self.request.user)	×
109
110	def get_owner_objects(self):	1✔
111	"""
112	Return objects to check for ownership permission
113	"""
114	return [(User, 'user')]	×
115
116	def list(self, request, args, *kwargs):	1✔
117	"""
118	Customized list view, that returns only the current user's data
119	"""
120	queryset = self.get_queryset()	×
121	serializer = self.serializer_class(queryset.first(), many=False)	×
122
123	return Response(serializer.data)	×
124
125	def retrieve(self, request, args, *kwargs):	1✔
126	return self.list(request, args, *kwargs)	×
127
128	def create(self, request, args, *kwargs):	1✔
129	serializer = self.get_serializer(request.user.userprofile, data=request.data)	1✔
130	serializer.is_valid(raise_exception=True)	1✔
131	serializer.save()	1✔
132	return Response(serializer.data)	1✔
133
134	def update(self, request, args, *kwargs):	1✔
135	return self.create(request, args, *kwargs)	×
136
137	def partial_update(self, request, args, *kwargs):	1✔
138	return self.create(request, args, *kwargs)	×
139
140	def destroy(self, request, args, *kwargs):	1✔
UNCOV 141	return Response(status=status.HTTP_405_METHOD_NOT_ALLOWED)	×
142
143	@action(detail=False, url_name='verify-email', url_path='verify-email')	1✔
144	def verify_email(self, request):	1✔
145	"""
146	Verify the user's email address
147	"""
148	email_obj = request.user.userprofile.get_allauth_email	1✔
149
150	if email_obj is None:	1✔
151	return Response({'result': 'not sent', 'message': 'The user has no associated email'})	×
152
153	if email_obj.verified:	1✔
154	return Response({'status': 'verified', 'message': 'This email is already verified'})	1✔
155
156	email_obj.send_confirmation(request)	1✔
157	return Response(	1✔
158	{'status': 'sent', 'message': f'A verification email was sent to {request.user.email}'}
159	)
160
161
162	class ApplicationVersionView(viewsets.ViewSet):	1✔
163	"""
164	Returns the application's version
165	"""
166
167	permission_classes = (AllowAny,)	1✔
168
169	@staticmethod	1✔
170	@extend_schema(	1✔
171	parameters=[],
172	responses={
173	200: OpenApiTypes.STR,
174	},
175	)
176	def get(request):	1✔
UNCOV 177	return Response(get_version())	×
178
179
180	class PermissionView(viewsets.ViewSet):	1✔
181	"""
182	Checks whether the user has a django permission
183	"""
184
185	permission_classes = (AllowAny,)	1✔
186
187	@staticmethod	1✔
188	@extend_schema(	1✔
189	parameters=[
190	OpenApiParameter(
191	'permission',
192	OpenApiTypes.STR,
193	OpenApiParameter.QUERY,
194	description='The name of the django permission such as "exercises.change_muscle"',
195	),
196	],
197	responses={
198	201: inline_serializer(
199	name='PermissionResponse',
200	fields={
201	'result': BooleanField(),
202	},
203	),
204	400: OpenApiResponse(
205	description="Please pass a permission name in the 'permission' parameter"
206	),
207	},
208	)
209	def get(request):	1✔
UNCOV 210	permission = request.query_params.get('permission')	×
211
UNCOV 212	if permission is None:	×
UNCOV 213	return Response(	×
214	"Please pass a permission name in the 'permission' parameter",
215	status=status.HTTP_400_BAD_REQUEST,
216	)
217
UNCOV 218	if request.user.is_anonymous:	×
UNCOV 219	return Response({'result': False})	×
220
UNCOV 221	return Response({'result': request.user.has_perm(permission)})	×
222
223
224	class RequiredApplicationVersionView(viewsets.ViewSet):	1✔
225	"""
226	Returns the minimum required version of flutter app to access this server.
227	"""
228
229	permission_classes = (AllowAny,)	1✔
230
231	@staticmethod	1✔
232	@extend_schema(	1✔
233	parameters=[],
234	responses={
235	200: OpenApiTypes.STR,
236	},
237	)
238	def get(request):	1✔
UNCOV 239	return Response(str(MIN_APP_VERSION))	×
240
241
242	class RequiredServerVersionView(viewsets.ViewSet):	1✔
243	"""
244	Returns the minimum required version of the server to perform sync requests
245	"""
246
247	permission_classes = (AllowAny,)	1✔
248
249	@staticmethod	1✔
250	@extend_schema(	1✔
251	parameters=[],
252	responses={
253	200: OpenApiTypes.STR,
254	},
255	)
256	def get(request):	1✔
UNCOV 257	return Response(str(MIN_SERVER_VERSION))	×
258
259
260	class UserAPILoginView(viewsets.ViewSet):	1✔
261	"""
262	API login endpoint. Returns a token that can subsequently passed in the
263	header.
264
265	Note that it is recommended to use token authorization instead.
266	"""
267
268	permission_classes = (AllowAny,)	1✔
269	queryset = User.objects.all()	1✔
270	serializer_class = UserLoginSerializer	1✔
271	throttle_scope = 'login'	1✔
272
273	def get(self, request):	1✔
274	return Response(data={'message': "You must send a 'username' and 'password' via POST"})	1✔
275
276	@extend_schema(	1✔
277	parameters=[],
278	responses={
279	status.HTTP_200_OK: inline_serializer(
280	name='loginSerializer',
281	fields={'token': CharField()},
282	),
283	},
284	)
285	def post(self, request):	1✔
286	serializer = self.serializer_class(data=request.data, request=request)	1✔
287	serializer.is_valid(raise_exception=True)	1✔
288
289	# This is a bit hacky, but saving the email or username as the username
290	# allows us to simply use the helpers.EmailAuthBackend backend which also
291	# uses emails
292	username = serializer.data.get('username', serializer.data.get('email', None))	1✔
293	data = {'username': username, 'password': serializer.data['password']}	1✔
294	form = UserLoginForm(data=data, authenticate_on_clean=False)	1✔
295
296	if not form.is_valid():	1✔
UNCOV 297	logger.info(f"Tried logging via API with unknown user : '{username}'")	×
UNCOV 298	return Response(	×
299	{'detail': 'Username or password unknown'},
300	status=status.HTTP_401_UNAUTHORIZED,
301	)
302
303	form.authenticate(request)	1✔
304	token = create_token(form.get_user())	1✔
305	return Response(	1✔
306	data={'token': token.key},
307	status=status.HTTP_200_OK,
308	headers={
309	'Deprecation': 'Sat, 01 Oct 2022 23:59:59 GMT',
310	},
311	)
312
313
314	class UserAPIRegistrationViewSet(viewsets.ViewSet):	1✔
315	"""
316	API endpoint
317	"""
318
319	# permission_classes = (AllowRegisterUser,)
320	serializer_class = UserRegistrationSerializer	1✔
321	throttle_scope = 'registration'	1✔
322
323	def get_queryset(self):	1✔
324	"""
325	Only allow access to appropriate objects
326	"""
UNCOV 327	return UserProfile.objects.filter(user=self.request.user)	×
328
329	@extend_schema(	1✔
330	parameters=[],
331	responses={
332	status.HTTP_200_OK: inline_serializer(
333	name='loginSerializer',
334	fields={'token': CharField()},
335	),
336	},
337	)
338	def post(self, request):	1✔
339	if not settings.WGER_SETTINGS['ALLOW_REGISTRATION']:	1✔
340	return Response(	1✔
341	{'message': 'Registration is not allowed on this instance'},
342	status=status.HTTP_200_OK,
343	)
344
345	data = request.data	1✔
346	serializer = self.serializer_class(data=data)	1✔
347	serializer.is_valid(raise_exception=True)	1✔
348	user = serializer.save()	1✔
349	# user.userprofile.added_by = request.user
350	user.userprofile.save()	1✔
351	token = create_token(user)	1✔
352
353	EmailAddress.objects.add_email(request, user, user.email, confirm=True)	1✔
354
355	return Response(	1✔
356	{'message': 'api user successfully registered', 'token': token.key},
357	status=status.HTTP_201_CREATED,
358	)
359
360
361	class LanguageViewSet(viewsets.ReadOnlyModelViewSet):	1✔
362	"""
363	API endpoint for the languages used in the application
364	"""
365
366	queryset = Language.objects.all()	1✔
367	serializer_class = LanguageSerializer	1✔
368	ordering_fields = '__all__'	1✔
369	filterset_fields = ('full_name', 'short_name')	1✔
370
371	@method_decorator(cache_page(settings.WGER_SETTINGS['EXERCISE_CACHE_TTL']))	1✔
372	def dispatch(self, request, args, *kwargs):	1✔
373	return super().dispatch(request, args, *kwargs)	1✔
374
375
376	class LicenseViewSet(viewsets.ReadOnlyModelViewSet):	1✔
377	"""
378	API endpoint for license objects
379	"""
380
381	queryset = License.objects.all()	1✔
382	serializer_class = LicenseSerializer	1✔
383	ordering_fields = '__all__'	1✔
384	filterset_fields = (	1✔
385	'full_name',
386	'short_name',
387	'url',
388	)
389
390
391	class RepetitionUnitViewSet(viewsets.ReadOnlyModelViewSet):	1✔
392	"""
393	API endpoint for repetition units objects
394	"""
395
396	queryset = RepetitionUnit.objects.all()	1✔
397	serializer_class = RepetitionUnitSerializer	1✔
398	ordering_fields = '__all__'	1✔
399	filterset_fields = ('name',)	1✔
400
401
402	class RoutineWeightUnitViewSet(viewsets.ReadOnlyModelViewSet):	1✔
403	"""
404	API endpoint for weight units objects
405	"""
406
407	queryset = WeightUnit.objects.all()	1✔
408	serializer_class = RoutineWeightUnitSerializer	1✔
409	ordering_fields = '__all__'	1✔
410	filterset_fields = ('name',)	1✔
411
412
413	@api_view(['POST'])	1✔
414	def check_language(request):	1✔
415	"""
416	Checks the language of a string
417	"""
UNCOV 418	serializer = LanguageCheckSerializer(data=request.data)	×
UNCOV 419	serializer.is_valid(raise_exception=True)	×
420
UNCOV 421	return Response({'result': True})	×

wger-project / wger / 24964072477

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