feat: replace enterprise logistration logic with openedx-filters hooks

common/djangoapps/third_party_auth/models.py

@@ -15,6 +15,7 @@
 from django.db import models
 from django.utils import timezone
 from django.utils.translation import gettext_lazy as _
+from openedx_filters.authentication.types import RunningPipelineKwargs
 from organizations.models import Organization
 from social_core.backends.base import BaseAuth
 from social_core.backends.oauth import OAuthAuth
@@ -317,16 +318,19 @@ def get_social_auth_uid(self, remote_id):
         return remote_id
 
     @classmethod
-    def get_register_form_data(cls, pipeline_kwargs):
+    def get_register_form_data(cls, pipeline_kwargs: RunningPipelineKwargs):
         """Gets dict of data to display on the register form.
 
         register_user uses this to populate
         the new account creation form with values supplied by the user's chosen
         provider, preventing duplicate data entry.
 
         Args:
-            pipeline_kwargs: dict of string -> object. Keyword arguments
-                accumulated by the pipeline thus far.
+            pipeline_kwargs (RunningPipelineKwargs): Keyword arguments accumulated by the
+                pipeline thus far. This method is reachable from pipeline steps of the
+                registration form filter, which may live in other repositories, so the
+                argument's declared shape is the shared cross-repository contract in
+                ``openedx_filters.authentication.types``.
 
         Returns:
             Dict of string -> string. Keys are names of form fields; values are

common/djangoapps/third_party_auth/pipeline.py

@@ -78,6 +78,7 @@ def B(*args, **kwargs):
 from django.shortcuts import redirect
 from django.urls import reverse
 from edx_django_utils.monitoring import set_custom_attribute
+from openedx_filters.authentication.types import RunningPipeline
 from social_core.exceptions import AuthException
 from social_core.pipeline import partial
 from social_core.utils import module_member, slugify
@@ -212,8 +213,15 @@ def get_unlink_form_name(self):
         return self.provider.provider_id + '_unlink_form'
 
 
-def get(request):
-    """Gets the running pipeline's data from the passed request."""
+def get(request) -> RunningPipeline | None:
+    """Gets the running pipeline's data from the passed request.
+
+    The returned mapping is the shared cross-repository contract for third-party auth
+    pipeline state: it is handed to pipeline steps of the login and registration form
+    filters, which may live in other repositories. See
+    ``openedx_filters.authentication.types.RunningPipeline`` for the declared shape, and
+    keep this function's dict literal in agreement with it.
+    """
     strategy = social_django.utils.load_strategy(request)
     token = strategy.session_get('partial_pipeline_token')
 
@@ -222,7 +230,7 @@ def get(request):
         token = strategy.session_get('partial_pipeline_token')
 
     partial_object = strategy.partial_load(token)
-    pipeline_data = None
+    pipeline_data: RunningPipeline | None = None
     if partial_object:
         pipeline_data = {'kwargs': partial_object.kwargs, 'backend': partial_object.backend}
     return pipeline_data

mypy.ini

@@ -6,6 +6,16 @@ plugins =
     mypy_django_plugin.main,
     mypy_drf_plugin.main
 files =
+    # Third-party auth pipeline state is a cross-repository contract (see
+    # openedx_filters.authentication.types.RunningPipeline). pipeline.py must be a check
+    # root rather than merely imported: follow_imports = silent suppresses errors inside
+    # unlisted modules, so the dict literal that produces that contract would otherwise
+    # go unchecked.
+    common/djangoapps/third_party_auth/pipeline.py,
+    common/djangoapps/third_party_auth/provider.py,
+    openedx/core/djangoapps/user_authn/views/utils.py,
+    openedx/core/djangoapps/user_authn/views/login_form.py,
+    openedx/core/djangoapps/user_authn/views/registration_form.py,
     cms/lib/xblock,
     cms/djangoapps/contentstore/rest_api/v2/views,
     cms/djangoapps/contentstore/xblock_storage_handlers,
@@ -64,6 +74,8 @@ ignore_missing_imports = True
 ignore_missing_imports = True
 [mypy-search.*]
 ignore_missing_imports = True
+[mypy-six.*]
+ignore_missing_imports = True
 [mypy-rules.*]
 ignore_missing_imports = True
 [mypy-web_fragments.*]

openedx/core/djangoapps/user_authn/api/tests/data_mock.py

@@ -112,3 +112,54 @@
         'extended_profile': []
     }
 }
+
+# Entries a plugin may contribute through the AuthnMFEContextGenerated filter. The
+# serializer declares no fields for these, so they are merged into the response as-is,
+# at whatever depth the contributor chose.
+EXTRA_CONTEXT_DATA = {
+    'brandingName': 'Acme Corp',
+    'brandingLogoUrl': 'https://example.com/acme-logo.png',
+    'brandingStrings': {
+        'welcome': 'Welcome, Acme learners!',
+    },
+}
+
+MFE_CONTEXT_WITH_EXTRA_CONTEXT_DATA = {
+    'context_data': {
+        'currentProvider': None,
+        'platformName': 'edX',
+        'providers': [],
+        'secondaryProviders': [],
+        'finishAuthUrl': None,
+        'errorMessage': None,
+        'registerFormSubmitButtonText': 'Create Account',
+        'autoSubmitRegForm': False,
+        'syncLearnerProfileData': False,
+        'countryCode': '',
+        'welcomePageRedirectUrl': '',
+        'pipeline_user_details': {},
+        'extra_context': EXTRA_CONTEXT_DATA,
+    },
+}
+
+SERIALIZED_MFE_CONTEXT_WITH_EXTRA_CONTEXT_DATA = {
+    'contextData': {
+        'currentProvider': None,
+        'platformName': 'edX',
+        'providers': [],
+        'secondaryProviders': [],
+        'finishAuthUrl': None,
+        'errorMessage': None,
+        'registerFormSubmitButtonText': 'Create Account',
+        'autoSubmitRegForm': False,
+        'syncLearnerProfileData': False,
+        'countryCode': '',
+        'welcomePageRedirectUrl': '',
+        'pipelineUserDetails': {},
+        **EXTRA_CONTEXT_DATA,
+    },
+    'registrationFields': {},
+    'optionalFields': {
+        'extended_profile': [],
+    },
+}

openedx/core/djangoapps/user_authn/api/tests/test_serializers.py

@@ -5,8 +5,10 @@
 from django.test import TestCase
 
 from openedx.core.djangoapps.user_authn.api.tests.data_mock import (
+    MFE_CONTEXT_WITH_EXTRA_CONTEXT_DATA,
     MFE_CONTEXT_WITH_TPA_DATA,
     MFE_CONTEXT_WITHOUT_TPA_DATA,
+    SERIALIZED_MFE_CONTEXT_WITH_EXTRA_CONTEXT_DATA,
     SERIALIZED_MFE_CONTEXT_WITH_TPA_DATA,
     SERIALIZED_MFE_CONTEXT_WITHOUT_TPA_DATA,
 )
@@ -44,3 +46,30 @@ def test_mfe_context_serializer_default_response(self):
             serialized_data,
             SERIALIZED_MFE_CONTEXT_WITHOUT_TPA_DATA
         )
+
+    def test_mfe_context_serializer_with_extra_context(self):
+        """
+        Test that entries the serializer declares no fields for are merged into contextData
+        from extra_context, preserving their nesting.
+        """
+        output_data = MFEContextSerializer(
+            MFE_CONTEXT_WITH_EXTRA_CONTEXT_DATA
+        ).data
+
+        assert output_data == SERIALIZED_MFE_CONTEXT_WITH_EXTRA_CONTEXT_DATA
+
+    def test_mfe_context_serializer_ignores_undeclared_context_keys(self):
+        """
+        Test that context entries outside extra_context are still dropped, so the response
+        shape stays under the serializer's control.
+        """
+        context = {
+            'context_data': {
+                **MFE_CONTEXT_WITHOUT_TPA_DATA['context_data'],
+                'skipRegistrationOptionalCheckboxes': True,
+            },
+        }
+
+        serialized_data = MFEContextSerializer(context).data
+
+        assert 'skipRegistrationOptionalCheckboxes' not in serialized_data['contextData']

openedx/core/djangoapps/user_authn/serializers.py

@@ -61,6 +61,18 @@ def get_pipelineUserDetails(self, obj):
             return PipelineUserDetailsSerializer(obj.get('pipeline_user_details')).data
         return {}
 
+    def to_representation(self, instance):
+        """
+        Serialize the declared fields, then merge in the context's ``extra_context`` entries.
+
+        ``extra_context`` holds entries contributed by plugins, which this serializer cannot
+        declare fields for. They are merged as-is: the contributor owns their shape, and no
+        coercion is applied.
+        """
+        representation = super().to_representation(instance)
+        representation.update(instance.get('extra_context') or {})
+        return representation
+
 
 class MFEContextSerializer(serializers.Serializer):
     """

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

@@ -7,8 +7,6 @@
 import hashlib
 import json
 import logging
-import re
-import urllib
 
 from django.conf import settings
 from django.contrib.auth import authenticate, get_user_model
@@ -29,6 +27,7 @@
 from eventtracking import tracker
 from openedx_events.learning.data import UserData, UserPersonalData
 from openedx_events.learning.signals import SESSION_LOGIN_COMPLETED
+from openedx_filters.authentication.filters import LoginAltRedirectURLRequested
 from openedx_filters.learning.filters import StudentLoginRequested
 from rest_framework import status
 from rest_framework.views import APIView
@@ -54,12 +53,12 @@
     is_require_third_party_auth_enabled,
     should_redirect_to_authn_microfrontend,
 )
+from openedx.core.djangoapps.user_authn.utils import is_safe_login_or_logout_redirect
 from openedx.core.djangoapps.user_authn.views.login_form import get_login_session_form
 from openedx.core.djangoapps.user_authn.views.password_reset import send_password_reset_email_for_user
-from openedx.core.djangoapps.user_authn.views.utils import API_V1, ENTERPRISE_ENROLLMENT_URL_REGEX, UUID4_REGEX
+from openedx.core.djangoapps.user_authn.views.utils import API_V1
 from openedx.core.djangoapps.util.user_messages import PageLevelMessages
 from openedx.core.djangolib.markup import HTML, Text
-from openedx.features.enterprise_support.api import activate_learner_enterprise, get_enterprise_learner_data_from_api
 
 log = logging.getLogger("edx.student")
 AUDIT_LOG = logging.getLogger("audit")
@@ -477,33 +476,45 @@ def finish_auth(request):
     )
 
 
-def enterprise_selection_page(request, user, next_url):
+def _get_alt_redirect_url(request, redirect_url, user):
     """
-    Updates redirect url to enterprise selection page if user is associated
-    with multiple enterprises otherwise return the next url.
+    Ask the configured pipeline steps for an alternative post-login redirect URL.
 
-    param:
-      next_url(string): The URL to redirect to after multiple enterprise selection or in case
-      the selection page is bypassed e.g when dealing with direct enrolment urls.
-    """
-    redirect_url = next_url
+    The pipeline is arbitrary configured code, so its answer is held to the same
+    open-redirect protections as a caller-supplied ``?next=`` parameter: an unsafe URL is
+    discarded and the caller's own destination is used instead.
 
-    response = get_enterprise_learner_data_from_api(user)
-    if response and len(response) > 1:
-        redirect_url = reverse("enterprise_select_active") + "/?success_url=" + urllib.parse.quote(next_url)
+    Arguments:
+        request (HttpRequest)
+        redirect_url (str): the destination the caller intends to send the user to.
+        user (User): the authenticated user.
 
-        # Check to see if next url has an enterprise in it. In this case if user is associated with
-        # that enterprise, activate that enterprise and bypass the selection page.
-        if re.match(ENTERPRISE_ENROLLMENT_URL_REGEX, urllib.parse.unquote(next_url)):
-            enterprise_in_url = re.search(UUID4_REGEX, next_url).group(0)
-            for enterprise in response:
-                if enterprise_in_url == str(enterprise["enterprise_customer"]["uuid"]):
-                    is_activated_successfully = activate_learner_enterprise(request, user, enterprise_in_url)
-                    if is_activated_successfully:
-                        redirect_url = next_url
-                    break
+    Returns: str
+        the alternative redirect url if safe, else the given redirect_url.
+    """
+    # .. filter_implemented_name: LoginAltRedirectURLRequested
+    # .. filter_type: org.openedx.authentication.login.alt_redirect_url.requested.v1
+    alt_redirect_url, __ = LoginAltRedirectURLRequested.run_filter(
+        redirect_url=redirect_url,
+        user=user,
+    )
+
+    if alt_redirect_url == redirect_url:
+        return redirect_url
+
+    if not alt_redirect_url or not is_safe_login_or_logout_redirect(
+        redirect_to=alt_redirect_url,
+        request_host=request.get_host(),
+        dot_client_id=request.POST.get("client_id"),
+        require_https=request.is_secure(),
+    ):
+        log.warning(
+            "Unsafe alternative redirect URL detected after login: '%(alt_redirect_url)s'",
+            {"alt_redirect_url": alt_redirect_url},
+        )
+        return redirect_url
 
-    return redirect_url
+    return alt_redirect_url
 
 
 @ensure_csrf_cookie
@@ -648,7 +659,8 @@ def login_user(request, api_version="v1"):  # pylint: disable=too-many-statement
         elif should_redirect_to_authn_microfrontend():
             next_url, root_url = get_next_url_for_login_page(request, include_host=True)
             redirect_url = get_redirect_url_with_host(
-                root_url, enterprise_selection_page(request, possibly_authenticated_user, finish_auth_url or next_url)
+                root_url,
+                _get_alt_redirect_url(request, finish_auth_url or next_url, possibly_authenticated_user),
             )
 
         if (