diff --git a/.github/workflows/pylint-checks.yml b/.github/workflows/pylint-checks.yml
index 03160b77a7e4..96d1c66b5928 100644
--- a/.github/workflows/pylint-checks.yml
+++ b/.github/workflows/pylint-checks.yml
@@ -21,7 +21,7 @@ jobs:
           - module-name: openedx-1
             path: "openedx/core/types/ openedx/core/djangoapps/ace_common/ openedx/core/djangoapps/agreements/ openedx/core/djangoapps/api_admin/ openedx/core/djangoapps/auth_exchange/ openedx/core/djangoapps/bookmarks/ openedx/core/djangoapps/cache_toolbox/ openedx/core/djangoapps/catalog/ openedx/core/djangoapps/ccxcon/ openedx/core/djangoapps/commerce/ openedx/core/djangoapps/common_initialization/ openedx/core/djangoapps/common_views/ openedx/core/djangoapps/config_model_utils/ openedx/core/djangoapps/content/ openedx/core/djangoapps/content_libraries/ openedx/core/djangoapps/content_staging/ openedx/core/djangoapps/contentserver/ openedx/core/djangoapps/cookie_metadata/ openedx/core/djangoapps/cors_csrf/ openedx/core/djangoapps/course_apps/ openedx/core/djangoapps/course_date_signals/ openedx/core/djangoapps/course_groups/ openedx/core/djangoapps/courseware_api/ openedx/core/djangoapps/crawlers/ openedx/core/djangoapps/credentials/ openedx/core/djangoapps/credit/ openedx/core/djangoapps/dark_lang/ openedx/core/djangoapps/debug/ openedx/core/djangoapps/discussions/ openedx/core/djangoapps/django_comment_common/ openedx/core/djangoapps/embargo/ openedx/core/djangoapps/enrollments/ openedx/core/djangoapps/external_user_ids/ openedx/core/djangoapps/zendesk_proxy/ openedx/core/djangolib/ openedx/core/lib/ openedx/core/djangoapps/course_live/"
           - module-name: openedx-2
-            path: "openedx/core/djangoapps/geoinfo/ openedx/core/djangoapps/header_control/ openedx/core/djangoapps/heartbeat/ openedx/core/djangoapps/lang_pref/ openedx/core/djangoapps/models/ openedx/core/djangoapps/monkey_patch/ openedx/core/djangoapps/oauth_dispatch/ openedx/core/djangoapps/olx_rest_api/ openedx/core/djangoapps/password_policy/ openedx/core/djangoapps/plugin_api/ openedx/core/djangoapps/plugins/ openedx/core/djangoapps/profile_images/ openedx/core/djangoapps/programs/ openedx/core/djangoapps/safe_sessions/ openedx/core/djangoapps/schedules/ openedx/core/djangoapps/service_status/ openedx/core/djangoapps/session_inactivity_timeout/ openedx/core/djangoapps/signals/ openedx/core/djangoapps/site_configuration/ openedx/core/djangoapps/system_wide_roles/ openedx/core/djangoapps/theming/ openedx/core/djangoapps/user_api/ openedx/core/djangoapps/user_authn/ openedx/core/djangoapps/util/ openedx/core/djangoapps/verified_track_content/ openedx/core/djangoapps/video_config/ openedx/core/djangoapps/video_pipeline/ openedx/core/djangoapps/waffle_utils/ openedx/core/djangoapps/xblock/ openedx/core/djangoapps/xmodule_django/ openedx/core/tests/ openedx/features/ openedx/testing/ openedx/tests/ openedx/envs/ openedx/core/djangoapps/notifications/ openedx/core/djangoapps/staticfiles/ openedx/core/djangoapps/content_tagging/"
+            path: "openedx/core/djangoapps/geoinfo/ openedx/core/djangoapps/header_control/ openedx/core/djangoapps/heartbeat/ openedx/core/djangoapps/lang_pref/ openedx/core/djangoapps/models/ openedx/core/djangoapps/monkey_patch/ openedx/core/djangoapps/oauth_dispatch/ openedx/core/djangoapps/olx_rest_api/ openedx/core/djangoapps/password_policy/ openedx/core/djangoapps/plugin_api/ openedx/core/djangoapps/plugins/ openedx/core/djangoapps/profile_images/ openedx/core/djangoapps/programs/ openedx/core/djangoapps/safe_sessions/ openedx/core/djangoapps/schedules/ openedx/core/djangoapps/service_status/ openedx/core/djangoapps/session_inactivity_timeout/ openedx/core/djangoapps/signals/ openedx/core/djangoapps/site_configuration/ openedx/core/djangoapps/system_wide_roles/ openedx/core/djangoapps/theming/ openedx/core/djangoapps/user_api/ openedx/core/djangoapps/user_authn/ openedx/core/djangoapps/util/ openedx/core/djangoapps/verified_track_content/ openedx/core/djangoapps/video_config/ openedx/core/djangoapps/video_pipeline/ openedx/core/djangoapps/waffle_utils/ openedx/core/djangoapps/xblock/ openedx/core/djangoapps/xmodule_django/ openedx/core/tests/ openedx/features/ openedx/testing/ openedx/tests/ openedx/envs/ openedx/core/djangoapps/notifications/ openedx/core/djangoapps/staticfiles/ openedx/core/djangoapps/content_tagging/ openedx/core/djangoapps/authz/"
           - module-name: common
             path: "common"
           - module-name: cms
diff --git a/.github/workflows/unit-test-shards.json b/.github/workflows/unit-test-shards.json
index cb9beeb3c6bf..2c1ffd4744b7 100644
--- a/.github/workflows/unit-test-shards.json
+++ b/.github/workflows/unit-test-shards.json
@@ -147,6 +147,7 @@
       "openedx/core/djangoapps/xblock/",
       "openedx/core/djangoapps/xmodule_django/",
       "openedx/core/djangoapps/zendesk_proxy/",
+      "openedx/core/djangoapps/authz/",
       "openedx/core/djangolib/",
       "openedx/core/lib/",
       "openedx/core/tests/",
