Ip filtering

api/urls.py

@@ -8,6 +8,7 @@
     path("tasks/<int:task_id>", default_view.task_detail),
     path("tasks/<int:task_id>/duplicate", default_view.duplicate_task),
     path("tasks/", default_view.task_detail),
+    path("classrooms-list/", default_view.classrooms_list),
     path("task-list", default_view.tasks_list_all),
     path("task-list/<subject_abbr>", default_view.tasks_list_all),
     path("student-list", default_view.student_list),

api/views/default.py

@@ -47,8 +47,15 @@
     assignedtask_results,
     current_semester,
     submit_assignment_path,
+    ClassroomIpRange,
+)
+from common.submit import (
+    SubmitRateLimited,
+    store_submit,
+    SubmitPastHardDeadline,
+    SubmitAfterFinal,
+    SubmitFromUnauthorizedIPError,
 )
-from common.submit import SubmitRateLimited, store_submit, SubmitPastHardDeadline, SubmitAfterFinal
 from common.upload import MAX_UPLOAD_FILECOUNT, TooManyFilesError
 from common.utils import is_teacher, points_to_color, inbus_search_user, user_from_inbus_person
 from quiz.models import EnrolledQuiz
@@ -597,6 +604,12 @@ def add_student_to_class(request, class_id):
     )
 
 
+def classrooms_list(request):
+    classrooms = ClassroomIpRange.objects.values("id", "name")
+
+    return JsonResponse(list(classrooms), safe=False)
+
+
 @user_passes_test(is_teacher)
 def task_detail(request, task_id=None):
     errors = []
@@ -719,7 +732,7 @@ def set_subject(task):
 
         for cl in data["classes"]:
             if cl.get("assigned", None):
-                AssignedTask.objects.update_or_create(
+                assigned_task, _ = AssignedTask.objects.update_or_create(
                     task_id=task.pk,
                     clazz_id=cl["id"],
                     defaults={
@@ -731,6 +744,15 @@ def set_subject(task):
                         "hard_deadline": cl.get("hard_deadline", False),
                     },
                 )
+
+                if "allowed_classrooms" in cl:
+                    classes = []
+                    for class_id in cl["allowed_classrooms"]:
+                        class_object = ClassroomIpRange.objects.get(pk=class_id)
+                        classes.append(class_object)
+
+                    assigned_task.allowed_classrooms.set(classes)
+
             else:
                 submits = Submit.objects.filter(
                     assignment__task_id=task.pk, assignment__clazz_id=cl["id"]
@@ -856,6 +878,9 @@ def is_allowed(path):
             item["deadline"] = assigned.deadline
             item["max_points"] = assigned.max_points
             item["hard_deadline"] = assigned.hard_deadline
+            item["allowed_classrooms"] = list(
+                assigned.allowed_classrooms.values_list("id", flat=True)
+            )
 
         result["classes"].append(item)
 
@@ -1024,6 +1049,12 @@ def create_submit(request: django.http.HttpRequest, task_assignment: int) -> Jso
                 "error": "The submission was sent after the final one, so it is not a valid submission."
             }
         )
