feat: libraries v2 support for studio perms XBlock service

cms/djangoapps/contentstore/utils.py

@@ -53,7 +53,7 @@
 from common.djangoapps.course_modes.models import CourseMode
 from common.djangoapps.edxmako.services import MakoService
 from common.djangoapps.student import auth
-from common.djangoapps.student.auth import STUDIO_EDIT_ROLES, has_studio_read_access, has_studio_write_access
+from common.djangoapps.student.auth import STUDIO_EDIT_ROLES, has_studio_write_access
 from common.djangoapps.student.models import CourseEnrollment
 from common.djangoapps.student.roles import CourseInstructorRole, CourseStaffRole, GlobalStaff
 from common.djangoapps.track import contexts
@@ -71,6 +71,7 @@
 from common.djangoapps.xblock_django.user_service import DjangoXBlockUserService
 from openedx.core import toggles as core_toggles
 from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
+from openedx.core.djangoapps.content.services import StudioPermissionsService
 from openedx.core.djangoapps.content_libraries.api import get_container
 from openedx.core.djangoapps.content_tagging.toggles import is_tagging_feature_disabled
 from openedx.core.djangoapps.credit.api import get_credit_requirements, is_credit_course
@@ -2243,27 +2244,6 @@ def get_group_configurations_context(course, store):
     return context
 
 
-class StudioPermissionsService:
-    """
-    Service that can provide information about a user's permissions.
-
-    Deprecated. To be replaced by a more general authorization service.
-
-    Only used by LegacyLibraryContentBlock (and library_tools.py).
-    """
-
-    def __init__(self, user):
-        self._user = user
-
-    def can_read(self, course_key):
-        """ Does the user have read access to the given course/library? """
-        return has_studio_read_access(self._user, course_key)
-
-    def can_write(self, course_key):
-        """ Does the user have read access to the given course/library? """
-        return has_studio_write_access(self._user, course_key)
-
-
 def track_course_update_event(course_key, user, course_update_content=None):
     """
     Track course update event

cms/djangoapps/contentstore/views/preview.py

@@ -29,6 +29,7 @@
 from common.djangoapps.student.models import anonymous_id_for_user
 from common.djangoapps.xblock_django.user_service import DjangoXBlockUserService
 from lms.djangoapps.lms_xblock.field_data import LmsFieldData
+from openedx.core.djangoapps.content.services import StudioPermissionsService
 from openedx.core.djangoapps.discussions.services import DiscussionConfigService
 from openedx.core.djangoapps.video_config.services import VideoConfigService
 from openedx.core.lib.cache_utils import CacheService
@@ -44,7 +45,7 @@
 from xmodule.util.sandboxing import SandboxService
 from xmodule.x_module import AUTHOR_VIEW, PREVIEW_VIEWS, STUDENT_VIEW, XModuleMixin
 
-from ..utils import StudioPermissionsService, get_visibility_partition_info
+from ..utils import get_visibility_partition_info
 from .access import get_user_role
 from .session_kv_store import SessionKeyValueStore
 

openedx/core/djangoapps/authz/tests/fixtures.py

@@ -0,0 +1,26 @@
+""" Fixtures for AuthZ-aware tests """
+import casbin
+import pkg_resources
+from openedx_authz.engine.enforcer import AuthzEnforcer
+from openedx_authz.engine.utils import migrate_policy_between_enforcers
+
+
+def seed_policies():
+    """Seed the database with AuthZ policies."""
+    global_enforcer = AuthzEnforcer.get_enforcer()
+    global_enforcer.load_policy()
+
+    model_path = pkg_resources.resource_filename(
+        "openedx_authz.engine",
+        "config/model.conf",
+    )
+
+    policy_path = pkg_resources.resource_filename(
+        "openedx_authz.engine",
+        "config/authz.policy",
+    )
+
+    migrate_policy_between_enforcers(
+        source_enforcer=casbin.Enforcer(model_path, policy_path),
+        target_enforcer=global_enforcer,
+    )

openedx/core/djangoapps/authz/tests/mixins.py

@@ -2,16 +2,14 @@
 
 from unittest.mock import patch
 
-import casbin
-import pkg_resources
 from openedx_authz.api.users import assign_role_to_user_in_scope
 from openedx_authz.constants.roles import COURSE_STAFF
 from openedx_authz.engine.enforcer import AuthzEnforcer
-from openedx_authz.engine.utils import migrate_policy_between_enforcers
 from rest_framework.test import APIClient
 
 from common.djangoapps.student.tests.factories import UserFactory
 from openedx.core import toggles as core_toggles
+from openedx.core.djangoapps.authz.tests.fixtures import seed_policies
 
 
 class CourseAuthoringAuthzTestMixin:
@@ -44,7 +42,7 @@ def tearDownClass(cls):
     def setUp(self):
         super().setUp()
 
-        self._seed_policies()
+        seed_policies()
 
         self.authorized_user = UserFactory(password=self.password)
         self.unauthorized_user = UserFactory(password=self.password)
@@ -76,27 +74,6 @@ def add_user_to_role_in_course(self, user, role, course_key):
         )
         AuthzEnforcer.get_enforcer().load_policy()
 
-    @classmethod
-    def _seed_policies(cls):
-        """Seed the database with AuthZ policies."""
-        global_enforcer = AuthzEnforcer.get_enforcer()
-        global_enforcer.load_policy()
-
-        model_path = pkg_resources.resource_filename(
-            "openedx_authz.engine",
-            "config/model.conf",
-        )
-
-        policy_path = pkg_resources.resource_filename(
-            "openedx_authz.engine",
-            "config/authz.policy",
-        )
-
-        migrate_policy_between_enforcers(
-            source_enforcer=casbin.Enforcer(model_path, policy_path),
-            target_enforcer=global_enforcer,
-        )
-
 
 class CourseAuthzTestMixin(CourseAuthoringAuthzTestMixin):
     """