@@ -227,6 +228,7 @@
       "openedx/core/djangoapps/xblock/",
       "openedx/core/djangoapps/xmodule_django/",
       "openedx/core/djangoapps/zendesk_proxy/",
+      "openedx/core/djangoapps/authz/",
       "openedx/core/lib/",
       "openedx/tests/"
     ]
diff --git a/cms/djangoapps/contentstore/api/tests/base.py b/cms/djangoapps/contentstore/api/tests/base.py
index a169c431e419..dd460734645c 100644
--- a/cms/djangoapps/contentstore/api/tests/base.py
+++ b/cms/djangoapps/contentstore/api/tests/base.py
@@ -18,6 +18,8 @@ class BaseCourseViewTest(SharedModuleStoreTestCase, APITestCase):
     Base test class for course data views.
     """
     view_name = None  # The name of the view to use in reverse() call in self.get_url()
+    course_key_arg_name = 'course_id'
+    extra_request_args = {}
 
     @classmethod
     def setUpClass(cls):
@@ -86,9 +88,10 @@ def get_url(self, course_id):
         """
         Helper function to create the url
         """
+        args = {
+            self.course_key_arg_name: course_id,
+        }
         return reverse(
             self.view_name,
-            kwargs={
-                'course_id': course_id
-            }
+            kwargs= args | self.extra_request_args
         )
diff --git a/cms/djangoapps/contentstore/api/tests/test_quality.py b/cms/djangoapps/contentstore/api/tests/test_quality.py
index 8625cbeb55e5..49ffdb07831c 100644
--- a/cms/djangoapps/contentstore/api/tests/test_quality.py
+++ b/cms/djangoapps/contentstore/api/tests/test_quality.py
@@ -2,9 +2,12 @@
 Tests for the course import API views
 """
 
-
+from rest_framework.test import APIClient
 from rest_framework import status
+from openedx_authz.constants.roles import COURSE_STAFF
 
+from common.djangoapps.student.tests.factories import UserFactory
+from openedx.core.djangoapps.authz.tests.mixins import CourseAuthzTestMixin
 from .base import BaseCourseViewTest
 
 
@@ -67,3 +70,48 @@ def test_student_fails(self):
         self.client.login(username=self.student.username, password=self.password)
         resp = self.client.get(self.get_url(self.course_key))
         self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+
+class CourseQualityAuthzTest(CourseAuthzTestMixin, BaseCourseViewTest):
+    """
+    Tests Course Quality API authorization using openedx-authz.
+    """
+
+    view_name = "courses_api:course_quality"
+    authz_roles_to_assign = [COURSE_STAFF.external_key]
+
+    def test_authorized_user_can_access(self):
+        """User with COURSE_STAFF role can access."""
+        resp = self.authorized_client.get(self.get_url(self.course_key))
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+    def test_unauthorized_user_cannot_access(self):
+        """User without role cannot access."""
+        resp = self.unauthorized_client.get(self.get_url(self.course_key))
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_role_scoped_to_course(self):
+        """Authorization should only apply to the assigned course."""
+        other_course = self.store.create_course("OtherOrg", "OtherCourse", "Run", self.staff.id)
+
+        resp = self.authorized_client.get(self.get_url(other_course.id))
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_staff_user_allowed_via_legacy(self):
+        """
+        Staff users should still pass through legacy fallback.
+        """
+        self.client.login(username=self.staff.username, password=self.password)
+
+        resp = self.client.get(self.get_url(self.course_key))
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+    def test_superuser_allowed(self):
+        """Superusers should always be allowed."""
+        superuser = UserFactory(is_superuser=True)
+
+        client = APIClient()
+        client.force_authenticate(user=superuser)
+
+        resp = client.get(self.get_url(self.course_key))
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
diff --git a/cms/djangoapps/contentstore/api/tests/test_validation.py b/cms/djangoapps/contentstore/api/tests/test_validation.py
index 8bf8e19b626c..2656bd379d9a 100644
--- a/cms/djangoapps/contentstore/api/tests/test_validation.py
+++ b/cms/djangoapps/contentstore/api/tests/test_validation.py
@@ -7,18 +7,23 @@
 
 import ddt
 import factory
+
 from django.conf import settings
 from django.contrib.auth import get_user_model
 from django.test.utils import override_settings
 from django.urls import reverse
 from rest_framework import status
 from rest_framework.test import APITestCase
+from rest_framework.test import APIClient
+from openedx.core.djangoapps.authz.tests.mixins import CourseAuthzTestMixin
+from openedx_authz.constants.roles import COURSE_STAFF
 
 from common.djangoapps.course_modes.models import CourseMode
 from common.djangoapps.course_modes.tests.factories import CourseModeFactory
 from common.djangoapps.student.tests.factories import StaffFactory, UserFactory
 from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase
 from xmodule.modulestore.tests.factories import BlockFactory, CourseFactory
+from cms.djangoapps.contentstore.api.tests.base import BaseCourseViewTest
 
 User = get_user_model()
 
@@ -272,3 +277,66 @@ def test_list_ready_to_update_reference_success(self, mock_block, mock_auth):
             {'usage_key': str(self.block2.location)},
         ])
         mock_auth.assert_called_once()
