diff --git a/cms/djangoapps/contentstore/views/block.py b/cms/djangoapps/contentstore/views/block.py index e606477ee222..4d60640cb2cf 100644 --- a/cms/djangoapps/contentstore/views/block.py +++ b/cms/djangoapps/contentstore/views/block.py @@ -12,6 +12,7 @@ from django.views.decorators.clickjacking import xframe_options_exempt from django.views.decorators.http import require_http_methods from opaque_keys.edx.keys import CourseKey +from openedx_authz.constants.permissions import COURSES_VIEW_COURSE from web_fragments.fragment import Fragment from cms.djangoapps.contentstore.utils import load_services_for_studio @@ -27,6 +28,8 @@ from common.djangoapps.edxmako.shortcuts import render_to_response, render_to_string from common.djangoapps.student.auth import has_studio_read_access, has_studio_write_access from common.djangoapps.util.json_request import JsonResponse, expect_json +from openedx.core.djangoapps.authz.constants import LegacyAuthoringPermission +from openedx.core.djangoapps.authz.decorators import user_has_course_permission from openedx.core.djangoapps.content_tagging.toggles import is_tagging_feature_disabled from openedx.core.lib.xblock_utils import hash_resource, request_token, wrap_xblock, wrap_xblock_aside from xmodule.modulestore.django import modulestore # lint-amnesty, pylint: disable=wrong-import-order @@ -329,7 +332,12 @@ def xblock_outline_handler(request, usage_key_string): a course. """ usage_key = usage_key_with_run(usage_key_string) - if not has_studio_read_access(request.user, usage_key.course_key): + if not user_has_course_permission( + request.user, + COURSES_VIEW_COURSE.identifier, + usage_key.course_key, + LegacyAuthoringPermission.READ, + ): raise PermissionDenied() response_format = request.GET.get("format", "html") diff --git a/cms/djangoapps/contentstore/views/tests/test_block.py b/cms/djangoapps/contentstore/views/tests/test_block.py index bb1206169189..c272f44d7c2a 100644 --- a/cms/djangoapps/contentstore/views/tests/test_block.py +++ b/cms/djangoapps/contentstore/views/tests/test_block.py @@ -19,6 +19,7 @@ from opaque_keys.edx.asides import AsideUsageKeyV2 from opaque_keys.edx.keys import CourseKey, UsageKey from opaque_keys.edx.locator import BlockUsageLocator, CourseLocator +from openedx_authz.constants.roles import COURSE_ADMIN, COURSE_AUDITOR, COURSE_EDITOR, COURSE_STAFF from openedx_events.content_authoring.data import DuplicatedXBlockData from openedx_events.content_authoring.signals import XBLOCK_DUPLICATED from openedx_events.testing import OpenEdxEventsTestMixin @@ -54,6 +55,7 @@ from common.djangoapps.xblock_django.user_service import DjangoXBlockUserService from common.test.utils import assert_dict_contains_subset from lms.djangoapps.lms_xblock.mixin import NONSENSICAL_ACCESS_RESTRICTION +from openedx.core.djangoapps.authz.tests.mixins import CourseAuthoringAuthzTestMixin from openedx.core.djangoapps.content_tagging import api as tagging_api from openedx.core.djangoapps.discussions.models import DiscussionsConfiguration from openedx.core.djangoapps.video_config.toggles import PUBLIC_VIDEO_SHARE @@ -3486,6 +3488,145 @@ def validate_xblock_info_consistency( self.assertIsNone(xblock_info.get("child_info", None)) # noqa: PT009 +@ddt.ddt +class TestXBlockOutlineHandlerAuthz(CourseAuthoringAuthzTestMixin, ItemTest): + """ + Unit tests for xblock_outline_handler authorization functionality. + """ + + def setUp(self): + super().setUp() + user_id = self.user.id + self.chapter = BlockFactory.create( + parent_location=self.course.location, + category="chapter", + display_name="Week 1", + user_id=user_id, + ) + self.sequential = BlockFactory.create( + parent_location=self.chapter.location, + category="sequential", + display_name="Lesson 1", + user_id=user_id, + ) + self.vertical = BlockFactory.create( + parent_location=self.sequential.location, + category="vertical", + display_name="Unit 1", + user_id=user_id, + ) + # Assign COURSE_STAFF role to authorized_user for the course + self.add_user_to_role_in_course( + self.authorized_user, + COURSE_STAFF.external_key, + self.course.id + ) + + def test_authorized_user_gets_json_response(self): + """ + Test that authorized user gets JSON response from xblock_outline_handler. + """ + outline_url = reverse_usage_url("xblock_outline_handler", self.usage_key) + + self.client.login(username=self.authorized_user.username, password=self.password) + resp = self.client.get(outline_url, HTTP_ACCEPT="application/json") + + assert resp.status_code == 200 + json_response = json.loads(resp.content.decode("utf-8")) + assert "id" in json_response + assert "display_name" in json_response + assert "child_info" in json_response + + @ddt.data( + COURSE_ADMIN.external_key, + COURSE_AUDITOR.external_key, + COURSE_EDITOR.external_key, + ) + def test_other_course_roles_can_view_outline(self, role_key): + """ + Test that course_admin, course_auditor, and course_editor roles + can access the outline (all have COURSES_VIEW_COURSE). + """ + role_user = UserFactory(password=self.password) + self.add_user_to_role_in_course(role_user, role_key, self.course.id) + + outline_url = reverse_usage_url("xblock_outline_handler", self.usage_key) + self.client.login(username=role_user.username, password=self.password) + resp = self.client.get(outline_url, HTTP_ACCEPT="application/json") + + assert resp.status_code == 200 + + def test_unauthorized_user_gets_permission_denied(self): + """ + Test that unauthorized user gets 403 response from xblock_outline_handler. + """ + outline_url = reverse_usage_url("xblock_outline_handler", self.usage_key) + + self.client.login(username=self.unauthorized_user.username, password=self.password) + resp = self.client.get(outline_url, HTTP_ACCEPT="application/json") + + assert resp.status_code == 403 + + def test_superuser_gets_json_response(self): + """ + Test that superuser gets JSON response from xblock_outline_handler. + """ + outline_url = reverse_usage_url("xblock_outline_handler", self.usage_key) + + self.client.login(username=self.super_user.username, password=self.password) + resp = self.client.get(outline_url, HTTP_ACCEPT="application/json") + + assert resp.status_code == 200 + json_response = json.loads(resp.content.decode("utf-8")) + assert "id" in json_response + assert "display_name" in json_response + assert "child_info" in json_response + + def test_staff_user_gets_json_response(self): + """ + Test that staff user gets JSON response from xblock_outline_handler. + """ + outline_url = reverse_usage_url("xblock_outline_handler", self.usage_key) + + self.client.login(username=self.staff_user.username, password=self.password) + resp = self.client.get(outline_url, HTTP_ACCEPT="application/json") + + assert resp.status_code == 200 + json_response = json.loads(resp.content.decode("utf-8")) + assert "id" in json_response + assert "display_name" in json_response + assert "child_info" in json_response + + def test_authorized_chapter_outline(self): + """ + Test that authorized user can access chapter-level outline. + """ + outline_url = reverse_usage_url("xblock_outline_handler", self.chapter.location) + + self.client.login(username=self.authorized_user.username, password=self.password) + resp = self.client.get(outline_url, HTTP_ACCEPT="application/json") + + assert resp.status_code == 200 + json_response = json.loads(resp.content.decode("utf-8")) + assert json_response["display_name"] == "Week 1" + assert "child_info" in json_response + # Verify that children are included (should have the sequential) + children = json_response["child_info"]["children"] + assert len(children) > 0 + assert children[0]["display_name"] == "Lesson 1" + + def test_unauthorized_chapter_outline(self): + """ + Test that unauthorized user cannot access chapter-level outline. + """ + outline_url = reverse_usage_url("xblock_outline_handler", self.chapter.location) + + self.client.login(username=self.unauthorized_user.username, password=self.password) + resp = self.client.get(outline_url, HTTP_ACCEPT="application/json") + + assert resp.status_code == 403 + + class TestGetMetadataWithProblemDefaults(ModuleStoreTestCase): """ Unit tests for _get_metadata_with_problem_defaults. diff --git a/openedx/core/djangoapps/content_tagging/auth.py b/openedx/core/djangoapps/content_tagging/auth.py index 24a64edc131a..78c4cf3e9533 100644 --- a/openedx/core/djangoapps/content_tagging/auth.py +++ b/openedx/core/djangoapps/content_tagging/auth.py @@ -11,6 +11,8 @@ from openedx.core import toggles as core_toggles +from .utils import get_context_key_from_key_string + log = logging.getLogger(__name__) @@ -39,3 +41,33 @@ def has_view_object_tags_access(user, object_id): # The obj arg expects a model, but we are passing an object oel_tagging_rules.ObjectTagPermissionItem(taxonomy=None, object_id=object_id), # type: ignore[arg-type] ) + + +def should_use_course_authz_for_object(object_id) -> tuple[bool, CourseKey | None]: + """ + Check if openedx-authz should be used for the given object based on the context key and toggle. + + This intentionally scopes to course objects only. For non-course objects + (e.g. content libraries), this returns (False, None), and callers are + expected to fall back to legacy django-rules permissions, which handle + library authz checks separately (see can_change_object_tag_objectid + in rules.py). + + Returns (should_use_authz, course_key) where: + - should_use_authz: True if authz should be used, False otherwise + - course_key: The CourseKey if object is a course and the toggle is active, None otherwise + """ + # Extract context_key and ensure it is a CourseKey (not a LibraryLocatorV2, etc.) + try: + context_key = get_context_key_from_key_string(object_id) + if not isinstance(context_key, CourseKey): + return False, None + except (ValueError, AttributeError): + return False, None + + # Check if toggle is active + if not core_toggles.enable_authz_course_authoring(context_key): + return False, None + + # Authz should be used for this course object + return True, context_key diff --git a/openedx/core/djangoapps/content_tagging/rest_api/v1/filters.py b/openedx/core/djangoapps/content_tagging/rest_api/v1/filters.py index 75bbde280e17..a51bf6c99812 100644 --- a/openedx/core/djangoapps/content_tagging/rest_api/v1/filters.py +++ b/openedx/core/djangoapps/content_tagging/rest_api/v1/filters.py @@ -8,6 +8,7 @@ from ...models import TaxonomyOrg from ...rules import get_admin_orgs, get_user_orgs +from ...utils import rules_cache class UserOrgFilterBackend(BaseFilterBackend): @@ -62,7 +63,30 @@ class ObjectTagTaxonomyOrgFilterBackend(BaseFilterBackend): Filter for ObjectTagViewSet to only show taxonomies that the user can view. """ - def filter_queryset(self, request, queryset, _): + def filter_queryset(self, request, queryset, view): + # Authz path: filter by course org only. + # The legacy validation layer (check_taxonomy_context_key_org in rules.py) enforces + # that tags can only be applied to a course from global taxonomies or taxonomies + # owned by the course's org. This is narrower than the legacy filter path below, + # which filters by all the user's orgs, but it matches the actual enforcement — + # a user can never apply an OrgA taxonomy to an OrgB course regardless of their + # permissions. + should_use_authz, course_key = getattr(view, '_authz_check', (False, None)) + if should_use_authz and course_key: + course_orgs = rules_cache.get_orgs([course_key.org]) if course_key.org else [] + return queryset.filter(taxonomy__enabled=True).filter( + Exists( + TaxonomyOrg.objects + .filter(taxonomy=OuterRef("taxonomy_id"), rel_type=TaxonomyOrg.RelType.OWNER) + .filter(Q(org=None) | Q(org__in=course_orgs)) + ) + ).prefetch_related('taxonomy__taxonomyorg_set') + + # Legacy path: filter by all the user's orgs. + # This is broader than necessary for this endpoint (see authz path comment above). + # Users may see taxonomies from orgs unrelated to the current course — those + # taxonomies can't have tags applied here (blocked by check_taxonomy_context_key_org), + # but they still appear in the response with can_tag_object=false. if oel_tagging.is_taxonomy_admin(request.user): return queryset.prefetch_related('taxonomy__taxonomyorg_set') diff --git a/openedx/core/djangoapps/content_tagging/rest_api/v1/serializers.py b/openedx/core/djangoapps/content_tagging/rest_api/v1/serializers.py index e15f04504a3e..22d875670817 100644 --- a/openedx/core/djangoapps/content_tagging/rest_api/v1/serializers.py +++ b/openedx/core/djangoapps/content_tagging/rest_api/v1/serializers.py @@ -4,14 +4,18 @@ from __future__ import annotations +from openedx_authz import api as authz_api +from openedx_authz.constants.permissions import COURSES_MANAGE_TAGS from openedx_tagging.rest_api.v1.serializers import ( ObjectTagMinimalSerializer, + ObjectTagsByTaxonomySerializer, TaxonomyListQueryParamsSerializer, TaxonomySerializer, ) from organizations.models import Organization from rest_framework import fields, serializers +from ...auth import should_use_course_authz_for_object from ...models import TaxonomyOrg @@ -95,12 +99,35 @@ class Meta: read_only_fields = ["orgs", "all_orgs"] +class ObjectTagOrgByTaxonomySerializer(ObjectTagsByTaxonomySerializer): + """ + Extend ObjectTagsByTaxonomySerializer to conditionally use openedx-authz for can_tag_object. + """ + + def can_tag_object(self, obj_tag) -> bool | None: + """ + Check if the user is authorized to tag the provided object. + Conditionally use openedx-authz for course objects with the toggle enabled. + """ + should_use_authz, course_key = should_use_course_authz_for_object(obj_tag.object_id) + if should_use_authz: + request = self.context.get('request') + if request and hasattr(request, 'user'): + return authz_api.is_user_allowed( + request.user.username, COURSES_MANAGE_TAGS.identifier, str(course_key) + ) + return False + + # Fall back to parent implementation + return super().can_tag_object(obj_tag) + + class ObjectTagCopiedMinimalSerializer(ObjectTagMinimalSerializer): """ Serializer for Object Tags. - This override `get_can_delete_objecttag` to avoid delete - object tags if is copied. + This overrides `can_delete_object_tag` to avoid deleting + object tags if they are copied and to conditionally use openedx-authz. """ is_copied = serializers.BooleanField(read_only=True) @@ -108,14 +135,25 @@ class ObjectTagCopiedMinimalSerializer(ObjectTagMinimalSerializer): class Meta(ObjectTagMinimalSerializer.Meta): fields = ObjectTagMinimalSerializer.Meta.fields + ["is_copied"] - def get_can_delete_objecttag(self, instance): + def can_delete_object_tag(self, instance) -> bool | None: """ - Verify if the user can delete the object tag. + Check if the user is authorized to delete the provided tag. - Override to return `False` if the object tag is copied. + Override to return `False` if the object tag is copied, + and conditionally use openedx-authz for course objects with the toggle enabled. """ if instance.is_copied: # The user can't delete copied tags. return False - return super().get_can_delete_objecttag(instance) + should_use_authz, course_key = should_use_course_authz_for_object(instance.object_id) + if should_use_authz: + request = self.context.get('request') + if request and hasattr(request, 'user'): + return authz_api.is_user_allowed( + request.user.username, COURSES_MANAGE_TAGS.identifier, str(course_key) + ) + return False + + # Fall back to parent implementation + return super().can_delete_object_tag(instance) diff --git a/openedx/core/djangoapps/content_tagging/rest_api/v1/tests/test_views.py b/openedx/core/djangoapps/content_tagging/rest_api/v1/tests/test_views.py index 35f209f8a7e1..0422845cc346 100644 --- a/openedx/core/djangoapps/content_tagging/rest_api/v1/tests/test_views.py +++ b/openedx/core/djangoapps/content_tagging/rest_api/v1/tests/test_views.py @@ -17,7 +17,7 @@ from edx_django_utils.cache import RequestCache from opaque_keys.edx.locator import BlockUsageLocator, CourseLocator, LibraryCollectionLocator, LibraryContainerLocator from openedx_authz.constants import permissions as authz_permissions -from openedx_authz.constants.roles import COURSE_STAFF +from openedx_authz.constants.roles import COURSE_AUDITOR, COURSE_EDITOR, COURSE_STAFF from openedx_tagging.models import Tag, Taxonomy from openedx_tagging.models.system_defined import SystemDefinedTaxonomy from openedx_tagging.rest_api.v1.serializers import TaxonomySerializer @@ -38,6 +38,7 @@ from openedx.core.djangoapps.authz.tests.mixins import CourseAuthzTestMixin from openedx.core.djangoapps.content_libraries.api import AccessLevel, create_library, set_library_user_permissions from openedx.core.djangoapps.content_tagging import api as tagging_api +from openedx.core.djangoapps.content_tagging.api import set_taxonomy_orgs from openedx.core.djangoapps.content_tagging.models import TaxonomyOrg from openedx.core.djangolib.testing.utils import skip_unless_cms from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase @@ -1979,19 +1980,19 @@ def test_get_copied_tags(self): assert response.data[str(object_id_2)]["taxonomies"] == expected_tags @ddt.data( - ('staff', 'courseA', 8), + ('staff', 'courseA', 10), ('staff', 'libraryA', 17), ('staff', 'collection_key', 17), - ("content_creatorA", 'courseA', 18, False), + ("content_creatorA", 'courseA', 20, False), ("content_creatorA", 'libraryA', 23, False), ("content_creatorA", 'collection_key', 23, False), ("library_staffA", 'libraryA', 23, False), # Library users can only view objecttags, not change them? ("library_staffA", 'collection_key', 23, False), ("library_userA", 'libraryA', 23, False), ("library_userA", 'collection_key', 23, False), - ("instructorA", 'courseA', 18), - ("course_instructorA", 'courseA', 18), - ("course_staffA", 'courseA', 18), + ("instructorA", 'courseA', 20), + ("course_instructorA", 'courseA', 20), + ("course_staffA", 'courseA', 20), ) @ddt.unpack def test_object_tags_query_count( @@ -2136,6 +2137,228 @@ def test_superuser_allowed(self): resp = client.get(self.get_url(self.course_key)) self.assertEqual(resp.status_code, status.HTTP_200_OK) # noqa: PT009 + +@skip_unless_cms +class TestObjectTagOrgViewWithAuthz(CourseAuthzTestMixin, SharedModuleStoreTestCase, APITestCase): + """ + Test ObjectTagOrgView with authz permissions. + """ + + authz_roles_to_assign = [COURSE_STAFF.external_key] + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.course = CourseFactory.create() + cls.course_key = cls.course.id + + def setUp(self): + super().setUp() + + # Create another course for cross-course scoping tests + self.other_course = self.store.create_course("OtherOrg", "OtherCourse", "Run", self.authorized_user.id) + self.other_course_key = self.other_course.id + + # Create taxonomy + self.taxonomy = tagging_api.create_taxonomy( + name="Test Taxonomy", + description="Test taxonomy for authz", + ) + set_taxonomy_orgs(self.taxonomy, all_orgs=True) + + # Create tags + self.tag1 = self.taxonomy.add_tag("Tag 1") + self.tag2 = self.taxonomy.add_tag("Tag 2") + + # Create auditor user with view-only permissions + self.auditor_user = UserFactory(password=self.password) + self.auditor_client = APIClient() + self.auditor_client.force_authenticate(user=self.auditor_user) + + # Assign auditor role to auditor_user + self.add_user_to_role_in_course( + self.auditor_user, + COURSE_AUDITOR.external_key, + self.course_key + ) + + def _update_tags_request(self, object_id, tags_data=None): + """Helper method to make PUT request to update tags.""" + if tags_data is None: + tags_data = [ + { + "taxonomy": self.taxonomy.pk, + "tags": ["Tag 1", "Tag 2"] + } + ] + + url = OBJECT_TAG_UPDATE_URL.format(object_id=object_id) + return url, {"tagsData": tags_data} + + def _get_tags_request(self, object_id): + """Helper method to make GET request to retrieve tags.""" + url = OBJECT_TAGS_URL.format(object_id=object_id) + return url + + def test_course_staff_can_update_tags(self): + """course_staff can update tags → 200""" + url, data = self._update_tags_request(str(self.course_key)) + response = self.authorized_client.put(url, data, format='json') + assert response.status_code == status.HTTP_200_OK + + def test_course_auditor_denied_update(self): + """course_auditor denied → 403""" + url, data = self._update_tags_request(str(self.course_key)) + response = self.auditor_client.put(url, data, format='json') + assert response.status_code == status.HTTP_403_FORBIDDEN + + def test_no_role_user_denied_update(self): + """No-role user denied → 403""" + url, data = self._update_tags_request(str(self.course_key)) + response = self.unauthorized_client.put(url, data, format='json') + assert response.status_code == status.HTTP_403_FORBIDDEN + + def test_superuser_allowed_update(self): + """Superuser allowed → 200""" + url, data = self._update_tags_request(str(self.course_key)) + response = self.super_client.put(url, data, format='json') + assert response.status_code == status.HTTP_200_OK + + def test_django_staff_allowed_update(self): + """Django is_staff allowed → 200""" + url, data = self._update_tags_request(str(self.course_key)) + response = self.staff_client.put(url, data, format='json') + assert response.status_code == status.HTTP_200_OK + + def test_cross_course_scoping_denied(self): + """course_staff for course A tags course B → 403""" + url, data = self._update_tags_request(str(self.other_course_key)) + response = self.authorized_client.put(url, data, format='json') + assert response.status_code == status.HTTP_403_FORBIDDEN + + def test_course_staff_sees_manage_permissions(self): + """course_staff sees can_tag_object=True, can_delete_objecttag=True""" + # First add some tags to the course + url, data = self._update_tags_request(str(self.course_key)) + response = self.authorized_client.put(url, data, format='json') + assert response.status_code == status.HTTP_200_OK + + # Now check permissions in GET response + url = self._get_tags_request(str(self.course_key)) + response = self.authorized_client.get(url) + assert response.status_code == status.HTTP_200_OK + + # Check serializer permissions in response data + taxonomies = response.data[str(self.course_key)]["taxonomies"] + assert len(taxonomies) == 1 + for taxonomy_data in taxonomies: + assert taxonomy_data.get('can_tag_object') is True + tags = taxonomy_data.get('tags', []) + assert len(tags) == 2 + for tag_data in tags: + assert tag_data.get('can_delete_objecttag') is True + + def test_course_auditor_sees_view_only_permissions(self): + """course_auditor sees can_tag_object=False, can_delete_objecttag=False""" + # First add some tags using authorized user + url, data = self._update_tags_request(str(self.course_key)) + response = self.authorized_client.put(url, data, format='json') + assert response.status_code == status.HTTP_200_OK + + # Now check permissions as auditor in GET response + url = self._get_tags_request(str(self.course_key)) + response = self.auditor_client.get(url) + assert response.status_code == status.HTTP_200_OK + + # Check serializer permissions in response data + taxonomies = response.data[str(self.course_key)]["taxonomies"] + assert len(taxonomies) == 1 + for taxonomy_data in taxonomies: + assert taxonomy_data.get('can_tag_object') is False + tags = taxonomy_data.get('tags', []) + assert len(tags) == 2 + for tag_data in tags: + assert tag_data.get('can_delete_objecttag') is False + + def test_no_role_user_denied_view(self): + """No-role user denied on view → 403 (checks view_course)""" + url = self._get_tags_request(str(self.course_key)) + response = self.unauthorized_client.get(url) + assert response.status_code == status.HTTP_403_FORBIDDEN + + def test_course_editor_sees_org_scoped_taxonomy_tags(self): + """ + A course editor (authz-only, no legacy org roles) can see tags from + an org-scoped taxonomy applied to their course. + + Reproduces the scenario from review: admin tags a course with an + org-scoped taxonomy, course editor should see those tags. + """ + # Create an Organization matching the course's org + course_org, _ = Organization.objects.get_or_create( + short_name=self.course_key.org + ) + + # Create an org-scoped taxonomy (not global) + org_taxonomy = tagging_api.create_taxonomy( + name="Org Taxonomy", + description="Taxonomy scoped to course org", + ) + set_taxonomy_orgs(org_taxonomy, orgs=[course_org]) + org_taxonomy.add_tag("OrgTag") + + # Admin tags the course with the org-scoped taxonomy + tagging_api.tag_object( + object_id=str(self.course_key), + taxonomy=org_taxonomy, + tags=["OrgTag"], + ) + + # Create a course editor with authz only (no legacy roles) + editor_user = UserFactory(password=self.password) + editor_client = APIClient() + editor_client.force_authenticate(user=editor_user) + self.add_user_to_role_in_course( + editor_user, COURSE_EDITOR.external_key, self.course_key + ) + + # Course editor should see the org-scoped taxonomy tags + url = self._get_tags_request(str(self.course_key)) + response = editor_client.get(url) + assert response.status_code == status.HTTP_200_OK + + taxonomies = response.data[str(self.course_key)]["taxonomies"] + org_tax_names = [t["name"] for t in taxonomies] + assert "Org Taxonomy" in org_tax_names + + def test_library_fallthrough_to_legacy(self): + """Library object_id falls through to legacy permissions""" + # Create organization for library + org, _ = Organization.objects.get_or_create(short_name="TestOrg") + + # Create library + library = create_library( + org=org, + slug="test-lib", + title="Test Library", + description="Test library for authz fallthrough", + ) + library_key = library.key + + # Grant library access to authorized_user + set_library_user_permissions( + library_key, + self.authorized_user, + AccessLevel.ADMIN_LEVEL + ) + + # Test that library requests fall through to legacy permissions + url = self._get_tags_request(str(library_key)) + response = self.authorized_client.get(url) + # Should succeed via legacy permissions, not authz + assert response.status_code == status.HTTP_200_OK + + @skip_unless_cms @ddt.ddt class TestDownloadTemplateView(APITestCase): diff --git a/openedx/core/djangoapps/content_tagging/rest_api/v1/views.py b/openedx/core/djangoapps/content_tagging/rest_api/v1/views.py index a52a8810b547..556ea06cfe1e 100644 --- a/openedx/core/djangoapps/content_tagging/rest_api/v1/views.py +++ b/openedx/core/djangoapps/content_tagging/rest_api/v1/views.py @@ -3,8 +3,13 @@ """ from __future__ import annotations +import functools +from typing import TYPE_CHECKING + from django.db.models import Count from django.http import StreamingHttpResponse +from openedx_authz import api as authz_api +from openedx_authz.constants.permissions import COURSES_MANAGE_TAGS, COURSES_VIEW_COURSE from openedx_events.content_authoring.data import ContentObjectChangedData, ContentObjectData from openedx_events.content_authoring.signals import CONTENT_OBJECT_ASSOCIATIONS_CHANGED, CONTENT_OBJECT_TAGS_CHANGED from openedx_tagging import rules as oel_tagging_rules @@ -12,6 +17,7 @@ from rest_framework import status from rest_framework.decorators import action from rest_framework.exceptions import PermissionDenied, ValidationError +from rest_framework.permissions import IsAuthenticated from rest_framework.response import Response from rest_framework.views import APIView @@ -26,16 +32,20 @@ get_unassigned_taxonomies, set_taxonomy_orgs, ) -from ...auth import has_view_object_tags_access +from ...auth import has_view_object_tags_access, should_use_course_authz_for_object from ...rules import get_admin_orgs from .filters import ObjectTagTaxonomyOrgFilterBackend, UserOrgFilterBackend from .serializers import ( ObjectTagCopiedMinimalSerializer, + ObjectTagOrgByTaxonomySerializer, TaxonomyOrgListQueryParamsSerializer, TaxonomyOrgSerializer, TaxonomyUpdateOrgBodySerializer, ) +if TYPE_CHECKING: + from opaque_keys.edx.keys import CourseKey + class TaxonomyOrgView(TaxonomyView): """ @@ -151,9 +161,70 @@ class ObjectTagOrgView(ObjectTagView): Refer to ObjectTagView docstring for usage details. """ + # Serializer overrides minimal_serializer_class = ObjectTagCopiedMinimalSerializer + object_tags_serializer_class = ObjectTagOrgByTaxonomySerializer + filter_backends = [ObjectTagTaxonomyOrgFilterBackend] + @functools.cached_property + def _authz_check(self) -> tuple[bool, CourseKey | None]: + """ + Cache the authz toggle + key-parsing result for the current object_id. + + Safe to cache per-instance because DRF creates a new view instance per request. + """ + object_id = self.kwargs.get('object_id') + if object_id: + return should_use_course_authz_for_object(object_id) + return False, None + + def get_permissions(self): + """ + Override get_permissions when using openedx-authz. + + When the toggle is enabled for course objects, we need to change the default + permission classes set by the parent ObjectTagView so that only openedx-authz + permissions are used. + """ + if self._authz_check[0]: + return [IsAuthenticated()] + + return super().get_permissions() + + def ensure_has_view_object_tag_permission(self, user, taxonomy, object_id): + """ + Check if user has permission to view object tags. + + This method is overridden to conditionally use openedx-authz when the toggle is enabled. + """ + should_use_authz, course_key = self._authz_check + if should_use_authz and not authz_api.is_user_allowed( + user.username, COURSES_VIEW_COURSE.identifier, str(course_key) + ): + raise PermissionDenied("You do not have permission to view object tags.") + if not should_use_authz: + # Fall back to parent implementation + super().ensure_has_view_object_tag_permission(user, taxonomy, object_id) + + def ensure_user_has_can_tag_object_permissions(self, user, tags_data, object_id): + """ + Check if user has permission to tag object for each taxonomy in tags_data. + + This method is overridden to conditionally use openedx-authz when the toggle is enabled. + + When using openedx-authz, if the user has manage tags permission for the course, + they can tag the object regardless of the taxonomy. + """ + should_use_authz, course_key = self._authz_check + if should_use_authz and not authz_api.is_user_allowed( + user.username, COURSES_MANAGE_TAGS.identifier, str(course_key) + ): + raise PermissionDenied("You do not have permission to manage object tags.") + if not should_use_authz: + # Fall back to parent implementation + super().ensure_user_has_can_tag_object_permissions(user, tags_data, object_id) + def update(self, request, *args, **kwargs) -> Response: """ Extend the update method to fire CONTENT_OBJECT_ASSOCIATIONS_CHANGED event