openedx/core/djangoapps/content/services.py

@@ -0,0 +1,39 @@
+"""
+Services for learning content
+"""
+from __future__ import annotations
+
+from opaque_keys.edx.locator import LibraryLocatorV2
+from openedx_authz import api as authz_api
+from openedx_authz.constants.permissions import EDIT_LIBRARY_CONTENT, VIEW_LIBRARY
+
+from common.djangoapps.student.auth import has_studio_read_access, has_studio_write_access
+
+
+class StudioPermissionsService:
+    """
+    Service that can provide information about a user's permissions.
+    """
+
+    def __init__(self, user):
+        self._user = user
+
+    def can_read(self, context_key):
+        """ Does the user have read access to the given course/library? """
+        if isinstance(context_key, LibraryLocatorV2):
+            return self._user.is_active and authz_api.is_user_allowed(
+                self._user,
+                VIEW_LIBRARY.identifier,
+                str(context_key),
+            )
+        return has_studio_read_access(self._user, context_key)
+
+    def can_write(self, context_key):
+        """ Does the user have write access to the given course/library? """
+        if isinstance(context_key, LibraryLocatorV2):
+            return self._user.is_active and authz_api.is_user_allowed(
+                self._user,
+                EDIT_LIBRARY_CONTENT.identifier,
+                str(context_key),
+            )
+        return has_studio_write_access(self._user, context_key)

openedx/core/djangoapps/content/tests/test_services.py

