[SE-5163] Apply security patches for xss and email search from logout page

openedx/core/djangoapps/user_api/accounts/permissions.py

@@ -32,3 +32,15 @@ class CanReplaceUsername(permissions.BasePermission):
     """
     def has_permission(self, request, view):
         return request.user.username == getattr(settings, "USERNAME_REPLACEMENT_WORKER", False)
+
+
+class CanGetAccountInfo(permissions.BasePermission):
+    """
+    Grants access to AccountViewSet if the requesting user is a superuser/staff
+    and requesting to get account info based on non-public information.
+    """
+
+    def has_permission(self, request, view):
+        return (request.GET.get('lms_user_id') is None and request.GET.get('email') is None) or (
+            request.user.is_staff or request.user.is_superuser
+        )

openedx/core/djangoapps/user_api/accounts/tests/test_views.py

@@ -16,6 +16,7 @@
 from django.test.utils import override_settings
 from django.urls import reverse
 from rest_framework.test import APIClient, APITestCase
+from rest_framework import status
 
 from openedx.core.djangoapps.oauth_dispatch.jwt import create_jwt_for_user
 from openedx.core.djangoapps.user_api.accounts import ACCOUNT_VISIBILITY_PREF_KEY
@@ -74,7 +75,7 @@ def send_get(self, client, query_parameters=None, expected_status=200):
         """
         Helper method for sending a GET to the server. Verifies the expected status and returns the response.
         """
-        url = self.url + '?' + query_parameters if query_parameters else self.url    # pylint: disable=no-member
+        url = self.url + '?' + query_parameters if query_parameters else self.url  # pylint: disable=no-member
         response = client.get(url)
         assert expected_status == response.status_code
         return response
@@ -325,22 +326,62 @@ def test_get_account_unknown_user(self, api_client, user):
         response = client.get(reverse("accounts_api", kwargs={'username': "does_not_exist"}))
         assert 404 == response.status_code
 
-    @ddt.data(
-        ("client", "user"),
-        ("staff_client", "staff_user"),
-    )
-    @ddt.unpack
-    def test_get_account_by_email(self, api_client, user):
+    def test_successful_get_account_by_email(self):
         """
-        Test that requesting a user email search works.
+        Test that request using email by a staff user successfully retrieves Account Info.
         """
+        api_client = "staff_client"
+        user = "staff_user"
         client = self.login_client(api_client, user)
         self.create_mock_profile(self.user)
         set_user_preference(self.user, ACCOUNT_VISIBILITY_PREF_KEY, PRIVATE_VISIBILITY)
 
         response = self.send_get(client, query_parameters=f'email={self.user.email}')
         self._verify_full_account_response(response)
 
+    def test_unsuccessful_get_account_by_email(self):
+        """
+        Test that request using email by a normal user fails to retrieve Account Info.
+        """
+        api_client = "client"
+        user = "user"
+        client = self.login_client(api_client, user)
+        self.create_mock_profile(self.user)
+        set_user_preference(self.user, ACCOUNT_VISIBILITY_PREF_KEY, PRIVATE_VISIBILITY)
+
+        response = self.send_get(
+            client, query_parameters=f'email={self.user.email}', expected_status=status.HTTP_403_FORBIDDEN
+        )
+        assert response.data.get('detail') == 'You do not have permission to perform this action.'
+
+    def test_successful_get_account_by_user_id(self):
+        """
+        Test that request using lms user id by a staff user successfully retrieves Account Info.
+        """
+        api_client = "staff_client"
+        user = "staff_user"
+        client = self.login_client(api_client, user)
+        self.create_mock_profile(self.user)
+        set_user_preference(self.user, ACCOUNT_VISIBILITY_PREF_KEY, PRIVATE_VISIBILITY)
+
+        response = self.send_get(client, query_parameters=f'lms_user_id={self.user.id}')
+        self._verify_full_account_response(response)
+
+    def test_unsuccessful_get_account_by_user_id(self):
+        """
+        Test that requesting using lms user id by a normal user fails to retrieve Account Info.
+        """
+        api_client = "client"
+        user = "user"
+        client = self.login_client(api_client, user)
+        self.create_mock_profile(self.user)
+        set_user_preference(self.user, ACCOUNT_VISIBILITY_PREF_KEY, PRIVATE_VISIBILITY)
+
+        response = self.send_get(
+            client, query_parameters=f'lms_user_id={self.user.id}', expected_status=status.HTTP_403_FORBIDDEN
+        )
+        assert response.data.get('detail') == 'You do not have permission to perform this action.'
+
     # Note: using getattr so that the patching works even if there is no configuration.
     # This is needed when testing CMS as the patching is still executed even though the
     # suite is skipped.
@@ -410,7 +451,7 @@ def verify_fields_visible_to_all_users(response):
         response = self.send_get(client, query_parameters='view=shared')
         verify_fields_visible_to_all_users(response)
 
-        response = self.send_get(client, query_parameters=f'view=shared&email={self.user.email}')
+        response = self.send_get(client, query_parameters=f'view=shared&username={self.user.username}')
         verify_fields_visible_to_all_users(response)
 
     @ddt.data(

openedx/core/djangoapps/user_api/accounts/views.py

@@ -76,7 +76,7 @@
     UserRetirementStatus
 )
 from .api import get_account_settings, update_account_settings
-from .permissions import CanDeactivateUser, CanReplaceUsername, CanRetireUser
+from .permissions import CanDeactivateUser, CanGetAccountInfo, CanReplaceUsername, CanRetireUser
 from .serializers import UserRetirementPartnerReportSerializer, UserRetirementStatusSerializer
 from .signals import USER_RETIRE_LMS_CRITICAL, USER_RETIRE_LMS_MISC, USER_RETIRE_MAILINGS
 from .utils import create_retirement_request_and_deactivate_account
@@ -280,7 +280,7 @@ class AccountViewSet(ViewSet):
     authentication_classes = (
         JwtAuthentication, BearerAuthenticationAllowInactiveUser, SessionAuthenticationAllowInactiveUser
     )
-    permission_classes = (permissions.IsAuthenticated,)
+    permission_classes = (permissions.IsAuthenticated, CanGetAccountInfo)
     parser_classes = (MergePatchParser,)
 
     def get(self, request):
@@ -292,10 +292,12 @@ def get(self, request):
     def list(self, request):
         """
         GET /api/user/v1/accounts?username={username1,username2}
