feat: enhance exam answer submission security and add tests for user authentication

generic/tests/test_exam_submit_security.py

@@ -0,0 +1,172 @@
+import json
+import pytest
+from django.urls import reverse
+from django.contrib.auth.models import User
+from django.test import Client
+from django.utils import timezone
+from anatomy.models import Exam as AnatomyExam, AnatomyQuestion, QuestionType, Structure
+from generic.models import CidUser, CidUserGroup
+
+@pytest.fixture
+def logged_in_user(db):
+    user = User.objects.create_user(username="user1", password="password", email="user1@test.com")
+    return user
+
+@pytest.fixture
+def logged_in_client(logged_in_user):
+    client = Client()
+    client.force_login(logged_in_user)
+    return client
+
+@pytest.fixture
+def anonymous_client(db):
+    return Client()
+
+@pytest.fixture
+def exam_data(db):
+    exam = AnatomyExam.objects.create(
+        name="Anatomy Exam Test",
+        exam_mode=True,
+        active=True,
+        open_access=False,
+        exam_open_access=True,  # allows open access so we don't have to populate cid/user lists for candidate test
+    )
+    qtype = QuestionType.objects.create(id=1, text="Test Type")
+    structure = Structure.objects.create(name="Test Structure")
+    question = AnatomyQuestion.objects.create(
+        question_type=qtype,
+        structure=structure,
+        image="test.jpg",
+        description="Test Question description",
+    )
+    question.exams.set([exam])
+    return exam, question
+
+@pytest.mark.django_db
+def test_authenticated_user_submit_success(logged_in_client, logged_in_user, exam_data):
+    exam, question = exam_data
+    payload = {
+        "eid": f"anatomy/{exam.pk}",
+        "cid": f"u-{logged_in_user.pk}",
+        "start_time": timezone.now().timestamp(),
+        "answers": json.dumps([
+            {
+                "eid": f"anatomy/{exam.pk}",
+                "cid": f"u-{logged_in_user.pk}",
+                "qid": question.pk,
+                "ans": "Normal",
+                "qidn": "1",
+            }
+        ])
+    }
+    response = logged_in_client.post(reverse("global_exam_answers_submit"), payload)
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert data["question_count"] == 1
+
+@pytest.mark.django_db
+def test_authenticated_user_submit_mismatch_other_user(logged_in_client, logged_in_user, exam_data):
+    exam, question = exam_data
+    payload = {
+        "eid": f"anatomy/{exam.pk}",
+        "cid": f"u-{logged_in_user.pk + 999}", # root cid doesn't match authenticated user
+        "start_time": timezone.now().timestamp(),
+        "answers": json.dumps([
+            {
+                "eid": f"anatomy/{exam.pk}",
+                "cid": f"u-{logged_in_user.pk}",
+                "qid": question.pk,
+                "ans": "Normal",
+                "qidn": "1",
+            }
+        ])
+    }
+    response = logged_in_client.post(reverse("global_exam_answers_submit"), payload)
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is False
+    assert "user / cid mismatch" in data["error"]
+
+@pytest.mark.django_db
+def test_authenticated_user_submit_mismatch_answer_level(logged_in_client, logged_in_user, exam_data):
+    exam, question = exam_data
+    payload = {
+        "eid": f"anatomy/{exam.pk}",
+        "cid": f"u-{logged_in_user.pk}",
+        "start_time": timezone.now().timestamp(),
+        "answers": json.dumps([
+            {
+                "eid": f"anatomy/{exam.pk}",
+                "cid": f"u-{logged_in_user.pk + 999}", # answer level cid doesn't match authenticated user
+                "qid": question.pk,
+                "ans": "Normal",
+                "qidn": "1",
+            }
+        ])
+    }
+    response = logged_in_client.post(reverse("global_exam_answers_submit"), payload)
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is False
+    assert "user / uid mismatch" in data["error"]
+
+@pytest.mark.django_db
+def test_anonymous_user_submit_user_prefix_fails(anonymous_client, exam_data):
+    exam, question = exam_data
+    payload = {
+        "eid": f"anatomy/{exam.pk}",
+        "cid": "u-1", # prefix u- is forbidden for anonymous users
+        "start_time": timezone.now().timestamp(),
+        "answers": json.dumps([
+            {
+                "eid": f"anatomy/{exam.pk}",
+                "cid": "u-1",
+                "qid": question.pk,
+                "ans": "Normal",
+                "qidn": "1",
+            }
+        ])
+    }
+    response = anonymous_client.post(reverse("global_exam_answers_submit"), payload)
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is False
+    assert "authentication required" in data["error"]
+
+@pytest.mark.django_db
+def test_candidate_submit_success(anonymous_client, exam_data):
+    exam, question = exam_data
+    cid_user = CidUser.objects.create(
+        cid=8888,
+        name="Candidate 1",
+        email="candidate1@test.com",
+        passcode="candidatepass",
+    )
+    # Associate CID user with the exam
+    exam.exam_open_access = False
+    group = CidUserGroup.objects.create(name="Test Group")
+    group.ciduser_set.add(cid_user)
+    exam.cid_user_groups.add(group)
+    exam.valid_cid_users.add(cid_user)
+
+    payload = {
+        "eid": f"anatomy/{exam.pk}",
+        "cid": "8888",
+        "start_time": timezone.now().timestamp(),
+        "answers": json.dumps([
+            {
+                "eid": f"anatomy/{exam.pk}",
+                "cid": "8888",
+                "passcode": "candidatepass",
+                "qid": question.pk,
+                "ans": "Normal",
+                "qidn": "1",
+            }
+        ])
+    }
+    response = anonymous_client.post(reverse("global_exam_answers_submit"), payload)
+    assert response.status_code == 200
+    data = response.json()
+    assert data["success"] is True
+    assert data["question_count"] == 1