+    except SubmitFromUnauthorizedIPError:
+        return JsonResponse(
+            {
+                "error": "The submission was sent from IP address that is not on the list of allowed IP addresses."
+            }
+        )
 
     url = (
         reverse(

common/admin.py

@@ -1,7 +1,11 @@
+import ipaddress
+
 from django.contrib import admin
 from django.contrib.auth.admin import UserAdmin
 from django.contrib.auth.models import User
 
+from django import forms
+
 import common.models as models
 import common.utils
 
@@ -82,6 +86,8 @@ class AssignedTaskAdmin(admin.ModelAdmin):
     autocomplete_fields = ["task", "clazz"]
     search_fields = ["task__name", "clazz__teacher__username", "clazz__subject_abbr"]
 
+    filter_horizontal = ["allowed_classrooms"]
+
     def formfield_for_foreignkey(self, db_field, request, **kwargs):
         if db_field.name == "clazz":
             kwargs["queryset"] = models.Class.objects.current_semester()
@@ -138,13 +144,64 @@ class SubmitAdmin(admin.ModelAdmin):
     autocomplete_fields = ["assignment"]
 
 
+class ClassroomAdminForm(forms.ModelForm):
+    use_cidr = forms.BooleanField(required=False, label="Enter CIDR instead of range")
+
+    ip_range_start = forms.GenericIPAddressField(required=False)
+    ip_range_end = forms.GenericIPAddressField(required=False)
+    cidr = forms.CharField(required=False, label="CIDR address")
+
+    class Meta:
+        model = models.ClassroomIpRange
+        fields = ("name", "ip_range_start", "ip_range_end", "cidr")
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+        if self.instance and self.instance.pk:
+            start_ip = ipaddress.ip_address(self.instance.ip_range_start)
+            end_ip = ipaddress.ip_address(self.instance.ip_range_end)
+
+            cidr = list(ipaddress.summarize_address_range(start_ip, end_ip))[0]
+
+            self.fields["cidr"].initial = str(cidr)
+
+    def clean(self):
+        cleaned_data = super().clean()
+
+        if cleaned_data.get("name") is None:
+            raise forms.ValidationError("Please enter classroom code")
+
+        cidr = cleaned_data.get("use_cidr")
+
+        if cidr:
+            cidr_value = cleaned_data.get("cidr")
+            if not cidr_value:
+                raise forms.ValidationError("Cannot read CIDR field value")
+
+            network = ipaddress.ip_network(cidr_value, strict=False)
+
+            cleaned_data["ip_range_start"] = network.network_address
+            cleaned_data["ip_range_end"] = network.broadcast_address
+        else:
+            if not cleaned_data.get("ip_range_start") or not cleaned_data.get("ip_range_end"):
+                raise forms.ValidationError("You didn't enter IP range")
+
+        return cleaned_data
+
+
+class ClassroomAdmin(admin.ModelAdmin):
+    form = ClassroomAdminForm
+
+
 admin.site.register(models.Task, TaskAdmin)
 admin.site.register(models.Class, ClassAdmin)
 admin.site.register(models.Submit, SubmitAdmin)
 admin.site.register(models.AssignedTask, AssignedTaskAdmin)
 admin.site.register(models.Semester)
 admin.site.register(models.Subject)
 admin.site.register(models.UserEventModel)
+admin.site.register(models.ClassroomIpRange, ClassroomAdmin)
 
 admin.site.unregister(User)
 admin.site.register(User, MyUserAdmin)

common/migrations/0030_classroomiprange_assignedtask_allowed_classrooms.py

@@ -0,0 +1,27 @@
+# Generated by Django 4.2.16 on 2025-10-23 12:06
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('common', '0029_submit_is_final_alter_usereventmodel_action'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='ClassroomIpRange',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('name', models.TextField()),
+                ('ip_range_start', models.GenericIPAddressField()),
+                ('ip_range_end', models.GenericIPAddressField()),
+            ],
+        ),
+        migrations.AddField(
+            model_name='assignedtask',
+            name='allowed_classrooms',
+            field=models.ManyToManyField(related_name='assignments', to='common.classroomiprange'),
+        ),
+    ]

common/models.py

@@ -1,3 +1,4 @@
+import ipaddress
 import os
 import re
 import logging
@@ -222,6 +223,15 @@ class Meta:
         verbose_name_plural = "classes"
 
 
+class ClassroomIpRange(models.Model):
+    name = models.TextField()
+    ip_range_start = models.GenericIPAddressField()
+    ip_range_end = models.GenericIPAddressField()
+
+    def __str__(self):
+        return f"{self.name}: {self.ip_range_start} – {self.ip_range_end}"
+
+
 class AssignedTask(models.Model):
     task = models.ForeignKey(Task, on_delete=models.CASCADE)
     clazz = models.ForeignKey(Class, on_delete=models.CASCADE)
@@ -230,6 +240,7 @@ class AssignedTask(models.Model):
     hard_deadline = models.BooleanField(default=False)
     max_points = models.IntegerField(null=True, blank=True)
     moss_url = models.URLField(null=True, blank=True, editable=False)
+    allowed_classrooms = models.ManyToManyField(ClassroomIpRange, related_name="assignments")
 
     def is_visible(self):
         return timezone.now() >= self.assigned
@@ -243,6 +254,22 @@ def is_past_deadline(self):
             and datetime.datetime.now(datetime.timezone.utc) > self.deadline
         )
 
+    def is_allowed_from_ip(self, ip: str) -> bool:
+        ip = ipaddress.ip_address(ip)
+
+        if not self.allowed_classrooms.all().exists():
+            return True
+
+        allowed = False
+
+        for classroom in self.allowed_classrooms.all():
+            start = ipaddress.ip_address(classroom.ip_range_start)
+            end = ipaddress.ip_address(classroom.ip_range_end)
+
+            allowed |= start <= ip <= end
+
+        return allowed
+
     def __str__(self):
         return f"{self.task.name} {self.clazz}"
 

common/submit.py

@@ -35,6 +35,11 @@ def __init__(self, message: str):
         super().__init__(message)
 
 
+class SubmitFromUnauthorizedIPError(Exception):
+    def __init__(self, message: str):
+        super().__init__(message)
+
+
 def store_submit(request: HttpRequest, assignment: AssignedTask) -> Submit:
     """
     Creates a new submit for the given `assignment` and the user logged in the `request`.
@@ -81,6 +86,12 @@ def store_submit(request: HttpRequest, assignment: AssignedTask) -> Submit:
     if client_ip:
         s.ip_address = client_ip
 
+    if assignment.allowed_classrooms:
+        if not assignment.is_allowed_from_ip(client_ip):
+            raise SubmitFromUnauthorizedIPError(
+                "It is forbidden to upload solutions from this IP address"
+            )
+
     solutions = request.FILES.getlist("solution")
     tmp = request.POST.get("paths", None)
     if tmp:

common/utils.py

@@ -1,5 +1,10 @@
 from datetime import timedelta
+
+from django.core.exceptions import PermissionDenied
 from django.http import HttpRequest
+from django.http.response import Http404
+from django.utils import timezone
+
 from .inbus import inbus
 import django.contrib.auth.models
 import re
@@ -81,3 +86,56 @@ def get_client_ip_address(request: HttpRequest) -> IPAddressString | None:
         return None
     else:
         return IPAddressString(client_ip)
+
+
+def ip_address_check(function):
+    """
+    Decorator that restricts access to specific IP addresses.
+
+    The decorated function must have the following parameters:
+        - request
+        - assignment_id
+
+    Access is granted if any of the following conditions are met:
+        - requesting user is a teacher
+        - assignment deadline has passed
+        - assignment has no IP address restrictions
+        - user is accessing from an allowed IP address
+
+    IP address check is performed using:
+        models.AssignedTask.is_allowed_from_ip(str)
+    """
+
+    def wrapper(*args, **kwargs):
+        # this import is here to prevent cyclic dependency (.models uses is_teacher)
+        from .models import AssignedTask
+
+        request = args[0]
+
+        if is_teacher(request.user):
+            return function(*args, **kwargs)
+
+        assignment_id = kwargs.get("assignment_id")
+
+        try:
+            assignment = AssignedTask.objects.get(pk=assignment_id)
+        except AssignedTask.DoesNotExist:
+            raise Http404(f"AssignedTask with id {assignment_id} not found")
+
+        # allow after deadline
+        if assignment.deadline is not None and timezone.now() > assignment.deadline:
+            return function(*args, **kwargs)
+
+        if assignment.allowed_classrooms:
+            x_forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR")
+            if x_forwarded_for:
+                ip = x_forwarded_for.split(",")[0].strip()
+            else:
+                ip = request.META.get("REMOTE_ADDR")
+
+            if assignment.is_allowed_from_ip(ip):
+                return function(*args, **kwargs)
+            else:
+                raise PermissionDenied("Access from this IP is not allowed")
+
+    return wrapper