diff --git a/cms/djangoapps/contentstore/rest_api/v2/views/tests/test_downstreams.py b/cms/djangoapps/contentstore/rest_api/v2/views/tests/test_downstreams.py index 0f3ba29673e6..3b0f7d5d763c 100644 --- a/cms/djangoapps/contentstore/rest_api/v2/views/tests/test_downstreams.py +++ b/cms/djangoapps/contentstore/rest_api/v2/views/tests/test_downstreams.py @@ -1760,3 +1760,60 @@ def test_auditor_cannot_sync_downstream(self): # _load_accessible_block permission denial (which returns 404, not 403, to avoid # leaking block existence). assert response.status_code == status.HTTP_404_NOT_FOUND + + +class PostDownstreamSyncAuthzViewTest( + CourseAuthoringAuthzTestMixin, + _BaseDownstreamViewTestMixin, + ImmediateOnCommitMixin, + SharedModuleStoreTestCase, +): + """ + AuthZ tests for: + POST /api/contentstore/v2/downstreams/{usage_key}/sync + + Verifies that a user with the ``course_staff`` authz role (which includes + ``courses.manage_library_updates``) can sync a downstream container even + when the user has **no** permissions on the source library. + """ + + def call_api(self, usage_key_string): + return self.authorized_client.post( + f"/api/contentstore/v2/downstreams/{usage_key_string}/sync" + ) + + def test_course_staff_can_sync_container_without_library_access(self): + """ + A user with Course Staff role (which carries + ``courses.manage_library_updates``) should be able to sync a + downstream container from its upstream library, even when the user + has no explicit permissions on the library. + """ + # Give the user Course Staff in authz so they get manage_library_updates + from openedx_authz.constants.roles import COURSE_STAFF + self.add_user_to_role_in_course( + self.authorized_user, + COURSE_STAFF.external_key, + self.course.id, + ) + + # Confirm the user has NO explicit permissions on the library. + assert lib_api.get_library_user_permissions( + self.library_key, self.authorized_user, + ) is None + + # The downstream_unit_key is linked to a container upstream in self.library. + # The unit was updated (display_name changed + republished) in setUp, + # so it is ready to sync. + response = self.call_api(self.downstream_unit_key) + + assert response.status_code == 200, ( + f"Expected 200 but got {response.status_code}: {getattr(response, 'data', '')}" + ) + + # Same test but for a block sync instead of a container one + response = self.call_api(self.downstream_html_key) + + assert response.status_code == 200, ( + f"Expected 200 but got {response.status_code}: {getattr(response, 'data', '')}" + ) diff --git a/cms/lib/xblock/test/test_upstream_sync.py b/cms/lib/xblock/test/test_upstream_sync.py index d32f2fb5809a..bbc3327508cd 100644 --- a/cms/lib/xblock/test/test_upstream_sync.py +++ b/cms/lib/xblock/test/test_upstream_sync.py @@ -2,8 +2,10 @@ Test CMS's upstream->downstream syncing system """ import datetime +from unittest.mock import patch import ddt +from openedx_content.models_api import Unit from organizations.api import ensure_organization from organizations.models import Organization @@ -17,10 +19,12 @@ sever_upstream_link, ) from cms.lib.xblock.upstream_sync_block import fetch_customizable_fields_from_block, sync_from_upstream_block +from cms.lib.xblock.upstream_sync_container import sync_from_upstream_container from common.djangoapps.student.tests.factories import UserFactory from openedx.core.djangoapps.content_libraries import api as libs from openedx.core.djangoapps.content_tagging import api as tagging_api from openedx.core.djangoapps.xblock import api as xblock +from openedx.core.djangoapps.xblock.data import CheckPerm from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase from xmodule.modulestore.tests.factories import BlockFactory, CourseFactory @@ -651,3 +655,55 @@ def test_sync_keep_customizaton_option(self): # data is overridden assert downstream.data == "Upstream content V2" assert downstream.downstream_customized == ["display_name"] + + def test_load_upstream_block_legacy_does_not_bypass_library_permission(self): + """ + When AuthZ is not enabled, _load_upstream_block falls through to the + library-level CAN_READ_AS_AUTHOR check. + """ + downstream = BlockFactory.create( + category="html", parent=self.unit, upstream=str(self.upstream_key) + ) + + # Get upstream xblock before patching + real_upstream = xblock.load_block(self.upstream_key, self.user) + + with patch("openedx.core.djangoapps.xblock.api.load_block") as mock_load_block: + mock_load_block.return_value = real_upstream + sync_from_upstream_block(downstream, self.user) + + mock_load_block.assert_called_once() + _, lb_kwargs = mock_load_block.call_args + assert lb_kwargs["check_permission"] == CheckPerm.CAN_READ_AS_AUTHOR, ( + "When the course-level permission is denied, the library block " + "should be loaded with CAN_READ_AS_AUTHOR, not with check_permission=None" + ) + + def test_sync_container_legacy_does_not_bypass_library_permission(self): + """ + When AuthZ is not enabled, sync_from_upstream_container falls through + to the library-level CAN_VIEW_THIS_CONTENT_LIBRARY check. + """ + upstream_container = libs.create_container( + self.library.key, Unit, "test-container", "Test Container Title", self.user.id, + ) + libs.publish_changes(self.library.key, self.user.id) + + downstream = BlockFactory.create( + category="vertical", + parent=self.unit, + upstream=str(upstream_container.container_key), + ) + + with patch( + "cms.lib.xblock.upstream_sync_container.lib_api.require_permission_for_library_key" + ) as mock_require_perm: + sync_from_upstream_container(downstream, self.user) + + mock_require_perm.assert_called_once() + _, rp_kwargs = mock_require_perm.call_args + assert rp_kwargs.get("permission") == libs.permissions.CAN_VIEW_THIS_CONTENT_LIBRARY, ( + "When the course-level permission is denied, the container sync " + "should enforce CAN_VIEW_THIS_CONTENT_LIBRARY via " + "require_permission_for_library_key" + ) diff --git a/cms/lib/xblock/upstream_sync_block.py b/cms/lib/xblock/upstream_sync_block.py index 85a18072144f..9ef788491e47 100644 --- a/cms/lib/xblock/upstream_sync_block.py +++ b/cms/lib/xblock/upstream_sync_block.py @@ -12,10 +12,13 @@ from django.core.exceptions import PermissionDenied from django.utils.translation import gettext_lazy as _ from opaque_keys.edx.locator import LibraryUsageLocatorV2 +from openedx_authz.constants.permissions import COURSES_MANAGE_LIBRARY_UPDATES from rest_framework.exceptions import NotFound from xblock.core import XBlock from xblock.fields import Scope +from openedx.core.djangoapps.authz.decorators import user_has_course_permission + from .upstream_sync import BadDownstream, BadUpstream, UpstreamLink if t.TYPE_CHECKING: @@ -94,6 +97,10 @@ def _load_upstream_block(downstream: XBlock, user: User) -> XBlock: library. This assumption may need to be relaxed in the future (see module docstring). If `downstream` lacks a valid+supported upstream link, this raises an UpstreamLinkException. + + If the user holds ``courses.manage_library_updates`` for the course that + owns ``downstream``, the library-level permission check is bypassed. + Otherwise the default ``CAN_READ_AS_AUTHOR`` check is applied. """ # We import load_block here b/c UpstreamSyncMixin is used by cms/envs, which loads before the djangoapps are ready. from openedx.core.djangoapps.xblock.api import ( # pylint: disable=wrong-import-order @@ -101,11 +108,23 @@ def _load_upstream_block(downstream: XBlock, user: User) -> XBlock: LatestVersion, load_block, ) + + # Try course-level permission first; fall back to library-level check. + course_key = downstream.usage_key.context_key + if course_key and user_has_course_permission( + user, + COURSES_MANAGE_LIBRARY_UPDATES.identifier, + course_key, + ): + check_perm = None + else: + check_perm = CheckPerm.CAN_READ_AS_AUTHOR + try: lib_block: XBlock = load_block( LibraryUsageLocatorV2.from_string(downstream.upstream), user, - check_permission=CheckPerm.CAN_READ_AS_AUTHOR, + check_permission=check_perm, version=LatestVersion.PUBLISHED, ) except (NotFound, PermissionDenied) as exc: diff --git a/cms/lib/xblock/upstream_sync_container.py b/cms/lib/xblock/upstream_sync_container.py index d6117509d238..3148eb1ce6ee 100644 --- a/cms/lib/xblock/upstream_sync_container.py +++ b/cms/lib/xblock/upstream_sync_container.py @@ -10,8 +10,10 @@ from django.utils.translation import gettext_lazy as _ # noqa: F401 from opaque_keys.edx.locator import LibraryContainerLocator +from openedx_authz.constants.permissions import COURSES_MANAGE_LIBRARY_UPDATES from xblock.core import XBlock +from openedx.core.djangoapps.authz.decorators import user_has_course_permission from openedx.core.djangoapps.content_libraries import api as lib_api from .upstream_sync import UpstreamLink @@ -37,15 +39,27 @@ def sync_from_upstream_container( Should children be handled in here? Maybe if sync_from_upstream_block were updated to handle static assets and also save changes to modulestore. + + The library-level permission check is skipped when the user holds + ``courses.manage_library_updates`` for ``downstream``'s course (derived + from ``downstream.usage_key.context_key``). """ link = UpstreamLink.get_for_block(downstream) # can raise UpstreamLinkException if not isinstance(link.upstream_key, LibraryContainerLocator): raise TypeError("sync_from_upstream_container() only supports Container upstreams, not containers") - lib_api.require_permission_for_library_key( # TODO: should permissions be checked at this low level? - link.upstream_key.lib_key, + + # Try course-level permission first; fall back to library-level check. + course_key = downstream.usage_key.context_key + if not (course_key and user_has_course_permission( user, - permission=lib_api.permissions.CAN_VIEW_THIS_CONTENT_LIBRARY, - ) + COURSES_MANAGE_LIBRARY_UPDATES.identifier, + course_key, + )): + lib_api.require_permission_for_library_key( # TODO: should permissions be checked at this low level? + link.upstream_key.lib_key, + user, + permission=lib_api.permissions.CAN_VIEW_THIS_CONTENT_LIBRARY, + ) upstream_meta = lib_api.get_container(link.upstream_key) upstream_children = lib_api.get_container_children(link.upstream_key, published=True) _update_customizable_fields(upstream=upstream_meta, downstream=downstream, only_fetch=False)