feat: apply OEP-66 (queryset-scoping pattern) across 6 standardized APIs

openedx/core/djangoapps/enrollments/v2/tests/test_views.py

@@ -14,11 +14,12 @@
 
 from django.test import override_settings
 from django.urls import reverse
+from opaque_keys.edx.keys import CourseKey
 from rest_framework import status
 from rest_framework.test import APITestCase
 
-from common.djangoapps.student.tests.factories import AdminFactory, UserFactory
-from openedx.core.djangoapps.enrollments.v2.views import EnrollmentViewSet
+from common.djangoapps.student.tests.factories import AdminFactory, CourseEnrollmentFactory, UserFactory
+from openedx.core.djangoapps.enrollments.v2.views import AdminEnrollmentScopingPolicy, EnrollmentViewSet
 from openedx.core.djangolib.testing.utils import skip_unless_lms
 
 API_KEY = "test-enrollment-v2-api-key"
@@ -287,3 +288,79 @@ def test_minimal_view_collapses_course_details_to_course_id(self, mock_list, moc
         assert {r["course_id"] for r in response.data["results"]} == {
             "course-v1:org+a+r", "course-v1:org+b+r",
         }
+
+
+# ---------------------------------------------------------------------------
+# OEP-66 — EnrollmentsAdminListView (GET /enrollments/)
+# ---------------------------------------------------------------------------
+@skip_unless_lms
+class TestEnrollmentsAdminListView(APITestCase):
+    """
+    OEP-66 + ADR 0026/0033 — regression tests for the admin enrollment list.
+
+    This endpoint adopted ``ScopedQuerysetMixin`` and moved its filtering from
+    ``get_queryset()`` into ``filter_queryset()``. These tests guard that split:
+    endpoint access, the record-visibility pass-through policy, the user-driven
+    filters, the 400-on-invalid-params path (validation now runs in
+    ``filter_queryset``), and the ADR 0033 ``Deprecation`` header. Uses real
+    ``CourseEnrollment`` rows (SQL, MongoDB-free).
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.admin = AdminFactory.create(password="test")
+        self.user = UserFactory.create(password="test")
+        self.url = reverse("v2:enrollment-v2-admin-list")
+        self.course_a = CourseKey.from_string("course-v1:edX+A+run")
+        self.course_b = CourseKey.from_string("course-v1:edX+B+run")
+        self.learner_a = UserFactory.create()
+        self.learner_b = UserFactory.create()
+        CourseEnrollmentFactory.create(user=self.learner_a, course_id=self.course_a)
+        CourseEnrollmentFactory.create(user=self.learner_b, course_id=self.course_b)
+
+    def test_unauthenticated_gets_401(self):
+        assert self.client.get(self.url).status_code == status.HTTP_401_UNAUTHORIZED
+
+    def test_non_admin_gets_403(self):
+        """Endpoint-access layer: IsAdminUser rejects a regular user."""
+        self.client.force_authenticate(user=self.user)
+        assert self.client.get(self.url).status_code == status.HTTP_403_FORBIDDEN
+
+    def test_admin_sees_all_rows(self):
+        """Record-visibility layer is a pass-through: admin sees every enrollment."""
+        self.client.force_authenticate(user=self.admin)
+        response = self.client.get(self.url)
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["count"] == 2
+
+    def test_filter_by_course_key_narrows(self):
+        """User-driven filter (filter_queryset) still narrows by course_key."""
+        self.client.force_authenticate(user=self.admin)
+        response = self.client.get(self.url, {"course_key": str(self.course_a)})
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["count"] == 1
+
+    def test_filter_by_username_narrows(self):
+        """User-driven filter (filter_queryset) still narrows by username."""
+        self.client.force_authenticate(user=self.admin)
+        response = self.client.get(self.url, {"username": self.learner_b.username})
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["count"] == 1
+
+    def test_invalid_course_key_gets_400(self):
+        """Form validation moved to filter_queryset must still surface as a 400."""
+        self.client.force_authenticate(user=self.admin)
+        response = self.client.get(self.url, {"course_key": "not-a-course-key"})
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+    def test_legacy_course_id_emits_deprecation_header(self):
+        """ADR 0033: the legacy ``course_id`` alias still emits the Deprecation header."""
+        self.client.force_authenticate(user=self.admin)
+        response = self.client.get(self.url, {"course_id": str(self.course_a)})
+        assert response.status_code == status.HTTP_200_OK
+        assert "Deprecation" in response.headers
+
+    def test_scoping_policy_is_passthrough(self):
+        """AdminEnrollmentScopingPolicy returns the queryset unchanged (no narrowing today)."""
+        sentinel = object()
+        assert AdminEnrollmentScopingPolicy().scope(sentinel, self.admin) is sentinel

openedx/core/djangoapps/enrollments/v2/views.py

@@ -79,6 +79,7 @@
 from openedx.core.djangoapps.user_api.accounts.permissions import CanRetireUser
 from openedx.core.lib.api.mixins import StandardizedErrorMixin
 from openedx.core.lib.api.permissions import ApiKeyHeaderPermissionIsAuthenticated
+from openedx.core.lib.api.scoping import ScopedQuerysetMixin, ScopingPolicy
 
 log = logging.getLogger(__name__)
 User = get_user_model()
@@ -627,6 +628,29 @@ def get(self, request, course_id=None):
 # ===========================================================================
 # EnrollmentsAdminListView — GET /enrollments/  (admin paginated list)
 # ===========================================================================
+class AdminEnrollmentScopingPolicy(ScopingPolicy):
+    """
+    OEP-66 record-visibility policy for the admin enrollment list.
+
+    This endpoint is a *platform-administration* tool gated by
+    ``IsAdminUser`` (see the view's ``permission_classes``). A platform admin's
+    accessible scope is every enrollment, so the policy returns the queryset
+    unchanged — all rows are within scope. Encapsulating this as an explicit
+    :class:`~openedx.core.lib.api.scoping.ScopingPolicy` keeps the three OEP-66
+    layers separate (endpoint access vs. record visibility vs. user filtering)
+    and provides the single seam at which a *narrower* visibility rule would be
+    introduced — for example, restricting a delegated (org-scoped) admin to
+    enrollments in their organization's courses by translating an openedx-authz
+    ``get_scopes_for_subject_and_permission`` scope set into a
+    ``course_id__in=...`` / ``course__org__in=...`` filter. No such narrowing is
+    applied today, so behavior is unchanged.
+    """
+
+    def scope(self, queryset, user):
+        # Platform admins (IsAdminUser) see all enrollments; nothing to narrow.
+        return queryset
+
+
 @extend_schema(
     tags=["openedx-platform-sdk"],
     summary="List all course enrollments (admin-only, paginated)",
@@ -661,8 +685,24 @@ def get(self, request, course_id=None):
         403: _RESP_FORBIDDEN,
     },
 )
-class EnrollmentsAdminListView(StandardizedErrorMixin, ListAPIView):
-    """Admin-only paginated enrollment list with OEP-68 filter aliases."""
+class EnrollmentsAdminListView(ScopedQuerysetMixin, StandardizedErrorMixin, ListAPIView):
+    """
+    Admin-only paginated enrollment list with OEP-68 filter aliases.
+
+    OEP-66 – queryset-scoping pattern for list endpoints (PR openedx-proposals#802).
+    This is an ORM-backed list endpoint, so it wires the three authorization
+    layers cleanly:
+
+    * **Endpoint access** — ``permission_classes = (IsAdminUser,)``.
+    * **Record visibility (queryset scoping)** — ``ScopedQuerysetMixin`` applies
+      :class:`AdminEnrollmentScopingPolicy` to the base ``queryset`` in
+      ``get_queryset()``. (Currently a pass-through for platform admins; see the
+      policy docstring for how to narrow it via an openedx-authz scope set.)
+    * **User-driven filtering** — ``filter_queryset`` applies the
+      caller-supplied ``course_key`` / ``course_keys`` / ``username`` / ``email``
+      / ``ordering`` params (validated by ``EnrollmentsAdminListForm``). It runs
+      *after* the scoping layer and only narrows the already-authorized queryset.
+    """
 
     # ADR 0034 — JWT + cross-domain session (BearerAuthenticationAllowInactiveUser
     # removed per OEP-0042). EnrollmentCrossDomainSessionAuth retained because the
@@ -677,6 +717,11 @@ class EnrollmentsAdminListView(StandardizedErrorMixin, ListAPIView):
     serializer_class = CourseEnrollmentsApiListSerializer
     pagination_class = EnrollmentsAdminListPagination
 
+    # OEP-66 — base (unscoped) queryset and the record-visibility policy that
+    # ScopedQuerysetMixin.get_queryset() applies on top of it.
+    queryset = CourseEnrollment.objects.all().select_related("user", "course")
+    scoping_policy = AdminEnrollmentScopingPolicy()
+
     # ADR 0033 §3 — whitelist of allowed values for the ``ordering`` param.
     ALLOWED_ORDERING_FIELDS = frozenset({"created", "-created", "id", "-id"})
 
@@ -686,12 +731,16 @@ class EnrollmentsAdminListView(StandardizedErrorMixin, ListAPIView):
         ("course_ids", "course_keys"),
     )
 
-    def get_queryset(self):
+    def filter_queryset(self, queryset):
+        """
+        OEP-66 user-driven filtering layer. Narrows the already-scoped
+        ``queryset`` (from ``ScopedQuerysetMixin.get_queryset()``) by the
+        caller-supplied query parameters; never widens it.
+        """
         form = EnrollmentsAdminListForm(self.request.query_params)
         if not form.is_valid():
             raise ValidationError(form.errors)
 
-        queryset = CourseEnrollment.objects.all().select_related("user", "course")
         course_id = form.cleaned_data.get("course_id")
         course_ids = form.cleaned_data.get("course_ids")
         usernames = form.cleaned_data.get("username")

openedx/core/lib/api/scoping.py

@@ -0,0 +1,71 @@
+"""
+OEP-66 queryset-scoping building blocks for DRF list endpoints.
+
+OEP-66 ("Separating Authorization Concerns in List Endpoints",
+openedx-proposals#802) prescribes keeping three authorization concerns
+separate on a list endpoint, each handled by a dedicated layer:
+
+* **Endpoint access** — DRF ``permission_classes`` (may this user call the
+  endpoint at all?). Returns ``403`` when denied.
+* **Record visibility (queryset scoping)** — a :class:`ScopingPolicy` applied
+  in ``get_queryset()`` by :class:`ScopedQuerysetMixin` (which rows may this
+  user see?). Rows the user may not see are simply absent from the response;
+  their absence is never a ``403``.
+* **User-driven filtering** — a ``django-filter`` ``FilterSet`` (of the visible
+  rows, which did the user ask for?). Narrows an already-authorized queryset
+  and must never widen it.
+
+This is *application-level queryset scoping*, not database-enforced
+`row-level security`_: it only constrains queries that go through the view's
+scoped queryset. Code that queries the model directly is not protected by it.
+
+A policy resolves the subject's accessible scopes in **one bulk lookup**
+(see openedx-authz ``get_scopes_for_subject_and_permission`` /
+``get_scopes_for_user_and_permission``) and turns that scope set into a
+``WHERE`` clause, rather than fetching every row and running an ``enforce``-style
+check per object.
+
+.. _row-level security: https://www.postgresql.org/docs/current/ddl-rowsecurity.html
+"""
+
+from abc import ABC, abstractmethod
+
+from django.core.exceptions import ImproperlyConfigured
+
+
+class ScopingPolicy(ABC):
+    """
+    Scopes a queryset to the rows a user is permitted to see (OEP-66).
+
+    A policy must not re-implement access rules; it delegates to the platform's
+    authorization engine (e.g. openedx-authz) to resolve the subject's
+    accessible scopes and translates that answer into a queryset filter. It is
+    kept separate from the view so the same visibility rule can be reused and
+    unit-tested independently of any endpoint.
+    """
+
+    @abstractmethod
+    def scope(self, queryset, user):
+        """Return ``queryset`` filtered to the rows visible to ``user``."""
+
+
+class ScopedQuerysetMixin:
+    """
+    Applies :attr:`scoping_policy` to the base queryset of a DRF view (OEP-66).
+
+    Mix into a ``GenericAPIView`` / ``ListAPIView`` (or a ``GenericViewSet``)
+    and set :attr:`scoping_policy` to a :class:`ScopingPolicy` instance. The
+    mixin runs the policy on top of the view's ``get_queryset()`` result so the
+    ``list`` response only contains rows within the user's accessible scopes.
+    """
+
+    scoping_policy = None  # a ScopingPolicy instance
+
+    def get_queryset(self):
+        """Return the base queryset scoped to the rows the request user may see."""
+        queryset = super().get_queryset()
+        if self.scoping_policy is None:
+            raise ImproperlyConfigured(
+                f"{type(self).__name__} uses ScopedQuerysetMixin but does not set a scoping_policy"
+            )
+        return self.scoping_policy.scope(queryset, self.request.user)

openedx/core/lib/api/tests/test_scoping.py

@@ -0,0 +1,69 @@
+"""Tests for the OEP-66 queryset-scoping building blocks (``scoping.py``)."""
+from unittest import mock
+
+import pytest
+from django.core.exceptions import ImproperlyConfigured
+from django.test import RequestFactory, TestCase
+from rest_framework.generics import GenericAPIView
+
+from common.djangoapps.student.tests.factories import UserFactory
+from openedx.core.lib.api.scoping import ScopedQuerysetMixin, ScopingPolicy
+
+
+class _RecordingPolicy(ScopingPolicy):
+    """Test policy that records its ``scope`` call and returns a sentinel."""
+
+    def __init__(self, returns):
+        self.returns = returns
+        self.calls = []
+
+    def scope(self, queryset, user):
+        self.calls.append((queryset, user))
+        return self.returns
+
+
+class ScopingPolicyTests(TestCase):
+    """``ScopingPolicy`` is an abstract base requiring ``scope``."""
+
+    def test_cannot_instantiate_without_scope(self):
+        with pytest.raises(TypeError):
+            ScopingPolicy()  # pylint: disable=abstract-class-instantiated
+
+    def test_subclass_implementing_scope_is_usable(self):
+        policy = _RecordingPolicy(returns="scoped")
+        assert policy.scope("base", user=None) == "scoped"
+
+
+class ScopedQuerysetMixinTests(TestCase):
+    """``ScopedQuerysetMixin.get_queryset`` applies the policy on top of ``super()``."""
+
+    def setUp(self):
+        super().setUp()
+        self.user = UserFactory()
+        self.request = RequestFactory().get("/")
+        self.request.user = self.user
+
+    def _build_view(self, policy):
+        """Return a mixin-based view instance wired with ``policy`` and the test request."""
+        test_request = self.request
+
+        class _View(ScopedQuerysetMixin, GenericAPIView):
+            scoping_policy = policy
+            request = test_request
+
+        return _View()
+
+    def test_applies_policy_to_super_queryset(self):
+        policy = _RecordingPolicy(returns="SCOPED")
+        view = self._build_view(policy)
+        with mock.patch.object(GenericAPIView, "get_queryset", return_value="BASE"):
+            result = view.get_queryset()
+        assert result == "SCOPED"
+        # The policy was handed the base queryset and the request user.
+        assert policy.calls == [("BASE", self.user)]
+
+    def test_missing_policy_raises_improperly_configured(self):
+        view = self._build_view(policy=None)
+        with mock.patch.object(GenericAPIView, "get_queryset", return_value="BASE"):
+            with pytest.raises(ImproperlyConfigured):
+                view.get_queryset()