+
+
+class CourseValidationAuthzTest(CourseAuthzTestMixin, BaseCourseViewTest):
+    """
+    Tests Course Validation API authorization using openedx-authz.
+    """
+
+    view_name = "courses_api:course_validation"
+    authz_roles_to_assign = [COURSE_STAFF.external_key]
+
+    def test_authorized_user_can_access(self):
+        """
+        User with COURSE_STAFF role should be allowed via AuthZ.
+        """
+        resp = self.authorized_client.get(self.get_url(self.course_key))
+
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+    def test_unauthorized_user_cannot_access(self):
+        """
+        User without permissions should be denied.
+        """
+        resp = self.unauthorized_client.get(self.get_url(self.course_key))
+
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_role_scoped_to_course(self):
+        """
+        Authorization should only apply to the assigned course scope.
+        """
+        other_course = self.store.create_course(
+            "OtherOrg",
+            "OtherCourse",
+            "Run",
+            self.staff.id,
+        )
+
+        resp = self.authorized_client.get(self.get_url(other_course.id))
+
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_staff_user_allowed_via_legacy(self):
+        """
+        Course staff should pass through legacy fallback when AuthZ denies.
+        """
+        self.client.login(username=self.staff.username, password=self.password)
+
+        resp = self.client.get(self.get_url(self.course_key))
+
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+    def test_superuser_allowed(self):
+        """
+        Superusers should always be allowed through legacy fallback.
+        """
+        superuser = UserFactory(is_superuser=True)
+
+        client = APIClient()
+        client.force_authenticate(user=superuser)
+
+        resp = client.get(self.get_url(self.course_key))
+
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
diff --git a/cms/djangoapps/contentstore/api/views/course_quality.py b/cms/djangoapps/contentstore/api/views/course_quality.py
index b301f5ac1420..8db1eaa2df61 100644
--- a/cms/djangoapps/contentstore/api/views/course_quality.py
+++ b/cms/djangoapps/contentstore/api/views/course_quality.py
@@ -7,13 +7,15 @@
 from rest_framework.generics import GenericAPIView
 from rest_framework.response import Response
 from scipy import stats
+from openedx_authz.constants.permissions import COURSES_VIEW_COURSE
 
 from openedx.core.lib.api.view_utils import DeveloperErrorViewMixin, view_auth_classes
 from openedx.core.lib.cache_utils import request_cached
 from openedx.core.lib.graph_traversals import traverse_pre_order
+from openedx.core.djangoapps.authz.decorators import authz_permission_required
 from xmodule.modulestore.django import modulestore  # lint-amnesty, pylint: disable=wrong-import-order
 
-from .utils import course_author_access_required, get_bool_param
+from .utils import get_bool_param
 
 log = logging.getLogger(__name__)
 
@@ -82,7 +84,7 @@ class CourseQualityView(DeveloperErrorViewMixin, GenericAPIView):
     # does not specify a serializer class.
     swagger_schema = None
 
-    @course_author_access_required
+    @authz_permission_required(COURSES_VIEW_COURSE.identifier, legacy_permission="read")
     def get(self, request, course_key):
         """
         Returns validation information for the given course.
diff --git a/cms/djangoapps/contentstore/api/views/course_validation.py b/cms/djangoapps/contentstore/api/views/course_validation.py
index d3565fc1688d..986576a29e72 100644
--- a/cms/djangoapps/contentstore/api/views/course_validation.py
+++ b/cms/djangoapps/contentstore/api/views/course_validation.py
@@ -11,6 +11,7 @@
 from rest_framework.response import Response
 from user_tasks.models import UserTaskStatus
 from user_tasks.views import StatusViewSet
+from openedx_authz.constants.permissions import COURSES_VIEW_COURSE
 
 from cms.djangoapps.contentstore.course_info_model import get_course_updates
 from cms.djangoapps.contentstore.tasks import migrate_course_legacy_library_blocks_to_item_bank
@@ -19,6 +20,7 @@
 from openedx.core.lib.api.authentication import BearerAuthenticationAllowInactiveUser
 from openedx.core.lib.api.serializers import StatusSerializerWithUuid
 from openedx.core.lib.api.view_utils import DeveloperErrorViewMixin, view_auth_classes
+from openedx.core.djangoapps.authz.decorators import authz_permission_required
 from xmodule.course_metadata_utils import DEFAULT_GRADING_POLICY  # lint-amnesty, pylint: disable=wrong-import-order
 from xmodule.modulestore.django import modulestore  # lint-amnesty, pylint: disable=wrong-import-order
 
@@ -80,7 +82,7 @@ class CourseValidationView(DeveloperErrorViewMixin, GenericAPIView):
     # does not specify a serializer class.
     swagger_schema = None
 
-    @course_author_access_required
+    @authz_permission_required(COURSES_VIEW_COURSE.identifier, legacy_permission="read")
     def get(self, request, course_key):
         """
         Returns validation information for the given course.
diff --git a/cms/djangoapps/contentstore/api/views/utils.py b/cms/djangoapps/contentstore/api/views/utils.py
index 5d7dd8ff06d1..3426cc5156ba 100644
--- a/cms/djangoapps/contentstore/api/views/utils.py
+++ b/cms/djangoapps/contentstore/api/views/utils.py
@@ -120,7 +120,7 @@ def course_author_access_required(view):
     Usage::
         @course_author_access_required
         def my_view(request, course_key):
-            # Some functionality ...
+            # Some functionality...
     """
     def _wrapper_view(self, request, course_id, *args, **kwargs):
         """
diff --git a/cms/djangoapps/contentstore/views/import_export.py b/cms/djangoapps/contentstore/views/import_export.py
index 22310faa1ae2..554d1108264d 100644
--- a/cms/djangoapps/contentstore/views/import_export.py
+++ b/cms/djangoapps/contentstore/views/import_export.py
@@ -11,6 +11,7 @@
 import re
 import shutil
 from wsgiref.util import FileWrapper
+from openedx_authz.constants.permissions import COURSES_EXPORT_COURSE, COURSES_IMPORT_COURSE
 
 from django.conf import settings
 from django.contrib.auth.decorators import login_required
@@ -32,8 +33,8 @@
 from user_tasks.conf import settings as user_tasks_settings
 from user_tasks.models import UserTaskArtifact, UserTaskStatus
 
+from openedx.core.djangoapps.authz.decorators import user_has_course_permission
 from common.djangoapps.edxmako.shortcuts import render_to_response
-from common.djangoapps.student.auth import has_course_author_access
 from common.djangoapps.util.json_request import JsonResponse
 from common.djangoapps.util.monitoring import monitor_import_failure
 from common.djangoapps.util.views import ensure_valid_course_key
@@ -87,7 +88,12 @@ def import_handler(request, course_key_string):
         successful_url = reverse_course_url('course_handler', courselike_key)
         context_name = 'context_course'
         courselike_block = modulestore().get_course(courselike_key)
