Add access checks to all endpoints

access/config.py

@@ -9,7 +9,7 @@
 import logging
 import os
 import time
-from typing import Any, ClassVar, Dict, Optional, List, Tuple, Union
+from typing import Any, ClassVar, Dict, Iterable, Optional, List, Tuple, Union
 
 from django.conf import settings
 from pydantic.error_wrappers import ValidationError
@@ -180,7 +180,27 @@ def static_path_to(self, *paths: str) -> Optional[str]:
         return os.path.join(self.data.static_dir, *paths)
 
     @staticmethod
-    def all():
+    def get_for(courses: Iterable[CourseModel]):
+        configs = []
+        for course in courses:
+            try:
+                config = CourseConfig.get(course.key)
+            except ConfigError:
+                LOGGER.exception("Failed to load course: %s", course.key)
+            except ValidationError as e:
+                LOGGER.exception("Failed to load course: %s", course.key)
+                LOGGER.exception(validation_error_str(e))
+            else:
+                configs.append(config)
+                warnings = validation_warning_str(config)
+                if warnings:
+                    LOGGER.warning("Warnings in course config: %s", course.key)
+                    LOGGER.warning(warnings)
+
+        return configs
+
+    @staticmethod
+    def all(courses: Optional[Iterable[Course]] = None):
         '''
         Gets all course configs.
 
@@ -192,25 +212,12 @@ def all():
         t = os.path.getmtime(settings.COURSES_PATH)
         if CourseConfig._dir_mtime < t:
             CourseConfig._courses.clear()
-            CourseConfig._dir_mtime = t
 
-            LOGGER.debug('Recreating course list.')
-            for course in CourseModel.objects.all():
-                try:
-                    config = CourseConfig.get(course.key)
-                except ConfigError:
-                    LOGGER.exception("Failed to load course: %s", course.key)
-                except ValidationError as e:
-                    LOGGER.exception("Failed to load course: %s", course.key)
-                    LOGGER.exception(validation_error_str(e))
-                else:
-                    warnings = validation_warning_str(config)
-                    if warnings:
-                        LOGGER.warning("Warnings in course config: %s", course.key)
-                        LOGGER.warning(warnings)
+            CourseConfig.get_for(CourseModel.objects.all())
 
-        return CourseConfig._courses.values()
+            CourseConfig._dir_mtime = t
 
+        return CourseConfig._courses.values()
 
     @staticmethod
     def get(course_key: str) -> Optional[CourseConfig]:

access/views.py

@@ -6,7 +6,7 @@
 from typing import Any, Dict, List, Optional, Tuple
 
 from aplus_auth.auth.django import Request
-from django.shortcuts import render
+from django.shortcuts import get_object_or_404, render
 from django.http import HttpRequest, HttpResponse, JsonResponse, Http404
 from django.utils import translation
 from django.urls import reverse
@@ -30,7 +30,11 @@ def index(request):
     '''
     Signals that the grader is ready and lists available courses.
     '''
-    course_configs = CourseConfig.all()
+    # Only show courses user has access to
+    courses = (course for course in Course.objects.all() if course.has_read_access(request, True))
+
+    course_configs = CourseConfig.get_for(courses)
+
     if request.is_ajax():
         return JsonResponse({
             "ready": True,
@@ -46,6 +50,10 @@ def course(request, course_key):
     '''
     Signals that the course is ready to be graded and lists available exercises.
     '''
+    course = get_object_or_404(Course, key=course_key)
+    if not course.has_read_access(request, True):
+        return HttpResponse(status=403)
+
     error = None
     course_config = None
     exercises = None
@@ -55,12 +63,7 @@ def course(request, course_key):
         error = str(e)
     else:
         if course_config is None:
-            try:
-                Course.objects.get(key=course_key)
-            except:
-                raise Http404()
-            else:
-                error = "Failed to load course config (has it been built and published?)"
+            error = "Failed to load course config (has it been built and published?)"
         else:
             exercises = course_config.get_exercise_list()
 
@@ -95,11 +98,7 @@ def protected(request: Request, course_key: str, path: str):
     if os.path.normpath(path).startswith("../"):
         raise Http404()
 
-    try:
-        course: Course = Course.objects.get(key=course_key)
-    except Course.DoesNotExist:
-        raise Http404()
-
+    course = get_object_or_404(Course, key=course_key)
     if not course.has_read_access(request, True):
         return HttpResponse(status=403)
 
@@ -115,6 +114,10 @@ def protected(request: Request, course_key: str, path: str):
 
 
 def serve_exercise_file(request, course_key, exercise_key, basename, dict_key, type):
+    course = get_object_or_404(Course, key=course_key)
+    if not course.has_read_access(request, True):
+        return HttpResponse(status=403)
+
     lang = request.GET.get('lang', None)
     try:
         (course, exercise, lang) = _get_course_exercise_lang(course_key, exercise_key, lang)
@@ -164,6 +167,10 @@ def aplus_json(request: HttpRequest, course_key: str):
     '''
     errors = []
 
