Cleanup of third-party login and auto-enrollment

common/djangoapps/auth_exchange/forms.py

@@ -10,6 +10,7 @@
 from provider.oauth2.models import Client
 from requests import HTTPError
 from social.backends import oauth as social_oauth
+from social.exceptions import AuthException
 
 from third_party_auth import pipeline
 
@@ -54,7 +55,7 @@ def clean(self):
         if self._errors:
             return {}
 
-        backend = self.request.social_strategy.backend
+        backend = self.request.backend
         if not isinstance(backend, social_oauth.BaseOAuth2):
             raise OAuthValidationError(
                 {
@@ -88,8 +89,8 @@ def clean(self):
 
         user = None
         try:
-            user = backend.do_auth(self.cleaned_data.get("access_token"))
-        except HTTPError:
+            user = backend.do_auth(self.cleaned_data.get("access_token"), allow_inactive_user=True)
+        except (HTTPError, AuthException):
             pass
         if user and isinstance(user, User):
             self.cleaned_data["user"] = user

common/djangoapps/auth_exchange/tests/test_forms.py

@@ -24,8 +24,11 @@ class AccessTokenExchangeFormTest(AccessTokenExchangeTestMixin):
     def setUp(self):
         super(AccessTokenExchangeFormTest, self).setUp()
         self.request = RequestFactory().post("dummy_url")
+        redirect_uri = 'dummy_redirect_url'
         SessionMiddleware().process_request(self.request)
-        self.request.social_strategy = social_utils.load_strategy(self.request, self.BACKEND)
+        self.request.social_strategy = social_utils.load_strategy(self.request)
+        # pylint: disable=no-member
+        self.request.backend = social_utils.load_backend(self.request.social_strategy, self.BACKEND, redirect_uri)
 
     def _assert_error(self, data, expected_error, expected_error_description):
         form = AccessTokenExchangeForm(request=self.request, data=data)

common/djangoapps/external_auth/tests/test_shib.py

@@ -20,6 +20,7 @@
     shib_login, course_specific_login, course_specific_register, _flatten_to_ascii
 )
 from mock import patch
+from urllib import urlencode
 
 from student.views import create_account, change_enrollment
 from student.models import UserProfile, CourseEnrollment
@@ -169,7 +170,7 @@ def test_shib_login(self):
                 if idp == "https://idp.stanford.edu/" and remote_user == 'withmap@stanford.edu':
                     self.assertIsInstance(response, HttpResponseRedirect)
                     self.assertEqual(request.user, user_w_map)
-                    self.assertEqual(response['Location'], '/')
+                    self.assertEqual(response['Location'], '/dashboard')
                     # verify logging:
                     self.assertEquals(len(audit_log_calls), 2)
                     self._assert_shib_login_is_logged(audit_log_calls[0], remote_user)
@@ -193,7 +194,7 @@ def test_shib_login(self):
                     self.assertIsNotNone(ExternalAuthMap.objects.get(user=user_wo_map))
                     self.assertIsInstance(response, HttpResponseRedirect)
                     self.assertEqual(request.user, user_wo_map)
-                    self.assertEqual(response['Location'], '/')
+                    self.assertEqual(response['Location'], '/dashboard')
                     # verify logging:
                     self.assertEquals(len(audit_log_calls), 2)
                     self._assert_shib_login_is_logged(audit_log_calls[0], remote_user)
@@ -242,7 +243,7 @@ def _base_test_extauth_auto_activate_user_with_flag(self, log_user_string="inact
         self.assertTrue(inactive_user.is_active)
         self.assertIsInstance(response, HttpResponseRedirect)
         self.assertEqual(request.user, inactive_user)
-        self.assertEqual(response['Location'], '/')
+        self.assertEqual(response['Location'], '/dashboard')
         # verify logging:
         self.assertEquals(len(audit_log_calls), 3)
         self._assert_shib_login_is_logged(audit_log_calls[0], log_user_string)
@@ -549,29 +550,20 @@ def test_shib_login_enrollment(self):
         # no enrollment before trying
         self.assertFalse(CourseEnrollment.is_enrolled(student, course.id))
         self.client.logout()
+        params = [
+            ('course_id', course.id.to_deprecated_string()),
+            ('enrollment_action', 'enroll'),
+            ('next', '/testredirect')
+        ]
         request_kwargs = {'path': '/shib-login/',
-                          'data': {'enrollment_action': 'enroll', 'course_id': course.id.to_deprecated_string(), 'next': '/testredirect'},
+                          'data': dict(params),
                           'follow': False,
                           'REMOTE_USER': 'testuser@stanford.edu',
                           'Shib-Identity-Provider': 'https://idp.stanford.edu/'}
         response = self.client.get(**request_kwargs)
-        # successful login is a redirect to "/"
+        # successful login is a redirect to the URL that handles auto-enrollment
         self.assertEqual(response.status_code, 302)
-        self.assertEqual(response['location'], 'http://testserver/testredirect')
-        # now there is enrollment
-        self.assertTrue(CourseEnrollment.is_enrolled(student, course.id))
-
-        # Clean up and try again with POST (doesn't happen with real production shib, doing this for test coverage)
-        self.client.logout()
-        CourseEnrollment.unenroll(student, course.id)
-        self.assertFalse(CourseEnrollment.is_enrolled(student, course.id))
-
-        response = self.client.post(**request_kwargs)
-        # successful login is a redirect to "/"
-        self.assertEqual(response.status_code, 302)
-        self.assertEqual(response['location'], 'http://testserver/testredirect')
-        # now there is enrollment
-        self.assertTrue(CourseEnrollment.is_enrolled(student, course.id))
+        self.assertEqual(response['location'], 'http://testserver/account/finish_auth?{}'.format(urlencode(params)))
 
 
 class ShibUtilFnTest(TestCase):

common/djangoapps/external_auth/views.py

@@ -22,6 +22,7 @@
 if settings.FEATURES.get('AUTH_USE_CAS'):
     from django_cas.views import login as django_cas_login
 
+from student.helpers import get_next_url_for_login_page
 from student.models import UserProfile
 
 from django.http import HttpResponse, HttpResponseRedirect, HttpRequest, HttpResponseForbidden
@@ -118,7 +119,8 @@ def openid_login_complete(request,
             external_domain,
             details,
             details.get('email', ''),
-            fullname
+            fullname,
+            retfun=functools.partial(redirect, get_next_url_for_login_page(request)),
         )
 
     return render_failure(request, 'Openid failure')
@@ -236,14 +238,6 @@ def _external_login_or_signup(request,
     login(request, user)
     request.session.set_expiry(0)
 
-    # Now to try enrollment
-    # Need to special case Shibboleth here because it logs in via a GET.
-    # testing request.method for extra paranoia
-    if uses_shibboleth and request.method == 'GET':
-        enroll_request = _make_shib_enrollment_request(request)
-        student.views.try_change_enrollment(enroll_request)
-    else:
-        student.views.try_change_enrollment(request)
     if settings.FEATURES['SQUELCH_PII_IN_LOGS']:
         AUDIT_LOG.info(u"Login success - user.id: {0}".format(user.id))
     else:
@@ -449,9 +443,7 @@ def ssl_login(request):
 
     (_user, email, fullname) = _ssl_dn_extract_info(cert)
 
-    redirect_to = request.GET.get('next')
-    if not redirect_to:
-        redirect_to = '/'
+    redirect_to = get_next_url_for_login_page(request)
     retfun = functools.partial(redirect, redirect_to)
     return _external_login_or_signup(
         request,
@@ -528,10 +520,8 @@ def shib_login(request):
 
     fullname = shib['displayName'] if shib['displayName'] else u'%s %s' % (shib['givenName'], shib['sn'])
 
-    redirect_to = request.REQUEST.get('next')
-    retfun = None
-    if redirect_to:
-        retfun = functools.partial(_safe_postlogin_redirect, redirect_to, request.get_host())
+    redirect_to = get_next_url_for_login_page(request)
+    retfun = functools.partial(_safe_postlogin_redirect, redirect_to, request.get_host())
 
     return _external_login_or_signup(
         request,
@@ -558,31 +548,6 @@ def _safe_postlogin_redirect(redirect_to, safehost, default_redirect='/'):
     return redirect(default_redirect)
 
 
-def _make_shib_enrollment_request(request):
-    """
-        Need this hack function because shibboleth logins don't happen over POST
-        but change_enrollment expects its request to be a POST, with
-        enrollment_action and course_id POST parameters.
-    """
-    enroll_request = HttpRequest()
-    enroll_request.user = request.user
-    enroll_request.session = request.session
-    enroll_request.method = "POST"
-
-    # copy() also makes GET and POST mutable
-    # See https://docs.djangoproject.com/en/dev/ref/request-response/#django.http.QueryDict.update
-    enroll_request.GET = request.GET.copy()
-    enroll_request.POST = request.POST.copy()
-
-    # also have to copy these GET parameters over to POST
-    if "enrollment_action" not in enroll_request.POST and "enrollment_action" in enroll_request.GET:
-        enroll_request.POST.setdefault('enrollment_action', enroll_request.GET.get('enrollment_action'))
-    if "course_id" not in enroll_request.POST and "course_id" in enroll_request.GET:
-        enroll_request.POST.setdefault('course_id', enroll_request.GET.get('course_id'))
-
-    return enroll_request
-
-
 def course_specific_login(request, course_id):
     """
        Dispatcher function for selecting the specific login method

common/djangoapps/student/helpers.py

@@ -4,9 +4,11 @@
 from pytz import UTC
 from django.utils.http import cookie_date
 from django.conf import settings
+from django.core.urlresolvers import reverse, NoReverseMatch
+import third_party_auth
+import urllib
 from verify_student.models import SoftwareSecurePhotoVerification  # pylint: disable=F0401
 from course_modes.models import CourseMode
-from student_account.helpers import auth_pipeline_urls  # pylint: disable=unused-import,import-error
 
 
 def set_logged_in_cookie(request, response):
@@ -199,3 +201,70 @@ def check_verify_status_by_course(user, course_enrollment_pairs, all_course_mode
             status_by_course[key]['verification_good_until'] = recent_verification_datetime.strftime("%m/%d/%Y")
 
     return status_by_course
+
+
+def auth_pipeline_urls(auth_entry, redirect_url=None):
+    """Retrieve URLs for each enabled third-party auth provider.
+
+    These URLs are used on the "sign up" and "sign in" buttons
+    on the login/registration forms to allow users to begin
+    authentication with a third-party provider.
+
+    Optionally, we can redirect the user to an arbitrary
+    url after auth completes successfully.  We use this
+    to redirect the user to a page that required login,
+    or to send users to the payment flow when enrolling
+    in a course.
+
+    Args:
+        auth_entry (string): Either `pipeline.AUTH_ENTRY_LOGIN` or `pipeline.AUTH_ENTRY_REGISTER`
+
+    Keyword Args:
+        redirect_url (unicode): If provided, send users to this URL
+            after they successfully authenticate.
+
+    Returns:
+        dict mapping provider IDs to URLs
+
+    """
+    if not third_party_auth.is_enabled():
+        return {}
+
+    return {
+        provider.NAME: third_party_auth.pipeline.get_login_url(provider.NAME, auth_entry, redirect_url=redirect_url)
+        for provider in third_party_auth.provider.Registry.enabled()
+    }
+
+
+# Query string parameters that can be passed to the "finish_auth" view to manage
+# things like auto-enrollment.
+POST_AUTH_PARAMS = ('course_id', 'enrollment_action', 'course_mode', 'email_opt_in')
+
+
+def get_next_url_for_login_page(request):
+    """
+    Determine the URL to redirect to following login/registration/third_party_auth
+
+    The user is currently on a login or reigration page.
+    If 'course_id' is set, or other POST_AUTH_PARAMS, we will need to send the user to the
+    /account/finish_auth/ view following login, which will take care of auto-enrollment in
+    the specified course.
+
+    Otherwise, we go to the ?next= query param or to the dashboard if nothing else is
+    specified.
+    """
+    redirect_to = request.GET.get('next', None)
+    if not redirect_to:
+        try:
+            redirect_to = reverse('dashboard')
+        except NoReverseMatch:
+            redirect_to = reverse('home')
+    if any(param in request.GET for param in POST_AUTH_PARAMS):
+        # Before we redirect to next/dashboard, we need to handle auto-enrollment:
+        params = [(param, request.GET[param]) for param in POST_AUTH_PARAMS if param in request.GET]
+        params.append(('next', redirect_to))  # After auto-enrollment, user will be sent to payment page or to this URL
+        redirect_to = '{}?{}'.format(reverse('finish_auth'), urllib.urlencode(params))
+        # Note: if we are resuming a third party auth pipeline, then the next URL will already
+        # be saved in the session as part of the pipeline state. That URL will take priority
+        # over this one.
+    return redirect_to

common/djangoapps/student/tests/test_login.py

@@ -278,24 +278,6 @@ def test_change_enrollment_200_no_redirect(self):
         self.assertIsNone(response_content["redirect_url"])
         self._assert_response(response, success=True)
 
-    def test_change_enrollment_200_redirect(self):
-        """
-        Tests that "redirect_url" is the content of the HttpResponse returned
-        by change_enrollment, if there is content
-        """
-        # add this post param to trigger a call to change_enrollment
-        extra_post_params = {"enrollment_action": "enroll"}
-        with patch('student.views.change_enrollment') as mock_change_enrollment:
-            mock_change_enrollment.return_value = HttpResponse("in/nature/there/is/nothing/melancholy")
-            response, _ = self._login_response(
-                'test@edx.org',
-                'test_password',
-                extra_post_params=extra_post_params,
-            )
-        response_content = json.loads(response.content)
-        self.assertEqual(response_content["redirect_url"], "in/nature/there/is/nothing/melancholy")
-        self._assert_response(response, success=True)
-
     def _login_response(self, email, password, patched_audit_log='student.views.AUDIT_LOG', extra_post_params=None):
         ''' Post the login info '''
         post_params = {'email': email, 'password': password}