Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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', '')}"
)
56 changes: 56 additions & 0 deletions cms/lib/xblock/test/test_upstream_sync.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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

Expand Down Expand Up @@ -651,3 +655,55 @@ def test_sync_keep_customizaton_option(self):
# data is overridden
assert downstream.data == "<html><body>Upstream content V2</body></html>"
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"
)
21 changes: 20 additions & 1 deletion cms/lib/xblock/upstream_sync_block.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -94,18 +97,34 @@ 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
CheckPerm,
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:
Expand Down
22 changes: 18 additions & 4 deletions cms/lib/xblock/upstream_sync_container.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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)
Expand Down
Loading