Only allow linking FEIDE to a single user

lego/apps/users/tests/test_student_confirmation_api.py

@@ -12,11 +12,18 @@
 class MockFeideOAUTH:
     _auth_url = "https://auth.mock-feide.no/auth"
 
+    def __init__(self, token="valid_token"):
+        self.token = token
+
     def authorize_redirect(self, request, redirect_url):
         return HttpResponseRedirect(self._auth_url)
 
     def authorize_access_token(self, request):
-        return "valid_token"
+        return _token(self.token)
+
+    def userinfo(self, **kwargs):
+        uid = f"{kwargs.get('token')['access_token']}@ntnu.no"
+        return {"https://n.feide.no/claims/eduPersonPrincipalName": uid}
 
 
 mockFeide = MockFeideOAUTH()
@@ -205,10 +212,7 @@ def test_with_unauthenticated_user(self, *args):
         response = self.client.get(_get_validate_url())
         self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
 
-    @mock.patch(
-        "lego.apps.users.views.oidc.oauth.feide.authorize_access_token",
-        return_value=_token(Token.DATA),
-    )
+    @mock.patch("lego.apps.users.views.oidc.oauth.feide", MockFeideOAUTH(Token.DATA))
     def test_data_1st(self, *args):
         response = self.client.get(_get_validate_url())
         self.assertEqual(response.status_code, status.HTTP_200_OK)
@@ -220,10 +224,7 @@ def test_data_1st(self, *args):
             self.user_without_student_confirmation.grade.id, self.grade_data_1.id
         )
 
-    @mock.patch(
-        "lego.apps.users.views.oidc.oauth.feide.authorize_access_token",
-        return_value=_token(Token.KOMTEK),
-    )
+    @mock.patch("lego.apps.users.views.oidc.oauth.feide", MockFeideOAUTH(Token.KOMTEK))
     def test_komtek_1st(self, *args):
         response = self.client.get(_get_validate_url())
 
@@ -235,8 +236,7 @@ def test_komtek_1st(self, *args):
         )
 
     @mock.patch(
-        "lego.apps.users.views.oidc.oauth.feide.authorize_access_token",
-        return_value=_token(Token.DATA_MASTER),
+        "lego.apps.users.views.oidc.oauth.feide", MockFeideOAUTH(Token.DATA_MASTER)
     )
     def test_data_4st(self, *args):
         response = self.client.get(_get_validate_url())
@@ -251,8 +251,7 @@ def test_data_4st(self, *args):
         )
 
     @mock.patch(
-        "lego.apps.users.views.oidc.oauth.feide.authorize_access_token",
-        return_value=_token(Token.MULTI_OTHER),
+        "lego.apps.users.views.oidc.oauth.feide", MockFeideOAUTH(Token.MULTI_OTHER)
     )
     def test_with_other_study_informatics(self, *args):
         response = self.client.get(_get_validate_url())
@@ -265,10 +264,7 @@ def test_with_other_study_informatics(self, *args):
         self.assertEqual(len(json.get("studyProgrammes")), len(multi_other_resp))
         self.assertIsNone(self.user_without_student_confirmation.grade)
 
-    @mock.patch(
-        "lego.apps.users.views.oidc.oauth.feide.authorize_access_token",
-        return_value=_token(Token.DATA),
-    )
+    @mock.patch("lego.apps.users.views.oidc.oauth.feide", MockFeideOAUTH(Token.DATA))
     def test_valid_study_existing_grade(self, *args):
         """
         You should keep your grade when re-authenticating
@@ -286,10 +282,7 @@ def test_valid_study_existing_grade(self, *args):
             self.user_with_student_confirmation.grade.id, self.grade_data_4.id
         )
 
-    @mock.patch(
-        "lego.apps.users.views.oidc.oauth.feide.authorize_access_token",
-        return_value=_token(Token.INDOK),
-    )
+    @mock.patch("lego.apps.users.views.oidc.oauth.feide", MockFeideOAUTH(Token.INDOK))
     def test_switch_to_indok(self, *args):
         """
         You should keep your validation status and grade when switching to indok
@@ -306,3 +299,28 @@ def test_switch_to_indok(self, *args):
             self.user_with_student_confirmation.grade.id, self.grade_data_4.id
         )
         self.assertTrue(self.user_with_student_confirmation.is_verified_student())
