refactor: modify REST API endpoints for LR MFE (#1760)

Co-authored-by: Maxwell Frank <[email protected]>

credentials/apps/records/rest_api/v1/permissions.py

@@ -0,0 +1,36 @@
+"""
+Custom permissions classes for use with DRF.
+"""
+from edx_rest_framework_extensions.permissions import get_username_param
+from rest_framework import permissions
+
+
+class IsPublic(permissions.BasePermission):
+    """
+    Allows access if URL is for a public record.
+    """
+
+    def has_permission(self, request, view):
+        """
+        Check to see what type of record (private/public) in the query parameter.
+        If it is not public, run a normal authentication permission check.
+        """
+        query_param_is_public = request.query_params.get("is_public", "")
+        is_public = query_param_is_public.lower() == "true"
+        if not is_public:
+            return bool(request.user and request.user.is_authenticated)
+        return True
+
+
+class CanAccessProgramRecord(permissions.BasePermission):
+    """
+    Allows access if the requesting user a staff member or a superuser.
+    """
+
+    def has_permission(self, request, view):
+        """
+        If there is a username query param, check for either superuser or staff access.
+        """
+        if get_username_param(request):
+            return request.user and (request.user.is_superuser or request.user.is_staff)
+        return True

credentials/apps/records/rest_api/v1/tests/test_views.py

@@ -58,6 +58,9 @@ def setUp(self):
             credential_content_type=self.program_credential_content_type,
             credential=self.program_cert,
         )
+        self.program_cert_record = ProgramCertRecordFactory(
+            uuid=self.program.uuid, program=self.program, user=self.user
+        )
 
     def serialize_program_records(self):
         request = APIRequestFactory(SERVER_NAME=self.site.domain).get("/")
@@ -69,13 +72,11 @@ def serialize_program_records(self):
             many=True,
         ).data
 
-    def serialize_program_record_details(self, is_public):
-        url = "/" + str(self.program.uuid) + "/?is_public=" + str(is_public)
+    def serialize_program_record_details(self, uuid, is_public):
+        url = f"/{uuid}/?is_public={is_public}"
         request = APIRequestFactory(SERVER_NAME=self.site.domain).get(url)
         return ProgramRecordSerializer(
-            get_program_details(
-                request_user=self.user, request_site=self.site, uuid=self.program.uuid, is_public=is_public
-            ),
+            get_program_details(request_user=self.user, request_site=self.site, uuid=uuid, is_public=is_public),
             context={"request": request},
         ).data
 
@@ -84,13 +85,32 @@ def test_deny_unauthenticated_user(self):
         response = self.client.get("/records/api/v1/program_records/")
         self.assertEqual(response.status_code, 401)
 
+    def test_details_deny_unauthenticated_user(self):
+        self.client.logout()
+        uuid = str(self.program.uuid).replace("-", "")
+        response = self.client.get(f"/records/api/v1/program_records/{uuid}/?is_public=false")
+        self.assertEqual(response.status_code, 401)
+
     def test_allow_authenticated_user(self):
         """Verify the endpoint requires an authenticated user."""
         self.client.logout()
         self.client.login(username=self.user.username, password=USER_PASSWORD)
         response = self.client.get("/records/api/v1/program_records/")
         self.assertEqual(response.status_code, 200)
 
+    def test_details_allow_authenticated_user(self):
+        self.client.logout()
+        self.client.login(username=self.user.username, password=USER_PASSWORD)
+        uuid = str(self.program.uuid).replace("-", "")
+        response = self.client.get(f"/records/api/v1/program_records/{uuid}/?is_public=false")
+        self.assertEqual(response.status_code, 200)
+
+    def test_allow_public_record(self):
+        public_record_uuid = str(self.program_cert_record.uuid).replace("-", "")
+
+        response = self.client.get(f"/records/api/v1/program_records/{public_record_uuid}/?is_public=true")
+        self.assertEqual(response.status_code, 200)
+
     def test_get(self):
         self.client.login(username=self.user.username, password=USER_PASSWORD)
         response = self.client.get("/records/api/v1/program_records/")
@@ -101,12 +121,13 @@ def test_get_private_details(self):
         self.client.login(username=self.user.username, password=USER_PASSWORD)
         uuid = str(self.program.uuid).replace("-", "")
 
-        response = self.client.get("/records/api/v1/program_records/" + uuid + "/?is_public=false")
+        response = self.client.get(f"/records/api/v1/program_records/{uuid}/?is_public=false")
+
         self.assertEqual(response.status_code, 200)
 
         # Initialize two variables to hold data we intend to mutate
         response_dict = response.data
-        serializer_dict = self.serialize_program_record_details(False)
+        serializer_dict = self.serialize_program_record_details(self.program.uuid, False)
         # We want to omit "last_updated" from the test as it will always be different
         del response_dict["record"]["program"]["last_updated"]
         del serializer_dict["record"]["program"]["last_updated"]
@@ -117,22 +138,70 @@ def test_get_private_details(self):
         self.assertFalse(response.data["is_public"], "Query paramater is set to public when it should be private")
 
     def test_get_public_details(self):
-        self.client.login(username=self.user.username, password=USER_PASSWORD)
-        # Create ProgramCertRecord for public request
-        ProgramCertRecordFactory(uuid=self.program.uuid, program=self.program, user=self.user)
-        uuid = str(self.program.uuid).replace("-", "")
+        uuid = str(self.program_cert_record.uuid).replace("-", "")
 
-        response = self.client.get("/records/api/v1/program_records/" + uuid + "/?is_public=true")
+        response = self.client.get(f"/records/api/v1/program_records/{uuid}/?is_public=true")
         self.assertEqual(response.status_code, 200)
 
         # Initialize two variables to hold data we intend to mutate
         response_dict = response.data
-        serializer_dict = self.serialize_program_record_details(True)
+        serializer_dict = self.serialize_program_record_details(uuid, True)
         # We want to omit "last_updated" from the test as it will always be different
         del response_dict["record"]["program"]["last_updated"]
         del serializer_dict["record"]["program"]["last_updated"]
         # Remove any "-" from the uuid in the serializer
         serializer_dict["uuid"] = serializer_dict["uuid"].replace("-", "")
 
         self.assertEqual(response_dict, serializer_dict)
-        self.assertTrue(response.data["is_public"], "Query paramater is set to private when it should be public")
+        self.assertTrue(response.data["is_public"], "Query parameter is set to private when it should be public")
+
+    def test_deny_support_get(self):
+        self.client.login(username=self.user.username, password=USER_PASSWORD)
+        uuid = str(self.program.uuid).replace("-", "")
+
+        # Create a new user to look up
+        UserFactory(username="another_user")
+
+        response = self.client.get(f"/records/api/v1/program_records/{uuid}/?username=another_user")
+        self.assertEqual(response.status_code, 403)
+        response = self.client.get("/records/api/v1/program_records/?username=edx")
+        self.assertEqual(response.status_code, 403)
+
+    def test_allow_support_staff_get(self):
+        # Reset user object to have staff permissions
+        self.user = UserFactory(is_staff=True)
+        self.client.login(username=self.user.username, password=USER_PASSWORD)
+        uuid = str(self.program.uuid).replace("-", "")
+
+        # Create a new user to look up
+        UserFactory(username="another_user")
+
+        response = self.client.get(f"/records/api/v1/program_records/{uuid}/?username=another_user")
+        self.assertEqual(response.status_code, 200)
+        response = self.client.get("/records/api/v1/program_records/?username=another_user")
+        self.assertEqual(response.status_code, 200)
+
+    def test_allow_support_superuser_get(self):
+        # Reset user object to have superuser permissions
+        self.user = UserFactory(is_superuser=True)
+        self.client.login(username=self.user.username, password=USER_PASSWORD)
+        uuid = str(self.program.uuid).replace("-", "")
+
+        # Create a new user to look up
+        UserFactory(username="another_user")
+
+        response = self.client.get(f"/records/api/v1/program_records/{uuid}/?username=another_user")
+        self.assertEqual(response.status_code, 200)
+        response = self.client.get("/records/api/v1/program_records/?username=another_user")
+        self.assertEqual(response.status_code, 200)
+
+    def test_support_invalid_username_get(self):
+        # Reset user object to have superuser and staff permissions
+        self.user = UserFactory(is_superuser=True, is_staff=True)
+        self.client.login(username=self.user.username, password=USER_PASSWORD)
+        uuid = str(self.program.uuid).replace("-", "")
+
+        response = self.client.get(f"/records/api/v1/program_records/{uuid}/?username=invalid_user")
+        self.assertEqual(response.status_code, 404)
+        response = self.client.get("/records/api/v1/program_records/?username=invalid_user")
+        self.assertEqual(response.status_code, 404)

