fixup! feat: introduce authz_permission_required decorator

Author: dwong2708

cms/djangoapps/contentstore/api/tests/test_quality.py

@@ -79,14 +79,29 @@ def test_student_fails(self):
         self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
 
 
-@patch.object(core_toggles.AUTHZ_COURSE_AUTHORING_FLAG, "is_enabled", return_value=True)
 class CourseQualityAuthzTest(BaseCourseViewTest):
     """
     Tests Course Quality API authorization using openedx-authz.
     """
 
     view_name = "courses_api:course_quality"
 
+    @classmethod
+    def setUpClass(cls):
+        cls.toggle_patcher = patch.object(
+            core_toggles.AUTHZ_COURSE_AUTHORING_FLAG,
+            "is_enabled",
+            return_value=True
+        )
+        cls.toggle_patcher.start()
+
+        super().setUpClass()
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.toggle_patcher.stop()
+        super().tearDownClass()
+
     def setUp(self):
         super().setUp()
 
@@ -137,24 +152,24 @@ def _seed_database_with_policies(cls):
             target_enforcer=global_enforcer,
         )
 
-    def test_authorized_user_can_access(self, mock_flag):
+    def test_authorized_user_can_access(self):
         """User with COURSE_STAFF role can access."""
         resp = self.authorized_client.get(self.get_url(self.course_key))
         self.assertEqual(resp.status_code, status.HTTP_200_OK)
 
-    def test_unauthorized_user_cannot_access(self, mock_flag):
+    def test_unauthorized_user_cannot_access(self):
         """User without role cannot access."""
         resp = self.unauthorized_client.get(self.get_url(self.course_key))
         self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
 
-    def test_role_scoped_to_course(self, mock_flag):
+    def test_role_scoped_to_course(self):
         """Authorization should only apply to the assigned course."""
         other_course = self.store.create_course("OtherOrg", "OtherCourse", "Run", self.staff.id)
 
         resp = self.authorized_client.get(self.get_url(other_course.id))
         self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
 
-    def test_staff_user_allowed_via_legacy(self, mock_flag):
+    def test_staff_user_allowed_via_legacy(self):
         """
         Staff users should still pass through legacy fallback.
         """
@@ -163,7 +178,7 @@ def test_staff_user_allowed_via_legacy(self, mock_flag):
         resp = self.client.get(self.get_url(self.course_key))
         self.assertEqual(resp.status_code, status.HTTP_200_OK)
 
-    def test_superuser_allowed(self, mock_flag):
+    def test_superuser_allowed(self):
         """Superusers should always be allowed."""
         superuser = UserFactory(is_superuser=True)
 

cms/djangoapps/contentstore/api/tests/test_validation.py

@@ -284,14 +284,29 @@ def test_list_ready_to_update_reference_success(self, mock_block, mock_auth):
         mock_auth.assert_called_once()
 
 
-@patch.object(core_toggles.AUTHZ_COURSE_AUTHORING_FLAG, "is_enabled", return_value=True)
 class CourseValidationAuthzTest(BaseCourseViewTest):
     """
     Tests Course Validation API authorization using openedx-authz.
     """
 
     view_name = "courses_api:course_validation"
 
+    @classmethod
+    def setUpClass(cls):
+        cls.toggle_patcher = patch.object(
+            core_toggles.AUTHZ_COURSE_AUTHORING_FLAG,
+            "is_enabled",
+            return_value=True
+        )
+        cls.toggle_patcher.start()
+
+        super().setUpClass()
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.toggle_patcher.stop()
+        super().tearDownClass()
+
     def setUp(self):
         super().setUp()
 
@@ -342,23 +357,23 @@ def _seed_database_with_policies(cls):
             target_enforcer=global_enforcer,
         )
 
-    def test_authorized_user_can_access(self, mock_flag):
+    def test_authorized_user_can_access(self):
         """
         User with COURSE_STAFF role should be allowed via AuthZ.
         """
         resp = self.authorized_client.get(self.get_url(self.course_key))
 
         self.assertEqual(resp.status_code, status.HTTP_200_OK)
 
-    def test_unauthorized_user_cannot_access(self, mock_flag):
+    def test_unauthorized_user_cannot_access(self):
         """
         User without permissions should be denied.
         """
         resp = self.unauthorized_client.get(self.get_url(self.course_key))
 
         self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
 
-    def test_role_scoped_to_course(self, mock_flag):
+    def test_role_scoped_to_course(self):
         """
         Authorization should only apply to the assigned course scope.
         """
@@ -373,7 +388,7 @@ def test_role_scoped_to_course(self, mock_flag):
 
         self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
 
-    def test_staff_user_allowed_via_legacy(self, mock_flag):
+    def test_staff_user_allowed_via_legacy(self):
         """
         Course staff should pass through legacy fallback when AuthZ denies.
         """
@@ -383,7 +398,7 @@ def test_staff_user_allowed_via_legacy(self, mock_flag):
 
         self.assertEqual(resp.status_code, status.HTTP_200_OK)
 
