feat: adding bulk user validation endpoint for admin console

Author: jacobo-dominguez-wgu

CHANGELOG.rst

@@ -14,6 +14,15 @@ Change Log
 Unreleased
 **********
 
+1.3.0 - 2026-04-01
+******************
+
+Added
+=====
+
+* Add ``GlobalPermission`` class for cross-scope permission checking without requiring specific scope parameters.
+* Add ``users/`` endpoint endpoint for bulk validation of user identifiers (usernames or emails).
+
 1.2.0 - 2026-03-30
 ******************
 

openedx_authz/__init__.py

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

openedx_authz/rest_api/v1/fields.py

@@ -25,3 +25,11 @@ def to_internal_value(self, data):
     def to_representation(self, value):
         """Convert string to lowercase"""
         return value.strip().lower()
+
+
+class UserValidationSummarySerializer(serializers.Serializer):  # pylint: disable=abstract-method
+    """Serializer for user validation summary statistics."""
+
+    total = serializers.IntegerField(help_text="Total number of users validated")
+    valid_count = serializers.IntegerField(help_text="Number of valid users found")
+    invalid_count = serializers.IntegerField(help_text="Number of invalid users")

openedx_authz/rest_api/v1/permissions.py

@@ -1,11 +1,14 @@
 """Permissions for the Open edX AuthZ REST API."""
 
+import logging
 from typing import ClassVar
 
 from rest_framework.permissions import BasePermission
 
 from openedx_authz import api
 
