feat: logout other sessions on email change

common/djangoapps/student/signals/receivers.py

@@ -105,8 +105,10 @@ def listen_for_verified_name_approved(sender, user_id, profile_name, **kwargs):
 
 
 @receiver(USER_EMAIL_CHANGED)
-def _listen_for_user_email_changed(sender, user, **kwargs):
-    """ If user has changed their email, update that in email Braze. """
+def _listen_for_user_email_changed(sender, user, request, **kwargs):
+    """ If user has changed their email, update that in session and Braze profile. """
+    request.session['email'] = user.email
+
     email = user.email
     user_id = user.id
     attributes = [{'email': email, 'external_id': user_id}]

common/djangoapps/student/tests/test_receivers.py

@@ -9,7 +9,7 @@
 from common.djangoapps.student.signals.signals import USER_EMAIL_CHANGED
 from common.djangoapps.student.tests.factories import CourseEnrollmentFactory, UserFactory, UserProfileFactory
 from lms.djangoapps.courseware.toggles import COURSEWARE_MICROFRONTEND_PROGRESS_MILESTONES
-from openedx.core.djangolib.testing.utils import skip_unless_lms
+from openedx.core.djangolib.testing.utils import skip_unless_lms, get_mock_request
 from openedx.features.name_affirmation_api.utils import is_name_affirmation_installed
 from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase  # lint-amnesty, pylint: disable=wrong-import-order
 
@@ -75,10 +75,14 @@ def test_listen_for_verified_name_approved(self):
     @patch('common.djangoapps.student.signals.receivers.get_braze_client')
     def test_listen_for_user_email_changed(self, mock_get_braze_client):
         """
-        Ensure that USER_EMAIL_CHANGED signal triggers correct calls to get_braze_client.
+        Ensure that USER_EMAIL_CHANGED signal triggers correct calls to
+        get_braze_client and update email in session.
         """
         user = UserFactory(email='[email protected]', username='jdoe')
+        request = get_mock_request(user=user)
+        request.session = self.client.session
 
-        USER_EMAIL_CHANGED.send(sender=None, user=user)
+        USER_EMAIL_CHANGED.send(sender=None, user=user, request=request)
 
         assert mock_get_braze_client.called
+        assert request.session.get('email') == user.email

common/djangoapps/student/views/management.py

@@ -910,7 +910,7 @@ def confirm_email_change(request, key):
 
         response = render_to_response("email_change_successful.html", address_context)
 
-        USER_EMAIL_CHANGED.send(sender=None, user=user)
+        USER_EMAIL_CHANGED.send(sender=None, user=user, request=request)
         return response
 
 

lms/envs/common.py

@@ -2224,6 +2224,9 @@ def _make_locale_paths(settings):  # pylint: disable=missing-function-docstring
     #'django.contrib.auth.middleware.AuthenticationMiddleware',
     'openedx.core.djangoapps.cache_toolbox.middleware.CacheBackedAuthenticationMiddleware',
 
+    # Middleware to flush user's session in other browsers when their email is changed.
+    'openedx.core.djangoapps.safe_sessions.middleware.EmailVerificationMiddleware',
+
     'common.djangoapps.student.middleware.UserStandingMiddleware',
     'openedx.core.djangoapps.contentserver.middleware.StaticContentServer',
 

openedx/core/djangoapps/safe_sessions/middleware.py

@@ -768,6 +768,39 @@ def _get_encrypted_request_headers(request):
         return encrypt_for_log(str(request.headers), getattr(settings, 'SAFE_SESSIONS_DEBUG_PUBLIC_KEY', None))
 
 
