Block API endpoints access to users when their password has been invalidated by a superuser

kobo/apps/accounts/mfa/views.py

@@ -4,9 +4,9 @@
 from django.db.models import QuerySet
 from django.urls import reverse
 from rest_framework.generics import ListAPIView
-from rest_framework.permissions import IsAuthenticated
 from trench.utils import get_mfa_model
 
+from kpi.permissions import IsAuthenticated
 from .forms import MfaLoginForm, MfaTokenForm
 from .serializers import UserMfaMethodSerializer
 

kobo/apps/accounts/tests/test_email.py

@@ -1,4 +1,3 @@
-from allauth.account.models import EmailAddress
 from django.core import mail
 from django.urls import reverse
 from model_bakery import baker

kobo/apps/accounts/tests/test_social_account.py

@@ -13,7 +13,7 @@ def test_list(self):
         account1 = baker.make('socialaccount.SocialAccount', user=self.user)
         account2 = baker.make('socialaccount.SocialAccount')
         # Auth, Count, Queryset
-        with self.assertNumQueries(3):
+        with self.assertNumQueries(4):
             res = self.client.get(self.url_list)
         self.assertContains(res, account1.uid)
         self.assertNotContains(res, account2.uid)

kobo/apps/accounts/views.py

@@ -1,9 +1,9 @@
 from allauth.account.models import EmailAddress
 from allauth.socialaccount.models import SocialAccount
 from rest_framework import mixins, status, viewsets
-from rest_framework.permissions import IsAuthenticated
 from rest_framework.response import Response
 
+from kpi.permissions import IsAuthenticated
 from .mixins import MultipleFieldLookupMixin
 from .serializers import EmailAddressSerializer, SocialAccountSerializer
 

kobo/apps/audit_log/permissions.py

@@ -1,7 +1,10 @@
 from rest_framework.permissions import IsAdminUser
 
+from kpi.mixins.validation_password_permission import ValidationPasswordPermissionMixin
 
-class SuperUserPermission(IsAdminUser):
+
+class SuperUserPermission(ValidationPasswordPermissionMixin, IsAdminUser):
 
     def has_permission(self, request, view):
+        self.validate_password(request)
         return bool(request.user and request.user.is_superuser)

kobo/apps/help/permissions.py

@@ -1,9 +1,12 @@
 # coding: utf-8
 from rest_framework import exceptions, permissions
 
+from kpi.mixins.validation_password_permission import ValidationPasswordPermissionMixin
+
 
 class InAppMessagePermissions(permissions.BasePermission):
     def has_permission(self, request, view):
+
         if not request.user.is_authenticated:
             # Deny access to anonymous users
             return False

kobo/apps/languages/views/base.py

@@ -1,8 +1,8 @@
 # coding: utf-8
 from rest_framework import viewsets
-from rest_framework.permissions import IsAuthenticated
 
 from kpi.filters import SearchFilter
+from kpi.permissions import IsAuthenticated
 
 
 class BaseViewSet(viewsets.ReadOnlyModelViewSet):

kobo/apps/organizations/permissions.py

@@ -1,12 +1,20 @@
 from rest_framework import permissions
 
+from kpi.mixins.validation_password_permission import ValidationPasswordPermissionMixin
 
-class IsOrgAdminOrReadOnly(permissions.BasePermission):
+
+class IsOrgAdminOrReadOnly(
+    ValidationPasswordPermissionMixin, permissions.BasePermission
+):
     """
     Object-level permission to only allow admin members of an object to edit it.
     Assumes the model instance has an `is_admin` attribute.
     """
 
+    def has_permission(self, request, view):
+        self.validate_password(request)
+        return super().has_permission(request=request, view=view)
+
     def has_object_permission(self, request, view, obj):
         # Read permissions are allowed to any request,
         # so we'll always allow GET, HEAD or OPTIONS requests.

kobo/apps/organizations/tests/test_organizations_api.py

@@ -42,7 +42,7 @@ def test_list(self):
         self._insert_data()
         organization2 = baker.make(Organization, id='org_abcd123')
         organization2.add_user(user=self.user, is_admin=True)
-        with self.assertNumQueries(2):
+        with self.assertNumQueries(3):
             res = self.client.get(self.url_list)
         self.assertContains(res, organization2.name)
 
@@ -61,7 +61,7 @@ def test_api_returns_org_data(self):
     def test_update(self):
         self._insert_data()
         data = {'name': 'edit'}