+
+    @mock.patch("lego.apps.users.views.oidc.oauth.feide", MockFeideOAUTH(Token.DATA))
+    def test_multiple_users_one_feide(self, *args):
+        """
+        It should only be allowed to auth a single user with a feide account
+        """
+        self.client.force_authenticate(self.user_with_student_confirmation)
+        response = self.client.get(_get_validate_url())
+
+        json = response.json()
+        self.assertEqual(json.get("status"), "success")
+        self.assertEqual(
+            self.user_with_student_confirmation.student_username, f"{Token.DATA}"
+        )
+
+        self.client.force_authenticate(self.user_without_student_confirmation)
+        response = self.client.get(_get_validate_url())
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        json = response.json()
+        self.assertEqual(json.get("status"), "error")
+        user_without_student_confirmation = User.objects.get(username="test2")
+        self.assertNotEqual(
+            self.user_with_student_confirmation.student_username,
+            user_without_student_confirmation.student_username,
+        )

lego/apps/users/validators.py

@@ -10,8 +10,10 @@
     "invalid",
 )
 
+STUDENT_USERNAME_REGEX = r"[a-zA-Z0-9_\.]+"
+
 student_username_validator = validators.RegexValidator(
-    r"^[a-zA-Z0-9_\.]+$",
+    rf"^{STUDENT_USERNAME_REGEX}$",
     "Enter a valid username. This value may contain only letters, numbers, _ and . "
     "characters.",
     "invalid",

lego/apps/users/views/oidc.py

@@ -1,5 +1,8 @@
+import re
+
 from django.conf import settings
 from django.core.cache import caches
+from django.db import IntegrityError, transaction
 from django.http import HttpResponse, JsonResponse
 from rest_framework import status, viewsets
 from rest_framework.decorators import action
@@ -15,6 +18,7 @@
 
 from lego.apps.users.models import User
 from lego.apps.users.serializers.student_confirmation import FeideAuthorizeSerializer
+from lego.apps.users.validators import STUDENT_USERNAME_REGEX
 
 log = get_logger()
 
@@ -37,6 +41,16 @@ def get_feide_groups(bearer: str) -> Response:
     )
 
 
+def get_uid_from_principalName(principal_name: str) -> str | None:
+    regex = rf"(?P<uid>{STUDENT_USERNAME_REGEX})@ntnu.no"
+
+    match = re.match(regex, principal_name)
+    if match is not None:
+        return match.group("uid")
+
+    return None
+
+
 class OIDCViewSet(viewsets.GenericViewSet):
     permission_classes = (IsAuthenticated,)
 
@@ -53,21 +67,47 @@ def authorize(self, request: Request) -> JsonResponse:
     def validate(self, request: Request) -> HttpResponse:
         programme_names = []
 
+        user: User = request.user  # type: ignore[assignment]
+
         try:
             token = oauth.feide.authorize_access_token(request)
-            validation_status = "success"
+            userinfo = oauth.feide.userinfo(token=token)
+            principal_name = userinfo[
+                "https://n.feide.no/claims/eduPersonPrincipalName"
+            ]
+            uid = get_uid_from_principalName(principal_name)
+            if uid is None:
+                sentry_sdk.capture_message(
+                    "Could not extract student username from principal_name",
+                    "fatal",
+                    principle_name=principal_name,
+                )
+                validation_status = "error"
+            else:
+                try:
+                    validation_status = "success"
+                    with transaction.atomic():
+                        user.student_username = uid
+                        user.save()
+                except IntegrityError:
+                    return JsonResponse(
+                        {
+                            "status": "error",
+                            "detail": f"The feide account {uid} is already linked to another user",
+                        },
+                        status=status.HTTP_400_BAD_REQUEST,
+                    )
         except MismatchingStateError as e:
             sentry_sdk.capture_message("Failed while validating access token", "fatal")
             sentry_sdk.capture_exception(error=e)
             return JsonResponse(
                 {
                     "status": "error",
                     "detail": "Error when validating OAUTH acccess token",
-                }
+                },
+                status=status.HTTP_400_BAD_REQUEST,
             )
 
-        user: User = request.user  # type: ignore[assignment]
-
         try:
             groups_res = get_feide_groups(token["access_token"])
             groups_res.raise_for_status()