+class EmailVerificationMiddleware(MiddlewareMixin):
+    """
+    Middleware responsible for invalidating user sessions
+    by detecting a mismatch between the user's email in
+    the session and request.user.email.
+
+    This middleware ensures that the sessions in other browsers
+    are invalidated (and remain validated in current browser)
+    when a user changes their email in one browser.
+
+    The user's email is stored in their session during login
+    and gets updated when the user changes their email.
+    This middleware checks for any mismatch between the stored email
+    and the current user's email in each request, and if found,
+    it invalidates/flushes the session and mark cookies for deletion in request
+    which are then deleted in the process_response of SafeSessionMiddleware.
+    """
+
+    def process_request(self, request):
+        """
+        Invalidate the user session if there's a mismatch
+        between the email in the user's session and request.user.email.
+        """
+        if request.user.is_authenticated:
+            user_session_email = request.session.get('email', None)
+
+            if user_session_email is not None and request.user.email != user_session_email:
+                # Flush the session and mark cookies for deletion.
+                request.session.flush()
+                request.user = AnonymousUser()
+                _mark_cookie_for_deletion(request)
+
+
 def obscure_token(value: Union[str, None]) -> Union[str, None]:
     """
     Return a short string that can be used to detect other occurrences

openedx/core/djangoapps/safe_sessions/tests/test_middleware.py

@@ -17,6 +17,7 @@
 from common.djangoapps.student.tests.factories import UserFactory
 
 from ..middleware import (
+    EmailVerificationMiddleware,
     SafeCookieData,
     SafeSessionMiddleware,
     mark_user_change_as_expected,
@@ -615,3 +616,100 @@ def test_user_change_with_no_ids(self):
         request.user = object()
         assert len(request.debug_user_changes) == 2
         assert "Changing request user but user has no id." in request.debug_user_changes[1]
+
+
+class TestEmailVerificationMiddleware(TestSafeSessionsLogMixin, TestCase):
+    """
+    Test class for EmailVerificationMiddleware
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.TEST_PASSWORD = 'Password1234'
+        self.user = UserFactory.create(email='[email protected]', password=self.TEST_PASSWORD)
+        self.addCleanup(set_current_request, None)
+        self.request = get_mock_request(self.user)
+        self.client.response = HttpResponse()
+        self.client.response.cookies = SimpleCookie()
+
+    @patch('openedx.core.djangoapps.safe_sessions.middleware._mark_cookie_for_deletion')
+    def test_process_request_not_authenticated(self, mock_mark_cookie_for_deletion):
+        """
+        Calls EmailVerificationMiddleware.process_request when no user is authenticated.
+        Verifies that session and cookies are not affected.
+        """
+        # Unauthenticated User
+        self.request.user = AnonymousUser()
+
+        # Call process_request without authenticating a user
+        EmailVerificationMiddleware(get_response=lambda request: None).process_request(self.request)
+
+        # Assert that session and cookies are not affected
+        # Assert that _mark_cookie_for_deletion not called
+        mock_mark_cookie_for_deletion.assert_not_called()
+
+    def test_process_request_authenticated_no_session_email(self):
+        """
+        Calls EmailVerificationMiddleware.process_request when a user is authenticated
+        but there is no email in the session. Verifies that session and cookies are not affected.
+        """
+        # Log in the user
+        self.client.login(email=self.user.email, password=self.TEST_PASSWORD)
+        self.request.session = self.client.session
+
+        # Ensure no email is stored in the session
+        self.assertIsNone(self.request.session.get('email'))
+
+        # Call process_request
+        EmailVerificationMiddleware(get_response=lambda request: None).process_request(self.request)
+
+        # Assert that the session and cookies are not affected
+        self.assertIsNone(self.request.session.get('email'))
+        self.assertEqual(len(self.client.response.cookies), 0)
+
+    def test_process_request_authenticated_matching_session_email(self):
+        """
+        Calls EmailVerificationMiddleware.process_request when a user is authenticated
+        and the session email matches request.user.email. Verifies that session and cookies are not affected.
+        """
+        # Log in the user
+        self.client.login(email=self.user.email, password=self.TEST_PASSWORD)
+        self.request.session = self.client.session
+        self.request.session['email'] = self.user.email  # Set the session email to match request.user.email
+        self.client.response.set_cookie(settings.SESSION_COOKIE_NAME, 'authenticated')  # Add some logged-in cookie
+
+        # Ensure email matches in the session
+        self.assertEqual(self.request.session.get('email'), self.user.email)
+        # Ensure session cookie exist
+        self.assertEqual(len(self.client.response.cookies), 1)
+
+        # Call process_request
+        EmailVerificationMiddleware(get_response=lambda request: None).process_request(self.request)
+
+        # Assert that the session and cookies are not affected
+        self.assertEqual(self.request.session.get('email'), self.user.email)
+        self.assertEqual(len(self.client.response.cookies), 1)
+        self.assertEqual(self.client.response.cookies[settings.SESSION_COOKIE_NAME].value, 'authenticated')
+
+    @patch('openedx.core.djangoapps.safe_sessions.middleware._mark_cookie_for_deletion')
+    def test_process_request_email_mismatch(self, mock_mark_cookie_for_deletion):
+        """
+        Calls EmailVerificationMiddleware.process_request with a mismatch between session email and user email.
+        Verifies that session is flushed and cookies are marked for deletion.
+        """
+        # Log in the user
+        self.client.login(email=self.user.email, password=self.TEST_PASSWORD)
+        self.request.session = self.client.session
+        self.request.session['email'] = self.user.email  # Set the session email to match request.user.email
+        self.client.response.set_cookie(settings.SESSION_COOKIE_NAME, 'authenticated')  # Add some logged-in cookie
+
+        # Modify session email to create a mismatch (Email changed)
+        self.request.session['email'] = '[email protected]'
+
+        # Call process_request
+        EmailVerificationMiddleware(get_response=lambda request: None).process_request(self.request)
+
+        # Assert that the session is flushed and cookies marked for deletion
+        mock_mark_cookie_for_deletion.assert_called()
+        assert self.request.session.get(SESSION_KEY) is None
+        assert self.request.user == AnonymousUser()

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

@@ -315,6 +315,9 @@ def _handle_successful_authentication_and_login(user, request):
         request.session.set_expiry(604800 * 4)
         log.debug("Setting user session expiry to 4 weeks")
 
+        # Store the user's email for session consistency (used by EmailVerificationMiddleware)
+        request.session['email'] = user.email
+
         # .. event_implemented_name: SESSION_LOGIN_COMPLETED
         SESSION_LOGIN_COMPLETED.send_event(
             user=UserData(

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

@@ -1205,3 +1205,21 @@ def test_missing_login_params(self, is_api_v1):
         # Missing both email and password
         response = self.client.post(url, {})
         self.assertHttpBadRequest(response)
+
+    def test_email_in_session(self):
+        # Login and check email in session
+        data = {
+            "email": self.EMAIL,
+            "password": self.PASSWORD,
+        }
+
+        response = self.client.post(self.url, data)
+        self.assertHttpOK(response)
+
+        # Ensure the login was successful
+        self.assertEqual(response.status_code, 200)
+
+        # Verify that the email is correctly stored in the session
+        session_data = self.client.session.load()
+        stored_email = session_data.get('email')
+        assert stored_email == self.EMAIL