-        GET /api/user/v1/accounts?email={user_email}
+        GET /api/user/v1/accounts?email={user_email} (Staff Only)
+        GET /api/user/v1/accounts?lms_user_id={lms_user_id} (Staff Only)
         """
         usernames = request.GET.get('username')
         user_email = request.GET.get('email')
+        lms_user_id = request.GET.get('lms_user_id')
         search_usernames = []
 
         if usernames:
@@ -307,6 +309,12 @@ def list(self, request):
             except (UserNotFound, User.DoesNotExist):
                 return Response(status=status.HTTP_404_NOT_FOUND)
             search_usernames = [user.username]
+        elif lms_user_id:
+            try:
+                user = User.objects.get(id=lms_user_id)
+            except (UserNotFound, User.DoesNotExist):
+                return Response(status=status.HTTP_404_NOT_FOUND)
+            search_usernames = [user.username]
         try:
             account_settings = get_account_settings(
                 request, search_usernames, view=request.query_params.get('view'))

openedx/core/djangoapps/user_authn/views/logout.py

@@ -1,6 +1,7 @@
 """ Views related to logout. """
 
 
+import bleach
 import re
 import urllib.parse as parse  # pylint: disable=import-error
 from urllib.parse import parse_qs, urlsplit, urlunsplit  # pylint: disable=import-error
@@ -57,7 +58,7 @@ def target(self):
         #  >> /courses/course-v1:ARTS+D1+2018_T/course/
         #  to handle this scenario we need to encode our URL using quote_plus and then unquote it again.
         if target_url:
-            target_url = parse.unquote(parse.quote_plus(target_url))
+            target_url = bleach.clean(parse.unquote(parse.quote_plus(target_url)))
 
         use_target_url = target_url and is_safe_login_or_logout_redirect(
             redirect_to=target_url,

openedx/core/djangoapps/user_authn/views/tests/test_logout.py

@@ -7,6 +7,7 @@
 import urllib
 from unittest import mock
 import ddt
+import bleach
 from django.conf import settings
 from django.test import TestCase
 from django.test.utils import override_settings
@@ -193,3 +194,21 @@ def test_learner_portal_logout_having_idp_logout_url(self):
                 'show_tpa_logout_link': True,
             }
             self.assertDictContainsSubset(expected, response.context_data)
+
+    @ddt.data(
+        ('%22%3E%3Cscript%3Ealert(%27xss%27)%3C/script%3E', 'edx.org'),
+    )
+    @ddt.unpack
+    def test_logout_redirect_failure_with_xss_vulnerability(self, redirect_url, host):
+        """
+        Verify that it will block the XSS attack on edX’s LMS logout page
+        """
+        url = '{logout_path}?redirect_url={redirect_url}'.format(
+            logout_path=reverse('logout'),
+            redirect_url=redirect_url
+        )
+        response = self.client.get(url, HTTP_HOST=host)
+        expected = {
+            'target': bleach.clean(urllib.parse.unquote(redirect_url)),
+        }
+        self.assertDictContainsSubset(expected, response.context_data)