+logger = logging.getLogger(__name__)
+
 
 class PermissionMeta(type(BasePermission)):
     """Metaclass that automatically registers permission classes by namespace.
@@ -290,3 +293,52 @@ def has_permission(self, request, view) -> bool:
             return self.validate_permissions(request, permissions, scope_value)
 
         return True
+
+
+class GlobalPermission(MethodPermissionMixin, BasePermission):
+    """Permission handler for global operations that don't target specific scopes.
+
+    This class implements permission checks for operations that require management
+    capabilities across any scope, rather than within a specific scope. It checks
+    if the user has the required permissions in ANY of their role assignments.
+    """
+
+    def has_permission(self, request, view) -> bool:
+        """Check if the user has required permissions in any scope.
+
+        Checks if the view method has @authz_permissions decorator and validates
+        that the user has at least one of the required permissions in any scope.
+        Superuser and staff users are automatically granted permission.
+
+        Returns:
+            bool: True if user has required permissions in any scope, False otherwise.
+        """
+        if request.user.is_superuser or request.user.is_staff:
+            return True
+        permissions = self.get_required_permissions(request, view)
+        if not permissions:
+            return False
+        return self.validate_global_permissions(request, permissions)
+
+    def validate_global_permissions(self, request, permissions: list[str]) -> bool:
+        """Validate that user has at least one of the required permissions in any scope.
+
+        Args:
+            request: The Django REST framework request object.
+            permissions: List of permission identifiers to check.
+
+        Returns:
+            bool: True if user has any required permission in any scope, False otherwise.
+        """
+        try:
+            username = request.user.username
+            user_assignments = api.get_user_role_assignments(username)
+            for assignment in user_assignments:
+                for role in assignment.roles:
+                    role_permissions = role.get_permission_identifiers()
+                    if any(perm in role_permissions for perm in permissions):
+                        return True
+            return False
+        except Exception:
+            logger.error(f"Error checking global permissions for user {username}", exc_info=True)
+            return False

openedx_authz/rest_api/v1/serializers.py

@@ -6,7 +6,11 @@
 from openedx_authz import api
 from openedx_authz.rest_api.data import SortField, SortOrder
 from openedx_authz.rest_api.utils import get_generic_scope
-from openedx_authz.rest_api.v1.fields import CommaSeparatedListField, LowercaseCharField
+from openedx_authz.rest_api.v1.fields import (
+    CommaSeparatedListField,
+    LowercaseCharField,
+    UserValidationSummarySerializer,
+)
 
 User = get_user_model()
 
@@ -203,3 +207,28 @@ def get_email(self, obj) -> str:
     def get_roles(self, obj: api.RoleAssignmentData) -> list[str]:
         """Get the roles for the given role assignment."""
         return [role.external_key for role in obj.roles]
+
+
+class UserValidationAPIViewSerializer(serializers.Serializer):  # pylint: disable=abstract-method
+    """Serializer for validating user existence."""
+
+    users = serializers.ListField(
+        child=serializers.CharField(max_length=255), allow_empty=False, help_text="List of user identifiers to validate"
+    )
+
+    def validate_users(self, value) -> list[str]:
+        """Eliminate duplicates preserving order"""
+        return list(dict.fromkeys(value))
+
+
+class UserValidationAPIViewResponseSerializer(serializers.Serializer):  # pylint: disable=abstract-method
+    """Serializer for user validation response."""
+
+    valid_users = serializers.ListField(
+        child=serializers.CharField(max_length=255), help_text="List of user identifiers that were found to be valid"
+    )
+    invalid_users = serializers.ListField(
+        child=serializers.CharField(max_length=255),
+        help_text="List of user identifiers that were not found or are invalid",
+    )
+    summary = UserValidationSummarySerializer(help_text="Summary statistics for the validation operation")

openedx_authz/rest_api/v1/urls.py

@@ -12,4 +12,5 @@
     ),
     path("roles/", views.RoleListView.as_view(), name="role-list"),
     path("roles/users/", views.RoleUserAPIView.as_view(), name="role-user-list"),
+    path("users/validate/", views.UserValidationAPIView.as_view(), name="user-validation"),
 ]

openedx_authz/rest_api/v1/views.py

@@ -26,7 +26,7 @@
     sort_users,
 )
 from openedx_authz.rest_api.v1.paginators import AuthZAPIViewPagination
-from openedx_authz.rest_api.v1.permissions import DynamicScopePermission
+from openedx_authz.rest_api.v1.permissions import DynamicScopePermission, GlobalPermission
 from openedx_authz.rest_api.v1.serializers import (
     AddUsersToRoleWithScopeSerializer,
     ListRolesWithScopeResponseSerializer,
@@ -36,6 +36,8 @@
     PermissionValidationSerializer,
     RemoveUsersFromRoleWithScopeSerializer,
     UserRoleAssignmentSerializer,
+    UserValidationAPIViewResponseSerializer,
+    UserValidationAPIViewSerializer,
 )
 
 logger = logging.getLogger(__name__)
@@ -449,3 +451,86 @@ def get(self, request: HttpRequest) -> Response:
         paginated_response_data = paginator.paginate_queryset(response_data, request)
         serialized_data = ListRolesWithScopeResponseSerializer(paginated_response_data, many=True)
         return paginator.get_paginated_response(serialized_data.data)
+
+
+@view_auth_classes()
+class UserValidationAPIView(APIView):
+    """API view for validating that provided user identifiers correspond to existing users.
+
+    This view allows clients to verify that a list of user identifiers (usernames or emails)
+    correspond to valid users in the system. It is designed to support bulk validation of multiple
+    user identifiers in a single request, providing a convenient way to check the validity of users before
+    performing operations such as role assignments.
+
+    **Endpoints**
+    - POST: Validate that the provided list of usernames or emails correspond to existing users
+
+    **Request Format (POST)**
+    - users: List of user identifiers (username or email)
+
+    **Response Format (POST)**
+
+    Returns HTTP 200 OK with::
+
+        {
+            "valid_users": ["john_doe", "[email protected]"],
+            "invalid_users": ["nonexistent_user"],
+            "summary": {
+                "total": 3,
+                "valid_count": 2,
+                "invalid_count": 1
+            }
+        }
+
+    **Authentication and Permissions**
+
+    - Requires authenticated user.
+    - Requires ``manage_library_team`` or ``manage_course_team`` permission in any scope.
+
+    **Example Request**
+
+    POST /api/authz/v1/users/validate/ ::
+
+        {
+            "users": ["john_doe", "[email protected]", "nonexistent_user"]
+        }
+    """
+
+    permission_classes = [GlobalPermission]
+
+    @apidocs.schema(
+        body=UserValidationAPIViewSerializer,
+        responses={
+            status.HTTP_200_OK: UserValidationAPIViewResponseSerializer,
+            status.HTTP_400_BAD_REQUEST: "The request data is invalid",
+            status.HTTP_401_UNAUTHORIZED: "The user is not authenticated",
+            status.HTTP_403_FORBIDDEN: "The user does not have the required permissions",
+        },
+    )
+    @authz_permissions([permissions.MANAGE_LIBRARY_TEAM.identifier, permissions.COURSES_MANAGE_COURSE_TEAM.identifier])
+    def post(self, request: HttpRequest) -> Response:
+        """Validates the provided usernames or emails correspond to existing users."""
+
+        request_serializer = UserValidationAPIViewSerializer(data=request.data)
+        request_serializer.is_valid(raise_exception=True)
+        serialized_request_data = request_serializer.validated_data
+        valid_users = []
+        invalid_users = []
+        for user_identifier in serialized_request_data["users"]:
+            try:
+                user = get_user_by_username_or_email(user_identifier)
+                valid_users.append(user_identifier)
+            except User.DoesNotExist:
+                invalid_users.append(user_identifier)
+
+        response_data = {
+            "valid_users": valid_users,
+            "invalid_users": invalid_users,
+            "summary": {
+                "total": len(serialized_request_data["users"]),
+                "valid_count": len(valid_users),
+                "invalid_count": len(invalid_users),
+            },
+        }
+        response_serializer = UserValidationAPIViewResponseSerializer(response_data)
+        return Response(response_serializer.data, status=status.HTTP_200_OK)