fix: otp flow

Pipfile

@@ -23,7 +23,7 @@ name = "pypi"
 # 5. Run `pipenv install --dev` in your terminal.
 
 [packages]
-codeforlife = "==0.24.14"
+codeforlife = {ref = "portal-backend-376", git = "https://github.com/ocadotechnology/codeforlife-package-python.git"}
 # 🚫 Don't add [packages] below that are inherited from the CFL package.
 pyjwt = "==2.6.0" # TODO: upgrade to latest version
 # TODO: Needed by RR. Remove when RR has moved to new system.
@@ -32,7 +32,7 @@ django-sekizai = "==4.1.0"
 django-classy-tags = "==4.1.0"
 
 [dev-packages]
-codeforlife = {version = "==0.24.14", extras = ["dev"]}
+codeforlife = {ref = "portal-backend-376", git = "https://github.com/ocadotechnology/codeforlife-package-python.git", extras = ["dev"]}
 # codeforlife = {file = "../codeforlife-package-python", editable = true, extras = ["dev"]}
 # 🚫 Don't add [dev-packages] below that are inherited from the CFL package.
 

src/api/serializers/auth_factor.py

@@ -3,19 +3,23 @@
 Created on 23/01/2024 at 11:05:37(+00:00).
 """
 
-import pyotp
+import re
+
 from codeforlife.serializers import ModelSerializer
 from codeforlife.user.models import AuthFactor, User
 from rest_framework import serializers
 
 
 # pylint: disable-next=missing-class-docstring,too-many-ancestors
 class AuthFactorSerializer(ModelSerializer[User, AuthFactor]):
+    otp = serializers.CharField(required=False, write_only=True)
+
     class Meta:
         model = AuthFactor
         fields = [
             "id",
             "type",
+            "otp",
         ]
         extra_kwargs = {
             "id": {"read_only": True},
@@ -33,24 +37,20 @@ def validate_type(self, value: str):
 
         return value
 
-    def create(self, validated_data):
-        user = self.request.auth_user
-        if not user.userprofile.otp_secret:
-            user.userprofile.otp_secret = pyotp.random_base32()
-            user.userprofile.save()
-
-        validated_data["user"] = user
-        return super().create(validated_data)
+    # pylint: disable-next=missing-function-docstring
+    def validate_otp(self, value: str):
+        if not re.match(r"^[0-9]{6}$", value):
+            raise serializers.ValidationError("Must be 6 digits", code="format")
+        if not self.request.auth_user.totp.verify(value):
+            raise serializers.ValidationError("Invalid OTP.", code="invalid")
 
-    def to_representation(self, instance):
-        representation = super().to_representation(instance)
+        return value
 
-        if (
-            self.view.action == "create"
-            and instance.type == AuthFactor.Type.OTP
-        ):
-            representation["totp_provisioning_uri"] = (
-                self.request.auth_user.totp_provisioning_uri
+    def validate(self, attrs):
+        if attrs["type"] == "otp" and "otp" not in attrs:
+            raise serializers.ValidationError(
+                "Current OTP required to enable OTP.",
+                code="otp__required",
             )
 
-        return representation
+        return attrs

src/api/serializers/auth_factor_test.py

@@ -3,10 +3,11 @@
 Created on 15/02/2024 at 15:44:25(+00:00).
 """
 
+from unittest.mock import Mock, patch
+
 from codeforlife.tests import ModelSerializerTestCase
 from codeforlife.user.models import AuthFactor, TeacherUser, User
 
-from ..views import AuthFactorViewSet
 from .auth_factor import AuthFactorSerializer
 
 # pylint: disable=missing-class-docstring
@@ -46,42 +47,41 @@ def test_validate_type__already_exists(self):
             },
         )
 
