Add PrairieTest integration

Author: inducer

.ci/run-mypy.sh

@@ -1,3 +1,3 @@
 #! /bin/bash
 
-mypy relate course accounts
+mypy relate course accounts prairietest

.ci/run-tests-for-ci.sh

@@ -2,13 +2,6 @@
 
 set -e
 
-PY_EXE=${PY_EXE:-$(poetry run which python)}
-
-echo "-----------------------------------------------"
-echo "Current directory: $(pwd)"
-echo "Python executable: ${PY_EXE}"
-echo "-----------------------------------------------"
-
 echo "i18n"
 # Testing i18n needs a local_settings file even though the rest of the tests
 #   don't use it

course/admin.py

@@ -23,10 +23,11 @@
 THE SOFTWARE.
 """
 
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from django import forms
 from django.contrib import admin
+from django.db.models import QuerySet
 from django.utils.translation import gettext_lazy as _, pgettext
 
 from course.constants import exam_ticket_states, participation_permission as pperm
@@ -56,9 +57,13 @@
 from relate.utils import string_concat
 
 
+if TYPE_CHECKING:
+    from accounts.models import User
+
+
 # {{{ permission helpers
 
-def _filter_courses_for_user(queryset, user):
+def _filter_courses_for_user(queryset: QuerySet, user: User) -> QuerySet:
     if user.is_superuser:
         return queryset
     z = queryset.filter(
@@ -67,7 +72,7 @@ def _filter_courses_for_user(queryset, user):
     return z
 
 
-def _filter_course_linked_obj_for_user(queryset, user):
+def _filter_course_linked_obj_for_user(queryset: QuerySet, user: User) -> QuerySet:
     if user.is_superuser:
         return queryset
     return queryset.filter(
@@ -76,7 +81,9 @@ def _filter_course_linked_obj_for_user(queryset, user):
             )
 
 
-def _filter_participation_linked_obj_for_user(queryset, user):
+def _filter_participation_linked_obj_for_user(
+            queryset: QuerySet, user: User
+        ) -> QuerySet:
     if user.is_superuser:
         return queryset
     return queryset.filter(

course/utils.py

@@ -54,7 +54,6 @@
     parse_date_spec,
 )
 from course.page.base import PageBase, PageContext
-from prairietest.utils import has_access_to_exam
 from relate.utils import (
     RelateHttpRequest,
     not_none,
@@ -164,6 +163,8 @@ def _eval_generic_conditions(
         now_datetime: datetime.datetime,
         flow_id: str,
         login_exam_ticket: ExamTicket | None,
+        *,
+        remote_ip_address: IPv4Address | IPv6Address | None = None,
         ) -> bool:
 
     if hasattr(rule, "if_before"):
@@ -189,6 +190,22 @@ def _eval_generic_conditions(
         if login_exam_ticket.exam.flow_id != flow_id:
             return False
 
+    if hasattr(rule, "if_has_prairietest_exam_access"):
+        if remote_ip_address is None:
+            return False
+        if participation is None:
+            return False
+
+        from prairietest.utils import has_access_to_exam
+        if not has_access_to_exam(
+                    course,
+                    participation.user.email,
+                    rule.if_has_prairietest_exam_access,
+                    now_datetime,
+                    remote_ip_address,
+                ):
+            return False
+
     return True
 
 
@@ -313,7 +330,8 @@ def get_session_start_rule(
     for rule in rules:
         if not _eval_generic_conditions(rule, course, participation,
                 now_datetime, flow_id=flow_id,
-                login_exam_ticket=login_exam_ticket):
+                login_exam_ticket=login_exam_ticket,
+                remote_ip_address=remote_ip_address):
             continue
 
         if not _eval_participation_tags_conditions(rule, participation):
@@ -401,9 +419,11 @@ def get_session_access_rule(
 
     for rule in rules:
         if not _eval_generic_conditions(
-                rule, session.course, session.participation,
-                now_datetime, flow_id=session.flow_id,
-                login_exam_ticket=login_exam_ticket):
+                    rule, session.course, session.participation,
+                    now_datetime, flow_id=session.flow_id,
+                    login_exam_ticket=login_exam_ticket,
+                    remote_ip_address=remote_ip_address,
+                ):
             continue
 
         if not _eval_participation_tags_conditions(rule, session.participation):
@@ -1050,7 +1070,7 @@ class FacilityFindingMiddleware:
     def __init__(self, get_response):
         self.get_response = get_response
 
-    def __call__(self, request):
+    def __call__(self, request: http.HttpRequest) -> http.HttpResponse:
         pretend_facilities = request.session.get("relate_pretend_facilities")
 
         if pretend_facilities is not None:
@@ -1071,6 +1091,7 @@ def __call__(self, request):
                     if remote_address in ip_network(str(ir)):
                         facilities.add(name)
 
+        request = cast(RelateHttpRequest, request)
         request.relate_facilities = frozenset(facilities)
 
         return self.get_response(request)

course/validation.py

@@ -576,6 +576,7 @@ def validate_session_start_rule(
                 ("if_has_fewer_sessions_than", int),
                 ("if_has_fewer_tagged_sessions_than", int),
                 ("if_signed_in_with_matching_exam_ticket", bool),
+                ("if_has_prairietest_exam_access", str),
                 ("tag_session", (str, type(None))),
                 ("may_start_new_session", bool),
                 ("may_list_existing_sessions", bool),
@@ -673,6 +674,7 @@ def validate_session_access_rule(
                 ("if_expiration_mode", str),
                 ("if_session_duration_shorter_than_minutes", (int, float)),
                 ("if_signed_in_with_matching_exam_ticket", bool),
+                ("if_has_prairietest_exam_access", str),
                 ("message", str),
                 ]
             )

local_settings_example.py

@@ -471,6 +471,17 @@
 #             "exams_only": True,
 #             },
 #     }
+#
+#    # Automatically get denied facilities from PrairieTest
+#    result = {}
+#
+#    from prairietest.utils import denied_ip_networks_at
+#    pt_facilities_networks = denied_ip_networks_at(now_datetime)
+#    for (course_id, facility_name), networks in pt_facilities_networks.items():
+#        fdata = result.setdefault(facility_name, {})
+#        fdata["exams_only"] = True
+#        fdata["ip_ranges"] = [*fdata.get("ip_ranges", []), *networks]
+#    return result
 
 
 RELATE_FACILITIES = {

prairietest/__init__.py

@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+
+__copyright__ = "Copyright (C) 2024 University of Illinois Board of Trustees"
+
+__license__ = """
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.
+"""
+
+from typing import TYPE_CHECKING
+
+from django import forms, http
+from django.contrib import admin
+from django.db.models import QuerySet
+from django.urls import reverse
+from django.utils.safestring import mark_safe
+
+from accounts.models import User
+from course.constants import participation_permission as pperm
+from prairietest.models import AllowEvent, DenyEvent, Facility, MostRecentDenyEvent
+
+
+if TYPE_CHECKING:
+    from accounts.models import User
+
+
+class FacilityAdminForm(forms.ModelForm):
+    class Meta:
+        model = Facility
+        fields = "__all__"
+        widgets = {
+            "secret": forms.PasswordInput(render_value=True),
+        }
+
+
+@admin.register(Facility)
+class FacilityAdmin(admin.ModelAdmin):
+    def get_queryset(self, request: http.HttpRequest) -> QuerySet:
+        assert request.user.is_authenticated
+
+        qs = super().get_queryset(request)
+        if request.user.is_superuser:
+            return qs
+        from course.admin import _filter_course_linked_obj_for_user
+        return _filter_course_linked_obj_for_user(qs, request.user)
+
+    def webhook_url(self, obj: Facility) -> str:
+        url = reverse(
+                "prairietest:webhook",
+                args=(obj.course.identifier, obj.identifier),
+            )
+        return mark_safe(
+            f"<tt>https://YOUR-HOST{url}</tt> Make sure to include the trailing slash!")
+
+    list_display = ["course", "identifier", "webhook_url"]
+    list_display_links = ["identifier"]
+    list_filter = ["course", "identifier"]
+
+    form = FacilityAdminForm
+
+
+def _filter_events_for_user(queryset: QuerySet, user: User) -> QuerySet:
+    if user.is_superuser:
+        return queryset
+    return queryset.filter(
+        facility__course__participations__user=user,
+        facility__course__participations__roles__permissions__permission=pperm.use_admin_interface)
+
+
+@admin.register(AllowEvent)
+class AllowEventAdmin(admin.ModelAdmin):
+    def get_queryset(self, request: http.HttpRequest) -> QuerySet:
+        assert request.user.is_authenticated
+
+        qs = super().get_queryset(request)
+        return _filter_events_for_user(qs, request.user)
+
+    list_display = [
+        "event_id", "facility", "user_uid", "start", "end", "exam_uuid"]
+    list_filter = ["facility", "user_uid", "exam_uuid"]
+
+
+@admin.register(DenyEvent)
+class DenyEventAdmin(admin.ModelAdmin):
+    def get_queryset(self, request: http.HttpRequest) -> QuerySet:
+        assert request.user.is_authenticated
+
+        qs = super().get_queryset(request)
+        return _filter_events_for_user(qs, request.user)
+
+    list_display = [
+        "event_id", "facility", "start", "end", "deny_uuid"]
+    list_filter = ["facility"]
+
+
+@admin.register(MostRecentDenyEvent)
+class MostRecentDenyEventAdmin(admin.ModelAdmin):
+    def get_queryset(self, request: http.HttpRequest) -> QuerySet:
+        assert request.user.is_authenticated
+
+        qs = super().get_queryset(request)
+        if request.user.is_superuser:
+            return qs
+        return qs.filter(
+            event__facility__course__participations__user=request.user,
+            event__facility__course__participations__roles__permissions__permission=pperm.use_admin_interface)
+
+    list_display = ["deny_uuid", "end"]