-    if not has_course_author_access(request.user, courselike_key):
+    if not user_has_course_permission(
+        user=request.user,
+        authz_permission=COURSES_IMPORT_COURSE.identifier,
+        course_key=courselike_key,
+        legacy_permission="write"
+    ):
         raise PermissionDenied()
 
     if 'application/json' in request.META.get('HTTP_ACCEPT', 'application/json'):
@@ -257,7 +263,12 @@ def import_status_handler(request, course_key_string, filename=None):
 
     """
     course_key = CourseKey.from_string(course_key_string)
-    if not has_course_author_access(request.user, course_key):
+    if not user_has_course_permission(
+        user=request.user,
+        authz_permission=COURSES_IMPORT_COURSE.identifier,
+        course_key=course_key,
+        legacy_permission="write"
+    ):
         raise PermissionDenied()
 
     # The task status record is authoritative once it's been created
@@ -318,7 +329,12 @@ def export_handler(request, course_key_string):
     a link appearing on the page once it's ready.
     """
     course_key = CourseKey.from_string(course_key_string)
-    if not has_course_author_access(request.user, course_key):
+    if not user_has_course_permission(
+        user=request.user,
+        authz_permission=COURSES_EXPORT_COURSE.identifier,
+        course_key=course_key,
+        legacy_permission="write"
+    ):
         raise PermissionDenied()
     library = isinstance(course_key, LibraryLocator)
     if library:
@@ -373,7 +389,12 @@ def export_status_handler(request, course_key_string):
     returned.
     """
     course_key = CourseKey.from_string(course_key_string)
-    if not has_course_author_access(request.user, course_key):
+    if not user_has_course_permission(
+        user=request.user,
+        authz_permission=COURSES_EXPORT_COURSE.identifier,
+        course_key=course_key,
+        legacy_permission="write"
+    ):
         raise PermissionDenied()
 
     # The task status record is authoritative once it's been created
@@ -435,7 +456,12 @@ def export_output_handler(request, course_key_string):
     filesystem instead of an external service like S3.
     """
     course_key = CourseKey.from_string(course_key_string)
-    if not has_course_author_access(request.user, course_key):
+    if not user_has_course_permission(
+        user=request.user,
+        authz_permission=COURSES_EXPORT_COURSE.identifier,
+        course_key=course_key,
+        legacy_permission="write"
+    ):
         raise PermissionDenied()
 
     task_status = _latest_task_status(request, course_key_string, export_output_handler)
diff --git a/cms/djangoapps/contentstore/views/tests/test_import_export.py b/cms/djangoapps/contentstore/views/tests/test_import_export.py
index 0f11338a4c59..f3512a9b845f 100644
--- a/cms/djangoapps/contentstore/views/tests/test_import_export.py
+++ b/cms/djangoapps/contentstore/views/tests/test_import_export.py
@@ -29,9 +29,13 @@
 from path import Path as path
 from storages.backends.s3boto3 import S3Boto3Storage
 from user_tasks.models import UserTaskStatus
+from rest_framework import status
+from rest_framework.test import APIClient
+
 
 from cms.djangoapps.contentstore import toggles
 from cms.djangoapps.contentstore import errors as import_error
+from cms.djangoapps.contentstore.api.tests.base import BaseCourseViewTest
 from cms.djangoapps.contentstore.storage import course_import_export_storage
 from cms.djangoapps.contentstore.tests.test_libraries import LibraryTestCase
 from cms.djangoapps.contentstore.tests.utils import CourseTestCase
@@ -39,8 +43,11 @@
 from cms.djangoapps.models.settings.course_metadata import CourseMetadata
 from common.djangoapps.student import auth
 from common.djangoapps.student.roles import CourseInstructorRole, CourseStaffRole
+from common.djangoapps.student.tests.factories import UserFactory
 from common.djangoapps.util import milestones_helpers
+from openedx.core.djangoapps.authz.tests.mixins import CourseAuthzTestMixin
 from openedx.core.lib.extract_archive import safe_extractall
+from openedx_authz.constants.roles import COURSE_STAFF
 from xmodule.contentstore.django import contentstore
 from xmodule.modulestore import LIBRARY_ROOT, ModuleStoreEnum
 from xmodule.modulestore.django import modulestore
@@ -1367,3 +1374,305 @@ def test_problem_content_on_course_export_import(self, problem_data, expected_pr
         )
 
         self.assert_problem_definition(dest_course.location, expected_problem_content)