-        with self.assertNumQueries(4):
+        with self.assertNumQueries(5):
             res = self.client.patch(self.url_detail, data)
         self.assertContains(res, data['name'])
 

kobo/apps/organizations/views.py

@@ -1,8 +1,7 @@
-from django.contrib.auth.models import User
 from django.db.models import QuerySet
 from rest_framework import viewsets
-from rest_framework.permissions import IsAuthenticated
 
+from kpi.permissions import IsAuthenticated
 from .models import Organization, create_organization
 from .permissions import IsOrgAdminOrReadOnly
 from .serializers import OrganizationSerializer

kobo/apps/project_views/views.py

@@ -6,7 +6,6 @@
 from django.db.models.query import QuerySet
 from django.http import Http404
 from rest_framework import viewsets
-from rest_framework.permissions import IsAuthenticated
 from rest_framework.decorators import action
 from rest_framework.response import Response
 
@@ -17,6 +16,7 @@
 )
 from kpi.mixins.object_permission import ObjectPermissionViewSetMixin
 from kpi.models import Asset, ProjectViewExportTask
+from kpi.permissions import IsAuthenticated
 from kpi.serializers.v2.asset import AssetMetadataListSerializer
 from kpi.serializers.v2.user import UserListSerializer
 from kpi.utils.object_permission import get_database_user

kobo/apps/stripe/views.py

@@ -14,7 +14,6 @@
 from djstripe.settings import djstripe_settings
 from organizations.utils import create_organization
 from rest_framework import mixins, status, viewsets
-from rest_framework.permissions import IsAuthenticated
 from rest_framework.response import Response
 from rest_framework.views import APIView
 
@@ -27,6 +26,7 @@
     ProductSerializer,
     SubscriptionSerializer,
 )
+from kpi.permissions import IsAuthenticated
 
 
 # Lists the one-time purchases made by the organization that the logged-in user owns

kobo/apps/subsequences/tests/test_submission_extras_api_post.py

@@ -302,10 +302,10 @@ def test_google_transcript_post(self, m1, m2):
             'submission': submission_id,
             'q1': {GOOGLETS: {'status': 'requested', 'languageCode': ''}}
         }
-        with self.assertNumQueries(52):
+        with self.assertNumQueries(53):
             res = self.client.post(url, data, format='json')
         self.assertContains(res, 'complete')
-        with self.assertNumQueries(21):
+        with self.assertNumQueries(22):
             self.client.post(url, data, format='json')
 
     @override_settings(CACHES={'default': {'BACKEND': 'django.core.cache.backends.locmem.LocMemCache'}})

kpi/exceptions.py

@@ -1,6 +1,7 @@
 # coding: utf-8
+from django.conf import settings
 from django.utils.translation import gettext_lazy as t
-from rest_framework import exceptions
+from rest_framework import exceptions, status
 
 
 class AbstractMethodError(NotImplementedError):
@@ -67,8 +68,18 @@ class ImportAssetException(Exception):
     pass
 
 
+class InvalidPasswordAPIException(exceptions.APIException):
+    status_code = status.HTTP_403_FORBIDDEN
+    default_detail = t(
+        'Your access is restricted. Please reclaim your access by '
+        'changing your password at '
+        '##koboform_url##/accounts/password/reset/.'
+    ).replace('##koboform_url##', settings.KOBOFORM_URL)
+    default_code = 'invalid_password'
+
+
 class InvalidSearchException(exceptions.APIException):
-    status_code = 400
+    status_code = status.HTTP_400_BAD_REQUEST
     default_detail = t('Invalid search. Please try again')
     default_code = 'invalid_search'
 
@@ -96,7 +107,7 @@ class KobocatBulkUpdateSubmissionsClientException(exceptions.ValidationError):
 
 
 class KobocatBulkUpdateSubmissionsException(exceptions.APIException):
-    status_code = 500
+    status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
     default_detail = t(
         'An error occurred trying to bulk update the submissions.')
     default_code = 'bulk_update_submissions_error'
@@ -121,7 +132,7 @@ def invalid_form_id(self):
 
 
 class KobocatDuplicateSubmissionException(exceptions.APIException):
-    status_code = 500
+    status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
     default_detail = t('An error occurred trying to duplicate the submission')
     default_code = 'submission_duplication_error'
 
@@ -135,7 +146,7 @@ class NotSupportedFormatException(Exception):
 
 
 class ObjectDeploymentDoesNotExist(exceptions.APIException):