generic/views.py

@@ -3294,8 +3294,21 @@ class ExamViews(View, LoginRequiredMixin):
     @method_decorator(csrf_exempt)
     def post_exam_answers(self, request):
         if request.method == "POST":
-            n = 0
+            root_cid = request.POST.get("cid")
+            if not root_cid:
+                return JsonResponse({"success": False, "error": "cid not defined"})
 
+            if request.user.is_authenticated:
+                expected_cid = f"u-{request.user.id}"
+                if root_cid != expected_cid:
+                    return JsonResponse({"success": False, "error": "user / cid mismatch"})
+            else:
+                if root_cid.startswith("u-"):
+                    return JsonResponse(
+                        {"success": False, "error": "authentication required for user submissions"}
+                    )
+
+            n = 0
             for answer in json.loads(request.POST.get("answers")):
                 logger.debug("Processing answer: {}", answer)
 
@@ -3303,25 +3316,28 @@ class ExamViews(View, LoginRequiredMixin):
                 uid = False
                 passcode = None
 
-                try:
-                    cid = int(answer["cid"])
-                    passcode = answer["passcode"]
-                except ValueError:
-                    if answer["cid"].startswith("u-"):
-                        cid = False
-                        uid = int(answer["cid"][2:])
+                ans_cid = answer.get("cid")
+                if not ans_cid:
+                    return JsonResponse({"success": False, "error": "answer cid not defined"})
 
-                        if not uid == request.user.id:
-                            return JsonResponse(
-                                {"success": False, "error": "user / uid mismatch"}
-                            )
-
-                    else:
+                if request.user.is_authenticated:
+                    if ans_cid != f"u-{request.user.id}":
+                        return JsonResponse({"success": False, "error": "user / uid mismatch"})
+                    uid = request.user.id
+                    cid = False
+                    passcode = None
+                else:
+                    if str(ans_cid).startswith("u-"):
                         return JsonResponse(
-                            {
-                                "success": False,
-                                "error": "cid or eid or answers not defined",
-                            }
+                            {"success": False, "error": "authentication required"}
+                        )
+                    try:
+                        cid = int(ans_cid)
+                        passcode = answer["passcode"]
+                        uid = False
+                    except (ValueError, KeyError):
+                        return JsonResponse(
+                            {"success": False, "error": "invalid cid or passcode"}
                         )
 
                 if not uid: