feat: remove enterprise imports from third_party_auth

common/djangoapps/third_party_auth/pipeline.py

@@ -64,15 +64,15 @@ def B(*args, **kwargs):
 import json
 from collections import OrderedDict
 from logging import getLogger
+from random import randint
 from smtplib import SMTPException
 from uuid import uuid4
-from random import randint
 
 import six
 import social_django
 from django.conf import settings
+from django.contrib.auth import REDIRECT_FIELD_NAME, logout
 from django.contrib.auth.models import User  # lint-amnesty, pylint: disable=imported-auth-user
-from django.contrib.auth import logout, REDIRECT_FIELD_NAME
 from django.core.mail.message import EmailMessage
 from django.http import HttpResponseBadRequest
 from django.shortcuts import redirect
@@ -84,6 +84,14 @@ def B(*args, **kwargs):
 
 from common.djangoapps import third_party_auth
 from common.djangoapps.edxmako.shortcuts import render_to_string
+from common.djangoapps.third_party_auth.utils import (
+    get_associated_user_by_email_response,
+    is_oauth_provider,
+    is_saml_provider,
+    user_exists
+)
+from common.djangoapps.track import segment
+from common.djangoapps.util.json_request import JsonResponse
 from lms.djangoapps.verify_student.models import SSOVerification
 from lms.djangoapps.verify_student.utils import earliest_allowed_verification_date
 from openedx.core.djangoapps.site_configuration import helpers as configuration_helpers
@@ -93,16 +101,6 @@ def B(*args, **kwargs):
 from openedx.core.djangoapps.user_authn.toggles import is_auto_generated_username_enabled
 from openedx.core.djangoapps.user_authn.utils import is_safe_login_or_logout_redirect
 from openedx.core.djangoapps.user_authn.views.utils import get_auto_generated_username
-from common.djangoapps.third_party_auth.utils import (
-    get_associated_user_by_email_response,
-    get_user_from_email,
-    is_enterprise_customer_user,
-    is_oauth_provider,
-    is_saml_provider,
-    user_exists,
-)
-from common.djangoapps.track import segment
-from common.djangoapps.util.json_request import JsonResponse
 
 from . import provider
 
@@ -790,71 +788,10 @@ def associate_by_email_if_saml(auth_entry, backend, details, user, strategy, *ar
 
     This association is done ONLY if the user entered the pipeline belongs to SAML provider.
     """
-    from openedx.features.enterprise_support.api import enterprise_is_enabled
-
-    def get_user():
-        """
-        This is the helper method to get the user from system by matching email.
-        """
-        user_details = {'email': details.get('email')} if details else None
-        return get_user_from_email(user_details or {})
-
-    @enterprise_is_enabled()
-    def associate_by_email_if_enterprise_user():
-        """
-        If the learner arriving via SAML is already linked to the enterprise customer linked to the same IdP,
-        they should not be prompted for their edX password.
-        """
-        try:
-            enterprise_customer_user = is_enterprise_customer_user(current_provider.provider_id, current_user)
-            logger.info(
-                '[Multiple_SSO_SAML_Accounts_Association_to_User] Enterprise user verification:'
-                'User Email: {email}, User ID: {user_id}, Provider ID: {provider_id},'
-                ' is_enterprise_customer_user: {enterprise_customer_user}'.format(
-                    email=current_user.email,
-                    user_id=current_user.id,
-                    provider_id=current_provider.provider_id,
-                    enterprise_customer_user=enterprise_customer_user,
-                )
-            )
-
-            if enterprise_customer_user:
-                # this is python social auth pipeline default method to automatically associate social accounts
-                # if the email already matches a user account.
-                association_response, user_is_active = get_associated_user_by_email_response(
-                    backend, details, user, *args, **kwargs)
-
-                if not user_is_active:
-                    logger.info(
-                        '[Multiple_SSO_SAML_Accounts_Association_to_User] User association account is not'
-                        ' active: User Email: {email}, User ID: {user_id}, Provider ID: {provider_id},'
-                        ' is_enterprise_customer_user: {enterprise_customer_user}'.format(
-                            email=current_user.email,
-                            user_id=current_user.id,
-                            provider_id=current_provider.provider_id,
-                            enterprise_customer_user=enterprise_customer_user
-                        )
-                    )
-                    return None
-
-                return association_response
-
-        except Exception as ex:  # pylint: disable=broad-except
-            logger.exception('[Multiple_SSO_SAML_Accounts_Association_to_User] Error in'
-                             ' saml multiple accounts association: User ID: %s, User Email: %s:,'
-                             'Provider ID: %s, Exception: %s', current_user.id, current_user.email,
-                             current_provider.provider_id, ex)
-
     saml_provider, current_provider = is_saml_provider(strategy.request.backend.name, kwargs)
 
     if saml_provider:
-        # get the user by matching email if the pipeline user is not available.
-        current_user = user if user else get_user()
-
-        # Verify that the user linked to enterprise customer of current identity provider and an active user
-        associate_response = associate_by_email_if_enterprise_user() if current_user else None
-        if associate_response:
-            return associate_response
+        return get_associated_user_by_email_response(backend, details, user, *args, **kwargs)
 
 
 def user_details_force_sync(auth_entry, strategy, details, user=None, *args, **kwargs):  # lint-amnesty, pylint: disable=keyword-arg-before-vararg

common/djangoapps/third_party_auth/saml.py

@@ -8,15 +8,15 @@
 import requests
 from django.contrib.sites.models import Site
 from django.http import Http404
-from django.utils.functional import cached_property
 from django.utils.datastructures import MultiValueDictKeyError
+from django.utils.functional import cached_property
 from django_countries import countries
 from onelogin.saml2.settings import OneLogin_Saml2_Settings
 from social_core.backends.saml import OID_EDU_PERSON_ENTITLEMENT, SAMLAuth, SAMLIdentityProvider
-from social_core.exceptions import AuthForbidden, AuthMissingParameter, AuthInvalidParameter
+from social_core.exceptions import AuthForbidden, AuthInvalidParameter, AuthMissingParameter
 
-from openedx.core.djangoapps.theming.helpers import get_current_request
 from common.djangoapps.third_party_auth.exceptions import IncorrectConfigurationException
+from openedx.core.djangoapps.theming.helpers import get_current_request
 
 STANDARD_SAML_PROVIDER_KEY = 'standard_saml_provider'
 SAP_SUCCESSFACTORS_SAML_KEY = 'sap_success_factors'
@@ -141,12 +141,17 @@ def auth_url(self):
 
     def disconnect(self, *args, **kwargs):
         """
-        Override of SAMLAuth.disconnect to unlink the learner from enterprise customer if associated.
+        Override to emit a signal when a user disconnects their SAML account.
         """
-        from openedx.features.enterprise_support.api import unlink_enterprise_user_from_idp
-        user = kwargs.get('user', None)
-        unlink_enterprise_user_from_idp(self.strategy.request, user, self.name)
-        return super().disconnect(*args, **kwargs)
+        from common.djangoapps.third_party_auth.signals import SocialAuthAccountDisconnected
+        result = super().disconnect(*args, **kwargs)
+        SocialAuthAccountDisconnected.send(
+            sender=self.__class__,
+            request=self.strategy.request,
+            user=self.strategy.request.user if self.strategy.request else None,
+            social=self,
+        )
+        return result
 
     def _check_entitlements(self, idp, attributes):
         """

common/djangoapps/third_party_auth/settings.py

@@ -12,7 +12,6 @@
 
 
 from django.conf import settings
-from openedx.features.enterprise_support.api import insert_enterprise_pipeline_elements
 
 
 def apply_settings(django_settings):
@@ -70,9 +69,6 @@ def apply_settings(django_settings):
         'common.djangoapps.third_party_auth.pipeline.ensure_redirect_url_is_safe',
     ]
 
-    # Add enterprise pipeline elements if the enterprise app is installed
-    insert_enterprise_pipeline_elements(django_settings.SOCIAL_AUTH_PIPELINE)
-
     # Required so that we can use unmodified PSA OAuth2 backends:
     django_settings.SOCIAL_AUTH_STRATEGY = 'common.djangoapps.third_party_auth.strategy.ConfigurationModelStrategy'
 

common/djangoapps/third_party_auth/signals.py

@@ -0,0 +1,8 @@
+"""
+Django signals for third_party_auth.
+"""
+from django.dispatch import Signal
+
+# Signal fired when a user disconnects a social auth provider account.
+# providing_args=["request", "user", "social"]
+SocialAuthAccountDisconnected = Signal()

common/djangoapps/third_party_auth/tests/test_utils.py

@@ -11,19 +11,14 @@
 from common.djangoapps.student.tests.factories import UserFactory
 from common.djangoapps.third_party_auth.tests.testutil import TestCase
 from common.djangoapps.third_party_auth.utils import (
+    convert_saml_slug_provider_id,
     get_associated_user_by_email_response,
     get_user_from_email,
-    is_enterprise_customer_user,
     is_oauth_provider,
     parse_metadata_xml,
-    user_exists,
-    convert_saml_slug_provider_id,
+    user_exists
 )
 from openedx.core.djangolib.testing.utils import skip_unless_lms
-from openedx.features.enterprise_support.tests.factories import (
-    EnterpriseCustomerIdentityProviderFactory,
-    EnterpriseCustomerUserFactory,
-)
 
 
 @ddt.ddt
@@ -65,26 +60,6 @@ def test_get_user(self):
         assert get_user_from_email({'email': 'test_user@example.com'})
         assert not get_user_from_email({'email': 'invalid@example.com'})
 
-    def test_is_enterprise_customer_user(self):
-        """
-        Verify that if user is an enterprise learner.
-        """
-        # Create users from factory
-
-        user = UserFactory(username='test_user', email='test_user@example.com')
-        other_user = UserFactory(username='other_user', email='other_user@example.com')
-        customer_idp = EnterpriseCustomerIdentityProviderFactory.create(
-            provider_id='the-provider',
-        )
-        customer = customer_idp.enterprise_customer
-        EnterpriseCustomerUserFactory.create(
-            enterprise_customer=customer,
-            user_id=user.id,
-        )
-
-        assert is_enterprise_customer_user('the-provider', user)
-        assert not is_enterprise_customer_user('the-provider', other_user)
-
     @ddt.data(
         ('saml-farkle', False),
         ('oa2-fergus', True),

common/djangoapps/third_party_auth/utils.py

@@ -11,16 +11,12 @@
 import requests
 from django.contrib.auth.models import User  # lint-amnesty, pylint: disable=imported-auth-user
 from django.utils.timezone import now
-from enterprise.models import EnterpriseCustomerIdentityProvider, EnterpriseCustomerUser
 from lxml import etree
 from onelogin.saml2.utils import OneLogin_Saml2_Utils
 from requests import exceptions
 from social_core.pipeline.social_auth import associate_by_email
-from common.djangoapps.student.models import (
-    email_exists_or_retired,
-    username_exists_or_retired
-)
 
+from common.djangoapps.student.models import email_exists_or_retired, username_exists_or_retired
 from common.djangoapps.third_party_auth.models import OAuth2ProviderConfig, SAMLProviderData
 from openedx.core.djangolib.markup import Text
 
@@ -235,14 +231,6 @@ def is_saml_provider(backend, kwargs):
             current_provider.slug in [saml_provider.slug for saml_provider in saml_providers_list]), current_provider
 
 
-def is_enterprise_customer_user(provider_id, user):
-    """ Verify that the user linked to enterprise customer of current identity provider"""
-    enterprise_idp = EnterpriseCustomerIdentityProvider.objects.get(provider_id=provider_id)
-
-    return EnterpriseCustomerUser.objects.filter(enterprise_customer=enterprise_idp.enterprise_customer,
-                                                 user_id=user.id).exists()
-
-
 def is_oauth_provider(backend_name, **kwargs):
     """
     Verify that the third party provider uses oauth