[Backport FC-0099] feat: add openedx-authz and update libraries enforcement points

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

@@ -53,13 +53,11 @@
 from django.db import IntegrityError, transaction
 from django.db.models import Q, QuerySet
 from django.utils.translation import gettext as _
-from opaque_keys.edx.locator import (
-    LibraryLocatorV2,
-    LibraryUsageLocatorV2,
-)
-from openedx_events.content_authoring.data import (
-    ContentLibraryData,
-)
+from opaque_keys.edx.locator import LibraryLocatorV2, LibraryUsageLocatorV2
+from openedx_authz import api as authz_api
+from openedx_authz.api import assign_role_to_user_in_scope
+from openedx_authz.constants import permissions as authz_permissions
+from openedx_events.content_authoring.data import ContentLibraryData
 from openedx_events.content_authoring.signals import (
     CONTENT_LIBRARY_CREATED,
     CONTENT_LIBRARY_DELETED,
@@ -70,14 +68,14 @@
 from organizations.models import Organization
 from user_tasks.models import UserTaskArtifact, UserTaskStatus
 from xblock.core import XBlock
-from openedx_authz.api import assign_role_to_user_in_scope
 
 from openedx.core.types import User as UserType
 
 from .. import permissions, tasks
 from ..constants import ALL_RIGHTS_RESERVED
 from ..models import ContentLibrary, ContentLibraryPermission
 from .exceptions import LibraryAlreadyExists, LibraryPermissionIntegrityError
+from .permissions import LEGACY_LIB_PERMISSIONS
 
 log = logging.getLogger(__name__)
 
@@ -109,6 +107,7 @@
     "revert_changes",
     "get_backup_task_status",
     "assign_library_role_to_user",
+    "user_has_permission_across_lib_authz_systems",
 ]
 
 
@@ -245,7 +244,18 @@ def user_can_create_library(user: AbstractUser) -> bool:
     """
     Check if the user has permission to create a content library.
     """
-    return user.has_perm(permissions.CAN_CREATE_CONTENT_LIBRARY)
+    library_permission = permissions.CAN_CREATE_CONTENT_LIBRARY
+    lib_permission_in_authz = _transform_legacy_lib_permission_to_authz_permission(library_permission)
+    # The authz_api.is_user_allowed check only validates permissions within a specific library context. Since
+    # creating a library is not tied to an existing one, we use user.has_perm (via Bridgekeeper) to check if the user
+    # can create libraries, meaning they have the course creator role. In the future, this should rely on a global (*)
+    # role defined in the Authorization Framework for instance-level resource creation.
+    has_perms = user.has_perm(library_permission) or authz_api.is_user_allowed(
+        user,
+        lib_permission_in_authz,
+        authz_api.data.GLOBAL_SCOPE_WILDCARD,
+    )
+    return has_perms
 
 
 def get_libraries_for_user(user, org=None, text_search=None, order=None) -> QuerySet[ContentLibrary]:
@@ -267,7 +277,11 @@ def get_libraries_for_user(user, org=None, text_search=None, order=None) -> Quer
             Q(learning_package__description__icontains=text_search)
         )
 
-    filtered = permissions.perms[permissions.CAN_VIEW_THIS_CONTENT_LIBRARY].filter(user, qs)
+    # Using distinct() temporarily to avoid duplicate results caused by overlapping permission checks
+    # between Bridgekeeper and the new authorization framework. This ensures correct results for now,
+    # but it should be removed once Bridgekeeper support is fully dropped and all permission logic
+    # is handled through openedx-authz.
+    filtered = permissions.perms[permissions.CAN_VIEW_THIS_CONTENT_LIBRARY].filter(user, qs).distinct()
 
     if order:
         order_query = 'learning_package__'
