[FC-0099] fix: load policy in role and permissions views (#122)

Author: BryanttV

CHANGELOG.rst

@@ -14,7 +14,13 @@ Change Log
 Unreleased
 **********
 
-*
+0.12.0 - 2025-10-30
+********************
+
+Changed
+=======
+
+* Load authorization policies in permission class.
 
 0.11.2 - 2025-10-30
 ********************

openedx_authz/__init__.py

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

openedx_authz/rest_api/v1/permissions.py

@@ -5,6 +5,7 @@
 from rest_framework.permissions import BasePermission
 
 from openedx_authz import api
+from openedx_authz.engine.enforcer import AuthzEnforcer
 
 
 class PermissionMeta(type(BasePermission)):
@@ -182,6 +183,7 @@ def has_permission(self, request, view) -> bool:
         """
         if request.user.is_superuser or request.user.is_staff:
             return True
+        AuthzEnforcer.get_enforcer().load_policy()
         return self._get_permission_instance(request).has_permission(request, view)
 
     def has_object_permission(self, request, view, obj) -> bool:
@@ -198,6 +200,7 @@ def has_object_permission(self, request, view, obj) -> bool:
         """
         if request.user.is_superuser or request.user.is_staff:
             return True
+        AuthzEnforcer.get_enforcer().load_policy()
         return self._get_permission_instance(request).has_object_permission(request, view, obj)
 
 

openedx_authz/rest_api/v1/views.py

@@ -16,6 +16,7 @@
 
 from openedx_authz import api
 from openedx_authz.constants import permissions
+from openedx_authz.engine.enforcer import AuthzEnforcer
 from openedx_authz.rest_api.data import RoleOperationError, RoleOperationStatus
 from openedx_authz.rest_api.decorators import authz_permissions, view_auth_classes
 from openedx_authz.rest_api.utils import (
@@ -102,23 +103,23 @@ class PermissionValidationMeView(APIView):
     )
     def post(self, request: HttpRequest) -> Response:
         """Validate one or more permissions for the authenticated user."""
+        AuthzEnforcer.get_enforcer().load_policy()
+
         serializer = PermissionValidationSerializer(data=request.data, many=True)
         serializer.is_valid(raise_exception=True)
+        data = serializer.validated_data
 
         username = request.user.username
         response_data = []
-        for perm in serializer.validated_data:
+        for permission in data:
             try:
-                action = perm["action"]
-                scope = perm["scope"]
+                action = permission["action"]
+                scope = permission["scope"]
                 allowed = api.is_user_allowed(username, action, scope)
                 response_data.append({"action": action, "scope": scope, "allowed": allowed})
             except ValueError as e:
                 logger.error(f"Error validating permission for user {username}: {e}")
-                return Response(
-                    data={"message": "Invalid scope format"},
-                    status=status.HTTP_400_BAD_REQUEST,
-                )
+                return Response(data={"message": "Invalid scope format"}, status=status.HTTP_400_BAD_REQUEST)
             except Exception as e:  # pylint: disable=broad-exception-caught
                 logger.error(f"Error validating permission for user {username}: {e}")
                 return Response(
@@ -283,15 +284,14 @@ def put(self, request: HttpRequest) -> Response:
         """Assign multiple users to a specific role within a scope."""
         serializer = AddUsersToRoleWithScopeSerializer(data=request.data)
         serializer.is_valid(raise_exception=True)
+        data = serializer.validated_data
 
-        role = serializer.validated_data["role"]
-        scope = serializer.validated_data["scope"]
         completed, errors = [], []
-        for user_identifier in serializer.validated_data["users"]:
+        for user_identifier in data["users"]:
             response_dict = {"user_identifier": user_identifier}
             try:
                 user = get_user_by_username_or_email(user_identifier)
-                result = api.assign_role_to_user_in_scope(user.username, role, scope)
+                result = api.assign_role_to_user_in_scope(user.username, data["role"], data["scope"])
                 if result:
                     response_dict["status"] = RoleOperationStatus.ROLE_ADDED
                     completed.append(response_dict)
@@ -330,15 +330,14 @@ def delete(self, request: HttpRequest) -> Response:
         """Remove multiple users from a specific role within a scope."""
         serializer = RemoveUsersFromRoleWithScopeSerializer(data=request.query_params)
         serializer.is_valid(raise_exception=True)
+        data = serializer.validated_data
 
-        role = serializer.validated_data["role"]
-        scope = serializer.validated_data["scope"]
         completed, errors = [], []
-        for user_identifier in serializer.validated_data["users"]:
+        for user_identifier in data["users"]:
             response_dict = {"user_identifier": user_identifier}
             try:
                 user = get_user_by_username_or_email(user_identifier)
-                result = api.unassign_role_from_user(user.username, role, scope)
+                result = api.unassign_role_from_user(user.username, data["role"], data["scope"])
                 if result:
                     response_dict["status"] = RoleOperationStatus.ROLE_REMOVED
                     completed.append(response_dict)

openedx_authz/tests/rest_api/test_views.py

@@ -39,10 +39,7 @@ class ViewTestMixin(BaseRolesTestCase):
     """Mixin providing common test utilities for view tests."""
 
     @classmethod
-    def _assign_roles_to_users(
-        cls,
-        assignments: list[dict] | None = None,
-    ):
+    def _assign_roles_to_users(cls, assignments: list[dict] | None = None):
         """Helper method to assign roles to multiple users.
 
         This method can be used to assign a role to a single user or multiple users
@@ -115,7 +112,7 @@ def setUpClass(cls):
             },
             {
                 "subject_name": "regular_7",
-                "role_name": "library_collaborator",
+                "role_name": "library_contributor",
                 "scope_name": "lib:Org3:LIB3",
             },
             {
@@ -168,9 +165,9 @@ def setUp(self):
         ([{"action": permissions.VIEW_LIBRARY.identifier, "scope": "lib:Org1:LIB1"}], [True]),
         # Single permission - denied (scope not assigned to user)
         ([{"action": permissions.VIEW_LIBRARY.identifier, "scope": "lib:Org2:LIB2"}], [False]),
-        # # Single permission - denied (action not assigned to user)
+        # Single permission - denied (action not assigned to user)
         ([{"action": "edit_library", "scope": "lib:Org1:LIB1"}], [False]),
-        # # Multiple permissions - mixed results
+        # Multiple permissions - mixed results
         (
             [
                 {"action": permissions.VIEW_LIBRARY.identifier, "scope": "lib:Org1:LIB1"},
@@ -793,9 +790,9 @@ def test_get_roles_pagination(self, query_params: dict, expected_count: int, has
         # Library Admin user
         ("regular_5", status.HTTP_200_OK),
         # Library Author user
-        # ("regular_6", status.HTTP_200_OK),  # TODO: uncomment this when we have the explicit permissions
-        # Library Collaborator user
-        # ("regular_7", status.HTTP_200_OK),  # TODO: uncomment this when we have the explicit permissions
+        ("regular_6", status.HTTP_200_OK),
+        # Library Contributor user
+        ("regular_7", status.HTTP_200_OK),
         # Library User user
         ("regular_8", status.HTTP_200_OK),
         # Regular user without permission