@@ -0,0 +1,103 @@
+"""
+Tests for content XBlock Services
+"""
+from django.contrib.auth import get_user_model
+from django.test import TransactionTestCase
+from opaque_keys.edx.locator import LibraryLocatorV2
+from organizations.models import Organization
+
+from common.djangoapps.student.auth import update_org_role
+from common.djangoapps.student.roles import OrgStaffRole
+from common.djangoapps.student.tests.factories import UserFactory
+from openedx.core.djangoapps.authz.tests.fixtures import seed_policies
+from openedx.core.djangoapps.content.services import StudioPermissionsService
+from openedx.core.djangoapps.content_libraries.api import (
+    AccessLevel,
+    ContentLibraryMetadata,
+    assign_library_role_to_user,
+    create_library,
+)
+from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
+from xmodule.modulestore.tests.factories import CourseFactory
+
+User = get_user_model()
+
+
+class StudioPermissionsServiceTestCase(ModuleStoreTestCase, TransactionTestCase):
+    """
+    Test the studio permissions service.
+    """
+
+    def setUp(self) -> None:
+        super().setUp()
+        seed_policies()
+        self.org = Organization.objects.create(name="Organization A", short_name="orgA")
+        self.staff = UserFactory.create(
+            is_staff=True,
+        )
+
+    def _create_privileged_org_user(self) -> User:
+        user = UserFactory.create()
+        update_org_role(self.staff, OrgStaffRole, user, [self.org.short_name])
+        return user
+
+    def test_user_can_read_course(self) -> None:
+        course = CourseFactory.create(org=self.org.short_name)
+        user = self._create_privileged_org_user()
+        service = StudioPermissionsService(user=user)
+        assert service.can_read(course.location)
+
+    def test_user_can_write_course(self) -> None:
+        course = CourseFactory.create(org=self.org.short_name)
+        user = self._create_privileged_org_user()
+        service = StudioPermissionsService(user=user)
+        assert service.can_write(course.location)
+
+    def test_user_cannot_read_course(self) -> None:
+        course = CourseFactory.create(org=self.org.short_name)
+        user = UserFactory.create()
+        service = StudioPermissionsService(user=user)
+        assert not service.can_read(course.location)
+
+    def test_user_cannot_write_course(self) -> None:
+        course = CourseFactory.create(org=self.org.short_name)
+        user = UserFactory.create()
+        service = StudioPermissionsService(user=user)
+        assert not service.can_write(course.location)
+
+    def _create_library(self) -> ContentLibraryMetadata:
+        return create_library(
+            org=self.org,
+            slug="lib",
+            title="Library Org",
+            description="This is a library from Org",
+        )
+
+    def _create_privileged_library_user(self, library_key: LibraryLocatorV2) -> User:
+        user = UserFactory.create()
+        assign_library_role_to_user(library_key, user, AccessLevel.ADMIN_LEVEL)
+        return user
+
+    def test_user_can_read_library(self) -> None:
+        library = self._create_library()
+        user = self._create_privileged_library_user(library.key)
+        service = StudioPermissionsService(user=user)
+        assert service.can_read(library.key)
+
+    def test_user_can_write_library(self) -> None:
+        library = self._create_library()
+        user = self._create_privileged_library_user(library.key)
+        service = StudioPermissionsService(user=user)
+        assert service.can_write(library.key)
+
+    def test_user_cannot_read_library(self) -> None:
+        library = self._create_library()
+        user = UserFactory.create()
+        service = StudioPermissionsService(user=user)
+        assert not service.can_read(library.key)
+
+    def test_user_cannot_write_library(self) -> None:
+        library = self._create_library()
+        user = UserFactory.create()
+        service = StudioPermissionsService(user=user)
+        assert not service.can_write(library.key)

openedx/core/djangoapps/content_libraries/api/libraries.py

@@ -415,7 +415,7 @@ def get_library(library_key: LibraryLocatorV2) -> ContentLibraryMetadata:
 
 
 def create_library(
-    org: str,
+    org: Organization,
     slug: str,
     title: str,
     description: str = "",

openedx/core/djangoapps/xblock/runtime/runtime.py

@@ -355,6 +355,13 @@ def service(self, block: XBlock, service_name: str):
             return DiscussionConfigService()
         elif service_name == 'xqueue':
             return XQueueService(block)
+        elif service_name == 'studio_user_permissions':
+            from openedx.core.djangoapps.content.services import StudioPermissionsService
+            if self.user is None:
+                raise RuntimeError(
+                    "Cannot access studio permissions service when there is no user bound to the XBlock."
+                )
+            return StudioPermissionsService(self.user)
 
         # Otherwise, fall back to the base implementation which loads services
         # defined in the constructor: