feat: add course authoring migration and rollback scripts

CHANGELOG.rst

@@ -14,6 +14,18 @@ Change Log
 Unreleased
 **********
 
+0.23.0 - 2026-02-18
+********************
+
+Added
+=====
+
+* Add authz_migrate_course_authoring command to migrate legacy CourseAccessRole data to the new Authz (Casbin-based) system
+* Add authz_rollback_course_authoring command to rollback Authz roles back to legacy CourseAccessRole
+* Support optional --delete flag for controlled cleanup of source permissions after successful migration
+* Add migrate_legacy_course_roles_to_authz and migrate_authz_to_legacy_course_roles service functions
+* Add unit tests to verify migration and command behavior
+
 Added
 =====
 

openedx_authz/__init__.py

@@ -4,6 +4,6 @@
 
 import os
 
-__version__ = "0.22.0"
+__version__ = "0.23.0"
 
 ROOT_DIRECTORY = os.path.dirname(os.path.abspath(__file__))

openedx_authz/engine/utils.py

@@ -5,17 +5,33 @@
 """
 
 import logging
+from collections import defaultdict
 
 from casbin import Enforcer
 
-from openedx_authz.api.users import assign_role_to_user_in_scope, batch_assign_role_to_users_in_scope
-from openedx_authz.constants.roles import LIBRARY_ADMIN, LIBRARY_AUTHOR, LIBRARY_USER
+from openedx_authz.api.data import CourseOverviewData
+from openedx_authz.api.users import (
+    assign_role_to_user_in_scope,
+    batch_assign_role_to_users_in_scope,
+    batch_unassign_role_from_users,
+    get_user_role_assignments,
+)
+from openedx_authz.constants.roles import (
+    LEGACY_COURSE_ROLE_EQUIVALENCES,
+    LIBRARY_ADMIN,
+    LIBRARY_AUTHOR,
+    LIBRARY_USER,
+)
 
 logger = logging.getLogger(__name__)
 
 GROUPING_POLICY_PTYPES = ["g", "g2", "g3", "g4", "g5", "g6"]
 
 
+# Map new roles back to legacy roles for rollback purposes
+COURSE_ROLE_EQUIVALENCES = {v: k for k, v in LEGACY_COURSE_ROLE_EQUIVALENCES.items()}
+
+
 def migrate_policy_between_enforcers(
     source_enforcer: Enforcer,
     target_enforcer: Enforcer,
@@ -151,3 +167,203 @@ def migrate_legacy_permissions(ContentLibraryPermission):
             )
 
     return permissions_with_errors
+
+
+def migrate_legacy_course_roles_to_authz(course_access_role_model, course_id_list, org_id, delete_after_migration):
+    """
+    Migrate legacy course role data to the new Casbin-based authorization model.
+    This function reads legacy permissions from the CourseAccessRole model
+    and assigns equivalent roles in the new authorization system.
+
+    The old Course permissions are stored in the CourseAccessRole model, it consists of the following columns:
+
+    - user: FK to User
+    - org: optional Organization string
+    - course_id: optional CourseKeyField of Course
+    - role: 'instructor' | 'staff' | 'limited_staff' | 'data_researcher'
+
+    In the new Authz model, this would roughly translate to:
+
+    - course_id: scope
+    - user: subject
+    - role: role
+
+    param course_access_role_model: It should be the CourseAccessRole model. This is passed in because the function
+    is intended to run within a Django migration context, where direct model imports can cause issues.
+    param course_id_list: Optional list of course IDs to filter the migration.
+    param org_id: Optional organization ID to filter the migration.
+    param delete_after_migration: Whether to delete successfully migrated legacy permissions after migration.
+    """
+    if not course_id_list and not org_id:
+        raise ValueError(
+            "At least one of course_id_list or org_id must be provided to limit the scope of the migration."
+        )
+    course_access_role_filter = {
+        "course_id__startswith": "course-v1:",
+    }
+
+    if org_id:
+        course_access_role_filter["org"] = org_id
+
+    if course_id_list and not org_id:
+        # Only filter by course_id if org_id is not provided,
+        # otherwise we will filter by org_id which is more efficient
+        course_access_role_filter["course_id__in"] = course_id_list
+
+    legacy_permissions = (
+        course_access_role_model.objects.filter(**course_access_role_filter).select_related("user").all()
+    )
+
+    # List to keep track of any permissions that could not be migrated
+    permissions_with_errors = []
+    permissions_with_no_errors = []
+
+    for permission in legacy_permissions:
+        # Migrate the permission to the new model
+
+        role = LEGACY_COURSE_ROLE_EQUIVALENCES.get(permission.role)
+        if role is None:
+            # This should not happen as there are no more access_levels defined
+            # in CourseAccessRole, log and skip
+            logger.error(f"Unknown access level: {permission.role} for User: {permission.user}")
+            permissions_with_errors.append(permission)
+            continue
+
+        # Permission applied to individual user
+        logger.info(
+            f"Migrating permission for User: {permission.user.username} "
+            f"to Role: {role} in Scope: {permission.course_id}"
+        )
+
+        is_user_added = assign_role_to_user_in_scope(
+            user_external_key=permission.user.username,
+            role_external_key=role,
+            scope_external_key=str(permission.course_id),
+        )
+
+        if not is_user_added:
+            logger.error(
+                f"Failed to migrate permission for User: {permission.user.username} "
+                f"to Role: {role} in Scope: {permission.course_id}"
+            )
+            permissions_with_errors.append(permission)
+            continue
+
+        permissions_with_no_errors.append(permission)
+
+    if delete_after_migration:
+        # Only delete permissions that were successfully migrated to avoid data loss.
+        course_access_role_model.objects.filter(id__in=[p.id for p in permissions_with_no_errors]).delete()
+        logger.info(f"Deleted {len(permissions_with_no_errors)} legacy permissions after successful migration.")
+        logger.info(f"Retained {len(permissions_with_errors)} legacy permissions that had errors during migration.")
+
+    return permissions_with_errors, permissions_with_no_errors
+
+
+def migrate_authz_to_legacy_course_roles(
+    course_access_role_model, user_subject_model, course_id_list, org_id, delete_after_migration
+):
+    """
+    Migrate permissions from the new Casbin-based authorization model back to the legacy CourseAccessRole model.
+    This function reads permissions from the Casbin enforcer and creates equivalent entries in the
+    CourseAccessRole model.
+
+    This is essentially the reverse of migrate_legacy_course_roles_to_authz and is intended
+    for rollback purposes in case of migration issues.
+
+    param course_access_role_model: It should be the CourseAccessRole model. This is passed in because the function
+    is intended to run within a Django migration context, where direct model imports can cause issues.
+    param user_subject_model: It should be the UserSubject model. This is passed in because the function
+    is intended to run within a Django migration context, where direct model imports can cause issues.
+    param course_id_list: Optional list of course IDs to filter the migration.
+    param org_id: Optional organization ID to filter the migration.
+    param delete_after_migration: Whether to unassign successfully migrated permissions
+    from the new model after migration.
+    """
+    if not course_id_list and not org_id:
+        raise ValueError(
+            "At least one of course_id_list or org_id must be provided to limit the scope of the rollback migration."
+        )
+
+    # 1. Get all users with course-related permissions in the new model by filtering
+    # UserSubjects that are linked to CourseScopes with a valid course overview.
+    course_subject_filter = {
+        "casbin_rules__scope__coursescope__course_overview__isnull": False,
+    }
+
+    if org_id:
+        course_subject_filter["casbin_rules__scope__coursescope__course_overview__org"] = org_id
+
+    if course_id_list and not org_id:
+        # Only filter by course_id if org_id is not provided,
+        # otherwise we will filter by org_id which is more efficient
+        course_subject_filter["casbin_rules__scope__coursescope__course_overview__id__in"] = course_id_list
+
+    course_subjects = user_subject_model.objects.filter(**course_subject_filter).select_related("user").distinct()
+
+    roles_with_errors = []
+    roles_with_no_errors = []
+    unassignments = defaultdict(list)
+
+    for course_subject in course_subjects:
+        user = course_subject.user
+        user_external_key = user.username
+
+        # 2. Get all role assignments for the user
+        role_assignments = get_user_role_assignments(user_external_key=user_external_key)
+
+        for assignment in role_assignments:
+            if not isinstance(assignment.scope, CourseOverviewData):
+                logger.error(f"Skipping role assignment for User: {user_external_key} due to missing course scope.")
+                continue
+
+            scope = assignment.scope.external_key
+
+            course_overview = assignment.scope.get_object()
+
+            for role in assignment.roles:
+                legacy_role = COURSE_ROLE_EQUIVALENCES.get(role.external_key)
+                if legacy_role is None:
+                    logger.error(f"Unknown role: {role} for User: {user_external_key}")
+                    roles_with_errors.append((user_external_key, role.external_key, scope))
+                    continue
+
+                try:
+                    # Create legacy CourseAccessRole entry
+                    course_access_role_model.objects.get_or_create(
+                        user=user,
+                        org=course_overview.org,
+                        course_id=scope,
+                        role=legacy_role,
+                    )
+                    roles_with_no_errors.append((user_external_key, role.external_key, scope))
+                except Exception as e:  # pylint: disable=broad-exception-caught
+                    logger.error(
+                        f"Error creating CourseAccessRole for User: "
+                        f"{user_external_key}, Role: {legacy_role}, Course: {scope}: {e}"
+                    )
+                    roles_with_errors.append((user_external_key, role.external_key, scope))
+                    continue
+
+                # If we successfully created the legacy role, we can add this role assignment
+                # to the unassignment list if delete_after_migration is True
+                if delete_after_migration:
+                    unassignments[(role.external_key, scope)].append(user_external_key)
+
+    # Once the loop is done, we can log summary of unassignments
+    # and perform batch unassignment if delete_after_migration is True
+    if delete_after_migration:
+        total_unassignments = sum(len(users) for users in unassignments.values())
+        logger.info(f"Total of {total_unassignments} role assignments unassigned after successful rollback migration.")
+        for (role_external_key, scope), users in unassignments.items():
+            logger.info(
+                f"Unassigned Role: {role_external_key} from {len(users)} users \n"
+                f"in Scope: {scope} after successful rollback migration."
+            )
+            batch_unassign_role_from_users(
+                users=users,
+                role_external_key=role_external_key,
+                scope_external_key=scope,
+            )
+
+    return roles_with_errors, roles_with_no_errors

openedx_authz/management/commands/authz_migrate_course_authoring.py

@@ -0,0 +1,87 @@
+"""
+Django management command to migrate legacy course authoring roles to the new Authz (Casbin-based) authorization system.
+"""
+
+from django.core.management.base import BaseCommand, CommandError
+from django.db import transaction
+
+from openedx_authz.engine.utils import migrate_legacy_course_roles_to_authz
+
+try:
+    from common.djangoapps.student.models import CourseAccessRole
+except ImportError:
+    CourseAccessRole = None  # type: ignore
+
+
+class Command(BaseCommand):
+    """
+    Django command to migrate legacy CourseAccessRole data
+    to the new Authz (Casbin-based) authorization system.
+    """
+
+    help = "Migrate legacy course authoring roles to the new Authz system."
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--delete",
+            action="store_true",
+            help="Delete legacy CourseAccessRole records after successful migration.",
+        )
+        parser.add_argument(
+            "--course-id-list",
+            nargs="+",
+            type=str,
+            help="Optional list of course IDs to filter the migration.",
+        )
+
+        parser.add_argument(
+            "--org-id",
+            type=str,
+            help="Optional organization ID to filter the migration.",
+        )
+
+    def handle(self, *args, **options):
+        delete_after_migration = options["delete"]
+        course_id_list = options.get("course_id_list")
+        org_id = options.get("org_id")
+
+        if not course_id_list and not org_id:
+            raise CommandError("You must specify either --course-id-list or --org-id to filter the migration.")
+
+        if course_id_list and org_id:
+            raise CommandError("You cannot use --course-id-list and --org-id together.")
+
+        self.stdout.write(self.style.WARNING("Starting legacy → Authz migration..."))
+
+        try:
+            if delete_after_migration:
+                confirm = input(
+                    "Are you sure you want to delete successfully migrated legacy roles? Type 'yes' to continue: "
+                )
+
+                if confirm != "yes":
+                    self.stdout.write(self.style.WARNING("Migration aborted."))
+                    return
+            with transaction.atomic():
+                errors, success = migrate_legacy_course_roles_to_authz(
+                    course_access_role_model=CourseAccessRole,
+                    course_id_list=course_id_list,
+                    org_id=org_id,
+                    delete_after_migration=delete_after_migration,
+                )
+
+                if errors:
+                    self.stdout.write(self.style.ERROR(f"Migration completed with {len(errors)} errors."))
+                else:
+                    self.stdout.write(
+                        self.style.SUCCESS(f"Migration completed successfully with {len(success)} roles migrated.")
+                    )
+
+                if delete_after_migration:
+                    self.stdout.write(self.style.SUCCESS(f"{len(success)} Legacy roles deleted successfully."))
+
+        except Exception as exc:
+            self.stdout.write(self.style.ERROR(f"Migration failed due to unexpected error: {exc}"))
+            raise
+
+        self.stdout.write(self.style.SUCCESS("Done."))