-    status_code = 400
+    status_code = status.HTTP_400_BAD_REQUEST
     default_detail = t('The specified object has not been deployed')
     default_code = 'deployment_does_not_exist'
 

kpi/mixins/validation_password_permission.py

@@ -0,0 +1,21 @@
+from ..exceptions import InvalidPasswordAPIException
+
+
+class ValidationPasswordPermissionMixin:
+
+    def validate_password(self, request):
+
+        if request.user.is_anonymous:
+            return
+
+        try:
+            extra_details = request.user.extra_details
+        except request.user.extra_details.RelatedObjectDoesNotExist:
+            # if user has not extra details, admin has not been able to set
+            # `validated_password`. Let's consider it as True.
+            return
+
+        if extra_details.validated_password:
+            return
+
+        raise InvalidPasswordAPIException

kpi/permissions.py

@@ -6,6 +6,7 @@
 from django.contrib.auth.models import User
 from django.http import Http404
 from rest_framework import exceptions, permissions
+from rest_framework.permissions import IsAuthenticated as DRFIsAuthenticated
 
 from kpi.constants import (
     PERM_ADD_SUBMISSIONS,
@@ -14,6 +15,7 @@
     PERM_VIEW_ASSET,
     PERM_VIEW_SUBMISSIONS,
 )
+from kpi.mixins.validation_password_permission import ValidationPasswordPermissionMixin
 from kpi.models.asset import Asset
 from kpi.utils.object_permission import get_database_user
 from kpi.utils.project_views import (
@@ -119,7 +121,9 @@ def has_object_permission(self, request, view, obj):
         return True
 
 
-class AssetPermission(permissions.DjangoObjectPermissions):
+class AssetPermission(
+    ValidationPasswordPermissionMixin, permissions.DjangoObjectPermissions
+):
 
     # Setting this to False allows real permission checking on AnonymousUser.
     # With the default of True, anonymous requests are categorically rejected.
@@ -130,6 +134,10 @@ class AssetPermission(permissions.DjangoObjectPermissions):
     perms_map['OPTIONS'] = perms_map['GET']
     perms_map['HEAD'] = perms_map['GET']
 
+    def has_permission(self, request, view):
+        self.validate_password(request)
+        return super().has_permission(request=request, view=view)
+
     def has_object_permission(self, request, view, obj):
 
         user = get_database_user(request.user)
@@ -151,7 +159,9 @@ def has_object_permission(self, request, view, obj):
         return super().has_object_permission(request, view, obj)
 
 
-class AssetNestedObjectPermission(BaseAssetNestedObjectPermission):
+class AssetNestedObjectPermission(
+    ValidationPasswordPermissionMixin, BaseAssetNestedObjectPermission
+):
     """
     Permissions for nested objects of Asset.
     Only owner and managers can have write access on these objects.
@@ -172,6 +182,7 @@ class AssetNestedObjectPermission(BaseAssetNestedObjectPermission):
     perms_map['DELETE'] = perms_map['POST']
 
     def has_permission(self, request, view):
+        self.validate_password(request)
         if not request.user:
             return False
         elif request.user.is_superuser:
@@ -274,6 +285,13 @@ class AssetVersionReadOnlyPermission(AssetNestedObjectPermission):
     }
 
 
+class IsAuthenticated(ValidationPasswordPermissionMixin, DRFIsAuthenticated):
+
+    def has_permission(self, request, view):
+        self.validate_password(request)
+        return super().has_permission(request=request, view=view)
+
+
 # FIXME: Name is no longer accurate.
 class IsOwnerOrReadOnly(AssetPermission):
     """

kpi/serializers/current_user.py

@@ -91,7 +91,7 @@ def get_validated_password(self, obj):
             extra_details = obj.extra_details
         except obj.extra_details.RelatedObjectDoesNotExist:
             # if user has not extra details, admin has not been able to set
-            # `validated_password`. Let's consider it True.
+            # `validated_password`. Let's consider it as True.
             return True
 
         return extra_details.validated_password

kpi/tests/api/v2/test_api_imports.py

@@ -18,6 +18,7 @@
 from kpi.urls.router_api_v2 import URL_NAMESPACE as ROUTER_URL_NAMESPACE
 from kpi.utils.strings import to_str
 
+
 class AssetImportTaskTest(BaseTestCase):
     fixtures = ['test_data']