feat: add rest api for roles and permissions

Author: BryanttV

openedx_authz/api/data.py

@@ -416,3 +416,16 @@ class RoleAssignmentData(AuthZData):
     subject: SubjectData = None  # Needs defaults to avoid value error from attrs
     roles: list[RoleData] = []
     scope: ScopeData = None
+
+
+@define
+class SubjectRoleAssignmentData(AuthZData):
+    """A subject role assignment is the assignment of one or more roles to a subject in a specific scope.
+
+    Attributes:
+        subject: The ID of the subject namespaced (e.g., 'user@john_doe').
+        roles: The roles assigned to the subject.
+    """
+
+    subject: SubjectData = None
+    roles: list[RoleData] = None

openedx_authz/api/permissions.py

@@ -63,6 +63,7 @@ def is_subject_allowed(
     Returns:
         bool: True if the subject has the specified permission in the scope, False otherwise.
     """
+    enforcer.load_policy()
     return enforcer.enforce(
         subject.namespaced_key, action.namespaced_key, scope.namespaced_key
     )

openedx_authz/api/roles.py

@@ -18,6 +18,8 @@
     RoleData,
     ScopeData,
     SubjectData,
+    UserData,
+    SubjectRoleAssignmentData,
 )
 from openedx_authz.api.permissions import get_permission_from_policy
 from openedx_authz.engine.enforcer import enforcer
@@ -114,6 +116,7 @@ def get_permissions_for_active_roles_in_scope(
         dict[str, list[PermissionData]]: A dictionary mapping the role external_key to its
         permissions and scopes.
     """
+    enforcer.load_policy()
     filtered_policy = enforcer.get_filtered_grouping_policy(
         GroupingPolicyIndex.SCOPE.value, scope.namespaced_key
     )
@@ -145,6 +148,7 @@ def get_role_definitions_in_scope(scope: ScopeData) -> list[RoleData]:
     Returns:
         list[Role]: A list of roles.
     """
+    enforcer.load_policy()
     policy_filtered = enforcer.get_filtered_policy(
         PolicyIndex.SCOPE.value, scope.namespaced_key
     )
@@ -190,6 +194,7 @@ def get_all_roles_in_scope(scope: ScopeData) -> list[list[str]]:
     Returns:
         list[list[str]]: A list of policies in the specified scope.
     """
+    enforcer.load_policy()
     return enforcer.get_filtered_grouping_policy(
         GroupingPolicyIndex.SCOPE.value, scope.namespaced_key
     )
@@ -204,6 +209,7 @@ def assign_role_to_subject_in_scope(
         subject: The ID of the subject.
         role: The role to assign.
     """
+    enforcer.load_policy()
     enforcer.add_role_for_user_in_domain(
         subject.namespaced_key,
         role.namespaced_key,
@@ -234,6 +240,7 @@ def unassign_role_from_subject_in_scope(
         role: The role to unassign.
         scope: The scope from which to unassign the role.
     """
+    enforcer.load_policy()
     enforcer.delete_roles_for_user_in_domain(
         subject.namespaced_key, role.namespaced_key, scope.namespaced_key
     )
@@ -291,6 +298,7 @@ def get_subject_role_assignments_in_scope(
     Returns:
         list[RoleAssignment]: A list of role assignments for the subject in the scope.
     """
+    enforcer.load_policy()
     # TODO: we still need to get the remaining data for the role like email, etc
     role_assignments = []
     for namespaced_key in enforcer.get_roles_for_user_in_domain(
@@ -372,3 +380,17 @@ def get_all_subject_role_assignments_in_scope(scope: ScopeData) -> list[RoleAssi
         )
 
     return list(role_assignments_per_subject.values())
+
+
+def get_subjects_for_role(role: RoleData) -> list[SubjectData]:
+    """Get all the subjects assigned to a specific role.
+
+    Args:
+        role (RoleData): The role to filter subjects.
+
+    Returns:
+        list[SubjectData]: A list of subjects assigned to the specified role.
+    """
+    enforcer.load_policy()
+    policies = enforcer.get_filtered_grouping_policy(GroupingPolicyIndex.ROLE.value, role.namespaced_key)
+    return [SubjectData(namespaced_key=policy[GroupingPolicyIndex.SUBJECT.value]) for policy in policies]

openedx_authz/api/users.py

@@ -19,6 +19,7 @@
     get_subject_role_assignments,
     get_subject_role_assignments_for_role_in_scope,
     get_subject_role_assignments_in_scope,
+    get_subjects_for_role,
     unassign_role_from_subject_in_scope,
 )
 
@@ -32,6 +33,7 @@
     "get_user_role_assignments_for_role_in_scope",
     "get_all_user_role_assignments_in_scope",
     "is_user_allowed",
+    "get_users_for_role",
 ]
 
 
@@ -183,3 +185,16 @@ def is_user_allowed(
         ActionData(external_key=action_external_key),
         ScopeData(external_key=scope_external_key),
     )
+
+
+def get_users_for_role(role_external_key: str) -> list[UserData]:
+    """Get all the users assigned to a specific role.
+
+    Args:
+        role_external_key (str): The role to filter users (e.g., 'library_admin').
+
+    Returns:
+        list[UserData]: A list of users assigned to the specified role.
+    """
+    users = get_subjects_for_role(RoleData(external_key=role_external_key))
+    return [UserData(namespaced_key=user.namespaced_key) for user in users]

openedx_authz/apps.py

@@ -17,12 +17,12 @@ class OpenedxAuthzConfig(AppConfig):
         "url_config": {
             "lms.djangoapp": {
                 "namespace": "openedx-authz",
-                "regex": r"^openedx-authz/",
+                "regex": r"^api/",
                 "relative_path": "urls",
             },
             "cms.djangoapp": {
                 "namespace": "openedx-authz",
-                "regex": r"^openedx-authz/",
+                "regex": r"^api/",
                 "relative_path": "urls",
             },
         },

openedx_authz/rest_api/enums.py

@@ -0,0 +1,35 @@
+"""Enums for the Open edX AuthZ REST API."""
+
+from enum import Enum
+
+
+class BaseEnum(str, Enum):
+    """Base enum class."""
+
+    @classmethod
+    def values(cls):
+        """List the values of the enum."""
+        return [e.value for e in cls]
+
+
+class SortField(BaseEnum):
+    """Enum for the fields to sort by."""
+
+    USERNAME = "username"
+    FULL_NAME = "full_name"
+    EMAIL = "email"
+
+
+class SortOrder(BaseEnum):
+    """Enum for the order to sort by."""
+
+    ASC = "asc"
+    DESC = "desc"
+
+
+class SearchField(BaseEnum):
+    """Enum for the fields allowed for text search filtering."""
+
+    USERNAME = "username"
+    FULL_NAME = "full_name"
+    EMAIL = "email"

openedx_authz/rest_api/urls.py

@@ -0,0 +1,9 @@
+"""Open edX AuthZ API URLs."""
+
+from django.urls import include, path
+
+from openedx_authz.rest_api.v1 import urls as v1_urls
+
+urlpatterns = [
+    path("v1/", include(v1_urls)),
+]

openedx_authz/rest_api/utils.py

@@ -0,0 +1,111 @@
+"""Utility functions for the Open edX AuthZ REST API."""
+
+from django.contrib.auth import get_user_model
+from django.db.models import Q
+from edx_rest_framework_extensions.auth.jwt.authentication import JwtAuthentication
+from edx_rest_framework_extensions.auth.session.authentication import SessionAuthenticationAllowInactiveUser
+from rest_framework.permissions import IsAuthenticated
+
+from openedx_authz.rest_api.enums import SearchField, SortField, SortOrder
+
+User = get_user_model()
+
+
+def view_auth_classes(is_authenticated=True):
+    """
+    Function and class decorator that abstracts the authentication and permission checks for api views.
+    """
+
+    def _decorator(func_or_class):
+        """
+        Requires either OAuth2 or Session-based authentication.
+        """
+        func_or_class.authentication_classes = [
+            JwtAuthentication,
+            SessionAuthenticationAllowInactiveUser,
+        ]
+        if is_authenticated:
+            func_or_class.permission_classes.insert(0, IsAuthenticated)
+        return func_or_class
+
+    return _decorator
+
+
+def get_user_by_username_or_email(username_or_email: str) -> User:
+    """
+    Retrieve a user by their username or email address.
+
+    Args:
+        username_or_email (str): The username or email address to search for.
+
+    Returns:
+        User: The User object if found and not retired.
+
+    Raises:
+        User.DoesNotExist: If no user matches the provided username or email,
+            or if the user has an associated retirement request.
+    """
+    user = User.objects.get(Q(email=username_or_email) | Q(username=username_or_email))
+    if hasattr(user, "userretirementrequest"):
+        raise User.DoesNotExist
+    return user
+
+
+def sort_users(
+    users: list[dict], sort_by: SortField = SortField.USERNAME, order: SortOrder = SortOrder.ASC
+) -> list[dict]:
+    """
+    Sort users by a given field and order.
+
+    Args:
+        users (list[dict]): The users to sort.
+        sort_by (SortField, optional): The field to sort by. Defaults to SortField.USERNAME.
+        order (SortOrder, optional): The order to sort by. Defaults to SortOrder.ASC.
+
+    Raises:
+        ValueError: If the sort field is invalid.
+        ValueError: If the sort order is invalid.
+
+    Returns:
+        list[dict]: The sorted users.
+    """
+    if sort_by not in SortField.values():
+        raise ValueError(f"Invalid field: '{sort_by}'. Must be one of {SortField.values()}")
+
+    if order not in SortOrder.values():
+        raise ValueError(f"Invalid order: '{order}'. Must be one of {SortOrder.values()}")
+
+    sorted_users = sorted(users, key=lambda user: (user.get(sort_by) or "").lower(), reverse=order == SortOrder.DESC)
+    return sorted_users
+
+
+def filter_users(users: list[dict], search: str | None, roles: list[str] | None) -> list[dict]:
+    """
+    Filter users by a case-insensitive search string and/or by roles.
+
+    Args:
+        users (list[dict]): The users to filter.
+        search (str | None): Optional search term matched against fields in ``SearchField``.
+        roles (list[str] | None): Optional list of roles; include users that have any of these roles.
+
+    Returns:
+        list[dict]: The filtered users, preserving the original order.
+    """
+    if not search and not roles:
+        return users
+
+    filtered_users = []
+    for user in users:
+        if search:
+            matches_search = any(search in (user.get(field) or "").lower() for field in SearchField.values())
+            if not matches_search:
+                continue
+
+        if roles:
+            matches_role = any(role in user.get("roles", []) for role in roles)
+            if not matches_role:
+                continue
+
+        filtered_users.append(user)
+
+    return filtered_users

openedx_authz/rest_api/v1/fields.py

@@ -0,0 +1,27 @@
+"""Fields serializer for the Open edX AuthZ REST API."""
+
+from rest_framework import serializers
+
+
+class CommaSeparatedListField(serializers.Field):
+    """Serializer for a comma-separated list of strings."""
+
+    def to_internal_value(self, data):
+        """Convert string separated by commas to list"""
+        return [item.strip().lower() for item in data.split(",") if item.strip()]
+
+    def to_representation(self, value):
+        """Convert list to string separated by commas"""
+        return ",".join(value).lower()
+
+
+class LowercaseCharField(serializers.CharField):
+    """Serializer for a lowercase string."""
+
+    def to_internal_value(self, data):
+        """Convert string to lowercase"""
+        return data.strip().lower()
+
+    def to_representation(self, value):
+        """Convert string to lowercase"""
+        return value.strip().lower()

openedx_authz/rest_api/v1/paginators.py

@@ -0,0 +1,11 @@
+"""Pagination classes for the REST API."""
+
+from rest_framework.pagination import PageNumberPagination
+
+
+class AuthZAPIViewPagination(PageNumberPagination):
+    """Pagination class for the AuthZ API views."""
+
+    page_size = 10
+    page_size_query_param = "page_size"
+    max_page_size = 100