-    def test_superuser_allowed(self, mock_flag):
+    def test_superuser_allowed(self):
         """
         Superusers should always be allowed through legacy fallback.
         """

openedx/core/djangoapps/authz/decorators.py

@@ -7,13 +7,19 @@
 from openedx_authz import api as authz_api
 from rest_framework import status
 
-from common.djangoapps.student.auth import has_course_author_access
+from common.djangoapps.student.auth import has_studio_write_access, has_studio_read_access
 from openedx.core import toggles as core_toggles
 from openedx.core.lib.api.view_utils import DeveloperErrorViewMixin
 
 log = logging.getLogger(__name__)
 
 
+legacy_permission_handler_map = {
+    "read": has_studio_read_access,
+    "write": has_studio_write_access,
+}
+
+
 def authz_permission_required(authz_permission, legacy_permission=None):
     """
     Decorator enforcing course author permissions via AuthZ
@@ -49,20 +55,23 @@ def user_has_course_permission(user, authz_permission, course_key, legacy_permis
     """
     Core authorization logic.
     """
+    if core_toggles.enable_authz_course_authoring(course_key):
+        # If AuthZ is enabled for this course, check the permission via AuthZ only.
+        is_user_allowed = authz_api.is_user_allowed(user.username, authz_permission, str(course_key))
+        log.info(
+            "AuthZ permission granted = {}".format(is_user_allowed),
+            extra={
+                "user_id": user.id,
+                "authz_permission": authz_permission,
+                "course_key": str(course_key),
+            },
+        )
+        return is_user_allowed
 
-    if is_authz_enabled(course_key):
-        if authz_api.is_user_allowed(user.username, authz_permission, str(course_key)):
-            log.info(
-                "AuthZ permission granted",
-                extra={
-                    "user_id": user.id,
-                    "authz_permission": authz_permission,
-                    "course_key": str(course_key),
-                },
-            )
-            return True
-
-    if legacy_permission and has_course_author_access(user, course_key):
+    # If AuthZ is not enabled for this course, fall back to legacy course author
+    # access check if legacy_permission is provided.
+    has_legacy_permission = legacy_permission_handler_map.get(legacy_permission)
+    if legacy_permission and has_legacy_permission and has_legacy_permission(user, course_key):
         log.info(
             "AuthZ fallback used",
             extra={
@@ -85,11 +94,6 @@ def user_has_course_permission(user, authz_permission, course_key, legacy_permis
     return False
 
 
-def is_authz_enabled(course_key):
-    """Check if AuthZ is enabled for this course."""
-    return core_toggles.AUTHZ_COURSE_AUTHORING_FLAG.is_enabled(course_key)
-
-
 def get_course_key(course_id):
     """
     Given a course_id string, attempts to parse it as a CourseKey.

openedx/core/djangoapps/authz/models.py

@@ -49,20 +49,20 @@ def test_view_executes_when_permission_granted(self):
             self.course_key,
         )
 
-    def test_view_executes_when_legacy_fallback(self):
+    def test_view_executes_when_legacy_fallback_read(self):
         """Decorator allows execution when AuthZ denies but legacy permission succeeds."""
         request = self._build_request()
 
         mock_view = Mock(return_value="success")
 
         with patch(
-            "openedx.core.djangoapps.authz.decorators.is_authz_enabled",
-            return_value=True,
+            "openedx.core.djangoapps.authz.decorators.core_toggles.enable_authz_course_authoring",
+            return_value=False,
         ), patch(
             "openedx.core.djangoapps.authz.decorators.authz_api.is_user_allowed",
-            return_value=False,
+            return_value=True,  # Should not be used when AuthZ is disabled, but set to True just in case
         ), patch(
-            "openedx.core.djangoapps.authz.decorators.has_course_author_access",
+            "openedx.core.djangoapps.authz.decorators.has_studio_read_access",
             return_value=True,
         ):
             decorated = authz_permission_required(
@@ -75,6 +75,32 @@ def test_view_executes_when_legacy_fallback(self):
         self.assertEqual(result, "success")
         mock_view.assert_called_once()
 
+    def test_view_executes_when_legacy_fallback_write(self):
+        """Decorator allows execution when AuthZ denies but legacy write permission succeeds."""
+        request = self._build_request()
+
+        mock_view = Mock(return_value="success")
+
+        with patch(
+            "openedx.core.djangoapps.authz.decorators.core_toggles.enable_authz_course_authoring",
+            return_value=False,
+        ), patch(
+            "openedx.core.djangoapps.authz.decorators.authz_api.is_user_allowed",
+            return_value=True,  # Should not be used when AuthZ is disabled, but set to True just in case
+        ), patch(
+            "openedx.core.djangoapps.authz.decorators.has_studio_write_access",
+            return_value=True,
+        ):
+            decorated = authz_permission_required(
+                "courses.edit",
+                legacy_permission="write"
+            )(mock_view)
+
+            result = decorated(self.view_instance, request, str(self.course_key))
+
+        self.assertEqual(result, "success")
+        mock_view.assert_called_once()
+
     def test_access_denied_when_permission_fails(self):
         """Decorator raises API error when permission fails."""
         request = self._build_request()