-    # test: create
-
-    def test_create(self):
-        """Enabling OTP will ensure the user's OTP-secret is set."""
-        user = self.uni_auth_factor_teacher_user
-        assert user.otp_secret
-
-        self.assert_create(
-            validated_data={"type": AuthFactor.Type.OTP},
-            context={"request": self.request_factory.post(user=user)},
-            new_data={
-                "user": {
-                    "id": user.id,
-                    "userprofile": {"otp_secret": user.otp_secret},
-                },
-            },
+    def test_validate_otp__format(self):
+        """OTP must be 6 digits."""
+        self.assert_validate_field(
+            name="otp",
+            value="12345",
+            error_code="format",
         )
 
-    # test: to representation
+    @patch("codeforlife.user.models.user.TOTP.verify", return_value=False)
+    def test_validate_otp__invalid(self, totp__verify: Mock):
+        """Cannot enable the OTP without providing the current OTP."""
+        user = TeacherUser.objects.filter(
+            # TODO: make otp_secret non-nullable
+            userprofile__otp_secret__isnull=False
+        ).first()
+        assert user
 
-    def test_to_representation(self):
-        """User's TOTP provisioning URI is returned when enabling OTP."""
-        assert self.multi_auth_factor_teacher_user.otp_secret
+        value = "123456"
 
-        self.assert_to_representation(
-            instance=AuthFactor(type=AuthFactor.Type.OTP),
-            new_data={
-                "totp_provisioning_uri": (
-                    self.multi_auth_factor_teacher_user.totp_provisioning_uri
-                ),
-            },
+        self.assert_validate_field(
+            name="otp",
+            value=value,
+            error_code="invalid",
             context={
-                "view": AuthFactorViewSet(action="create"),
                 "request": self.request_factory.post(
                     user=self.multi_auth_factor_teacher_user
-                ),
+                )
             },
-            non_model_fields={"totp_provisioning_uri"},
+        )
+
+        totp__verify.assert_called_once_with(value)
+
+    def test_validate__otp__required(self):
+        """Current OTP is required when enabling OTP."""
+        self.assert_validate(
+            attrs={"type": AuthFactor.Type.OTP.value},
+            error_code="otp__required",
         )

src/api/views/auth_factor.py

@@ -3,10 +3,13 @@
 Created on 23/01/2024 at 11:04:44(+00:00).
 """
 
+import pyotp
 from codeforlife.permissions import AllowNone
+from codeforlife.request import Request
+from codeforlife.response import Response
 from codeforlife.user.models import AuthFactor, User
 from codeforlife.user.permissions import IsTeacher
-from codeforlife.views import ModelViewSet
+from codeforlife.views import ModelViewSet, action
 
 from ..serializers import AuthFactorSerializer
 
@@ -22,7 +25,12 @@ class AuthFactorViewSet(ModelViewSet[User, AuthFactor]):
     def get_queryset(self):
         queryset = AuthFactor.objects.all()
         user = self.request.teacher_user
-        if user.teacher.school and user.teacher.is_admin:
+
+        if (
+            self.action in ["list", "destroy"]
+            and user.teacher.school
+            and user.teacher.is_admin
+        ):
             return queryset.filter(
                 user__new_teacher__school=user.teacher.school
             )
@@ -35,3 +43,14 @@ def get_permissions(self):
             return [AllowNone()]
 
         return [IsTeacher()]
+
+    @action(detail=False, methods=["post"])
+    def generate_otp_provisioning_uri(self, request: Request[User]):
+        """Generate a time-based one-time-password provisioning URI."""
+        # TODO: make otp_secret non-nullable and delete code block
+        user = request.auth_user
+        if not user.userprofile.otp_secret:
+            user.userprofile.otp_secret = pyotp.random_base32()
+            user.userprofile.save(update_fields=["otp_secret"])
+
+        return Response(user.totp_provisioning_uri, content_type="text/plain")

src/api/views/auth_factor_test.py

@@ -3,6 +3,8 @@
 Created on 23/01/2024 at 11:22:16(+00:00).
 """
 
+from unittest.mock import patch
+
 from codeforlife.permissions import AllowNone
 from codeforlife.tests import ModelViewSetTestCase
 from codeforlife.user.models import (
@@ -13,6 +15,7 @@
     User,
 )
 from codeforlife.user.permissions import IsTeacher
+from pyotp import TOTP
 
 from .auth_factor import AuthFactorViewSet
 
@@ -33,9 +36,41 @@ def setUp(self):
 
     # test: get queryset
 
-    def test_get_queryset__admin(self):
+    def test_get_queryset__list__admin(self):
+        """
+        Can list the author factors of all teachers in your school if you are
+        an admin.
+        """
+        user = self.mfa_non_admin_school_teacher_user
+        admin_school_teacher_user = AdminSchoolTeacherUser.objects.filter(
+            new_teacher__school=user.teacher.school
+        ).first()
+        assert admin_school_teacher_user
+
+        self.assert_get_queryset(
+            action="list",
+            values=list(
+                user.auth_factors.all()
+                | admin_school_teacher_user.auth_factors.all()
+            ),
+            request=self.client.request_factory.get(
+                user=admin_school_teacher_user
+            ),
+        )
+
+    def test_get_queryset__list__non_admin(self):
+        """Can only list your own auth factors if you are not an admin."""
+        user = self.mfa_non_admin_school_teacher_user
+
+        self.assert_get_queryset(
+            action="list",
+            values=list(user.auth_factors.all()),
+            request=self.client.request_factory.get(user=user),
+        )
+
+    def test_get_queryset__destroy__admin(self):
         """
-        Can query the author factors of all teachers in your school if you are
+        Can destroy the author factors of all teachers in your school if you are
         an admin.
         """
         user = self.mfa_non_admin_school_teacher_user
@@ -45,6 +80,7 @@ def test_get_queryset__admin(self):
         assert admin_school_teacher_user
 
         self.assert_get_queryset(
+            action="destroy",
             values=list(
                 user.auth_factors.all()
                 | admin_school_teacher_user.auth_factors.all()
@@ -54,11 +90,22 @@ def test_get_queryset__admin(self):
             ),
         )
 
-    def test_get_queryset__non_admin(self):
-        """Can only query your own auth factors if you are not an admin."""
+    def test_get_queryset__destroy__non_admin(self):
+        """Can only destroy your own auth factors if you are not an admin."""
         user = self.mfa_non_admin_school_teacher_user
 
         self.assert_get_queryset(
+            action="destroy",
+            values=list(user.auth_factors.all()),
+            request=self.client.request_factory.get(user=user),
+        )
+
+    def test_get_queryset__generate_otp_provisioning_uri(self):
+        """Can only generate an OTP provisioning URI yourself."""
+        user = self.mfa_non_admin_school_teacher_user
+
+        self.assert_get_queryset(
+            action="generate_otp_provisioning_uri",
             values=list(user.auth_factors.all()),
             request=self.client.request_factory.get(user=user),
         )
@@ -85,7 +132,13 @@ def test_get_permissions__destroy(self):
         """Only a teacher-user can disable an auth factor."""
         self.assert_get_permissions([IsTeacher()], action="destroy")
 
-    # test: generic actions
+    def test_get_permissions__generate_otp_provisioning_uri(self):
+        """Only a teacher-user can generate a OTP provisioning URI."""
+        self.assert_get_permissions(
+            [IsTeacher()], action="generate_otp_provisioning_uri"
+        )
+
+    # test: actions
 
     def test_list(self):
         """Can list enabled auth-factors."""
@@ -113,3 +166,28 @@ def test_destroy(self):
 
         self.client.login_as(user)
         self.client.destroy(auth_factor)
+
+    def test_generate_otp_provisioning_uri(self):
+        """Can successfully generate a OTP provisioning URI."""
+        user = TeacherUser.objects.exclude(
+            auth_factors__type__in=[AuthFactor.Type.OTP]
+        ).first()
+        assert user
+
+        # TODO: normalize password to "password"
+        self.client.login_as(user, password="abc123")
+
+        with patch.object(
+            TOTP, "provisioning_uri", return_value=user.totp_provisioning_uri
+        ) as provisioning_uri:
+            response = self.client.post(
+                self.reverse_action("generate_otp_provisioning_uri")
+            )
+
+            provisioning_uri.assert_called_once_with(
+                name=user.email,
+                issuer_name="Code for Life",
+            )
+
+            assert response.data == provisioning_uri.return_value
+            assert response.content_type == "text/plain"