+
+
+class ImportAuthzTest(CourseAuthzTestMixin, BaseCourseViewTest):
+    """
+    Tests Course Import Course authorization using openedx-authz.
+    """
+
+    view_name = 'import_handler'
+    course_key_arg_name = 'course_key_string'
+    authz_roles_to_assign = [COURSE_STAFF.external_key]
+
+
+    def setUp(self):
+        super().setUp()
+
+        self.content_dir = path(tempfile.mkdtemp())
+        self.addCleanup(shutil.rmtree, self.content_dir)
+
+        # Create tar test files -----------------------------------------------
+        # OK course:
+        good_dir = tempfile.mkdtemp(dir=self.content_dir)
+        # test course being deeper down than top of tar file
+        embedded_dir = os.path.join(good_dir, "grandparent", "parent")
+        os.makedirs(os.path.join(embedded_dir, "course"))
+        with open(os.path.join(embedded_dir, "course.xml"), "w+") as f:
+            f.write('<course url_name="2013_Spring" org="EDx" course="0.00x"/>')
+
+        with open(os.path.join(embedded_dir, "course", "2013_Spring.xml"), "w+") as f:
+            f.write('<course></course>')
+
+        self.file_to_upload = os.path.join(self.content_dir, "good.tar.gz")
+        with tarfile.open(self.file_to_upload, "w:gz") as gtar:
+            gtar.add(good_dir)
+
+    def import_file_in_course(self, client, course_key: str = None):
+        """Helper method to import provided file in the course."""
+        with open(self.file_to_upload, 'rb') as file_data:
+            args = {"name": self.file_to_upload, "course-data": [file_data]}
+            course_key = course_key or str(self.course_key)
+            url = self.get_url(course_key)
+            return client.post(url, args)
+
+    def test_authorized_user_can_access(self):
+        """User with COURSE_STAFF role can access."""
+        self.authorized_client.login(username=self.authorized_user.username, password=self.password)
+        resp = self.import_file_in_course(self.authorized_client)
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+    def test_unauthorized_user_cannot_access(self):
+        """User without role cannot access."""
+        self.unauthorized_client.login(username=self.unauthorized_user.username, password=self.password)
+        resp = self.import_file_in_course(self.unauthorized_client)
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_role_scoped_to_course(self):
+        """Authorization should only apply to the assigned course."""
+        other_course = self.store.create_course("OtherOrg", "OtherCourse", "Run", self.staff.id)
+        self.authorized_client.login(username=self.authorized_user.username, password=self.password)
+        resp = self.import_file_in_course(self.authorized_client, other_course.id)
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_staff_user_allowed_via_legacy(self):
+        """
+        Staff users should still pass through legacy fallback.
+        """
+        self.client.login(username=self.staff.username, password=self.password)
+        resp = self.import_file_in_course(self.client)
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+    def test_superuser_allowed(self):
+        """Superusers should always be allowed."""
+        superuser = UserFactory(is_superuser=True, username='superuser', password=self.password)
+
+        client = APIClient()
+        client.login(username=superuser.username, password=self.password)
+
+        resp = self.import_file_in_course(client)
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+
+class ImportStatusAuthzTest(CourseAuthzTestMixin, BaseCourseViewTest):
+    """
+    Tests Course Import Course Staus authorization using openedx-authz.
+    """
+
+    view_name = 'import_status_handler'
+    course_key_arg_name = 'course_key_string'
+    extra_request_args = {'filename': 'test.xml'}
+    authz_roles_to_assign = [COURSE_STAFF.external_key]
+
+    def test_authorized_user_can_access(self):
+        """User with COURSE_STAFF role can access."""
+        self.authorized_client.login(username=self.authorized_user.username, password=self.password)
+        resp = self.authorized_client.get(self.get_url(self.course_key))
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+    def test_unauthorized_user_cannot_access(self):
+        """User without role cannot access."""
+        self.unauthorized_client.login(username=self.unauthorized_user.username, password=self.password)
+        resp = self.unauthorized_client.get(self.get_url(self.course_key))
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_role_scoped_to_course(self):
+        """Authorization should only apply to the assigned course."""
+        other_course = self.store.create_course("OtherOrg", "OtherCourse", "Run", self.staff.id)
+
+        self.authorized_client.login(username=self.authorized_user.username, password=self.password)
+        resp = self.authorized_client.get(self.get_url(other_course.id))
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_staff_user_allowed_via_legacy(self):
+        """
+        Staff users should still pass through legacy fallback.
+        """
+        self.client.login(username=self.staff.username, password=self.password)
+
+        resp = self.client.get(self.get_url(self.course_key))
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+    def test_superuser_allowed(self):
+        """Superusers should always be allowed."""
+        superuser = UserFactory(is_superuser=True, username='superuser', password=self.password)
+
+        client = APIClient()
+        client.login(username=superuser.username, password=self.password)
+
+        resp = client.get(self.get_url(self.course_key))
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+
+class ExportStatusAuthzTest(CourseAuthzTestMixin, BaseCourseViewTest):
+    """
+    Tests Course Export Course Status authorization using openedx-authz.
+    """
+
+    view_name = 'export_status_handler'
+    course_key_arg_name = 'course_key_string'
+    authz_roles_to_assign = [COURSE_STAFF.external_key]
+
+    def test_authorized_user_can_access(self):
+        """User with COURSE_STAFF role can access."""
+        self.authorized_client.login(username=self.authorized_user.username, password=self.password)
+        resp = self.authorized_client.get(self.get_url(self.course_key))
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+    def test_unauthorized_user_cannot_access(self):
+        """User without role cannot access."""
+        self.unauthorized_client.login(username=self.unauthorized_user.username, password=self.password)
+        resp = self.unauthorized_client.get(self.get_url(self.course_key))
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_role_scoped_to_course(self):
+        """Authorization should only apply to the assigned course."""
+        other_course = self.store.create_course("OtherOrg", "OtherCourse", "Run", self.staff.id)
+
+        self.authorized_client.login(username=self.authorized_user.username, password=self.password)
+        resp = self.authorized_client.get(self.get_url(other_course.id))
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_staff_user_allowed_via_legacy(self):
+        """
+        Staff users should still pass through legacy fallback.
+        """
+        self.client.login(username=self.staff.username, password=self.password)
+
+        resp = self.client.get(self.get_url(self.course_key))
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+    def test_superuser_allowed(self):
+        """Superusers should always be allowed."""
+        superuser = UserFactory(is_superuser=True, username='superuser', password=self.password)
+
+        client = APIClient()
+        client.login(username=superuser.username, password=self.password)
+
+        resp = client.get(self.get_url(self.course_key))
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+
+class ExportAuthzTest(CourseAuthzTestMixin, BaseCourseViewTest):
+    """
+    Tests Course Export Course authorization using openedx-authz.
+    """
+
+    view_name = 'export_handler'
+    course_key_arg_name = 'course_key_string'
+    authz_roles_to_assign = [COURSE_STAFF.external_key]
+
+    def test_authorized_user_can_access(self):
+        """User with COURSE_STAFF role can access."""
+        self.authorized_client.login(username=self.authorized_user.username, password=self.password)
+        resp = self.authorized_client.post(self.get_url(self.course_key))
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+    def test_unauthorized_user_cannot_access(self):
+        """User without role cannot access."""
+        self.unauthorized_client.login(username=self.unauthorized_user.username, password=self.password)
+        resp = self.unauthorized_client.post(self.get_url(self.course_key))
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_role_scoped_to_course(self):
+        """Authorization should only apply to the assigned course."""
+        other_course = self.store.create_course("OtherOrg", "OtherCourse", "Run", self.staff.id)
+
+        self.authorized_client.login(username=self.authorized_user.username, password=self.password)
+        resp = self.authorized_client.post(self.get_url(other_course.id))
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_staff_user_allowed_via_legacy(self):
+        """
+        Staff users should still pass through legacy fallback.
+        """
+        self.client.login(username=self.staff.username, password=self.password)
+
+        resp = self.client.post(self.get_url(self.course_key))
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+    def test_superuser_allowed(self):
+        """Superusers should always be allowed."""
+        superuser = UserFactory(is_superuser=True, username='superuser', password=self.password)
+
+        client = APIClient()
+        client.login(username=superuser.username, password=self.password)
+
+        resp = client.post(self.get_url(self.course_key))
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+
+class ExportOutputAuthzTest(CourseAuthzTestMixin, BaseCourseViewTest):
+    """
+    Tests Course Export Course Output authorization using openedx-authz.
+    """
+
+    view_name = 'export_output_handler'
+    course_key_arg_name = 'course_key_string'
+    authz_roles_to_assign = [COURSE_STAFF.external_key]
+
+    def _mock_artifact(self, spec=None, file_url=None):
+        """
+        Creates a Mock of the UserTaskArtifact model for testing exports handler
+        code without touching the database.
+        """
+        mock_artifact = Mock()
+        mock_artifact.file.name = 'testfile.tar.gz'
+        mock_artifact.file.storage = Mock(spec=spec)
+        mock_artifact.file.storage.url.return_value = file_url
+        return mock_artifact
+
+    def test_authorized_user_can_access(self):
+        """User with COURSE_STAFF role can access."""
+        self.authorized_client.login(username=self.authorized_user.username, password=self.password)
+        self.authorized_client.post(reverse_course_url('export_handler', self.course_key))
+        resp = self.authorized_client.get(self.get_url(self.course_key))
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+    def test_unauthorized_user_cannot_access(self):
+        """User without role cannot access."""
+        self.unauthorized_client.login(username=self.unauthorized_user.username, password=self.password)
+        resp = self.unauthorized_client.get(self.get_url(self.course_key))
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_role_scoped_to_course(self):
+        """Authorization should only apply to the assigned course."""
+        other_course = self.store.create_course("OtherOrg", "OtherCourse", "Run", self.staff.id)
+        self.authorized_client.login(username=self.authorized_user.username, password=self.password)
+        resp = self.authorized_client.get(self.get_url(other_course.id))
+        self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_staff_user_allowed_via_legacy(self):
+        """
+        Staff users should still pass through legacy fallback.
+        """
+        self.client.login(username=self.staff.username, password=self.password)
+        self.client.post(reverse_course_url('export_handler', self.course_key))
+        resp = self.client.get(self.get_url(self.course_key))
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+
+    @patch('user_tasks.models.UserTaskArtifact.objects.get')
+    @patch('cms.djangoapps.contentstore.views.import_export._latest_task_status')
+    @patch('cms.djangoapps.contentstore.views.import_export.course_import_export_storage')
+    def test_superuser_allowed(
+        self,
+        mock_get_user_task_artifact,
+        mock_latest_task_status,
+        mock_storage,
+    ):
+        """Superusers should always be allowed."""
+        mock_latest_task_status.return_value = Mock(state=UserTaskStatus.SUCCEEDED)
+        mock_get_user_task_artifact.return_value = self._mock_artifact(
+            file_url='/path/to/testfile.tar.gz',
+        )
+        mock_tarball = Mock()
+        mock_tarball.name = 'testfile.tar.gz'
+        mock_storage.open.return_value = mock_tarball
+        mock_storage.size.return_value = 0
+
+        superuser = UserFactory(is_superuser=True, username='superuser', password=self.password)
+
+        client = APIClient()
+        client.login(username=superuser.username, password=self.password)
+        resp = client.get(self.get_url(self.course_key))
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
\ No newline at end of file
diff --git a/lms/envs/common.py b/lms/envs/common.py
index 27e0306711e1..4e8b0323bed7 100644
--- a/lms/envs/common.py
+++ b/lms/envs/common.py
@@ -2018,6 +2018,9 @@
     # Notifications
     'openedx.core.djangoapps.notifications',
 