+    course = get_object_or_404(Course, key=course_key)
+    if not course.has_read_access(request, True):
+        return HttpResponse(status=403)
+
     try:
         config = CourseConfig.load_from_store(course_key)
     except ConfigError as e:
@@ -234,6 +241,10 @@ def children_recursion(config: CourseConfig, parent: Parent) -> List[Dict[str, A
 
 @login_required
 def publish(request: HttpRequest, course_key: str) -> HttpResponse:
+    course = get_object_or_404(Course, key=course_key)
+    if not course.has_write_access(request, True):
+        return HttpResponse(status=403)
+
     try:
         errors = builder.publish(course_key)
     except Exception as e:

gitmanager/models.py

@@ -1,8 +1,8 @@
 from enum import Enum
 
-from aplus_auth.auth.django import Request
 from aplus_auth.payload import Permission
 from django.db import models
+from django.http.request import HttpRequest
 
 from util.login_required import has_access
 
@@ -24,16 +24,16 @@ class Course(models.Model):
     class META:
         ordering = ['key']
 
-    def has_access(self, request: Request, permission: Permission, default: bool = False) -> bool:
+    def has_access(self, request: HttpRequest, permission: Permission, default: bool = False) -> bool:
         if self.remote_id is None:
             return default
 
         return has_access(request, permission, self.remote_id)
 
-    def has_write_access(self, request: Request, default: bool = False):
+    def has_write_access(self, request: HttpRequest, default: bool = False):
         return self.has_access(request, Permission.WRITE, default)
 
-    def has_read_access(self, request: Request, default: bool = False):
+    def has_read_access(self, request: HttpRequest, default: bool = False):
         return self.has_access(request, Permission.READ, default)
 
 

gitmanager/views.py

@@ -22,8 +22,10 @@
 
 @login_required
 def courses(request):
+    courses = (course for course in Course.objects.all() if course.has_read_access(request, True))
+
     return render(request, 'gitmanager/courses.html', {
-        'courses': Course.objects.all(),
+        'courses': courses,
         'ssh_key': ssh_key,
     })
 
@@ -32,6 +34,8 @@ def courses(request):
 def edit(request, key = None):
     if key:
         course = get_object_or_404(Course, key=key)
+        if not course.has_write_access(request, True):
+            return HttpResponse(status=403)
         form = CourseForm(request.POST or None, instance=course)
     else:
         course = None
@@ -40,6 +44,8 @@ def edit(request, key = None):
         if name not in ("email_on_error", "update_automatically"):
             form.fields[name].widget.attrs = {'class': 'form-control'}
     if request.method == 'POST' and form.is_valid():
+        if "remote_id" in request.POST and not has_access(request, Permission.WRITE, request.POST["remote_id"]):
+            return HttpResponse(f"No access to instance id {request.POST['remote_id']}", status=403)
         form.save()
         return redirect('manager-courses')
     return render(request, 'gitmanager/edit.html', {
@@ -142,6 +148,8 @@ def put(self, request: Request, key: str, **kwargs) -> HttpResponse:
 @login_required
 def updates(request, key):
     course = get_object_or_404(Course, key=key)
+    if not course.has_read_access(request, True):
+        return HttpResponse(status=403)
     return render(request, 'gitmanager/updates.html', {
         'course': course,
         'updates': course.updates.order_by('-request_time').all(),
@@ -152,9 +160,11 @@ def updates(request, key):
 @login_required
 def build_log_json(request, key):
     try:
-        course = Course.objects.get(key=key)
+        course = get_object_or_404(Course, key=key)
     except Course.DoesNotExist:
         return JsonResponse({})
+    if not course.has_read_access(request, True):
+        return HttpResponse(status=403)
     latest_update = course.updates.order_by("-updated_time")[0]
     return JsonResponse({
         'build_log': latest_update.log,

util/login_required.py

@@ -1,16 +1,17 @@
 from typing import Callable
 
 from aplus_auth import settings as auth_settings
-from aplus_auth.auth.django import login_required as login_required_base, Request
+from aplus_auth.auth.django import login_required as login_required_base
 from aplus_auth.payload import Permission
+from django.http import HttpRequest
 from django.http.response import HttpResponseBase
 
 
 ViewType = Callable[..., HttpResponseBase]
 login_required: Callable[[ViewType],ViewType] = login_required_base(redirect_url="/login?referer={url}")
 
 
-def has_access(request: Request, permission: Permission, instance_id: int) -> bool:
+def has_access(request: HttpRequest, permission: Permission, instance_id: int) -> bool:
     if auth_settings().DISABLE_LOGIN_CHECKS:
         return True