credentials/apps/records/rest_api/v1/views.py

@@ -1,21 +1,32 @@
+import logging
+
+from django.contrib.auth import get_user_model
 from edx_rest_framework_extensions.auth.jwt.authentication import JwtAuthentication
-from rest_framework import mixins, permissions, status, viewsets
+from rest_framework import mixins, status, viewsets
 from rest_framework.authentication import SessionAuthentication
 from rest_framework.response import Response
 
 from credentials.apps.records.api import get_program_details
 from credentials.apps.records.models import ProgramCertRecord
+from credentials.apps.records.rest_api.v1.permissions import CanAccessProgramRecord, IsPublic
 from credentials.apps.records.rest_api.v1.serializers import ProgramRecordSerializer, ProgramSerializer
 from credentials.apps.records.utils import get_user_program_data
 
 
+User = get_user_model()
+log = logging.getLogger(__name__)
+
+
 class ProgramRecordsViewSet(mixins.RetrieveModelMixin, mixins.ListModelMixin, viewsets.GenericViewSet):
 
     authentication_classes = (
         JwtAuthentication,
         SessionAuthentication,
     )
-    permission_classes = (permissions.IsAuthenticated,)
+    permission_classes = (
+        IsPublic,
+        CanAccessProgramRecord,
+    )
 
     def list(self, request, *args, **kwargs):
         """
@@ -28,9 +39,26 @@ def list(self, request, *args, **kwargs):
         Returns:
             response(dict): Information about the user's enrolled programs
         """
+        # Check for a username query parameter.
+        # If there is one in the request, we will pass it into the `get_user_program_data` function.
+        # If there is no username query parameter, we instead pass the username from the user in the request.
+        query_param_username = request.query_params.get("username", "")
+        username = request.user.username
+
+        if query_param_username:
+            try:
+                User.objects.get(username=query_param_username)
+            except User.DoesNotExist:
+                log.error(f'A user matching the username "{query_param_username}" does not exist')
+                return Response(status=status.HTTP_404_NOT_FOUND)
+            else:
+                # Overwrite username variable once we know a User with that username exists
+                username = query_param_username
+
         programs = get_user_program_data(
-            request.user.username, request.site, include_empty_programs=False, include_retired_programs=True
+            username, request.site, include_empty_programs=False, include_retired_programs=True
         )
+
         serializer = ProgramSerializer(programs, many=True)
         return Response({"enrolled_programs": serializer.data})
 
@@ -46,17 +74,35 @@ def retrieve(self, request, *args, **kwargs):
         Returns:
             response(dict): Details about a user's progress in a given program
         """
-        # query parameters come through as a string and we need to convert it to a boolean
+        # Query parameters come through as a string and we need to convert it to a boolean
         query_param_is_public = request.query_params.get("is_public", "")
         is_public = query_param_is_public.lower() == "true"
 
+        # Check for a username query parameter.
+        # If there is one in the request, we will fetch the User associated with that username.
+        # This Django User object is passed into the `get_program_details` function.
+        # If there is no username query parameter, we instead pass the user from the request.
+        query_param_username = request.query_params.get("username", "")
+        user = request.user
+
         try:
+            if query_param_username:
+                try:
+                    searched_user = User.objects.get(username=query_param_username)
+                except User.DoesNotExist:
+                    log.error(f'A user matching the username "{query_param_username}" does not exist')
+                    return Response(status=status.HTTP_404_NOT_FOUND)
+                else:
+                    # Overwrite user variable to be the fetched User
+                    user = searched_user
+
             program = get_program_details(
-                request_user=request.user,
+                request_user=user,
                 request_site=request.site,
                 uuid=kwargs["pk"],
                 is_public=is_public,
             )
+
         except ProgramCertRecord.DoesNotExist:
             return Response(status=status.HTTP_404_NOT_FOUND)