+    # Authz
+    'openedx.core.djangoapps.authz.apps.AuthzConfig',
+
     'openedx_events',
 
     # Core models to represent courses
diff --git a/openedx/core/djangoapps/authz/__init__.py b/openedx/core/djangoapps/authz/__init__.py
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/openedx/core/djangoapps/authz/apps.py b/openedx/core/djangoapps/authz/apps.py
new file mode 100644
index 000000000000..c1fc84dc751c
--- /dev/null
+++ b/openedx/core/djangoapps/authz/apps.py
@@ -0,0 +1,9 @@
+"""Django app configuration for authz app."""
+
+from django.apps import AppConfig
+
+
+class AuthzConfig(AppConfig):
+    default_auto_field = 'django.db.models.BigAutoField'
+    name = 'openedx.core.djangoapps.authz'
+    verbose_name = "Authz"
diff --git a/openedx/core/djangoapps/authz/decorators.py b/openedx/core/djangoapps/authz/decorators.py
new file mode 100644
index 000000000000..18ceae47111d
--- /dev/null
+++ b/openedx/core/djangoapps/authz/decorators.py
@@ -0,0 +1,108 @@
+"""Decorators for AuthZ-based permissions enforcement."""
+import logging
+from functools import wraps
+
+from opaque_keys import InvalidKeyError
+from opaque_keys.edx.keys import CourseKey, UsageKey
+from openedx_authz import api as authz_api
+from rest_framework import status
+
+from common.djangoapps.student.auth import has_studio_write_access, has_studio_read_access
+from openedx.core import toggles as core_toggles
+from openedx.core.lib.api.view_utils import DeveloperErrorViewMixin
+
+log = logging.getLogger(__name__)
+
+
+legacy_permission_handler_map = {
+    "read": has_studio_read_access,
+    "write": has_studio_write_access,
+}
+
+
+def authz_permission_required(authz_permission, legacy_permission=None):
+    """
+    Decorator enforcing course author permissions via AuthZ
+    with optional legacy fallback.
+    """
+
+    def decorator(view_func):
+
+        @wraps(view_func)
+        def _wrapped_view(self, request, course_id, *args, **kwargs):
+            course_key = get_course_key(course_id)
+
+            if not user_has_course_permission(
+                request.user,
+                authz_permission,
+                course_key,
+                legacy_permission
+            ):
+                raise DeveloperErrorViewMixin.api_error(
+                    status_code=status.HTTP_403_FORBIDDEN,
+                    developer_message="The requesting user does not have course author permissions.",
+                    error_code="user_permissions",
+                )
+
+            return view_func(self, request, course_key, *args, **kwargs)
+
+        return _wrapped_view
+
+    return decorator
+
+
+def user_has_course_permission(user, authz_permission, course_key, legacy_permission=None):
+    """
+    Core authorization logic.
+    """
+    if core_toggles.enable_authz_course_authoring(course_key):
+        # If AuthZ is enabled for this course, check the permission via AuthZ only.
+        is_user_allowed = authz_api.is_user_allowed(user.username, authz_permission, str(course_key))
+        log.info(
+            "AuthZ permission granted = {}".format(is_user_allowed),
+            extra={
+                "user_id": user.id,
+                "authz_permission": authz_permission,
+                "course_key": str(course_key),
+            },
+        )
+        return is_user_allowed
+
+    # If AuthZ is not enabled for this course, fall back to legacy course author
+    # access check if legacy_permission is provided.
+    has_legacy_permission = legacy_permission_handler_map.get(legacy_permission)
+    if legacy_permission and has_legacy_permission and has_legacy_permission(user, course_key):
+        log.info(
+            "AuthZ fallback used",
+            extra={
+                "user_id": user.id,
+                "authz_permission": authz_permission,
+                "legacy_permission": legacy_permission,
+                "course_key": str(course_key),
+            },
+        )
+        return True
+
+    log.info(
+        "AuthZ permission denied",
+        extra={
+            "user_id": user.id,
+            "authz_permission": authz_permission,
+            "course_key": str(course_key),
+        },
+    )
+    return False
+
+
+def get_course_key(course_id):
+    """
+    Given a course_id string, attempts to parse it as a CourseKey.
+    If that fails, attempts to parse it as a UsageKey and extract the course key from it.
+    """
+    try:
+        return CourseKey.from_string(course_id)
+    except InvalidKeyError:
+        # If the course_id doesn't match the COURSE_KEY_PATTERN, it might be a usage key.
+        # Attempt to parse it as such and extract the course key.
+        usage_key = UsageKey.from_string(course_id)
+        return usage_key.course_key
diff --git a/openedx/core/djangoapps/authz/migrations/__init__.py b/openedx/core/djangoapps/authz/migrations/__init__.py
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/openedx/core/djangoapps/authz/tests/__init__.py b/openedx/core/djangoapps/authz/tests/__init__.py
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/openedx/core/djangoapps/authz/tests/mixins.py b/openedx/core/djangoapps/authz/tests/mixins.py
new file mode 100644
index 000000000000..c31d5dc98eba
--- /dev/null
+++ b/openedx/core/djangoapps/authz/tests/mixins.py
@@ -0,0 +1,86 @@
+""" Mixins for testing course-scoped AuthZ endpoints. """
+
+import casbin
+import pkg_resources
+
+from unittest.mock import patch
+from rest_framework.test import APIClient
+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 openedx.core import toggles as core_toggles
+from common.djangoapps.student.tests.factories import UserFactory
+
+
+class CourseAuthzTestMixin:
+    """
+    Reusable mixin for testing course-scoped AuthZ endpoints.
+    """
+
+    authz_roles_to_assign = [COURSE_STAFF.external_key]
+
+    @classmethod
+    def setUpClass(cls):
+        cls.toggle_patcher = patch.object(
+            core_toggles.AUTHZ_COURSE_AUTHORING_FLAG,
+            "is_enabled",
+            return_value=True
+        )
+        cls.toggle_patcher.start()
+
+        super().setUpClass()
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.toggle_patcher.stop()
+        super().tearDownClass()
+
+    def setUp(self):
+        super().setUp()
+
+        self._seed_database_with_policies()
+
+        self.authorized_user = UserFactory(username='authorized', password='test')
+        self.unauthorized_user = UserFactory(username='unauthorized', password='test')
+
+        for role in self.authz_roles_to_assign:
+            assign_role_to_user_in_scope(
+                self.authorized_user.username,
+                role,
+                str(self.course_key)
+            )
+
+        AuthzEnforcer.get_enforcer().load_policy()
+
+        self.authorized_client = APIClient()
+        self.authorized_client.force_authenticate(user=self.authorized_user)
+
+        self.unauthorized_client = APIClient()
+        self.unauthorized_client.force_authenticate(user=self.unauthorized_user)
+
+    def tearDown(self):
+        super().tearDown()
+        AuthzEnforcer.get_enforcer().clear_policy()
+
+    @classmethod
+    def _seed_database_with_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,
+        )
diff --git a/openedx/core/djangoapps/authz/tests/test_decorators.py b/openedx/core/djangoapps/authz/tests/test_decorators.py
new file mode 100644
index 000000000000..0c8dca53fadd
--- /dev/null
+++ b/openedx/core/djangoapps/authz/tests/test_decorators.py
@@ -0,0 +1,155 @@
+"""Tests for authz decorators."""
+from unittest.mock import Mock, patch
+
+from django.test import RequestFactory, TestCase
+from opaque_keys.edx.locator import BlockUsageLocator, CourseLocator
+
+from openedx.core.djangoapps.authz.decorators import authz_permission_required, get_course_key
+from openedx.core.lib.api.view_utils import DeveloperErrorResponseException
+
+
+class AuthzPermissionRequiredDecoratorTests(TestCase):
+    """
+    Tests focused on the authz_permission_required decorator behavior.
+    """
+
+    def setUp(self):
+        self.factory = RequestFactory()
+        self.course_key = CourseLocator("TestX", "TST101", "2025")
+
+        self.user = Mock()
+        self.user.username = "testuser"
+        self.user.id = 1
+
+        self.view_instance = Mock()
+
+    def _build_request(self):
+        request = self.factory.get("/test")
+        request.user = self.user
+        return request
+
+    def test_view_executes_when_permission_granted(self):
+        """Decorator allows execution when permission check passes."""
+        request = self._build_request()
+
+        mock_view = Mock(return_value="success")
+
+        with patch(
+            "openedx.core.djangoapps.authz.decorators.user_has_course_permission",
+            return_value=True,
+        ):
+            decorated = authz_permission_required("courses.view")(mock_view)
+
+            result = decorated(self.view_instance, request, str(self.course_key))
+
+        self.assertEqual(result, "success")
+        mock_view.assert_called_once_with(
+            self.view_instance,
+            request,
+            self.course_key,
+        )
+
+    def test_view_executes_when_legacy_fallback_read(self):
+        """Decorator allows execution when AuthZ denies but legacy permission succeeds."""
+        request = self._build_request()
+
+        mock_view = Mock(return_value="success")
+
+        with patch(
+            "openedx.core.djangoapps.authz.decorators.core_toggles.enable_authz_course_authoring",
+            return_value=False,
+        ), patch(
+            "openedx.core.djangoapps.authz.decorators.authz_api.is_user_allowed",
+            return_value=True,  # Should not be used when AuthZ is disabled, but set to True just in case
+        ), patch(
+            "openedx.core.djangoapps.authz.decorators.has_studio_read_access",
+            return_value=True,
+        ):
+            decorated = authz_permission_required(
+                "courses.view",
+                legacy_permission="read"
+            )(mock_view)
+
+            result = decorated(self.view_instance, request, str(self.course_key))
+
+        self.assertEqual(result, "success")
+        mock_view.assert_called_once()
+
+    def test_view_executes_when_legacy_fallback_write(self):
+        """Decorator allows execution when AuthZ denies but legacy write permission succeeds."""
+        request = self._build_request()
+
+        mock_view = Mock(return_value="success")
+
+        with patch(
+            "openedx.core.djangoapps.authz.decorators.core_toggles.enable_authz_course_authoring",
+            return_value=False,
+        ), patch(
+            "openedx.core.djangoapps.authz.decorators.authz_api.is_user_allowed",
+            return_value=True,  # Should not be used when AuthZ is disabled, but set to True just in case
+        ), patch(
+            "openedx.core.djangoapps.authz.decorators.has_studio_write_access",
+            return_value=True,
+        ):
+            decorated = authz_permission_required(
+                "courses.edit",
+                legacy_permission="write"
+            )(mock_view)
+
+            result = decorated(self.view_instance, request, str(self.course_key))
+
+        self.assertEqual(result, "success")
+        mock_view.assert_called_once()
+
+    def test_access_denied_when_permission_fails(self):
+        """Decorator raises API error when permission fails."""
+        request = self._build_request()
+
+        mock_view = Mock()
+
+        with patch(
+            "openedx.core.djangoapps.authz.decorators.user_has_course_permission",
+            return_value=False,
+        ):
+            decorated = authz_permission_required("courses.view")(mock_view)
+
+            with self.assertRaises(DeveloperErrorResponseException) as context:
+                decorated(self.view_instance, request, str(self.course_key))
+
+        self.assertEqual(context.exception.response.status_code, 403)
+        mock_view.assert_not_called()
+
+    def test_decorator_preserves_function_name(self):
+        """Decorator preserves wrapped function metadata."""
+
+        def sample_view(self, request, course_key):
+            return "ok"
+
+        decorated = authz_permission_required("courses.view")(sample_view)
+
+        self.assertEqual(decorated.__name__, "sample_view")
+
+
+class GetCourseKeyTests(TestCase):
+    """Tests for the get_course_key function used in the authz decorators."""
+
+    def setUp(self):
+        self.course_key = CourseLocator("TestX", "TST101", "2025")
+
+    def test_course_key_string(self):
+        """Valid course key string returns CourseKey."""
+        result = get_course_key(str(self.course_key))
+
+        self.assertEqual(result, self.course_key)
+
+    def test_usage_key_string(self):
+        """UsageKey string resolves to course key."""
+        usage_key = BlockUsageLocator(
+            self.course_key,
+            "html",
+            "block1"
+        )
+
+        result = get_course_key(str(usage_key))
+
+        self.assertEqual(result, self.course_key)
diff --git a/openedx/core/djangoapps/content_tagging/auth.py b/openedx/core/djangoapps/content_tagging/auth.py
index 4a1157c3bff3..75e91f2faaf0 100644
--- a/openedx/core/djangoapps/content_tagging/auth.py
+++ b/openedx/core/djangoapps/content_tagging/auth.py
@@ -1,12 +1,37 @@
 """
 Functions to validate the access in content tagging actions
 """
+import logging
 
-
+from opaque_keys import InvalidKeyError
+from opaque_keys.edx.keys import CourseKey
+from openedx_authz import api as authz_api
+from openedx_authz.constants.permissions import COURSES_EXPORT_TAGS
 from openedx_tagging import rules as oel_tagging_rules
 
+from openedx.core import toggles as core_toggles
+
+log = logging.getLogger(__name__)
+
 
 def has_view_object_tags_access(user, object_id):
+    """
+    Check if the user has access to view object tags for the given object.
+    """
+    # If authz is enabled, check for the export tags authz permission
+    course_key = None
+    try:
+        course_key = CourseKey.from_string(object_id)
+    except InvalidKeyError:
+        log.warning("Invalid course key %s", object_id)
+
+    if course_key and core_toggles.enable_authz_course_authoring(course_key) and not authz_api.is_user_allowed(
+        user.username, COURSES_EXPORT_TAGS.identifier, str(course_key)
+    ):
+        return False
+
+
+    # Always check for tagging permissions
     return user.has_perm(
         "oel_tagging.view_objecttag",
         # The obj arg expects a model, but we are passing an object