Limit page access to users which can access course

server/config/settings/base.py

@@ -515,7 +515,7 @@ if "django_redis.cache.RedisCache" in env("IT_DJANGO_CACHE_BACKEND", default="")
 
 CACHES["api_page_cache"] = {
     "BACKEND": "django.core.cache.backends.db.DatabaseCache",
-    "LOCATION": "django_cache_learning_path",
+    "LOCATION": "django_cache_table_api_page",
 }
 
 # OAuth/OpenId Connect

server/config/urls.py

@@ -21,7 +21,7 @@ from vbv_lernwelt.core.views import (
 )
 from vbv_lernwelt.course.views import (
     mark_course_completion,
-    page_api_view,
+    course_page_api_view,
     request_course_completion,
 )
 from wagtail import urls as wagtail_urls
@@ -59,7 +59,8 @@ urlpatterns = [
             name="generate_web_component_icons"),
 
     # course
-    path(r"api/course/page/<slug:slug>/", page_api_view, name="page_api_view"),
+    path(r"api/course/page/<slug:slug>/", course_page_api_view,
+         name="course_page_api_view"),
     path(r"api/course/completion/mark/", mark_course_completion,
          name="mark_course_completion"),
     path(r"api/course/completion/<course_id>/", request_course_completion,

server/vbv_lernwelt/core/utils.py

@@ -2,6 +2,7 @@ import logging
 
 import structlog
 from django.conf import settings
+from django.core.cache import caches
 from rest_framework.throttling import UserRateThrottle
 from structlog.types import EventDict
 
@@ -50,3 +51,16 @@ def first_true(iterable, default=False, pred=None):
     # first_true([a,b,c], x) --> a or b or c or x
     # first_true([a,b], x, f) --> a if f(a) else b if f(b) else x
     return next(filter(pred, iterable), default)
+
+
+def get_api_page_cache():
+    return caches["api_page_cache"]
+
+
+def api_page_cache_get_or_set(key, func, timeout=60 * 60 * 8):
+    cache = get_api_page_cache()
+    value = cache.get(key)
+    if value is None:
+        value = func()
+        cache.set(key, value, timeout=timeout)
+    return value

server/vbv_lernwelt/course/permissions.py

@@ -1,12 +1,21 @@
-from rest_framework import permissions
+from vbv_lernwelt.course.models import CourseSessionUser
 
 
-class CourseAccessPermission(permissions.BasePermission):
+def has_course_access(request, obj):
-    def has_object_permission(self, request, view, obj):
+    if request.user.is_superuser:
-        # Read permissions are allowed to any request,
-        # so we'll always allow GET, HEAD or OPTIONS requests.
-        if request.method in permissions.SAFE_METHODS:
         return True
 
-        # Instance must have an attribute named `owner`.
+    course = obj.specific.get_course()
-        return obj.owner == request.user
+
+    # attached to CourseSession
+    course_session = CourseSessionUser.objects.filter(
+        course_session__course_id=course.id, user=request.user
+    ).exists()
+
+    if course_session:
+        return True
+
+    # TODO is trainer/expert of session
+
+    # TODO check school class access
+    return False

server/vbv_lernwelt/course/tests/test_completion_api.py

@@ -14,8 +14,8 @@ class CourseCompletionApiTestCase(APITestCase):
     def setUp(self) -> None:
         create_default_users()
         create_test_course()
-        self.user = User.objects.get(username="student")
+        self.user = User.objects.get(username="admin")
-        self.client.login(username="student", password="test")
+        self.client.login(username="admin", password="test")
 
     def test_completeLearningContent_works(self):
         learning_content = LearningContent.objects.get(title="Fachcheck Fahrzeug")

server/vbv_lernwelt/course/views.py

@@ -1,10 +1,12 @@
 import structlog
-from django.views.decorators.cache import cache_page
 from rest_framework.decorators import api_view
+from rest_framework.exceptions import PermissionDenied
 from rest_framework.response import Response
 from wagtail.models import Page
 
-from vbv_lernwelt.course.models import CourseCompletion, CoursePage
+from vbv_lernwelt.core.utils import api_page_cache_get_or_set
+from vbv_lernwelt.course.models import CourseCompletion
+from vbv_lernwelt.course.permissions import has_course_access
 from vbv_lernwelt.course.serializers import CourseCompletionSerializer
 from vbv_lernwelt.learnpath.utils import get_wagtail_type
 
@@ -12,12 +14,20 @@ logger = structlog.get_logger(__name__)
 
 
 @api_view(["GET"])
-@cache_page(60 * 60 * 8, cache="api_page_cache")
+def course_page_api_view(request, slug):
-def page_api_view(request, slug):
     try:
         page = Page.objects.get(slug=slug, locale__language_code="de-CH")
-        serializer = page.specific.get_serializer_class()(page.specific)
+        if not has_course_access(request, page):
-        return Response(serializer.data)
+            raise PermissionDenied()
+
+        data = api_page_cache_get_or_set(
+            key=request.get_full_path(),
+            func=lambda: page.specific.get_serializer_class()(page.specific).data,
+        )
+
+        return Response(data)
+    except PermissionDenied as e:
+        raise e
     except Exception as e:
         logger.error(e)
         return Response({"error": str(e)}, status=404)
@@ -25,22 +35,32 @@ def page_api_view(request, slug):
 
 @api_view(["GET"])
 def request_course_completion(request, course_id):
+    try:
         response_data = CourseCompletionSerializer(
             CourseCompletion.objects.filter(user=request.user, course_id=course_id),
             many=True,
         ).data
 
         return Response(status=200, data=response_data)
+    except PermissionDenied as e:
+        raise e
+    except Exception as e:
+        logger.error(e)
+        return Response({"error": str(e)}, status=404)
 
 
 @api_view(["POST"])
 def mark_course_completion(request):
+    try:
         page_key = request.data.get("page_key")
         completion_status = request.data.get("completion_status", "success")
 
         page = Page.objects.get(translation_key=page_key, locale__language_code="de-CH")
+        if not has_course_access(request, page):
+            raise PermissionDenied()
+
         page_type = get_wagtail_type(page.specific)
-    course = CoursePage.objects.ancestor_of(page).first().specific.course
+        course = page.specific.get_course()
 
         cc, created = CourseCompletion.objects.get_or_create(
             user=request.user,
@@ -70,3 +90,8 @@ def mark_course_completion(request):
         )
 
         return Response(status=200, data=response_data)
+    except PermissionDenied as e:
+        raise e
+    except Exception as e:
+        logger.error(e)
+        return Response({"error": str(e)}, status=404)