@@ -332,7 +346,7 @@ def require_permission_for_library_key(library_key: LibraryLocatorV2, user: User
     library_obj = ContentLibrary.objects.get_by_key(library_key)
     # obj should be able to read any valid model object but mypy thinks it can only be
     # "User | AnonymousUser | None"
-    if not user.has_perm(permission, obj=library_obj):  # type:ignore[arg-type]
+    if not user_has_permission_across_lib_authz_systems(user, permission, library_obj):
         raise PermissionDenied
 
     return library_obj
@@ -750,3 +764,90 @@ def get_backup_task_status(
         result['file'] = artifact.file
 
     return result
+
+
+def _transform_legacy_lib_permission_to_authz_permission(permission: str) -> str:
+    """
+    Transform a legacy content library permission to an openedx-authz permission.
+    """
+    # There is no dedicated permission or role for can_create_content_library in openedx-authz yet,
+    # so we reuse the same permission to rely on user.has_perm via Bridgekeeper.
+    return {
+        permissions.CAN_CREATE_CONTENT_LIBRARY: permissions.CAN_CREATE_CONTENT_LIBRARY,
+        permissions.CAN_DELETE_THIS_CONTENT_LIBRARY: authz_permissions.DELETE_LIBRARY.identifier,
+        permissions.CAN_EDIT_THIS_CONTENT_LIBRARY: authz_permissions.EDIT_LIBRARY_CONTENT.identifier,
+        permissions.CAN_EDIT_THIS_CONTENT_LIBRARY_TEAM: authz_permissions.MANAGE_LIBRARY_TEAM.identifier,
+        permissions.CAN_VIEW_THIS_CONTENT_LIBRARY: authz_permissions.VIEW_LIBRARY.identifier,
+        permissions.CAN_VIEW_THIS_CONTENT_LIBRARY_TEAM: authz_permissions.VIEW_LIBRARY_TEAM.identifier,
+    }.get(permission, permission)
+
+
+def _transform_authz_permission_to_legacy_lib_permission(permission: str) -> str:
+    """
+    Transform an openedx-authz permission to a legacy content library permission.
+    """
+    return {
+        authz_permissions.PUBLISH_LIBRARY_CONTENT.identifier: permissions.CAN_EDIT_THIS_CONTENT_LIBRARY,
+        authz_permissions.CREATE_LIBRARY_COLLECTION.identifier: permissions.CAN_EDIT_THIS_CONTENT_LIBRARY,
+        authz_permissions.EDIT_LIBRARY_COLLECTION.identifier: permissions.CAN_EDIT_THIS_CONTENT_LIBRARY,
+        authz_permissions.DELETE_LIBRARY_COLLECTION.identifier: permissions.CAN_EDIT_THIS_CONTENT_LIBRARY,
+    }.get(permission, permission)
+
+
+def user_has_permission_across_lib_authz_systems(
+    user: UserType,
+    permission: str | authz_api.data.PermissionData,
+    library_obj: ContentLibrary,
+) -> bool:
+    """
+    Check whether a user has a given permission on a content library across both the
+    legacy edx-platform permission system and the newer openedx-authz system.
+
+    The provided permission name is normalized to both systems (legacy and authz), and
+    authorization is granted if either:
+    - the user holds the legacy object-level permission on the ContentLibrary instance, or
+    - the openedx-authz API allows the user for the corresponding permission on the library.
+
+    **Note:**
+    Temporary: this function uses Bridgekeeper-based logic for cases not yet modeled in openedx-authz.
+
+    Current gaps covered here:
+    - CAN_CREATE_CONTENT_LIBRARY: we call user.has_perm via Bridgekeeper to verify the user is a course creator.
+    - CAN_VIEW_THIS_CONTENT_LIBRARY: we respect the allow_public_read flag via Bridgekeeper.
+
+    Replace these with authz_api.is_user_allowed once openedx-authz supports
+    these conditions natively (including global (*) roles).
+
+    Args:
+        user: The Django user (or user-like object) to check.
+        permission: The permission identifier (either a legacy codename or an openedx-authz name).
+        library_obj: The ContentLibrary instance to check against.
+
+    Returns:
+        bool: True if the user is authorized by either system; otherwise False.
+    """
+    if isinstance(permission, authz_api.data.PermissionData):
+        permission = permission.identifier
+    if _is_legacy_permission(permission):
+        legacy_permission = permission
+        authz_permission = _transform_legacy_lib_permission_to_authz_permission(permission)
+    else:
+        authz_permission = permission
+        legacy_permission = _transform_authz_permission_to_legacy_lib_permission(permission)
+    return (
+        # Check both the legacy and the new openedx-authz permissions
+        user.has_perm(perm=legacy_permission, obj=library_obj)
+        or authz_api.is_user_allowed(
+            user,
+            authz_permission,
+            str(library_obj.library_key),
+        )
+    )
+
+
+def _is_legacy_permission(permission: str) -> bool:
+    """
+    Determine if the specified library permission is part of the legacy
+    or the new openedx-authz system.
+    """
+    return permission in LEGACY_LIB_PERMISSIONS

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

@@ -12,3 +12,13 @@
     CAN_VIEW_THIS_CONTENT_LIBRARY,
     CAN_VIEW_THIS_CONTENT_LIBRARY_TEAM
 )
+
+LEGACY_LIB_PERMISSIONS = frozenset({
+    CAN_CREATE_CONTENT_LIBRARY,
+    CAN_DELETE_THIS_CONTENT_LIBRARY,
+    CAN_EDIT_THIS_CONTENT_LIBRARY,
+    CAN_EDIT_THIS_CONTENT_LIBRARY_TEAM,
+    CAN_LEARN_FROM_THIS_CONTENT_LIBRARY,
+    CAN_VIEW_THIS_CONTENT_LIBRARY,
+    CAN_VIEW_THIS_CONTENT_LIBRARY_TEAM,
+})

openedx/core/djangoapps/content_libraries/permissions.py

@@ -2,8 +2,12 @@
 Permissions for Content Libraries (v2, Learning-Core-based)
 """
 from bridgekeeper import perms, rules
-from bridgekeeper.rules import Attribute, ManyRelation, Relation, blanket_rule, in_current_groups
+from bridgekeeper.rules import Attribute, ManyRelation, Relation, blanket_rule, in_current_groups, Rule
 from django.conf import settings
+from django.db.models import Q
+
+from openedx_authz import api as authz_api
+from openedx_authz.constants.permissions import VIEW_LIBRARY
 
 from openedx.core.djangoapps.content_libraries.models import ContentLibraryPermission
 
@@ -54,6 +58,154 @@ def is_course_creator(user):
 
     return get_course_creator_status(user) == 'granted'
 
+
+class HasPermissionInContentLibraryScope(Rule):
+    """Bridgekeeper rule that checks content library permissions via the openedx-authz system.
+
+    This rule integrates the openedx-authz authorization system (backed by Casbin) with
+    Bridgekeeper's declarative permission system. It checks if a user has been granted a
+    specific permission (action) through their role assignments in the authorization system.
+
+    The rule works by:
+    1. Querying the authorization system to find library scopes where the user has this permission
+    2. Parsing the library keys (org/slug) from the scopes
+    3. Building database filters to match ContentLibrary models with those org/slug combinations
+
+    Attributes:
+        permission (PermissionData): The permission object representing the action to check
+            (e.g., 'view', 'edit'). This is used to look up scopes in the authorization system.
+
+        filter_keys (list[str]): The Django model fields to use when building QuerySet filters.
+            Defaults to ['org', 'slug'] for ContentLibrary models.
+
+            These fields are used to construct the Q object filters that match libraries
+            based on the parsed components from library keys in authorization scopes.
+
+            For ContentLibrary, library keys have the format 'lib:ORG:SLUG', which maps to:
+            - 'org' -> filters on org__short_name (related Organization model)
+            - 'slug' -> filters on slug field
+
+            If filtering by different fields is needed, pass a custom list. For example:
+            - ['org', 'slug'] - default for ContentLibrary (filters by org and slug)
+            - ['id'] - filter by primary key (for other models)
+
+    Examples:
+        Basic usage with default filter_keys:
+            >>> from bridgekeeper import perms
+            >>> from openedx.core.djangoapps.content_libraries.permissions import HasPermissionInContentLibraryScope
+            >>>
+            >>> # Uses default filter_keys=['org', 'slug'] for ContentLibrary
+            >>> can_view = HasPermissionInContentLibraryScope('view_library')
+            >>> perms['libraries.view_library'] = can_view
+
+        Compound permissions with boolean operators:
+            >>> from bridgekeeper.rules import Attribute
+            >>>
+            >>> is_active = Attribute('is_active', True)
+            >>> is_staff = Attribute('is_staff', True)
+            >>> can_view = HasPermissionInContentLibraryScope('view_library')
+            >>>
+            >>> # User must be active AND (staff OR have explicit permission)
+            >>> perms['libraries.view_library'] = is_active & (is_staff | can_view)
+
+        QuerySet filtering (efficient, database-level):
+            >>> from openedx.core.djangoapps.content_libraries.models import ContentLibrary
+            >>>
+            >>> # Gets all libraries user can view in a single SQL query
+            >>> visible_libraries = perms['libraries.view_library'].filter(
+            ...     request.user,
+            ...     ContentLibrary.objects.all()
+            ... )
+
+        Individual object checks:
+            >>> library = ContentLibrary.objects.get(org__short_name='DemoX', slug='CSPROB')
+            >>> if perms['libraries.view_library'].check(request.user, library):
+            ...     # User can view this specific library
+
+    Note:
+        The library keys in authorization scopes must have the format 'lib:ORG:SLUG'
+        to match the ContentLibrary model's org.short_name and slug fields.
+        For example, scope 'lib:DemoX:CSPROB' matches a library with
+        org.short_name='DemoX' and slug='CSPROB'.
+    """
+
+    def __init__(self, permission: authz_api.PermissionData, filter_keys: list[str] | None = None):
+        """Initialize the rule with the action and filter keys to filter on.
+
+        Args:
+            permission (PermissionData): The permission to check (e.g., 'view', 'edit').
+            filter_keys (list[str]): The model fields to filter on when building QuerySet filters.
+                Defaults to ['org', 'slug'] for ContentLibrary.
+        """
+        self.permission = permission
+        self.filter_keys = filter_keys if filter_keys is not None else ["org", "slug"]
+
+    def query(self, user):
+        """Convert this rule to a Django Q object for QuerySet filtering.
+
+        Args:
+            user: The Django user object (must have a 'username' attribute).
+
+        Returns:
+            Q: A Django Q object that can be used to filter a QuerySet.
+               The Q object combines multiple conditions using OR (|) operators,
+               where each condition matches a library's org and slug fields:
+               Q(org__short_name='OrgA' & slug='lib-a') | Q(org__short_name='OrgB' & slug='lib-b')
+
+        Example:
+            >>> # User has 'view' permission in scopes: ['lib:OrgA:lib-a', 'lib:OrgB:lib-b']
+            >>> rule = HasPermissionInContentLibraryScope('view', filter_keys=['org', 'slug'])
+            >>> q = rule.query(user)
+            >>> # Results in: Q(org__short_name='OrgA', slug='lib-a') | Q(org__short_name='OrgB', slug='lib-b')
+            >>>
+            >>> # Apply to queryset
+            >>> libraries = ContentLibrary.objects.filter(q)
+            >>> # SQL: SELECT * FROM content_library
+            >>> #      WHERE (org.short_name='OrgA' AND slug='lib-a')
+            >>> #         OR (org.short_name='OrgB' AND slug='lib-b')
+        """
+        scopes = authz_api.get_scopes_for_user_and_permission(
+            user.username,
+            self.permission.identifier
+        )
+
+        library_keys = [scope.library_key for scope in scopes]
+
+        if not library_keys:
+            return Q(pk__in=[])  # No access, return Q that matches nothing
+
+        # Build Q object: OR together (org AND slug) conditions for each library
+        query = Q()
+        for library_key in library_keys:
+            query |= Q(org__short_name=library_key.org, slug=library_key.slug)
+
+        return query
+
+    def check(self, user, instance, *args, **kwargs):  # pylint: disable=arguments-differ
+        """Check if user has permission for a specific object instance.
+
+        This method is used for checking permission on individual objects rather
+        than filtering a QuerySet. It extracts the scope from the object and
+        checks if the user has the required permission in that scope via Casbin.
+
+        Args:
+            user: The Django user object (must have a 'username' attribute).
+            instance: The Django model instance to check permission for.
+            *args: Additional positional arguments (for compatibility with parent signature).
+            **kwargs: Additional keyword arguments (for compatibility with parent signature).
+
+        Returns:
+            bool: True if the user has the permission in the object's scope,
+                  False otherwise.
+
+        Example:
+            >>> rule = HasPermissionInContentLibraryScope('view')
+            >>> can_view = rule.check(user, library)
+            >>> # Checks if user has 'view' permission in scope 'lib:DemoX:CSPROB'
+        """
+        return authz_api.is_user_allowed(user.username, self.permission.identifier, str(instance.library_key))
+
+
 ########################### Permissions ###########################
 
 # Is the user allowed to view XBlocks from the specified content library
@@ -87,7 +239,9 @@ def is_course_creator(user):
     is_global_staff |
     # Libraries with "public read" permissions can be accessed only by course creators
     (Attribute('allow_public_read', True) & is_course_creator) |
-    # Otherwise the user must be part of the library's team
+    # Users can access libraries within their authorized scope (via Casbin/role-based permissions)
+    HasPermissionInContentLibraryScope(VIEW_LIBRARY) |
+    # Fallback to: the user must be part of the library's team (legacy permission system)
     has_explicit_read_permission_for_library
 )
 

openedx/core/djangoapps/content_libraries/rest_api/blocks.py

@@ -9,6 +9,7 @@
 from django.utils.decorators import method_decorator
 from drf_yasg.utils import swagger_auto_schema
 from opaque_keys.edx.locator import LibraryLocatorV2, LibraryUsageLocatorV2
+from openedx_authz.constants import permissions as authz_permissions
 from openedx_learning.api import authoring as authoring_api
 from rest_framework import status
 from rest_framework.exceptions import NotFound, ValidationError
@@ -238,7 +239,7 @@ def post(self, request, usage_key_str):
         api.require_permission_for_library_key(
             key.lib_key,
             request.user,
-            permissions.CAN_EDIT_THIS_CONTENT_LIBRARY
+            authz_permissions.PUBLISH_LIBRARY_CONTENT
         )
         api.publish_component_changes(key, request.user)
         return Response({})